Require the active sandbox socket identity at message dispatch (COL-128)

packages/control-plane/src/sandbox/lifecycle/manager.test.ts

@@ -121,6 +121,7 @@ function createMockSandbox(
     tunnel_urls: null,
     ttyd_url: null,
     ttyd_token: null,
+    active_socket_id: null,
     created_at: Date.now() - 60000,
     spawn_failure_count: 0,
     last_spawn_failure: 0,

packages/control-plane/src/session/components.ts

@@ -775,6 +775,7 @@ export function createSessionRuntime(platform: SessionPlatform, env: Env): Sessi
     send: (ws, message) => wsManager.send(ws, message),
     getClient: (ws) => connectionAuthenticator.getClientInfo(ws),
     close: (ws, code, reason) => wsManager.close(ws, code, reason),
+    isActiveSandbox: (ws) => wsManager.isActiveSandboxSocket(ws),
     clearSandboxIfMatch: (ws) => wsManager.clearSandboxSocketIfMatch(ws),
     removeClient: (ws) => wsManager.removeClient(ws),
     hasParticipant: (participantId) =>

packages/control-plane/src/session/http/handlers/child-summary.handler.test.ts

@@ -67,6 +67,7 @@ function createSandbox(overrides: Partial<SandboxRow> = {}): SandboxRow {
     tunnel_urls: null,
     ttyd_url: null,
     ttyd_token: null,
+    active_socket_id: null,
     created_at: 1,
     ...overrides,
   };

packages/control-plane/src/session/http/handlers/session-lifecycle.handler.test.ts

@@ -62,6 +62,7 @@ function createSandbox(overrides: Partial<SandboxRow> = {}): SandboxRow {
     tunnel_urls: null,
     ttyd_url: null,
     ttyd_token: null,
+    active_socket_id: null,
     created_at: 1,
     ...overrides,
   };

packages/control-plane/src/session/message-router.ts

@@ -56,11 +56,23 @@ export class SessionMessageRouter<Connection, Client extends ConnectedClient> {
     // The wire protocol is JSON text; binary frames have always been ignored.
     if (typeof message !== "string") return;
 
-    if (this.deps.sockets.classify(connection).kind === "sandbox") {
-      await this.handleSandboxMessage(message);
-    } else {
+    const classified = this.deps.sockets.classify(connection);
+    if (classified.kind !== "sandbox") {
       await this.handleClientMessage(connection, message);
+      return;
+    }
+    if (!this.deps.sockets.isActiveSandbox(connection)) {
+      // A replaced bridge keeps its tags until its close completes. A frame
+      // from it proves it is still open, so close it again instead of
+      // letting it mutate the session.
+      this.deps.log.debug("Ignoring frame from a replaced sandbox socket", {
+        sandbox_id: classified.sandboxId,
+        socket_id: classified.socketId,
+      });
+      this.deps.sockets.close(connection, 1000, "Sandbox socket replaced");
+      return;
     }
+    await this.handleSandboxMessage(message);
   }
 
   private async handleSandboxMessage(message: string): Promise<void> {

packages/control-plane/src/session/ports.ts

@@ -16,9 +16,14 @@ export interface ConnectedClient {
   lastFetchHistoryAtMs?: number;
 }
 
-/** Result of classifying an opaque runtime connection. */
+/**
+ * Result of classifying an opaque runtime connection. A sandbox socket
+ * carries the sandbox it authenticated as and its own accept-time identity;
+ * only the socket whose identity the session persisted as active is
+ * authoritative (see `SocketRegistry.isActiveSandbox`).
+ */
 export type ConnectionClassification =
-  | { kind: "sandbox"; sandboxId?: string }
+  | { kind: "sandbox"; sandboxId?: string; socketId?: string }
   | { kind: "client"; wsId?: string };
 
 /** Wall and monotonic time sources used by session application code. */
@@ -33,6 +38,12 @@ export interface SocketRegistry<Connection, Client extends ConnectedClient> {
   send(connection: Connection, message: ServerMessage): boolean;
   getClient(connection: Connection): Client | null;
   close(connection: Connection, code: number, reason: string): void;
+  /**
+   * Whether `connection` is the sandbox socket the session currently
+   * dispatches to. A replaced bridge keeps its tags until its close
+   * completes; its frames carry no authority.
+   */
+  isActiveSandbox(connection: Connection): boolean;
   clearSandboxIfMatch(connection: Connection): boolean;
   removeClient(connection: Connection): Client | null;
   hasParticipant(participantId: string): boolean;

packages/control-plane/src/session/sandbox-repository.test.ts

@@ -131,6 +131,8 @@ describe("SandboxRepository", () => {
       expect(mock.calls[0].query).toContain("vnc_password = NULL");
       // A replacement sandbox must not inherit the predecessor's runtime.
       expect(mock.calls[0].query).toContain("runtime_version = NULL");
+      // ...nor its bridge: the predecessor's socket loses dispatch authority here.
+      expect(mock.calls[0].query).toContain("active_socket_id = NULL");
       expect(mock.calls[0].params).toEqual(["spawning", 1000, "modal-sb-1"]);
     });
 
@@ -146,6 +148,32 @@ describe("SandboxRepository", () => {
     });
   });
 
+  describe("active socket id", () => {
+    const query = `SELECT active_socket_id FROM sandbox LIMIT 1`;
+
+    it("reads null before any bridge has connected", () => {
+      mock.setData(query, [{ active_socket_id: null }]);
+      expect(repository.getActiveSocketId()).toBeNull();
+    });
+
+    it("reads null without a sandbox row", () => {
+      expect(repository.getActiveSocketId()).toBeNull();
+    });
+
+    it("reads the persisted identity", () => {
+      mock.setData(query, [{ active_socket_id: "sbws-1" }]);
+      expect(repository.getActiveSocketId()).toBe("sbws-1");
+    });
+
+    it("writes the identity to the session's one sandbox row", () => {
+      repository.setActiveSocketId("sbws-2");
+
+      expect(mock.calls.length).toBe(1);
+      expect(mock.calls[0].query).toContain("UPDATE sandbox SET active_socket_id = ?");
+      expect(mock.calls[0].params).toEqual(["sbws-2"]);
+    });
+  });
+
   describe("updateSandboxAuthTokenHash", () => {
     const query = `UPDATE sandbox SET auth_token_hash = ? WHERE modal_sandbox_id = ?`;
 

packages/control-plane/src/session/sandbox-repository.ts

@@ -137,14 +137,33 @@ export class SandboxRepository {
          tunnel_urls = NULL,
          ttyd_url = NULL,
          ttyd_token = NULL,
-         runtime_version = NULL
+         runtime_version = NULL,
+         active_socket_id = NULL
        WHERE id = (SELECT id FROM sandbox LIMIT 1)`,
       data.status,
       data.createdAt,
       data.modalSandboxId
     );
   }
 
+  /**
+   * The bridge socket the session dispatches to, by its `socket:<id>` tag.
+   * Null once a spawn reserves a new identity or before any bridge connects.
+   */
+  getActiveSocketId(): string | null {
+    const result = this.sql.exec(`SELECT active_socket_id FROM sandbox LIMIT 1`);
+    const rows = this.rows<{ active_socket_id: string | null }>(result);
+    return rows[0]?.active_socket_id ?? null;
+  }
+
+  /** Make `socketId` the socket the session dispatches to; every earlier socket loses authority. */
+  setActiveSocketId(socketId: string): void {
+    this.sql.exec(
+      `UPDATE sandbox SET active_socket_id = ? WHERE id = (SELECT id FROM sandbox LIMIT 1)`,
+      socketId
+    );
+  }
+
   /**
    * Phase 2 of the two-phase spawn write (#1589): publish the reserved
    * identity's hash. Scoped to that identity so a delayed publisher cannot

packages/control-plane/src/session/schema.test.ts

@@ -273,6 +273,13 @@ describe("applyMigrations", () => {
     ]);
   });
 
+  it("adds sandbox.active_socket_id for fresh and migrated DOs", () => {
+    expect(SCHEMA_SQL).toContain("active_socket_id TEXT");
+
+    const migration = MIGRATIONS.find((entry) => entry.id === 48);
+    expect(migration?.run).toBe("ALTER TABLE sandbox ADD COLUMN active_socket_id TEXT");
+  });
+
   it("keeps repository context consistent at the session table boundary", () => {
     expect(SCHEMA_SQL).toContain("(repo_owner IS NULL) = (repo_name IS NULL)");
     expect(SCHEMA_SQL).toContain("repo_owner IS NOT NULL");

packages/control-plane/src/session/schema.ts

@@ -189,6 +189,7 @@ CREATE TABLE IF NOT EXISTS sandbox (
   tunnel_urls TEXT,                                 -- JSON mapping of port -> tunnel URL for extra ports
   ttyd_url TEXT,                                    -- ttyd proxy tunnel URL
   ttyd_token TEXT,                                  -- Encrypted JWT token for ttyd auth
+  active_socket_id TEXT,                            -- Bridge socket the session dispatches to (socket:<id> tag)
   created_at INTEGER NOT NULL
 );
 
@@ -648,6 +649,11 @@ export const MIGRATIONS: readonly SchemaMigration[] = [
     description: "Persist terminal message projections awaiting retry",
     run: TERMINAL_MESSAGE_PROJECTION_TABLE_SQL,
   },
+  {
+    id: 48,
+    description: "Add active_socket_id to sandbox",
+    run: `ALTER TABLE sandbox ADD COLUMN active_socket_id TEXT`,
+  },
 ];
 
 /**

packages/control-plane/src/session/server.test.ts

@@ -44,6 +44,7 @@ function createHarness() {
     send: vi.fn(() => true),
     getClient: vi.fn(() => currentClient),
     close: vi.fn(),
+    isActiveSandbox: vi.fn(() => true),
     clearSandboxIfMatch: vi.fn(() => true),
     removeClient: vi.fn(() => client),
     hasParticipant: vi.fn(() => false),
@@ -320,6 +321,29 @@ describe("SessionServer", () => {
     });
   });
 
+  it("refuses frames from a replaced sandbox socket and closes it again", async () => {
+    const { server, messageDeps, sockets, log, setConnectionKind } = createHarness();
+    setConnectionKind("sandbox");
+    vi.mocked(sockets.isActiveSandbox).mockReturnValue(false);
+
+    await server.onMessage(
+      "sandbox",
+      JSON.stringify({
+        type: "heartbeat",
+        sandboxId: "sandbox-1",
+        timestamp: 1000,
+        status: "ready",
+      })
+    );
+
+    expect(messageDeps.processSandboxEvent).not.toHaveBeenCalled();
+    expect(sockets.close).toHaveBeenCalledWith("sandbox", 1000, "Sandbox socket replaced");
+    expect(log.debug).toHaveBeenCalledWith(
+      "Ignoring frame from a replaced sandbox socket",
+      expect.objectContaining({ sandbox_id: "sandbox-1" })
+    );
+  });
+
   it("schedules sandbox reconnect checks and always reciprocates close", async () => {
     const { server, sockets, sandbox, setConnectionKind } = createHarness();
     setConnectionKind("sandbox");

packages/control-plane/src/session/types.ts

@@ -170,6 +170,8 @@ export interface SandboxRow {
   tunnel_urls: string | null; // JSON mapping of port -> tunnel URL
   ttyd_url: string | null;
   ttyd_token: string | null;
+  /** The `socket:<id>` tag of the bridge socket the session dispatches to. */
+  active_socket_id: string | null;
   created_at: number;
 }