fix(relay): order replacement sockets explicitly

Wall-clock timestamps can tie during rapid daemon socket replacement. Persist a per-connection generation so forwarding ownership follows acceptance order, including when replacing pre-generation sockets during rollout.

packages/relay/src/cloudflare-adapter.test.ts

@@ -230,6 +230,74 @@ describe("RelayDurableObject control nudge/reset behavior", () => {
   });
 });
 
+describe("RelayDurableObject data forwarding", () => {
+  it("forwards a client frame only to the newest daemon socket during replacement", () => {
+    const connectionId = "clt_replacing_daemon";
+    const client = createMockSocket({
+      version: "2",
+      role: "client",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 1,
+    });
+    const daemonSockets = Array.from({ length: 6 }, (_, index) =>
+      createMockSocket({
+        version: "2",
+        role: "server",
+        connectionId,
+        serverId: "srv_test",
+        generation: index + 1,
+        createdAt: 1,
+      }),
+    );
+    const { state, setTagSockets } = createMockState();
+    setTagSockets(`server:${connectionId}`, daemonSockets);
+
+    const relay = new RelayDurableObject(state as unknown as DurableObjectStateArg);
+    relay.webSocketMessage(client as unknown as WebSocket, "create_agent_request");
+
+    for (const staleSocket of daemonSockets.slice(0, -1)) {
+      expect(staleSocket.send).not.toHaveBeenCalled();
+    }
+    expect(daemonSockets.at(-1)?.send).toHaveBeenCalledOnce();
+    expect(daemonSockets.at(-1)?.send).toHaveBeenCalledWith("create_agent_request");
+  });
+
+  it("prefers a generated replacement over rollout sockets without generations", () => {
+    const connectionId = "clt_rollout";
+    const client = createMockSocket({
+      version: "2",
+      role: "client",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 1,
+    });
+    const staleSocket = createMockSocket({
+      version: "2",
+      role: "server",
+      connectionId,
+      serverId: "srv_test",
+      createdAt: 2,
+    });
+    const replacementSocket = createMockSocket({
+      version: "2",
+      role: "server",
+      connectionId,
+      serverId: "srv_test",
+      generation: 1,
+      createdAt: 2,
+    });
+    const { state, setTagSockets } = createMockState();
+    setTagSockets(`server:${connectionId}`, [staleSocket, replacementSocket]);
+
+    const relay = new RelayDurableObject(state as unknown as DurableObjectStateArg);
+    relay.webSocketMessage(client as unknown as WebSocket, "create_agent_request");
+
+    expect(staleSocket.send).not.toHaveBeenCalled();
+    expect(replacementSocket.send).toHaveBeenCalledWith("create_agent_request");
+  });
+});
+
 describe("relay worker endpoint routing", () => {
   it("routes missing v to legacy v1 isolated DO ids", async () => {
     const fetch = vi.fn(

packages/relay/src/cloudflare-adapter.ts

@@ -86,6 +86,29 @@ function getString(record: Record<string, unknown>, key: string): string | undef
   return typeof value === "string" ? value : undefined;
 }
 
+function getSocketGeneration(ws: WebSocket): number {
+  const attachment = deserializeAttachment(ws);
+  if (!isRecord(attachment)) return 0;
+  return typeof attachment.generation === "number" ? attachment.generation : 0;
+}
+
+function selectNewestSocket(sockets: WebSocket[]): WebSocket | null {
+  let newest: WebSocket | null = null;
+  let newestGeneration = Number.NEGATIVE_INFINITY;
+  for (const socket of sockets) {
+    const generation = getSocketGeneration(socket);
+    if (generation > newestGeneration) {
+      newest = socket;
+      newestGeneration = generation;
+    }
+  }
+  return newest;
+}
+
+function nextSocketGeneration(sockets: WebSocket[]): number {
+  return sockets.reduce((highest, socket) => Math.max(highest, getSocketGeneration(socket)), 0) + 1;
+}
+
 function getGlobalWebSocketPair(): (new () => WebSocketPair) | undefined {
   // Access WebSocketPair from global scope (Cloudflare Workers runtime)
   // Use Reflect to access global property without type assertions
@@ -352,6 +375,9 @@ export class RelayDurableObject {
 
     const isServerControl = role === "server" && !resolvedConnectionId;
     const isServerData = role === "server" && !!resolvedConnectionId;
+    const serverDataGeneration = isServerData
+      ? nextSocketGeneration(this.state.getWebSockets(`server:${resolvedConnectionId}`))
+      : undefined;
 
     // Close any existing server-side connection with the same identity.
     // - server-control: single per serverId
@@ -377,6 +403,7 @@ export class RelayDurableObject {
       role,
       version: CURRENT_RELAY_VERSION,
       connectionId: resolvedConnectionId || null,
+      ...(serverDataGeneration === undefined ? {} : { generation: serverDataGeneration }),
       createdAt: Date.now(),
     };
     serializeAttachment(server, attachment);
@@ -484,12 +511,11 @@ export class RelayDurableObject {
         this.bufferFrame(connectionId, message);
         return;
       }
-      for (const target of servers) {
-        try {
-          target.send(message);
-        } catch (error) {
-          console.error(`[Relay DO] Failed to forward client->server(${connectionId}):`, error);
-        }
+      const target = selectNewestSocket(servers);
+      try {
+        target?.send(message);
+      } catch (error) {
+        console.error(`[Relay DO] Failed to forward client->server(${connectionId}):`, error);
       }
       return;
     }

packages/relay/src/types.ts

@@ -24,5 +24,7 @@ export interface RelaySessionAttachment {
    * independent socket + E2EE channel per connected connection.
    */
   connectionId?: string | null;
+  /** Monotonic acceptance order for v2 daemon data sockets sharing a connectionId. */
+  generation?: number;
   createdAt: number;
 }

packages/server/src/server/session.ts

@@ -366,10 +366,6 @@ type FetchAgentsResponsePageInfo = FetchAgentsResponsePayload["pageInfo"];
 type AgentUpdatesFilter = FetchAgentsRequestFilter;
 type CreateAgentRequestMessage = Extract<SessionInboundMessage, { type: "create_agent_request" }>;
 
-interface CreatedAgentRequest {
-  agent: AgentSnapshotPayload;
-}
-
 interface ResolvedSessionCreateAgentIntent {
   config: AgentSessionConfig;
   intent: CreateAgentIntent;
@@ -680,10 +676,6 @@ export class Session {
   private readonly hubExecutionController: HubExecutionController | null;
   private readonly workspaceScripts: WorkspaceScriptsService;
   private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch;
-  private readonly inFlightAgentCreationsByClientMessageId = new Map<
-    string,
-    Promise<CreatedAgentRequest>
-  >();
 
   constructor(options: SessionOptions) {
     const {
@@ -3110,86 +3102,10 @@ export class Session {
    * Handle create agent request
    */
   private async handleCreateAgentRequest(msg: CreateAgentRequestMessage): Promise<void> {
-    const clientMessageId = msg.clientMessageId?.trim();
-    const existingCreation = clientMessageId
-      ? this.inFlightAgentCreationsByClientMessageId.get(clientMessageId)
-      : undefined;
-    if (existingCreation) {
-      try {
-        const created = await existingCreation;
-        this.emitCreatedAgentResponse(msg.requestId, created.agent);
-      } catch (error) {
-        this.emitFailedAgentCreateResponse(msg.requestId, error);
-      }
-      return;
-    }
-
-    const creation = this.createAgentForRequest(msg);
-    if (clientMessageId) {
-      this.inFlightAgentCreationsByClientMessageId.set(clientMessageId, creation);
-    }
-    try {
-      const created = await creation;
-      this.emitCreatedAgentResponse(msg.requestId, created.agent);
-    } catch (error) {
-      const wireError = toWorktreeWireError(error);
-      this.sessionLogger.error({ err: error }, "Failed to create agent");
-      this.emitFailedAgentCreateResponse(msg.requestId, wireError);
-      this.emit({
-        type: "activity_log",
-        payload: {
-          id: uuidv4(),
-          timestamp: new Date(),
-          type: "error",
-          content: `Failed to create agent: ${wireError.message}`,
-        },
-      });
-    } finally {
-      if (
-        clientMessageId &&
-        this.inFlightAgentCreationsByClientMessageId.get(clientMessageId) === creation
-      ) {
-        this.inFlightAgentCreationsByClientMessageId.delete(clientMessageId);
-      }
-    }
-  }
-
-  private emitCreatedAgentResponse(
-    requestId: string | undefined,
-    agent: AgentSnapshotPayload,
-  ): void {
-    if (!requestId) return;
-    this.emit({
-      type: "status",
-      payload: {
-        status: "agent_created",
-        agentId: agent.id,
-        requestId,
-        agent,
-      },
-    });
-  }
-
-  private emitFailedAgentCreateResponse(requestId: string | undefined, error: unknown): void {
-    if (!requestId) return;
-    const wireError = toWorktreeWireError(error);
-    this.emit({
-      type: "status",
-      payload: {
-        status: "agent_create_failed",
-        requestId,
-        error: wireError.message,
-        errorCode: wireError.code,
-      },
-    });
-  }
-
-  private async createAgentForRequest(
-    msg: CreateAgentRequestMessage,
-  ): Promise<CreatedAgentRequest> {
     const {
       config,
       worktreeName,
+      requestId,
       initialPrompt,
       clientMessageId,
       outputSchema,
@@ -3289,19 +3205,50 @@ export class Session {
         agentId: snapshot.id,
         createdWorktree,
       });
-      const agentPayload = await this.buildAgentPayload(liveSnapshot);
+      if (requestId) {
+        const agentPayload = await this.buildAgentPayload(liveSnapshot);
+        this.emit({
+          type: "status",
+          payload: {
+            status: "agent_created",
+            agentId: liveSnapshot.id,
+            requestId,
+            agent: agentPayload,
+          },
+        });
+      }
 
       this.sessionLogger.info(
         { agentId: snapshot.id, provider: snapshot.provider },
         `Created agent ${snapshot.id} (${snapshot.provider})`,
       );
-      return { agent: agentPayload };
     } catch (error) {
       await this.createAgentLifecycleDispatch.cleanupCreatedWorktreeAfterFailedAgentCreate({
         createdWorktree: createdWorktreeForCleanup,
         createdAgentId,
       });
-      throw error;
+      const wireError = toWorktreeWireError(error);
+      this.sessionLogger.error({ err: error }, "Failed to create agent");
+      if (requestId) {
+        this.emit({
+          type: "status",
+          payload: {
+            status: "agent_create_failed",
+            requestId,
+            error: wireError.message,
+            errorCode: wireError.code,
+          },
+        });
+      }
+      this.emit({
+        type: "activity_log",
+        payload: {
+          id: uuidv4(),
+          timestamp: new Date(),
+          type: "error",
+          content: `Failed to create agent: ${wireError.message}`,
+        },
+      });
     }
   }
 

packages/server/src/server/session.workspaces.test.ts

@@ -872,7 +872,7 @@ test("client heartbeat clears attention for the focused terminal", async () => {
   });
 });
 
-test("create_agent_request deduplicates concurrent draft submission retries", async () => {
+test("create_agent_request keeps requested child cwd when grouped under an existing parent workspace", async () => {
   const workdir = mkdtempSync(path.join(tmpdir(), "paseo-create-agent-cwd-"));
   try {
     const parent = path.join(workdir, "parent");
@@ -981,26 +981,14 @@ test("create_agent_request deduplicates concurrent draft submission retries", as
         terminalManager: null,
       }),
     );
-    await Promise.all([
-      session.handleMessage({
-        type: "create_agent_request",
-        requestId: "req-create-child-1",
-        clientMessageId: "draft-submission-1",
-        config: { provider: "codex", cwd: child },
-        attachments: [],
-      }),
-      session.handleMessage({
-        type: "create_agent_request",
-        requestId: "req-create-child-2",
-        clientMessageId: "draft-submission-1",
-        config: { provider: "codex", cwd: child },
-        attachments: [],
-      }),
-    ]);
+    await session.handleMessage({
+      type: "create_agent_request",
+      requestId: "req-create-child",
+      config: { provider: "codex", cwd: child },
+      attachments: [],
+    });
 
-    const createdAgents = agentManager.listAgents();
-    expect(createdAgents).toHaveLength(1);
-    const [createdAgent] = createdAgents;
+    const [createdAgent] = agentManager.listAgents();
     expect(createdAgent?.cwd).toBe(child);
     const createdWorkspace = await workspaceRegistry.get(createdAgent!.workspaceId!);
     expect(createdWorkspace).not.toBeNull();
@@ -1019,22 +1007,10 @@ test("create_agent_request deduplicates concurrent draft submission retries", as
       projectKey: createdWorkspace!.projectId,
       checkout: { cwd: child },
     });
-    expect(
-      emitted.filter((message) => message.type === "status").map((message) => message.payload),
-    ).toEqual(
-      expect.arrayContaining([
-        expect.objectContaining({
-          status: "agent_created",
-          requestId: "req-create-child-1",
-          agent: expect.objectContaining({ cwd: child }),
-        }),
-        expect.objectContaining({
-          status: "agent_created",
-          requestId: "req-create-child-2",
-          agent: expect.objectContaining({ cwd: child }),
-        }),
-      ]),
-    );
+    expect(findByType(emitted, "status")?.payload).toMatchObject({
+      status: "agent_created",
+      agent: { cwd: child },
+    });
   } finally {
     rmSync(workdir, { recursive: true, force: true });
   }