diff --git a/BRANCH_DETAILS.md b/BRANCH_DETAILS.md new file mode 100644 index 00000000000..19beb7d089a --- /dev/null +++ b/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` diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 89f903c4f89..74bea5920f2 100644 --- a/apps/server/src/server.test.ts +++ b/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; diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index ebcf65e4b47..fcdc9102900 100644 --- a/apps/server/src/ws.ts +++ b/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, }); } diff --git a/apps/web/src/composerDraftStore.test.ts b/apps/web/src/composerDraftStore.test.ts index 3e4106c583f..815e8b81b5d 100644 --- a/apps/web/src/composerDraftStore.test.ts +++ b/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, { diff --git a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx b/apps/web/src/routes/_chat.$environmentId.$threadId.tsx index 5ac4665cf8a..ea560d9f81c 100644 --- a/apps/web/src/routes/_chat.$environmentId.$threadId.tsx +++ b/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; diff --git a/apps/web/src/state/entities.test.ts b/apps/web/src/state/entities.test.ts index 0d7611ec41a..053bcac0bf9 100644 --- a/apps/web/src/state/entities.test.ts +++ b/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()), + 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); }); }); diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 7bca3118237..61c1fcdc9e0 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -20,6 +20,7 @@ import type { import type { EnvironmentId, ThreadId } from "@t3tools/contracts"; import { Atom } from "effect/unstable/reactivity"; import { useMemo } from "react"; +import { useComposerDraftStore } from "../composerDraftStore"; import { appAtomRegistry } from "../rpc/atomRegistry"; import { environmentProjects } from "./projects"; import { environmentServerConfigsAtom } from "./server"; @@ -146,12 +147,11 @@ export function useThreadDetail(ref: ScopedThreadRef | null): EnvironmentThread ); } -export function useThreadStatus(ref: ScopedThreadRef | null): EnvironmentThreadStatus { - return useAtomValue( - ref === null ? EMPTY_THREAD_STATUS_ATOM : environmentThreadDetails.statusAtom(ref), - ); -} - +/** + * Returns the detail ref unless the caller is waiting for a server shell that + * has not appeared yet. This is the canonical detail-subscription gate; local + * draft callers adapt their draft and shell readiness into these options. + */ export function resolveThreadDetailRef( ref: ScopedThreadRef | null, options: { @@ -162,6 +162,41 @@ export function resolveThreadDetailRef( return ref !== null && (!options.waitForShell || options.shellExists) ? ref : null; } +type ThreadDetailReadiness = { + readonly hasLocalDraft: boolean; + readonly hasServerShell: boolean; +}; + +function resolveReadyThreadDetailRef( + ref: ScopedThreadRef | null, + input: ThreadDetailReadiness, +): ScopedThreadRef | null { + return resolveThreadDetailRef(ref, { + shellExists: input.hasServerShell, + waitForShell: input.hasLocalDraft, + }); +} + +export function useThreadDetailWhenReady( + ref: ScopedThreadRef | null, + input: ThreadDetailReadiness, +): EnvironmentThread | null { + return useThreadDetail(resolveReadyThreadDetailRef(ref, input)); +} + +export function useThreadStatus(ref: ScopedThreadRef | null): EnvironmentThreadStatus { + return useAtomValue( + ref === null ? EMPTY_THREAD_STATUS_ATOM : environmentThreadDetails.statusAtom(ref), + ); +} + +export function useThreadStatusWhenReady( + ref: ScopedThreadRef | null, + input: ThreadDetailReadiness, +): EnvironmentThreadStatus { + return useThreadStatus(resolveReadyThreadDetailRef(ref, input)); +} + /** Detail collections composed with shell-authoritative thread/workspace metadata. */ export function useThread( ref: ScopedThreadRef | null, @@ -175,10 +210,13 @@ export function useThread( }, ): EnvironmentThread | null { const shell = useThreadShell(ref); + const hasLocalDraft = useComposerDraftStore((store) => + ref === null ? false : store.getDraftThreadByRef(ref) !== null, + ); const detail = useThreadDetail( resolveThreadDetailRef(ref, { shellExists: shell !== null, - waitForShell: options?.waitForShell === true, + waitForShell: hasLocalDraft || options?.waitForShell === true, }), ); return useMemo(() => mergeEnvironmentThread(detail, shell), [detail, shell]); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.test.ts b/packages/client-runtime/src/state/threadSnapshotHttp.test.ts new file mode 100644 index 00000000000..b939eaab0d4 --- /dev/null +++ b/packages/client-runtime/src/state/threadSnapshotHttp.test.ts @@ -0,0 +1,82 @@ +import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; + +import { PrimaryConnectionTarget, type PreparedConnection } from "../connection/model.ts"; +import { remoteHttpClientLayer } from "../rpc/http.ts"; +import { ThreadSnapshotLoader, threadSnapshotLoaderLayer } from "./threadSnapshotHttp.ts"; + +const TARGET = new PrimaryConnectionTarget({ + environmentId: EnvironmentId.make("environment-1"), + label: "Test environment", + httpBaseUrl: "https://environment.example.test", + wsBaseUrl: "wss://environment.example.test", +}); +const PREPARED: PreparedConnection = { + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: TARGET.httpBaseUrl, + socketUrl: TARGET.wsBaseUrl, + httpAuthorization: null, + target: TARGET, +}; +const THREAD_ID = ThreadId.make("thread-1"); + +function provideLoader(fetchFn: typeof fetch) { + return Effect.provide( + threadSnapshotLoaderLayer.pipe(Layer.provide(remoteHttpClientLayer(fetchFn))), + ); +} + +describe("ThreadSnapshotLoader", () => { + it.effect("preserves the decoded thread_not_found response", () => { + const fetchFn = (() => + Promise.resolve( + Response.json( + { + _tag: "EnvironmentResourceNotFoundError", + code: "not_found", + reason: "thread_not_found", + traceId: "trace-thread-not-found", + }, + { status: 404 }, + ), + )) satisfies typeof fetch; + + return Effect.gen(function* () { + const loader = yield* ThreadSnapshotLoader; + const error = yield* Effect.flip(loader.load(PREPARED, THREAD_ID)); + + expect(error).toMatchObject({ + _tag: "EnvironmentResourceNotFoundError", + code: "not_found", + reason: "thread_not_found", + traceId: "trace-thread-not-found", + }); + }).pipe(provideLoader(fetchFn)); + }); + + it.effect("maps a transient HTTP failure to the socket fallback", () => { + const fetchFn = (() => + Promise.resolve( + Response.json( + { + _tag: "EnvironmentInternalError", + code: "internal_error", + reason: "orchestration_thread_snapshot_failed", + traceId: "trace-thread-snapshot-failed", + }, + { status: 500 }, + ), + )) satisfies typeof fetch; + + return Effect.gen(function* () { + const loader = yield* ThreadSnapshotLoader; + const snapshot = yield* loader.load(PREPARED, THREAD_ID); + + expect(Option.isNone(snapshot)).toBe(true); + }).pipe(provideLoader(fetchFn)); + }); +}); diff --git a/packages/client-runtime/src/state/threadSnapshotHttp.ts b/packages/client-runtime/src/state/threadSnapshotHttp.ts index 6acc3b5d8a4..70e7fb40f4a 100644 --- a/packages/client-runtime/src/state/threadSnapshotHttp.ts +++ b/packages/client-runtime/src/state/threadSnapshotHttp.ts @@ -1,9 +1,14 @@ -import type { OrchestrationThreadDetailSnapshot, ThreadId } from "@t3tools/contracts"; +import { + EnvironmentResourceNotFoundError, + type OrchestrationThreadDetailSnapshot, + type ThreadId, +} from "@t3tools/contracts"; import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; import { HttpClient } from "effect/unstable/http"; import type { PreparedConnection } from "../connection/model.ts"; @@ -77,11 +82,15 @@ export const fetchEnvironmentThreadSnapshot = Effect.fn( export type FetchEnvironmentThreadSnapshotError = RemoteEnvironmentRequestError; +const isThreadSnapshotNotFoundError = Schema.is(EnvironmentResourceNotFoundError); + /** - * Loads a thread's detail snapshot over HTTP, returning `Option.none()` when it - * cannot be loaded (so the caller falls back to the socket-embedded snapshot). - * Decouples the thread state machine from the underlying HTTP + DPoP details and - * keeps them out of test contexts. + * Loads a thread's detail snapshot over HTTP, returning `Option.none()` for + * transient failures so the caller can fall back to the socket-embedded + * snapshot. An authoritative `thread_not_found` response remains in the error + * channel so the thread state machine can terminate the subscription. + * Decouples the thread state machine from the underlying HTTP + DPoP details + * and keeps them out of test contexts. */ export class ThreadSnapshotLoader extends Context.Service< ThreadSnapshotLoader, @@ -90,7 +99,10 @@ export class ThreadSnapshotLoader extends Context.Service< prepared: PreparedConnection, threadId: ThreadId, window?: ThreadSnapshotWindow, - ) => Effect.Effect>; + ) => Effect.Effect< + Option.Option, + EnvironmentResourceNotFoundError + >; } >()("@t3tools/client-runtime/state/threadSnapshotHttp/ThreadSnapshotLoader") {} @@ -116,27 +128,22 @@ export const threadSnapshotLoaderLayer: Layer.Layer< }).pipe( Effect.map(Option.some), Effect.provideService(HttpClient.HttpClient, httpClient), - // A genuinely missing thread (404) is expected — the socket - // subscription is the source of truth for thread existence and will - // surface the deletion — so don't treat it as an error worth warning - // about; just defer to the socket path. - Effect.catchTags({ - EnvironmentResourceNotFoundError: () => - Effect.logDebug( - "Thread snapshot not found over HTTP; deferring to the socket subscription.", - ).pipe( - Effect.annotateLogs({ threadId }), - Effect.as(Option.none()), - ), - }), - Effect.catchCause((cause) => - Effect.logWarning( + // Preserve the declared 404 in the error channel. It is authoritative + // for this resource and must terminate the thread subscription rather + // than fall into its retry loop. + Effect.catchCause((cause) => { + for (const reason of cause.reasons) { + if (Cause.isFailReason(reason) && isThreadSnapshotNotFoundError(reason.error)) { + return Effect.fail(reason.error); + } + } + return Effect.logWarning( "Could not load the thread snapshot over HTTP; using the socket snapshot instead.", ).pipe( Effect.annotateLogs({ threadId, cause: Cause.pretty(cause) }), Effect.as(Option.none()), - ), - ), + ); + }), ), }); }), diff --git a/packages/client-runtime/src/state/threads-pagination.test.ts b/packages/client-runtime/src/state/threads-pagination.test.ts index 62cad18f89e..34f835e3aca 100644 --- a/packages/client-runtime/src/state/threads-pagination.test.ts +++ b/packages/client-runtime/src/state/threads-pagination.test.ts @@ -1,4 +1,5 @@ import { + EnvironmentResourceNotFoundError, EnvironmentId, EventId, ORCHESTRATION_WS_METHODS, @@ -133,6 +134,7 @@ type LoaderResponse = Option.Option; const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (options?: { readonly paginationCapability?: boolean; readonly initialResponse?: LoaderResponse; + readonly olderResponseError?: EnvironmentResourceNotFoundError; /** Cached snapshot returned by the cache store (simulates a warm cache). */ readonly cached?: OrchestrationThreadDetailSnapshot; }) { @@ -141,6 +143,7 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt const loaderWindows = yield* Ref.make>([]); const lastSubscribeInput = yield* Ref.make | undefined>(undefined); const savedThreads = yield* Ref.make>([]); + const removedThreads = yield* Ref.make>([]); // Older-page responses resolve through deferreds so tests can interleave // live events with an in-flight page fetch. const pendingPageResponses = yield* Queue.unbounded>(); @@ -174,10 +177,12 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt ? Effect.succeed( options?.initialResponse ?? Option.none(), ) - : Deferred.make().pipe( - Effect.tap((deferred) => Queue.offer(pendingPageResponses, deferred)), - Effect.flatMap(Deferred.await), - ), + : options?.olderResponseError !== undefined + ? Effect.fail(options.olderResponseError) + : Deferred.make().pipe( + Effect.tap((deferred) => Queue.offer(pendingPageResponses, deferred)), + Effect.flatMap(Deferred.await), + ), ), ), }); @@ -197,7 +202,8 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt Effect.succeed(options?.cached !== undefined ? Option.some(options.cached) : Option.none()), saveThread: (_environmentId, thread) => Ref.update(savedThreads, (current) => [...current, thread]), - removeThread: () => Effect.void, + removeThread: (_environmentId, threadId) => + Ref.update(removedThreads, (current) => [...current, threadId]), loadServerConfig: () => Effect.succeed(Option.none()), saveServerConfig: () => Effect.void, loadVcsRefs: () => Effect.succeed(Option.none()), @@ -231,6 +237,7 @@ const makeHarness = Effect.fn("TestThreadPagination.makeHarness")(function* (opt loaderWindows, lastSubscribeInput, savedThreads, + removedThreads, threadState, }; }); @@ -339,6 +346,29 @@ describe("thread pagination state", () => { }), ); + it.effect( + "keeps an authoritative missing-thread response terminal while loading older turns", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + initialResponse: Option.some(WINDOWED_SNAPSHOT), + olderResponseError: new EnvironmentResourceNotFoundError({ + code: "not_found", + reason: "thread_not_found", + traceId: "trace-older-thread-not-found", + }), + }); + yield* harness.awaitState((value) => Option.isSome(value.page)); + + expect(requestOlderThreadTurns(TARGET.environmentId, THREAD_ID)).toBe(true); + const state = yield* harness.awaitState((value) => value.status === "deleted"); + + expect(Option.isNone(state.data)).toBe(true); + expect(Option.isNone(state.page)).toBe(true); + expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]); + }), + ); + it.effect("discards an in-flight older page when a revert rewrites history", () => Effect.gen(function* () { const harness = yield* makeHarness({ initialResponse: Option.some(WINDOWED_SNAPSHOT) }); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index c2df434e8e7..9555c4f44fb 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -1,7 +1,9 @@ import { + EnvironmentResourceNotFoundError, EnvironmentId, EventId, ORCHESTRATION_WS_METHODS, + OrchestrationThreadNotFoundError, ProjectId, ProviderInstanceId, ThreadId, @@ -132,6 +134,7 @@ function awaitThreadState( const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (options?: { readonly cached?: OrchestrationThread; readonly httpSnapshot?: Option.Option; + readonly httpSnapshotError?: EnvironmentResourceNotFoundError; readonly completionMarker?: boolean; }) { const inputs = yield* Queue.unbounded(); @@ -181,10 +184,14 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const snapshotLoader = ThreadSnapshotLoader.of({ load: (_prepared, threadId) => Ref.update(loaderCalls, (count) => count + 1).pipe( - Effect.as( - threadId === THREAD_ID - ? (options?.httpSnapshot ?? Option.none()) - : Option.none(), + Effect.andThen( + threadId === THREAD_ID && options?.httpSnapshotError !== undefined + ? Effect.fail(options.httpSnapshotError) + : Effect.succeed( + threadId === THREAD_ID + ? (options?.httpSnapshot ?? Option.none()) + : Option.none(), + ), ), ), }); @@ -420,6 +427,78 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("keeps an authoritative HTTP thread_not_found response terminal", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + httpSnapshotError: new EnvironmentResourceNotFoundError({ + code: "not_found", + reason: "thread_not_found", + traceId: "trace-thread-not-found", + }), + }); + + const state = yield* awaitThreadState( + harness.observed, + (value) => value.status === "deleted", + ); + yield* Queue.offer(harness.wakeups, "application-active"); + yield* harness.replaceSession; + yield* TestClock.adjust("1 second"); + for (let attempt = 0; attempt < 100; attempt += 1) { + yield* Effect.yieldNow; + } + + expect({ + status: state.status, + hasData: Option.isSome(state.data), + loaderCalls: yield* Ref.get(harness.loaderCalls), + subscriptionCount: yield* Ref.get(harness.subscriptionCount), + removedThreads: yield* Ref.get(harness.removedThreads), + }).toEqual({ + status: "deleted", + hasData: false, + loaderCalls: 1, + subscriptionCount: 0, + removedThreads: [THREAD_ID], + }); + }), + ); + + it.effect("keeps a socket thread_not_found fallback terminal for a warm cache", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD }); + yield* Queue.offer( + harness.inputs, + new OrchestrationThreadNotFoundError({ + threadId: THREAD_ID, + }), + ); + + const state = yield* awaitThreadState( + harness.observed, + (value) => value.status === "deleted", + ); + yield* TestClock.adjust("1 second"); + for (let attempt = 0; attempt < 100; attempt += 1) { + yield* Effect.yieldNow; + } + + expect({ + status: state.status, + hasData: Option.isSome(state.data), + loaderCalls: yield* Ref.get(harness.loaderCalls), + subscriptionCount: yield* Ref.get(harness.subscriptionCount), + removedThreads: yield* Ref.get(harness.removedThreads), + }).toEqual({ + status: "deleted", + hasData: false, + loaderCalls: 0, + subscriptionCount: 1, + removedThreads: [THREAD_ID], + }); + }), + ); + it.effect("ignores replayed thread events at or below the snapshot sequence", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); @@ -455,7 +534,26 @@ describe("EnvironmentThreads", () => { }), ); - it.effect("does not resurrect a deleted thread when the app returns to the foreground", () => + it.effect("does not persist a snapshot queued before deletion", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ cached: BASE_THREAD }); + yield* Queue.offer(harness.inputs, snapshot(BASE_THREAD)); + yield* awaitThreadState( + harness.observed, + (value) => value.status === "live" && Option.isSome(value.data), + ); + yield* Queue.offer(harness.inputs, deleted()); + yield* awaitThreadState(harness.observed, (value) => value.status === "deleted"); + + yield* TestClock.adjust("500 millis"); + yield* Effect.yieldNow; + + expect(yield* Ref.get(harness.savedThreads)).toEqual([]); + expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]); + }), + ); + + it.effect("does not resubscribe a deleted thread when the app returns to the foreground", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD, @@ -477,7 +575,7 @@ describe("EnvironmentThreads", () => { } const latest = yield* Ref.get(harness.latest); - expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); expect(yield* Ref.get(harness.loaderCalls)).toBe(0); expect(latest.status).toBe("deleted"); expect(Option.isNone(latest.data)).toBe(true); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 4ba5a0e9df1..9035f689313 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,5 +1,7 @@ import { + type EnvironmentResourceNotFoundError, ORCHESTRATION_WS_METHODS, + OrchestrationThreadNotFoundError, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, type OrchestrationThreadDetailPage, @@ -13,6 +15,7 @@ import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schema from "effect/Schema"; import * as Semaphore from "effect/Semaphore"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; @@ -126,6 +129,14 @@ function formatThreadError(cause: Cause.Cause): string { : "Could not synchronize the thread."; } +const isOrchestrationThreadNotFoundError = Schema.is(OrchestrationThreadNotFoundError); + +function isThreadNotFoundCause(cause: Cause.Cause): boolean { + return cause.reasons.some( + (reason) => Cause.isFailReason(reason) && isOrchestrationThreadNotFoundError(reason.error), + ); +} + function shouldPersistThread(thread: OrchestrationThread): boolean { const status = thread.session?.status; return status !== "starting" && status !== "running"; @@ -186,20 +197,29 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make readonly epoch: number; } | null>(null); const persistence = yield* Queue.sliding(1); + const persistenceLock = yield* Semaphore.make(1); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationThreadDetailSnapshot, ) { - yield* cache.saveThread(environmentId, snapshot).pipe( - Effect.catch((error) => - Effect.logWarning("Could not persist the thread cache.").pipe( - Effect.annotateLogs({ - environmentId, - threadId, - error: error.message, - }), - ), - ), + yield* persistenceLock.withPermits(1)( + Effect.gen(function* () { + const current = yield* SubscriptionRef.get(state); + if (current.status === "deleted") { + return; + } + yield* cache.saveThread(environmentId, snapshot).pipe( + Effect.catch((error) => + Effect.logWarning("Could not persist the thread cache.").pipe( + Effect.annotateLogs({ + environmentId, + threadId, + error: error.message, + }), + ), + ), + ); + }), ); }); @@ -251,7 +271,6 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make })), ), ); - const setThread = Effect.fn("EnvironmentThreadState.setThread")(function* ( thread: OrchestrationThread, // "keep" preserves the current page state (live events touch only loaded @@ -291,7 +310,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make } }); - const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { + const setDeletedLocked = Effect.fn("EnvironmentThreadState.setDeletedLocked")(function* () { yield* Ref.set(awaitingCompletion, false); yield* Ref.update(historyEpoch, (epoch) => epoch + 1); yield* SubscriptionRef.set(state, { @@ -300,23 +319,49 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make error: Option.none(), page: Option.none(), }); - yield* cache.removeThread(environmentId, threadId).pipe( - Effect.catch((error) => - Effect.logWarning("Could not remove the cached thread.").pipe( - Effect.annotateLogs({ - environmentId, - threadId, - error: error.message, - }), + yield* persistenceLock.withPermits(1)( + cache.removeThread(environmentId, threadId).pipe( + Effect.catch((error) => + Effect.logWarning("Could not remove the cached thread.").pipe( + Effect.annotateLogs({ + environmentId, + threadId, + error: error.message, + }), + ), ), ), ); }); + const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { + yield* applyLock.withPermits(1)(setDeletedLocked()); + }); + const handleHttpThreadNotFound = Effect.fn("EnvironmentThreadState.handleHttpThreadNotFound")( + function* (error: EnvironmentResourceNotFoundError) { + yield* Effect.logDebug( + "Thread snapshot was not found over HTTP; terminating the subscription.", + ).pipe( + Effect.annotateLogs({ + environmentId, + threadId, + reason: error.reason, + traceId: error.traceId, + }), + ); + yield* setDeleted(); + return yield* Effect.interrupt; + }, + ); + const handleStreamError = (cause: Cause.Cause) => + isThreadNotFoundCause(cause) ? setDeleted() : setStreamError(cause); // Body of applyItem, running under applyLock. const applyItemLocked = Effect.fn("EnvironmentThreadState.applyItemLocked")(function* ( item: OrchestrationThreadStreamItem, ) { + if ((yield* SubscriptionRef.get(state)).status === "deleted") { + return; + } if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => @@ -347,7 +392,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const current = yield* SubscriptionRef.get(state); if (Option.isNone(current.data)) { if (item.event.type === "thread.deleted") { - yield* setDeleted(); + yield* setDeletedLocked(); } return; } @@ -364,7 +409,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make if (result.kind === "updated") { yield* setThread(result.thread, "keep"); } else if (result.kind === "deleted") { - yield* setDeleted(); + yield* setDeletedLocked(); } // The event may have advanced the live state past a parked page's // watermark; merge it as soon as that happens. @@ -485,7 +530,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make turnLimit: OLDER_THREAD_PAGE_USER_TURN_LIMIT, beforeCursor: page.beforeCursor, }; - const response = yield* snapshotLoader.load(prepared, threadId, window); + const response = yield* snapshotLoader.load(prepared, threadId, window).pipe( + Effect.catchTags({ + EnvironmentResourceNotFoundError: handleHttpThreadNotFound, + }), + ); // Staleness check and merge run under the same lock as stream-item // application, so a revert/snapshot cannot land between them (TOCTOU // review finding) — anything that rewrites history bumps the epoch @@ -556,6 +605,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make subscribeDynamic( ORCHESTRATION_WS_METHODS.subscribeThread, Effect.fn("EnvironmentThreadState.makeSubscribeInput")(function* (session) { + let current = yield* SubscriptionRef.get(state); + if (current.status === "deleted") { + return yield* Effect.interrupt; + } + const config = yield* session.initialConfig.pipe( Effect.orElseSucceed( () => @@ -574,7 +628,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* Ref.set(awaitingCompletion, supportsCompletionMarker); yield* setSynchronizing; - let current = yield* SubscriptionRef.get(state); + current = yield* SubscriptionRef.get(state); + if (current.status === "deleted") { + return yield* Effect.interrupt; + } // A windowed cache resuming against a server without pagination is a // trap: afterSequence resume keeps only the window, and the missing // older turns can never be loaded (the server has no cursor reads). @@ -590,7 +647,10 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* SubscriptionRef.set(lastSequence, 0); current = yield* SubscriptionRef.get(state); } - if (Option.isNone(current.data) && current.status !== "deleted") { + if (current.status === "deleted") { + return yield* Effect.interrupt; + } + if (Option.isNone(current.data)) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -605,11 +665,17 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }), ), ); - const httpSnapshot = yield* snapshotLoader.load( - prepared, - threadId, - supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : undefined, - ); + const httpSnapshot = yield* snapshotLoader + .load( + prepared, + threadId, + supportsPagination ? { turnLimit: INITIAL_THREAD_USER_TURN_LIMIT } : undefined, + ) + .pipe( + Effect.catchTags({ + EnvironmentResourceNotFoundError: handleHttpThreadNotFound, + }), + ); if (Option.isSome(httpSnapshot)) { yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); current = yield* SubscriptionRef.get(state); @@ -637,7 +703,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }; }), { - onExpectedFailure: setStreamError, + onExpectedFailure: handleStreamError, retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index cd9f3a74787..e82abcca3ba 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -1706,6 +1706,17 @@ export class OrchestrationGetSnapshotError extends Schema.TaggedErrorClass()( + "OrchestrationThreadNotFoundError", + { + threadId: ThreadId, + }, +) { + override get message(): string { + return `Thread ${this.threadId} was not found`; + } +} + export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass()( "OrchestrationDispatchCommandError", { diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index 51c65f50e1a..57b169d85ec 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -61,6 +61,7 @@ import { OrchestrationGetSnapshotError, OrchestrationSearchThreadsError, OrchestrationSearchThreadsInput, + OrchestrationThreadNotFoundError, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, OrchestrationRpcSchemas, @@ -928,7 +929,11 @@ export const WsOrchestrationSubscribeThreadRpc = Rpc.make( { payload: OrchestrationRpcSchemas.subscribeThread.input, success: OrchestrationRpcSchemas.subscribeThread.output, - error: Schema.Union([OrchestrationGetSnapshotError, EnvironmentAuthorizationError]), + error: Schema.Union([ + OrchestrationGetSnapshotError, + OrchestrationThreadNotFoundError, + EnvironmentAuthorizationError, + ]), stream: true, }, );