Treat missing thread snapshots as terminal

BRANCH_DETAILS.md

@@ -0,0 +1,31 @@
+# Thread Detail Subscription Reliability
+
+Thread-detail synchronization distinguishes an authoritative missing resource from a transient snapshot failure across both HTTP snapshot loading and WebSocket snapshot fallback so stale thread state cannot enter an unbounded subscription retry loop.
+
+Expected behavior:
+
+- An HTTP `thread_not_found` response clears the persisted detail cache and marks the client thread state deleted.
+- When a bounded WebSocket resume falls back to a fresh snapshot, a dedicated `OrchestrationThreadNotFoundError` applies the same cache removal and deleted-state transition for a warm cached thread.
+- Cache removal is serialized with snapshot persistence, and persistence rechecks deleted state under the same lock, so a queued or in-flight save cannot resurrect an authoritatively deleted thread.
+- The missing-thread subscription terminates before opening or retrying its WebSocket stream, including after session replacement and application-foreground resubscription signals.
+- `resolveThreadDetailRef` is the canonical web detail-subscription gate. `useThread` waits for the shell when either automatic draft-store detection or an explicit `waitForShell` request identifies a pre-creation thread, while direct detail/status consumers such as the server-thread route map their local-draft readiness through the same resolver. Draft workspace-mode changes before shell creation preserve lookup by the reserved thread ref, so this guard remains active while switching between current-checkout and new-worktree modes. The expected pre-creation HTTP 404 therefore cannot mark the draft deleted, and the new shell starts fresh synchronization after the first send.
+- Other HTTP snapshot failures remain transient and fall back to the socket snapshot path. Other WebSocket snapshot failures remain transient and retain the existing retry behavior.
+
+Primary files:
+
+- `packages/client-runtime/src/state/threadSnapshotHttp.ts`
+- `packages/client-runtime/src/state/threads.ts`
+- `packages/client-runtime/src/state/threads-sync.test.ts`
+- `packages/contracts/src/orchestration.ts`
+- `packages/contracts/src/rpc.ts`
+- `apps/server/src/ws.ts`
+- `apps/server/src/server.test.ts`
+- `apps/web/src/composerDraftStore.test.ts`
+- `apps/web/src/state/entities.ts`
+- `apps/web/src/state/entities.test.ts`
+- `apps/web/src/routes/_chat.$environmentId.$threadId.tsx`
+
+## Development Ports
+
+- Web: `5741`
+- Server/WebSocket: `13781`

apps/server/src/server.test.ts

@@ -6195,6 +6195,36 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest)),
   );
 
+  it.effect("subscribeThread identifies a missing large-gap fallback", () =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest({
+        layers: {
+          orchestrationEngine: {
+            latestSequence: Effect.succeed(100_000),
+          },
+          projectionSnapshotQuery: {
+            getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
+          },
+        },
+      });
+
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const result = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) =>
+          client[ORCHESTRATION_WS_METHODS.subscribeThread]({
+            threadId: defaultThreadId,
+            afterSequence: 5,
+          }).pipe(Stream.runCollect),
+        ).pipe(Effect.result),
+      );
+
+      assertTrue(result._tag === "Failure");
+      assertTrue(result.failure._tag === "OrchestrationThreadNotFoundError");
+      assert.equal(result.failure.threadId, defaultThreadId);
+      assert.equal(result.failure.message, `Thread ${defaultThreadId} was not found`);
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
   it.effect("subscribeThread replaces a cursor ahead of the authoritative head", () =>
     Effect.gen(function* () {
       let readEventsCalls = 0;

apps/server/src/ws.ts

@@ -28,6 +28,7 @@ import {
   type OrchestrationThreadStreamItem,
   OrchestrationGetFullThreadDiffError,
   OrchestrationGetSnapshotError,
+  OrchestrationThreadNotFoundError,
   OrchestrationSearchThreadsError,
   OrchestrationGetTurnDiffError,
   ORCHESTRATION_WS_METHODS,
@@ -1412,9 +1413,8 @@ const makeWsRpcLayer = (
                 );
 
               if (Option.isNone(snapshot)) {
-                return yield* new OrchestrationGetSnapshotError({
-                  message: `Thread ${input.threadId} was not found`,
-                  cause: input.threadId,
+                return yield* new OrchestrationThreadNotFoundError({
+                  threadId: input.threadId,
                 });
               }
 

apps/web/src/composerDraftStore.test.ts

@@ -1123,6 +1123,38 @@ describe("composerDraftStore project draft thread mapping", () => {
     });
   });
 
+  it("keeps a pre-shell draft discoverable by thread ref across workspace mode changes", () => {
+    const store = useComposerDraftStore.getState();
+    const threadRef = scopeThreadRef(TEST_ENVIRONMENT_ID, threadId);
+    store.setProjectDraftThreadId(projectRef, draftId, {
+      threadId,
+      envMode: "local",
+      worktreePath: null,
+    });
+
+    store.setDraftThreadContext(draftId, {
+      envMode: "worktree",
+      worktreePath: null,
+    });
+    const worktreeDraft = useComposerDraftStore.getState().getDraftThread(draftId);
+    expect(useComposerDraftStore.getState().getDraftThreadByRef(threadRef)).toBe(worktreeDraft);
+    expect(worktreeDraft).toMatchObject({
+      envMode: "worktree",
+      worktreePath: null,
+    });
+
+    store.setDraftThreadContext(draftId, {
+      envMode: "local",
+      worktreePath: null,
+    });
+    const localDraft = useComposerDraftStore.getState().getDraftThread(draftId);
+    expect(useComposerDraftStore.getState().getDraftThreadByRef(threadRef)).toBe(localDraft);
+    expect(localDraft).toMatchObject({
+      envMode: "local",
+      worktreePath: null,
+    });
+  });
+
   it("stores the start-from-origin choice with the draft thread", () => {
     const store = useComposerDraftStore.getState();
     store.setProjectDraftThreadId(projectRef, draftId, {

apps/web/src/routes/_chat.$environmentId.$threadId.tsx

@@ -9,9 +9,9 @@ import { resolveThreadSyncPhase } from "../threadSync";
 import { SidebarInset } from "~/components/ui/sidebar";
 import {
   useEnvironmentThreadRefs,
-  useThreadDetail,
+  useThreadDetailWhenReady,
   useThreadShell,
-  useThreadStatus,
+  useThreadStatusWhenReady,
 } from "../state/entities";
 import { useEnvironmentQuery } from "../state/query";
 import { environmentShell } from "../state/shell";
@@ -25,17 +25,19 @@ function ChatThreadRouteView() {
     threadRef === null ? null : environmentShell.stateAtom(threadRef.environmentId),
   );
   const serverThreadShell = useThreadShell(threadRef);
-  const serverThreadDetail = useThreadDetail(threadRef);
-  const serverThreadStatus = useThreadStatus(threadRef);
-  const environmentThreadRefs = useEnvironmentThreadRefs(threadRef?.environmentId ?? null);
-  const bootstrapComplete = shell.data?.snapshot._tag === "Some";
-  const environmentHasServerThreads = environmentThreadRefs.length > 0;
-  const draftThreadExists = useComposerDraftStore((store) =>
-    threadRef ? store.getDraftThreadByRef(threadRef) !== null : false,
-  );
   const draftThread = useComposerDraftStore((store) =>
     threadRef ? store.getDraftThreadByRef(threadRef) : null,
   );
+  const draftThreadExists = draftThread !== null;
+  const detailReadiness = {
+    hasLocalDraft: draftThreadExists,
+    hasServerShell: serverThreadShell !== null,
+  };
+  const serverThreadDetail = useThreadDetailWhenReady(threadRef, detailReadiness);
+  const serverThreadStatus = useThreadStatusWhenReady(threadRef, detailReadiness);
+  const environmentThreadRefs = useEnvironmentThreadRefs(threadRef?.environmentId ?? null);
+  const bootstrapComplete = shell.data?.snapshot._tag === "Some";
+  const environmentHasServerThreads = environmentThreadRefs.length > 0;
   const environmentHasDraftThreads = useComposerDraftStore((store) => {
     if (!threadRef) {
       return false;

apps/web/src/state/entities.test.ts

@@ -1,15 +1,141 @@
-import { scopeThreadRef } from "@t3tools/client-runtime/environment";
-import { EnvironmentId, ThreadId } from "@t3tools/contracts";
-import { describe, expect, it } from "vite-plus/test";
+import { createElement } from "react";
+import { renderToStaticMarkup } from "react-dom/server";
+import { beforeEach, describe, expect, it, vi } from "vite-plus/test";
 
-import { resolveThreadDetailRef } from "./entities";
+import { EnvironmentId, ThreadId, type ScopedThreadRef } from "@t3tools/contracts";
 
-const threadRef = scopeThreadRef(EnvironmentId.make("environment-1"), ThreadId.make("thread-1"));
+const mocks = vi.hoisted(() => ({
+  hasLocalDraft: false,
+  shell: null as object | null,
+  detailAtom: vi.fn((ref: ScopedThreadRef) => ({ kind: "detail", ref })),
+  statusAtom: vi.fn((ref: ScopedThreadRef) => ({ kind: "status", ref })),
+  threadShellAtom: vi.fn((ref: ScopedThreadRef) => ({ kind: "shell", ref })),
+}));
+
+vi.mock("@effect/atom-react", () => ({
+  useAtomValue: (atom: { readonly kind?: string }) => {
+    if (atom.kind === "shell") return mocks.shell;
+    if (atom.kind === "detail") return { id: "detail" };
+    if (atom.kind === "status") return "deleted";
+    return null;
+  },
+}));
+
+vi.mock("@t3tools/client-runtime/state/threads", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("@t3tools/client-runtime/state/threads")>()),
+  mergeEnvironmentThread: (detail: unknown, shell: unknown) => ({ detail, shell }),
+}));
+
+vi.mock("../composerDraftStore", () => ({
+  useComposerDraftStore: (
+    selector: (store: { getDraftThreadByRef: () => object | null }) => unknown,
+  ) =>
+    selector({
+      getDraftThreadByRef: () => (mocks.hasLocalDraft ? {} : null),
+    }),
+}));
+
+vi.mock("./threads", () => ({
+  environmentThreadDetails: {
+    detailAtom: mocks.detailAtom,
+    statusAtom: mocks.statusAtom,
+  },
+  environmentThreadShells: {
+    threadShellAtom: mocks.threadShellAtom,
+  },
+}));
+
+import {
+  resolveThreadDetailRef,
+  useThread,
+  useThreadDetailWhenReady,
+  useThreadStatusWhenReady,
+} from "./entities";
+
+const THREAD_REF: ScopedThreadRef = {
+  environmentId: EnvironmentId.make("environment-1"),
+  threadId: ThreadId.make("thread-1"),
+};
+
+beforeEach(() => {
+  mocks.hasLocalDraft = false;
+  mocks.shell = null;
+  mocks.detailAtom.mockClear();
+  mocks.statusAtom.mockClear();
+  mocks.threadShellAtom.mockClear();
+});
+
+describe("thread detail subscription", () => {
+  it("starts the actual detail hook immediately for an ordinary thread", () => {
+    function Probe() {
+      useThread(THREAD_REF);
+      return null;
+    }
+
+    renderToStaticMarkup(createElement(Probe));
+    expect(mocks.detailAtom).toHaveBeenCalledOnce();
+    expect(mocks.detailAtom).toHaveBeenCalledWith(THREAD_REF);
+  });
+
+  it("starts the actual detail hook only after a local draft receives its shell", () => {
+    mocks.hasLocalDraft = true;
+
+    function Probe() {
+      useThread(THREAD_REF);
+      return null;
+    }
+
+    renderToStaticMarkup(createElement(Probe));
+    expect(mocks.detailAtom).not.toHaveBeenCalled();
+
+    mocks.shell = { id: THREAD_REF.threadId };
+    renderToStaticMarkup(createElement(Probe));
+    expect(mocks.detailAtom).toHaveBeenCalledOnce();
+    expect(mocks.detailAtom).toHaveBeenCalledWith(THREAD_REF);
+  });
+
+  it("preserves explicit shell gating for callers without a local draft", () => {
+    function Probe() {
+      useThread(THREAD_REF, { waitForShell: true });
+      return null;
+    }
+
+    renderToStaticMarkup(createElement(Probe));
+    expect(mocks.detailAtom).not.toHaveBeenCalled();
+
+    mocks.shell = { id: THREAD_REF.threadId };
+    renderToStaticMarkup(createElement(Probe));
+    expect(mocks.detailAtom).toHaveBeenCalledOnce();
+    expect(mocks.detailAtom).toHaveBeenCalledWith(THREAD_REF);
+  });
+
+  it("gates direct detail state consumers until a local draft receives its shell", () => {
+    function Probe({ hasServerShell }: { readonly hasServerShell: boolean }) {
+      const readiness = {
+        hasLocalDraft: true,
+        hasServerShell,
+      };
+      useThreadDetailWhenReady(THREAD_REF, readiness);
+      useThreadStatusWhenReady(THREAD_REF, readiness);
+      return null;
+    }
+
+    renderToStaticMarkup(createElement(Probe, { hasServerShell: false }));
+    expect(mocks.detailAtom).not.toHaveBeenCalled();
+    expect(mocks.statusAtom).not.toHaveBeenCalled();
+
+    renderToStaticMarkup(createElement(Probe, { hasServerShell: true }));
+    expect(mocks.detailAtom).toHaveBeenCalledOnce();
+    expect(mocks.detailAtom).toHaveBeenCalledWith(THREAD_REF);
+    expect(mocks.statusAtom).toHaveBeenCalledOnce();
+    expect(mocks.statusAtom).toHaveBeenCalledWith(THREAD_REF);
+  });
+});
 
 describe("resolveThreadDetailRef", () => {
   it("does not subscribe to a reserved draft thread before it enters the shell index", () => {
     expect(
-      resolveThreadDetailRef(threadRef, {
+      resolveThreadDetailRef(THREAD_REF, {
         shellExists: false,
         waitForShell: true,
       }),
@@ -18,19 +144,19 @@ describe("resolveThreadDetailRef", () => {
 
   it("subscribes once the reserved draft thread enters the shell index", () => {
     expect(
-      resolveThreadDetailRef(threadRef, {
+      resolveThreadDetailRef(THREAD_REF, {
         shellExists: true,
         waitForShell: true,
       }),
-    ).toBe(threadRef);
+    ).toBe(THREAD_REF);
   });
 
   it("keeps direct server-thread lookups enabled when the shell has not loaded it", () => {
     expect(
-      resolveThreadDetailRef(threadRef, {
+      resolveThreadDetailRef(THREAD_REF, {
         shellExists: false,
         waitForShell: false,
       }),
-    ).toBe(threadRef);
+    ).toBe(THREAD_REF);
   });
 });