Refactor web orchestration sync to incremental events and isolated recovery

apps/web/src/components/ChatView.browser.tsx

@@ -2,8 +2,11 @@
 import "../index.css";
 
 import {
+  EventId,
   ORCHESTRATION_WS_METHODS,
+  ORCHESTRATION_WS_CHANNELS,
   type MessageId,
+  type OrchestrationEvent,
   type OrchestrationReadModel,
   type ProjectId,
   type ServerConfig,
@@ -57,6 +60,8 @@ interface TestFixture {
 let fixture: TestFixture;
 const wsRequests: WsRequestEnvelope["body"][] = [];
 let customWsRpcResolver: ((body: WsRequestEnvelope["body"]) => unknown | undefined) | null = null;
+let wsClient: { send: (message: string) => void } | null = null;
+let pushSequence = 1;
 const wsLink = ws.link(/ws(s)?:\/\/.*/);
 
 interface ViewportSpec {
@@ -336,6 +341,79 @@ function addThreadToSnapshot(
   };
 }
 
+function createThreadCreatedEvent(threadId: ThreadId, sequence: number): OrchestrationEvent {
+  return {
+    sequence,
+    eventId: EventId.makeUnsafe(`event-thread-created-${sequence}`),
+    aggregateKind: "thread",
+    aggregateId: threadId,
+    occurredAt: NOW_ISO,
+    commandId: null,
+    causationEventId: null,
+    correlationId: null,
+    metadata: {},
+    type: "thread.created",
+    payload: {
+      threadId,
+      projectId: PROJECT_ID,
+      title: "New thread",
+      modelSelection: {
+        provider: "codex",
+        model: "gpt-5",
+      },
+      runtimeMode: "full-access",
+      interactionMode: "default",
+      branch: "main",
+      worktreePath: null,
+      createdAt: NOW_ISO,
+      updatedAt: NOW_ISO,
+    },
+  };
+}
+
+function sendOrchestrationDomainEvent(event: OrchestrationEvent): void {
+  if (!wsClient) {
+    throw new Error("WebSocket client not connected");
+  }
+  wsClient.send(
+    JSON.stringify({
+      type: "push",
+      sequence: pushSequence++,
+      channel: ORCHESTRATION_WS_CHANNELS.domainEvent,
+      data: event,
+    }),
+  );
+}
+
+async function waitForWsClient(): Promise<{ send: (message: string) => void }> {
+  let client: { send: (message: string) => void } | null = null;
+  await vi.waitFor(
+    () => {
+      client = wsClient;
+      expect(client).toBeTruthy();
+    },
+    { timeout: 8_000, interval: 16 },
+  );
+  if (!client) {
+    throw new Error("WebSocket client not connected");
+  }
+  return client;
+}
+
+async function promoteDraftThreadViaDomainEvent(threadId: ThreadId): Promise<void> {
+  await waitForWsClient();
+  fixture.snapshot = addThreadToSnapshot(fixture.snapshot, threadId);
+  sendOrchestrationDomainEvent(
+    createThreadCreatedEvent(threadId, fixture.snapshot.snapshotSequence),
+  );
+  await vi.waitFor(
+    () => {
+      expect(useComposerDraftStore.getState().draftThreadsByThreadId[threadId]).toBeUndefined();
+    },
+    { timeout: 8_000, interval: 16 },
+  );
+}
+
 function createDraftOnlySnapshot(): OrchestrationReadModel {
   const snapshot = createSnapshotForTargetUser({
     targetMessageId: "msg-user-draft-target" as MessageId,
@@ -500,10 +578,12 @@ function resolveWsRpc(body: WsRequestEnvelope["body"]): unknown {
 
 const worker = setupWorker(
   wsLink.addEventListener("connection", ({ client }) => {
+    wsClient = client;
+    pushSequence = 1;
     client.send(
       JSON.stringify({
         type: "push",
-        sequence: 1,
+        sequence: pushSequence++,
         channel: WS_CHANNELS.serverWelcome,
         data: fixture.welcome,
       }),
@@ -875,7 +955,7 @@ describe("ChatView timeline estimator parity (full app)", () => {
     useStore.setState({
       projects: [],
       threads: [],
-      threadsHydrated: false,
+      bootstrapComplete: false,
     });
   });
 
@@ -1800,21 +1880,16 @@ describe("ChatView timeline estimator parity (full app)", () => {
       // The composer editor should be present for the new draft thread.
       await waitForComposerEditor();
 
-      // Simulate the snapshot sync arriving from the server after the draft
-      // thread has been promoted to a server thread (thread.create + turn.start
-      // succeeded). The snapshot now includes the new thread, and the sync
-      // should clear the draft without disrupting the route.
-      const { syncServerReadModel } = useStore.getState();
-      syncServerReadModel(addThreadToSnapshot(fixture.snapshot, newThreadId));
-
-      // Clear the draft now that the server thread exists (mirrors EventRouter behavior).
-      useComposerDraftStore.getState().clearDraftThread(newThreadId);
+      // Simulate the steady-state promotion path: the server emits
+      // `thread.created`, the client materializes the thread incrementally,
+      // and the draft is cleared by live batch effects.
+      await promoteDraftThreadViaDomainEvent(newThreadId);
 
       // The route should still be on the new thread — not redirected away.
       await waitForURL(
         mounted.router,
         (path) => path === newThreadPath,
-        "New thread should remain selected after snapshot sync clears the draft.",
+        "New thread should remain selected after server thread promotion clears the draft.",
       );
 
       // The empty thread view and composer should still be visible.
@@ -2136,9 +2211,7 @@ describe("ChatView timeline estimator parity (full app)", () => {
       );
       const promotedThreadId = promotedThreadPath.slice(1) as ThreadId;
 
-      const { syncServerReadModel } = useStore.getState();
-      syncServerReadModel(addThreadToSnapshot(fixture.snapshot, promotedThreadId));
-      useComposerDraftStore.getState().clearDraftThread(promotedThreadId);
+      await promoteDraftThreadViaDomainEvent(promotedThreadId);
 
       const freshThreadPath = await triggerChatNewShortcutUntilPath(
         mounted.router,