transport: fix zero-state reconnect duplicate delivery; add P formal verification

.gitignore

@@ -7,3 +7,6 @@ example
 
 # Nix
 /.direnv/
+
+# Nix build symlinks
+/result

.prettierignore

@@ -3,3 +3,8 @@ node_modules
 .claude
 protobuf/gen
 testUtil/fixtures/gen
+verification/p/PGenerated/
+verification/p/PCheckerOutput/
+verification/p/PObs/PGenerated/
+verification/p/verified/PGenerated/
+flake.lock

PROTOCOL.md

@@ -235,6 +235,10 @@ interface ControlHandshakeRequest {
   expectedSessionState: {
     nextExpectedSeq: number; // integer
     nextSentSeq: number; // integer
+    // whether the client considers this a reconnection to a session that was
+    // previously connected. Optional for wire compatibility; servers MUST
+    // treat an absent flag as `false`.
+    isReconnect?: boolean;
   };
   metadata?: unknown;
 }
@@ -626,8 +630,11 @@ The server will send an error response if either:
   - the client wanted a reconnection to a specific session but the server doesn't know about it
   - the client is in the future (`client.nextSentSeq > server.ack`)
   - server is in the future (`server.seq > client.nextExpectedSeq`)
+  - the client marked the handshake as a reconnection (`isReconnect: true`) but the server has no session for it. The explicit flag matters in the _zero-state window_: a client that has sent messages but never received anything back still has `nextSentSeq: 0, nextExpectedSeq: 0`, which is otherwise indistinguishable from a brand-new session. Without the flag, a server that lost the session (restart or grace expiry) would accept such a reconnect as a new session, the client would replay its send buffer believing the reconnect was transparent, and handlers that already processed those messages would execute them a second time — while the original callers never learn anything went wrong. Rejecting instead yields the normal hard-reconnect semantics: the client starts a fresh session and in-flight calls resolve with `UNEXPECTED_DISCONNECT`.
 
-When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection.
+When the client receives a status with `ok: false`, it should consider the handshake failed and close the connection. For the retriable code (`SESSION_STATE_MISMATCH`) the client MAY automatically reconnect, but MUST do so with a **fresh session** (a new session id and zeroed session state), resolving any in-flight calls of the old session with `UNEXPECTED_DISCONNECT`; retrying the same session would be rejected identically forever. For fatal codes the client MUST NOT reconnect automatically.
+
+Handshakes are **connection-scoped**: a handshake request or response is only meaningful on the connection that carried it. A client MUST ignore a handshake response that does not belong to its current connection attempt, and a server MUST bound the lifetime of un-handshaken connections (`handshakeTimeoutMs`), so a handshake request cannot outlive its connection. This scoping is load-bearing for the `isReconnect` guard: its correctness argument relies on a `isReconnect: false` request never being processed after the session it names has connected and transferred data (see `verification/p/verified/SessionReconnect.p`, where dropping this assumption breaks the no-duplicate-delivery proof).
 
 ### Re-handshaking (live credential refresh)
 

__tests__/properties/session.property.test.ts

@@ -20,6 +20,7 @@ import type { Connection } from '../../transport/connection';
 import type { ServerTransport } from '../../transport/server';
 import { closeAllConnections, numberOfConnections } from '../../testUtil';
 import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport';
+import { traceLogFn, traceSideOf } from '../../testUtil/fixtures/trace';
 import type { TestTransportOptions } from '../../testUtil/fixtures/transports';
 import {
   advanceFakeTimersByConnectionBackoff,
@@ -169,11 +170,7 @@ function setup(opts?: TestTransportOptions): {
 
   const violations: Array<string> = [];
   for (const t of [clientTransport, serverTransport]) {
-    t.bindLogger((msg, ctx, level) => {
-      if (ctx?.tags?.includes('invariant-violation')) {
-        violations.push(`[${level}] ${msg}`);
-      }
-    }, 'debug');
+    t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug');
   }
 
   createServer(serverTransport, services);
@@ -590,11 +587,10 @@ describe('re-handshake under faults', () => {
 
         const violations: Array<string> = [];
         for (const t of [clientTransport, serverTransport]) {
-          t.bindLogger((msg, ctx, level) => {
-            if (ctx?.tags?.includes('invariant-violation')) {
-              violations.push(`[${level}] ${msg}`);
-            }
-          }, 'debug');
+          t.bindLogger(
+            traceLogFn(traceSideOf(t.clientId), violations),
+            'debug',
+          );
         }
 
         createServer(serverTransport, metadataServices);

__tests__/properties/streams.property.test.ts

@@ -11,6 +11,7 @@ import {
   WritableImpl,
 } from '../../router/streams';
 import { createMockTransportNetwork } from '../../testUtil/fixtures/mockTransport';
+import { traceLogFn, traceSideOf } from '../../testUtil/fixtures/trace';
 import { cleanupTransports } from '../../testUtil/fixtures/cleanup';
 import type {
   ProvidedClientTransportOptions,
@@ -117,11 +118,7 @@ async function withNetwork(
 
   const violations: Array<string> = [];
   for (const t of [clientTransport, serverTransport]) {
-    t.bindLogger((msg, ctx, level) => {
-      if (ctx?.tags?.includes('invariant-violation')) {
-        violations.push(`[${level}] ${msg}`);
-      }
-    }, 'debug');
+    t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug');
   }
 
   createServer(serverTransport, services);

__tests__/zerostate.test.ts

@@ -0,0 +1,126 @@
+import { describe, expect, test } from 'vitest';
+import { Type } from 'typebox';
+import {
+  Ok,
+  Procedure,
+  UNEXPECTED_DISCONNECT_CODE,
+  createServiceSchema,
+} from '../router';
+import { createClient } from '../router/client';
+import { createServer } from '../router/server';
+import { createMockTransportNetwork } from '../testUtil/fixtures/mockTransport';
+import { traceLogFn, traceSideOf } from '../testUtil/fixtures/trace';
+import {
+  advanceFakeTimersByConnectionBackoff,
+  cleanupTransports,
+  waitFor,
+} from '../testUtil/fixtures/cleanup';
+
+/**
+ * The "zero-state window": a client that has sent messages the server accepted
+ * and DELIVERED TO HANDLERS, but that has received nothing back (no response,
+ * no heartbeat -- so its session still reads `nextSentSeq: 0,
+ * nextExpectedSeq: 0`), reconnects after the server lost the session (restart
+ * or grace expiry). Such a handshake is indistinguishable from a brand-new
+ * session by the seq counters alone, so without an explicit reconnect marker
+ * the server accepts it as new, the client replays its send buffer, and the
+ * handler executes the same request a second time -- while the original call
+ * never resolves.
+ *
+ * This scenario was found by the P model of the protocol
+ * (verification/p/README.md, finding 1). The expected behavior asserted here
+ * is that of a client that marks reconnection attempts: the server rejects
+ * the unknown session with SESSION_STATE_MISMATCH, the client starts a fresh
+ * session, and the in-flight call resolves with UNEXPECTED_DISCONNECT --
+ * exactly the documented hard-reconnect semantics, and never a duplicate
+ * handler execution.
+ */
+describe('zero-state reconnect to a server that lost the session', () => {
+  test('does not re-execute handlers; in-flight calls resolve with UNEXPECTED_DISCONNECT', async () => {
+    const invocations: Array<string> = [];
+
+    const ServiceSchema = createServiceSchema();
+    const ZeroStateService = ServiceSchema.define({
+      work: Procedure.rpc({
+        requestInit: Type.Object({ id: Type.String() }),
+        responseData: Type.Object({}),
+        async handler({ ctx, reqInit }) {
+          invocations.push(reqInit.id);
+          // hang until abort so nothing (response or ack) ever flows back to
+          // the client, keeping the client's session in the zero-state window
+          await new Promise<void>((resolve) => {
+            ctx.signal.addEventListener('abort', () => {
+              resolve();
+            });
+          });
+
+          return Ok({});
+        },
+      }),
+    });
+    const services = { svc: ZeroStateService };
+
+    // long heartbeat interval: a server heartbeat would ack the request and
+    // take the client out of the zero-state window, masking the scenario
+    const quietHeartbeats = {
+      heartbeatIntervalMs: 60_000,
+      heartbeatsUntilDead: 2,
+    };
+    const network = createMockTransportNetwork({
+      client: {
+        ...quietHeartbeats,
+        maxJitterMs: 0,
+        baseIntervalMs: 10,
+        attemptBudgetCapacity: 100,
+      },
+      server: quietHeartbeats,
+    });
+
+    const clientTransport = network.getClientTransport('client');
+    const serverTransport = network.getServerTransport('SERVER');
+    const violations: Array<string> = [];
+    for (const t of [clientTransport, serverTransport]) {
+      t.bindLogger(traceLogFn(traceSideOf(t.clientId), violations), 'debug');
+    }
+
+    createServer(serverTransport, services);
+    const client = createClient<typeof services>(clientTransport, 'SERVER');
+
+    try {
+      // the handler runs on the first server, but the client hears nothing
+      const pending = client.svc.work.rpc({ id: 'once' });
+      await waitFor(() => expect(invocations).toStrictEqual(['once']));
+
+      // the server loses all state; the client's session (and its send
+      // buffer holding the request) survives within its grace period
+      await network.restartServer();
+      const secondServer = network.getServerTransport('SERVER');
+      secondServer.bindLogger(traceLogFn('server', violations), 'debug');
+      createServer(secondServer, services);
+
+      await advanceFakeTimersByConnectionBackoff();
+
+      // the reconnect must be treated as a hard reconnect, not a fresh
+      // session: the caller learns its call died...
+      const result = await pending;
+      expect(result.ok).toBe(false);
+      if (!result.ok) {
+        expect(result.payload.code).toBe(UNEXPECTED_DISCONNECT_CODE);
+      }
+
+      // ...and the handler must never have executed the same request twice
+      expect(invocations).toStrictEqual(['once']);
+
+      // the fresh session works: new calls reach the new server
+      const again = client.svc.work.rpc({ id: 'later' });
+      await waitFor(() => expect(invocations).toStrictEqual(['once', 'later']));
+      clientTransport.hardDisconnect();
+      await again;
+
+      expect(violations).toStrictEqual([]);
+    } finally {
+      await cleanupTransports([clientTransport, serverTransport]);
+      await network.cleanup();
+    }
+  });
+});