From 4b23d1d28bab12491c0fab0e9b5aef8d232e798d Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 09:26:19 +0200 Subject: [PATCH 1/7] fix(sync): stop retrying missing threads --- apps/mobile/src/connection/runtime.ts | 11 +- apps/server/src/server.test.ts | 25 +++ apps/server/src/ws.ts | 1 + apps/web/src/connection/runtime.ts | 11 +- .../src/state/shell-sync.test.ts | 14 ++ packages/client-runtime/src/state/shell.ts | 46 ++++-- .../src/state/shellMembership.test.ts | 61 ++++++++ .../src/state/shellMembership.ts | 55 +++++++ .../src/state/threads-atoms.test.ts | 6 +- .../src/state/threads-sync.test.ts | 144 +++++++++++++++++- packages/client-runtime/src/state/threads.ts | 42 ++++- packages/contracts/src/orchestration.ts | 1 + 12 files changed, 391 insertions(+), 26 deletions(-) create mode 100644 packages/client-runtime/src/state/shellMembership.test.ts create mode 100644 packages/client-runtime/src/state/shellMembership.ts diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index 3d4bf4944f1..56e8c71667b 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -1,5 +1,8 @@ import { Connection } from "@t3tools/client-runtime/connection"; -import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell"; +import { + environmentShellMembershipLayer, + shellSnapshotLoaderLayer, +} from "@t3tools/client-runtime/state/shell"; import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -11,7 +14,11 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); -const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); +const snapshotLoaderLayer = Layer.mergeAll( + threadSnapshotLoaderLayer, + shellSnapshotLoaderLayer, + environmentShellMembershipLayer, +); type ConnectionLayerSource = | typeof Connection.layer diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 6122c498145..c57033d3bc4 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5662,6 +5662,31 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("tags a missing thread subscription as not found", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + 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, + }).pipe(Stream.runCollect), + ).pipe(Effect.result), + ); + + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); + assert.equal(result.failure.reason, "not-found"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("buffers shell events published while the fallback snapshot loads", () => Effect.gen(function* () { const liveEvents = yield* PubSub.unbounded(); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 08b1770a0a2..0c70a26cfb1 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1429,6 +1429,7 @@ const makeWsRpcLayer = ( return yield* new OrchestrationGetSnapshotError({ message: `Thread ${input.threadId} was not found`, cause: input.threadId, + reason: "not-found", }); } diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 3d4bf4944f1..56e8c71667b 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -1,5 +1,8 @@ import { Connection } from "@t3tools/client-runtime/connection"; -import { shellSnapshotLoaderLayer } from "@t3tools/client-runtime/state/shell"; +import { + environmentShellMembershipLayer, + shellSnapshotLoaderLayer, +} from "@t3tools/client-runtime/state/shell"; import { threadSnapshotLoaderLayer } from "@t3tools/client-runtime/state/threads"; import * as Layer from "effect/Layer"; import { Atom } from "effect/unstable/reactivity"; @@ -11,7 +14,11 @@ const providedConnectionPlatformLayer = connectionPlatformLayer.pipe( Layer.provide(runtimeContextLayer), ); -const snapshotLoaderLayer = Layer.merge(threadSnapshotLoaderLayer, shellSnapshotLoaderLayer); +const snapshotLoaderLayer = Layer.mergeAll( + threadSnapshotLoaderLayer, + shellSnapshotLoaderLayer, + environmentShellMembershipLayer, +); type ConnectionLayerSource = | typeof Connection.layer diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index a4514d5f0e6..bfbe3c3362e 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -23,6 +23,7 @@ import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts"; +import { EnvironmentShellMembership } from "./shellMembership.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -94,10 +95,22 @@ describe("environment shell synchronization", () => { const snapshotLoader = ShellSnapshotLoader.of({ load: () => Effect.succeed(Option.none()), }); + const shellMembership = yield* Ref.make<"unknown" | "authoritative">("unknown"); const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), + Effect.provideService( + EnvironmentShellMembership, + EnvironmentShellMembership.of({ + getThreadMembership: () => + Ref.get(shellMembership).pipe( + Effect.map((status) => (status === "authoritative" ? "absent" : "unknown")), + ), + setAuthoritative: () => Ref.set(shellMembership, "authoritative"), + setUnknown: () => Ref.set(shellMembership, "unknown"), + }), + ), ); yield* SubscriptionRef.set(supervisorState, { @@ -145,6 +158,7 @@ describe("environment shell synchronization", () => { const state = yield* SubscriptionRef.get(shellState); expect(state.status).toBe("live"); expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT); + expect(yield* Ref.get(shellMembership)).toBe("authoritative"); }), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 6ccb11797f5..d0a8f00becf 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -23,6 +23,7 @@ import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribeDynamic } from "../rpc/client.ts"; import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; +import { EnvironmentShellMembership } from "./shellMembership.ts"; import type { EnvironmentCatalogState } from "./connections.ts"; import { followStreamInEnvironment } from "./runtime.ts"; @@ -52,6 +53,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ShellSnapshotLoader; + const shellMembership = yield* Effect.serviceOption(EnvironmentShellMembership); const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const cachedSnapshot = yield* cache.loadShell(environmentId).pipe( @@ -72,6 +74,15 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }); const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); + const setMembershipUnknown = Option.match(shellMembership, { + onNone: () => Effect.void, + onSome: (service) => service.setUnknown(environmentId), + }); + const setMembershipAuthoritative = (snapshot: OrchestrationShellSnapshot) => + Option.match(shellMembership, { + onNone: () => Effect.void, + onSome: (service) => service.setAuthoritative(environmentId, snapshot), + }); const persist = Effect.fn("EnvironmentShellState.persist")(function* ( snapshot: OrchestrationShellSnapshot, @@ -95,6 +106,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const setDisconnected = Ref.set(awaitingCompletion, false).pipe( + Effect.andThen(setMembershipUnknown), Effect.andThen( SubscriptionRef.update(state, (current) => ({ ...current, @@ -106,18 +118,22 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ...current, status: "synchronizing" as const, error: Option.none(), - })); - const setReady = SubscriptionRef.update(state, (current) => - current.status === "live" - ? current - : { - ...current, - status: "synchronizing" as const, - error: Option.none(), - }, - ); + })).pipe(Effect.andThen(setMembershipUnknown)); + const setReady = Effect.gen(function* () { + const current = yield* SubscriptionRef.get(state); + if (current.status === "live") { + return; + } + yield* setMembershipUnknown; + yield* SubscriptionRef.set(state, { + ...current, + status: "synchronizing" as const, + error: Option.none(), + }); + }); const setStreamError = (error: unknown) => Ref.set(awaitingCompletion, false).pipe( + Effect.andThen(setMembershipUnknown), Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")), Effect.annotateLogs({ environmentId, @@ -142,6 +158,10 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ? { ...current, status: "live" as const, error: Option.none() } : current, ); + const current = yield* SubscriptionRef.get(state); + if (current.status === "live" && Option.isSome(current.snapshot)) { + yield* setMembershipAuthoritative(current.snapshot.value); + } return; } @@ -166,6 +186,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") status: waiting ? "synchronizing" : "live", error: Option.none(), }); + if (!waiting) { + yield* setMembershipAuthoritative(nextSnapshot); + } yield* Queue.offer(persistence, nextSnapshot); }); @@ -233,6 +256,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Effect.forkScoped, ); + yield* Effect.addFinalizer(() => setMembershipUnknown); + return state; }); @@ -385,4 +410,5 @@ export * from "./models.ts"; export * from "./shellCommands.ts"; export * from "./shellReducer.ts"; export * from "./shellSnapshotHttp.ts"; +export * from "./shellMembership.ts"; export * from "./snapshots.ts"; diff --git a/packages/client-runtime/src/state/shellMembership.test.ts b/packages/client-runtime/src/state/shellMembership.test.ts new file mode 100644 index 00000000000..c2567b3cbf3 --- /dev/null +++ b/packages/client-runtime/src/state/shellMembership.test.ts @@ -0,0 +1,61 @@ +import { + EnvironmentId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationShellSnapshot, +} from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { EnvironmentShellMembership, environmentShellMembershipLayer } from "./shellMembership.ts"; + +const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); +const THREAD_ID = ThreadId.make("thread-1"); +const SNAPSHOT: OrchestrationShellSnapshot = { + snapshotSequence: 1, + projects: [], + threads: [ + { + id: THREAD_ID, + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: "default", + branch: "main", + worktreePath: null, + latestTurn: null, + createdAt: "2026-04-01T00:00:00.000Z", + updatedAt: "2026-04-01T00:00:00.000Z", + archivedAt: null, + latestUserMessageAt: null, + hasPendingApprovals: false, + hasPendingUserInput: false, + hasActionableProposedPlan: false, + session: null, + }, + ], + updatedAt: "2026-04-01T00:00:00.000Z", +}; + +describe("EnvironmentShellMembership", () => { + it.effect("resets authoritative membership to unknown", () => + Effect.gen(function* () { + const membership = yield* EnvironmentShellMembership; + + expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("unknown"); + yield* membership.setAuthoritative(ENVIRONMENT_ID, SNAPSHOT); + expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("present"); + expect( + yield* membership.getThreadMembership(ENVIRONMENT_ID, ThreadId.make("missing-thread")), + ).toBe("absent"); + + yield* membership.setUnknown(ENVIRONMENT_ID); + expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("unknown"); + }).pipe(Effect.provide(environmentShellMembershipLayer)), + ); +}); diff --git a/packages/client-runtime/src/state/shellMembership.ts b/packages/client-runtime/src/state/shellMembership.ts new file mode 100644 index 00000000000..d94049b089a --- /dev/null +++ b/packages/client-runtime/src/state/shellMembership.ts @@ -0,0 +1,55 @@ +import type { EnvironmentId, OrchestrationShellSnapshot, ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Ref from "effect/Ref"; + +export type ShellThreadMembership = "unknown" | "present" | "absent"; + +export class EnvironmentShellMembership extends Context.Service< + EnvironmentShellMembership, + { + readonly getThreadMembership: ( + environmentId: EnvironmentId, + threadId: ThreadId, + ) => Effect.Effect; + readonly setAuthoritative: ( + environmentId: EnvironmentId, + snapshot: OrchestrationShellSnapshot, + ) => Effect.Effect; + readonly setUnknown: (environmentId: EnvironmentId) => Effect.Effect; + } +>()("@t3tools/client-runtime/state/shellMembership/EnvironmentShellMembership") {} + +export const environmentShellMembershipLayer = Layer.effect( + EnvironmentShellMembership, + Effect.gen(function* () { + const snapshots = yield* Ref.make>( + new Map(), + ); + + return EnvironmentShellMembership.of({ + getThreadMembership: (environmentId, threadId) => + Ref.get(snapshots).pipe( + Effect.map((current) => { + const snapshot = current.get(environmentId); + if (snapshot === undefined) { + return "unknown"; + } + return snapshot.threads.some((thread) => thread.id === threadId) ? "present" : "absent"; + }), + ), + setAuthoritative: (environmentId, snapshot) => + Ref.update(snapshots, (current) => new Map(current).set(environmentId, snapshot)), + setUnknown: (environmentId) => + Ref.update(snapshots, (current) => { + if (!current.has(environmentId)) { + return current; + } + const next = new Map(current); + next.delete(environmentId); + return next; + }), + }); + }), +); diff --git a/packages/client-runtime/src/state/threads-atoms.test.ts b/packages/client-runtime/src/state/threads-atoms.test.ts index 0a9b2cd21b1..4e938338975 100644 --- a/packages/client-runtime/src/state/threads-atoms.test.ts +++ b/packages/client-runtime/src/state/threads-atoms.test.ts @@ -7,11 +7,15 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; import type { EnvironmentCacheStore } from "../platform/persistence.ts"; import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; import { createEnvironmentThreadStateAtoms, type ThreadSnapshotLoader } from "./threads.ts"; +import type { EnvironmentShellMembership } from "./shellMembership.ts"; describe("createEnvironmentThreadStateAtoms", () => { it("retains thread state across short subscriber gaps", () => { const runtime = Atom.runtime(Layer.empty) as unknown as Atom.AtomRuntime< - EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader, + | EnvironmentRegistry + | EnvironmentCacheStore + | EnvironmentShellMembership + | ThreadSnapshotLoader, never >; const threads = createEnvironmentThreadStateAtoms(runtime); diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index f3c4c4e6338..5e21f2140fa 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -2,6 +2,7 @@ import { EnvironmentId, EventId, ORCHESTRATION_WS_METHODS, + OrchestrationGetSnapshotError, ProjectId, ProviderInstanceId, ThreadId, @@ -36,6 +37,7 @@ import { ThreadSnapshotLoader, type EnvironmentThreadState, } from "./threads.ts"; +import { EnvironmentShellMembership, type ShellThreadMembership } from "./shellMembership.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -131,6 +133,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o readonly cached?: OrchestrationThread; readonly httpSnapshot?: Option.Option; readonly completionMarker?: boolean; + readonly shellMembership?: ShellThreadMembership; }) { const inputs = yield* Queue.unbounded(); const observed = yield* Queue.unbounded(); @@ -217,10 +220,16 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o saveVcsRefs: () => Effect.void, clear: () => Effect.void, }); + const shellMembership = EnvironmentShellMembership.of({ + getThreadMembership: () => Effect.succeed(options?.shellMembership ?? "unknown"), + setAuthoritative: () => Effect.void, + setUnknown: () => Effect.void, + }); const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ThreadSnapshotLoader, snapshotLoader), + Effect.provideService(EnvironmentShellMembership, shellMembership), Effect.provideService( ConnectionWakeups.ConnectionWakeups, ConnectionWakeups.ConnectionWakeups.of({ changes: Stream.fromQueue(wakeups) }), @@ -269,6 +278,19 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" }); +const missingThreadError = () => + new OrchestrationGetSnapshotError({ + message: `Thread ${THREAD_ID} was not found`, + cause: THREAD_ID, + reason: "not-found", + }); + +const unclassifiedSnapshotError = () => + new OrchestrationGetSnapshotError({ + message: `Failed to load thread ${THREAD_ID}`, + cause: THREAD_ID, + }); + const titleUpdated = (title: string, sequence = 2): OrchestrationThreadStreamItem => ({ kind: "event", event: { @@ -467,19 +489,63 @@ describe("EnvironmentThreads", () => { expect(yield* Ref.get(harness.loaderCalls)).toBe(0); yield* Queue.offer(harness.wakeups, "application-active"); - for (let attempt = 0; attempt < 100; attempt += 1) { - if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; - yield* Effect.yieldNow; - } + yield* Queue.offer( + harness.inputs, + snapshot({ ...BASE_THREAD, title: "Stale socket thread" }), + ); + for (let attempt = 0; attempt < 100; attempt += 1) yield* Effect.yieldNow; 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); }), ); + it.effect("stops cold hydration when the synchronized shell excludes the thread", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ shellMembership: "absent" }); + yield* Queue.offer(harness.inputs, missingThreadError()); + + 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(Option.isNone(state.data)).toBe(true); + expect(yield* Ref.get(harness.loaderCalls)).toBe(1); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); + expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]); + }), + ); + + it.effect( + "removes warm cached data and stops subscribing when the shell excludes the thread", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ + cached: BASE_THREAD, + shellMembership: "absent", + }); + yield* Queue.offer(harness.inputs, missingThreadError()); + + 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(Option.isNone(state.data)).toBe(true); + expect(yield* Ref.get(harness.loaderCalls)).toBe(0); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); + expect(yield* Ref.get(harness.removedThreads)).toEqual([THREAD_ID]); + }), + ); + it.effect("preserves data after a domain failure and resumes on a replacement session", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); @@ -523,13 +589,13 @@ describe("EnvironmentThreads", () => { it.effect("recovers from a transient domain failure without replacing the session", () => Effect.gen(function* () { - const harness = yield* makeHarness(); - yield* Queue.offer(harness.inputs, new Error("thread not found yet")); + const harness = yield* makeHarness({ shellMembership: "present" }); + yield* Queue.offer(harness.inputs, missingThreadError()); const failed = yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.error), ); - expect(Option.getOrThrow(failed.error)).toBe("thread not found yet"); + expect(Option.getOrThrow(failed.error)).toContain("was not found"); expect(yield* Ref.get(harness.subscriptionCount)).toBe(1); yield* TestClock.adjust("250 millis"); @@ -561,6 +627,68 @@ describe("EnvironmentThreads", () => { }), ); + it.effect("keeps retrying a missing thread while shell membership is unknown", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ shellMembership: "unknown" }); + yield* Queue.offer(harness.inputs, missingThreadError()); + + yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.error)); + yield* TestClock.adjust("250 millis"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + yield* Queue.offer( + harness.inputs, + snapshot({ ...BASE_THREAD, title: "Recovered without shell authority" }), + ); + + const recovered = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Recovered without shell authority", + ); + + expect(Option.isNone(recovered.error)).toBe(true); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.loaderCalls)).toBe(2); + }), + ); + + it.effect( + "keeps unclassified snapshot failures retryable when the shell excludes the thread", + () => + Effect.gen(function* () { + const harness = yield* makeHarness({ shellMembership: "absent" }); + yield* Queue.offer(harness.inputs, unclassifiedSnapshotError()); + + yield* awaitThreadState(harness.observed, (value) => Option.isSome(value.error)); + yield* TestClock.adjust("250 millis"); + for (let attempt = 0; attempt < 100; attempt += 1) { + if ((yield* Ref.get(harness.subscriptionCount)) >= 2) break; + yield* Effect.yieldNow; + } + yield* Queue.offer( + harness.inputs, + snapshot({ ...BASE_THREAD, title: "Recovered from unclassified failure" }), + ); + + const recovered = yield* awaitThreadState( + harness.observed, + (value) => + value.status === "live" && + Option.isSome(value.data) && + value.data.value.title === "Recovered from unclassified failure", + ); + + expect(Option.isNone(recovered.error)).toBe(true); + expect(yield* Ref.get(harness.subscriptionCount)).toBe(2); + expect(yield* Ref.get(harness.loaderCalls)).toBe(2); + }), + ); + it.effect("does not overwrite a live snapshot when the supervisor becomes ready", () => Effect.gen(function* () { const harness = yield* makeHarness({ cached: BASE_THREAD }); diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 196229cc8b1..51840452568 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,5 +1,6 @@ import { ORCHESTRATION_WS_METHODS, + OrchestrationGetSnapshotError, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, type OrchestrationThreadDetailSnapshot, @@ -7,10 +8,12 @@ import { type ThreadId as ThreadIdType, } from "@t3tools/contracts"; import * as Cause from "effect/Cause"; +import * as Deferred from "effect/Deferred"; 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 Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import { Atom } from "effect/unstable/reactivity"; @@ -25,6 +28,7 @@ import { ThreadSnapshotLoader } from "./threadSnapshotHttp.ts"; import { parseThreadKey, threadKey } from "./entities.ts"; import { applyThreadDetailEvent } from "./threadReducer.ts"; import { THREAD_STATE_IDLE_TTL_MS } from "./threadRetention.ts"; +import { EnvironmentShellMembership } from "./shellMembership.ts"; import { followStreamInEnvironment } from "./runtime.ts"; import { EMPTY_ENVIRONMENT_THREAD_STATE, @@ -43,6 +47,17 @@ function formatThreadError(cause: Cause.Cause): string { : "Could not synchronize the thread."; } +const isOrchestrationGetSnapshotError = Schema.is(OrchestrationGetSnapshotError); + +function isMissingThreadFailure(cause: Cause.Cause): boolean { + return cause.reasons.some( + (reason) => + reason._tag === "Fail" && + isOrchestrationGetSnapshotError(reason.error) && + reason.error.reason === "not-found", + ); +} + function shouldPersistThread(thread: OrchestrationThread): boolean { const status = thread.session?.status; return status !== "starting" && status !== "running"; @@ -54,6 +69,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const supervisor = yield* EnvironmentSupervisor; const cache = yield* EnvironmentCacheStore; const snapshotLoader = yield* ThreadSnapshotLoader; + const shellMembership = yield* EnvironmentShellMembership; const wakeups = yield* Effect.serviceOption(ConnectionWakeups.ConnectionWakeups); const environmentId = supervisor.target.environmentId; const cached = yield* cache.loadThread(environmentId, threadId).pipe( @@ -81,6 +97,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ); const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); + const deletedSignal = yield* Deferred.make(); const persist = Effect.fn("EnvironmentThreadState.persist")(function* ( snapshot: OrchestrationThreadDetailSnapshot, @@ -177,6 +194,20 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ), ), ); + yield* Deferred.succeed(deletedSignal, undefined); + }); + + const handleStreamFailure = Effect.fn("EnvironmentThreadState.handleStreamFailure")(function* ( + cause: Cause.Cause, + ) { + if (isMissingThreadFailure(cause)) { + const membership = yield* shellMembership.getThreadMembership(environmentId, threadId); + if (membership === "absent") { + yield* setDeleted(); + return; + } + } + yield* setStreamError(cause); }); const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* ( @@ -291,11 +322,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make }; }), { - onExpectedFailure: setStreamError, + onExpectedFailure: handleStreamFailure, retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEach(applyItem)), + ).pipe(Stream.runForEach(applyItem), Effect.raceFirst(Deferred.await(deletedSignal))), ); yield* Effect.addFinalizer(() => @@ -322,7 +353,11 @@ export function threadStateChanges(environmentId: EnvironmentIdType, threadId: T export function createEnvironmentThreadStateAtoms( runtime: Atom.AtomRuntime< - EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader | R, + | EnvironmentRegistry + | EnvironmentCacheStore + | EnvironmentShellMembership + | ThreadSnapshotLoader + | R, E >, ) { @@ -347,6 +382,7 @@ export function createEnvironmentThreadStateAtoms( export * from "./archivedThreads.ts"; export * from "./checkpointDiff.ts"; export * from "./threadSnapshotHttp.ts"; +export * from "./shellMembership.ts"; export * from "./composerPathSearch.ts"; export * from "./threadCommands.ts"; export * from "./threadDetail.ts"; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index c0c9c62d080..5d9df042b16 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -1292,6 +1292,7 @@ export class OrchestrationGetSnapshotError extends Schema.TaggedErrorClass Date: Tue, 21 Jul 2026 09:32:58 +0200 Subject: [PATCH 2/7] refactor(sync): model missing threads explicitly --- apps/server/src/server.test.ts | 4 ++-- apps/server/src/ws.ts | 7 +++---- .../client-runtime/src/state/threads-sync.test.ts | 7 +++---- packages/client-runtime/src/state/threads.ts | 9 +++------ packages/contracts/src/orchestration.ts | 12 +++++++++++- packages/contracts/src/rpc.ts | 7 ++++++- 6 files changed, 28 insertions(+), 18 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index c57033d3bc4..9b46c04e540 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5682,8 +5682,8 @@ it.layer(NodeServices.layer)("server router seam", (it) => { ); assertTrue(result._tag === "Failure"); - assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); - assert.equal(result.failure.reason, "not-found"); + assertTrue(result.failure._tag === "OrchestrationThreadNotFoundError"); + assert.equal(result.failure.threadId, defaultThreadId); }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0c70a26cfb1..0a73f33125a 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -34,6 +34,7 @@ import { type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, + OrchestrationThreadNotFoundError, OrchestrationGetTurnDiffError, ORCHESTRATION_WS_METHODS, type ProjectId, @@ -1426,10 +1427,8 @@ const makeWsRpcLayer = ( ); if (Option.isNone(snapshot)) { - return yield* new OrchestrationGetSnapshotError({ - message: `Thread ${input.threadId} was not found`, - cause: input.threadId, - reason: "not-found", + return yield* new OrchestrationThreadNotFoundError({ + threadId: input.threadId, }); } diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 5e21f2140fa..458807cbbe0 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -3,6 +3,7 @@ import { EventId, ORCHESTRATION_WS_METHODS, OrchestrationGetSnapshotError, + OrchestrationThreadNotFoundError, ProjectId, ProviderInstanceId, ThreadId, @@ -279,10 +280,8 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem => const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" }); const missingThreadError = () => - new OrchestrationGetSnapshotError({ - message: `Thread ${THREAD_ID} was not found`, - cause: THREAD_ID, - reason: "not-found", + new OrchestrationThreadNotFoundError({ + threadId: THREAD_ID, }); const unclassifiedSnapshotError = () => diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 51840452568..c742eb675a9 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -1,6 +1,6 @@ import { ORCHESTRATION_WS_METHODS, - OrchestrationGetSnapshotError, + OrchestrationThreadNotFoundError, type EnvironmentId as EnvironmentIdType, type OrchestrationThread, type OrchestrationThreadDetailSnapshot, @@ -47,14 +47,11 @@ function formatThreadError(cause: Cause.Cause): string { : "Could not synchronize the thread."; } -const isOrchestrationGetSnapshotError = Schema.is(OrchestrationGetSnapshotError); +const isOrchestrationThreadNotFoundError = Schema.is(OrchestrationThreadNotFoundError); function isMissingThreadFailure(cause: Cause.Cause): boolean { return cause.reasons.some( - (reason) => - reason._tag === "Fail" && - isOrchestrationGetSnapshotError(reason.error) && - reason.error.reason === "not-found", + (reason) => reason._tag === "Fail" && isOrchestrationThreadNotFoundError(reason.error), ); } diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 5d9df042b16..9fe788c0dd3 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -1292,10 +1292,20 @@ 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 0356aa1807b..ccee5e812af 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -52,6 +52,7 @@ import { OrchestrationGetFullThreadDiffError, OrchestrationGetFullThreadDiffInput, OrchestrationGetSnapshotError, + OrchestrationThreadNotFoundError, OrchestrationGetTurnDiffError, OrchestrationGetTurnDiffInput, OrchestrationReplayEventsError, @@ -648,7 +649,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, }, ); From 2c8a172d2ef043e795aea6983e81b9db88b754c9 Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 09:41:16 +0200 Subject: [PATCH 3/7] fix(sync): reject stale shell membership writes --- .../src/state/shell-sync.test.ts | 2 +- packages/client-runtime/src/state/shell.ts | 29 +++++++++------- .../src/state/shellMembership.test.ts | 13 ++++++-- .../src/state/shellMembership.ts | 33 ++++++++++++------- .../src/state/threads-sync.test.ts | 2 +- 5 files changed, 52 insertions(+), 27 deletions(-) diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index bfbe3c3362e..1c763414c9d 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -108,7 +108,7 @@ describe("environment shell synchronization", () => { Effect.map((status) => (status === "authoritative" ? "absent" : "unknown")), ), setAuthoritative: () => Ref.set(shellMembership, "authoritative"), - setUnknown: () => Ref.set(shellMembership, "unknown"), + setUnknown: () => Ref.set(shellMembership, "unknown").pipe(Effect.as(1)), }), ), ); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index d0a8f00becf..f316ba40e07 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -23,7 +23,7 @@ import { EnvironmentCacheStore } from "../platform/persistence.ts"; import { subscribeDynamic } from "../rpc/client.ts"; import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; -import { EnvironmentShellMembership } from "./shellMembership.ts"; +import { EnvironmentShellMembership, type ShellMembershipRevision } from "./shellMembership.ts"; import type { EnvironmentCatalogState } from "./connections.ts"; import { followStreamInEnvironment } from "./runtime.ts"; @@ -74,14 +74,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }); const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); - const setMembershipUnknown = Option.match(shellMembership, { - onNone: () => Effect.void, + const activeMembershipRevision = yield* Ref.make(0); + const invalidateMembership = Option.match(shellMembership, { + onNone: () => Effect.succeed(0), onSome: (service) => service.setUnknown(environmentId), }); const setMembershipAuthoritative = (snapshot: OrchestrationShellSnapshot) => Option.match(shellMembership, { onNone: () => Effect.void, - onSome: (service) => service.setAuthoritative(environmentId, snapshot), + onSome: (service) => + Ref.get(activeMembershipRevision).pipe( + Effect.flatMap((revision) => service.setAuthoritative(environmentId, snapshot, revision)), + ), }); const persist = Effect.fn("EnvironmentShellState.persist")(function* ( @@ -106,7 +110,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const setDisconnected = Ref.set(awaitingCompletion, false).pipe( - Effect.andThen(setMembershipUnknown), + Effect.andThen(invalidateMembership), Effect.andThen( SubscriptionRef.update(state, (current) => ({ ...current, @@ -114,17 +118,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") })), ), ); - const setSynchronizing = SubscriptionRef.update(state, (current) => ({ + const setSynchronizingState = SubscriptionRef.update(state, (current) => ({ ...current, status: "synchronizing" as const, error: Option.none(), - })).pipe(Effect.andThen(setMembershipUnknown)); + })); + const setSynchronizing = invalidateMembership.pipe(Effect.andThen(setSynchronizingState)); const setReady = Effect.gen(function* () { const current = yield* SubscriptionRef.get(state); if (current.status === "live") { return; } - yield* setMembershipUnknown; + yield* invalidateMembership; yield* SubscriptionRef.set(state, { ...current, status: "synchronizing" as const, @@ -133,7 +138,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }); const setStreamError = (error: unknown) => Ref.set(awaitingCompletion, false).pipe( - Effect.andThen(setMembershipUnknown), + Effect.andThen(invalidateMembership), Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")), Effect.annotateLogs({ environmentId, @@ -208,7 +213,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Effect.orElseSucceed(() => false), ); yield* Ref.set(awaitingCompletion, supportsCompletionMarker); - yield* setSynchronizing; + const membershipRevision = yield* invalidateMembership; + yield* Ref.set(activeMembershipRevision, membershipRevision); + yield* setSynchronizingState; const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( @@ -256,7 +263,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") Effect.forkScoped, ); - yield* Effect.addFinalizer(() => setMembershipUnknown); + yield* Effect.addFinalizer(() => invalidateMembership); return state; }); diff --git a/packages/client-runtime/src/state/shellMembership.test.ts b/packages/client-runtime/src/state/shellMembership.test.ts index c2567b3cbf3..6bb15ee905b 100644 --- a/packages/client-runtime/src/state/shellMembership.test.ts +++ b/packages/client-runtime/src/state/shellMembership.test.ts @@ -48,14 +48,23 @@ describe("EnvironmentShellMembership", () => { const membership = yield* EnvironmentShellMembership; expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("unknown"); - yield* membership.setAuthoritative(ENVIRONMENT_ID, SNAPSHOT); + const firstRevision = yield* membership.setUnknown(ENVIRONMENT_ID); + yield* membership.setAuthoritative(ENVIRONMENT_ID, SNAPSHOT, firstRevision); expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("present"); expect( yield* membership.getThreadMembership(ENVIRONMENT_ID, ThreadId.make("missing-thread")), ).toBe("absent"); - yield* membership.setUnknown(ENVIRONMENT_ID); + const currentRevision = yield* membership.setUnknown(ENVIRONMENT_ID); expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("unknown"); + + // Simulate a snapshot write that was paused before a disconnect invalidated + // its subscription revision. Resuming it cannot restore stale authority. + yield* membership.setAuthoritative(ENVIRONMENT_ID, SNAPSHOT, firstRevision); + expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("unknown"); + + yield* membership.setAuthoritative(ENVIRONMENT_ID, SNAPSHOT, currentRevision); + expect(yield* membership.getThreadMembership(ENVIRONMENT_ID, THREAD_ID)).toBe("present"); }).pipe(Effect.provide(environmentShellMembershipLayer)), ); }); diff --git a/packages/client-runtime/src/state/shellMembership.ts b/packages/client-runtime/src/state/shellMembership.ts index d94049b089a..561f19f4ce4 100644 --- a/packages/client-runtime/src/state/shellMembership.ts +++ b/packages/client-runtime/src/state/shellMembership.ts @@ -5,6 +5,12 @@ import * as Layer from "effect/Layer"; import * as Ref from "effect/Ref"; export type ShellThreadMembership = "unknown" | "present" | "absent"; +export type ShellMembershipRevision = number; + +interface EnvironmentMembershipState { + readonly revision: ShellMembershipRevision; + readonly snapshot: OrchestrationShellSnapshot | undefined; +} export class EnvironmentShellMembership extends Context.Service< EnvironmentShellMembership, @@ -16,39 +22,42 @@ export class EnvironmentShellMembership extends Context.Service< readonly setAuthoritative: ( environmentId: EnvironmentId, snapshot: OrchestrationShellSnapshot, + revision: ShellMembershipRevision, ) => Effect.Effect; - readonly setUnknown: (environmentId: EnvironmentId) => Effect.Effect; + readonly setUnknown: (environmentId: EnvironmentId) => Effect.Effect; } >()("@t3tools/client-runtime/state/shellMembership/EnvironmentShellMembership") {} export const environmentShellMembershipLayer = Layer.effect( EnvironmentShellMembership, Effect.gen(function* () { - const snapshots = yield* Ref.make>( + const states = yield* Ref.make>( new Map(), ); return EnvironmentShellMembership.of({ getThreadMembership: (environmentId, threadId) => - Ref.get(snapshots).pipe( + Ref.get(states).pipe( Effect.map((current) => { - const snapshot = current.get(environmentId); + const snapshot = current.get(environmentId)?.snapshot; if (snapshot === undefined) { return "unknown"; } return snapshot.threads.some((thread) => thread.id === threadId) ? "present" : "absent"; }), ), - setAuthoritative: (environmentId, snapshot) => - Ref.update(snapshots, (current) => new Map(current).set(environmentId, snapshot)), - setUnknown: (environmentId) => - Ref.update(snapshots, (current) => { - if (!current.has(environmentId)) { + setAuthoritative: (environmentId, snapshot, revision) => + Ref.update(states, (current) => { + const state = current.get(environmentId); + if (state === undefined || state.revision !== revision) { return current; } - const next = new Map(current); - next.delete(environmentId); - return next; + return new Map(current).set(environmentId, { revision, snapshot }); + }), + setUnknown: (environmentId) => + Ref.modify(states, (current) => { + const revision = (current.get(environmentId)?.revision ?? 0) + 1; + return [revision, new Map(current).set(environmentId, { revision, snapshot: undefined })]; }), }); }), diff --git a/packages/client-runtime/src/state/threads-sync.test.ts b/packages/client-runtime/src/state/threads-sync.test.ts index 458807cbbe0..cbcb656905d 100644 --- a/packages/client-runtime/src/state/threads-sync.test.ts +++ b/packages/client-runtime/src/state/threads-sync.test.ts @@ -224,7 +224,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o const shellMembership = EnvironmentShellMembership.of({ getThreadMembership: () => Effect.succeed(options?.shellMembership ?? "unknown"), setAuthoritative: () => Effect.void, - setUnknown: () => Effect.void, + setUnknown: () => Effect.succeed(0), }); const threadState = yield* makeEnvironmentThreadState(THREAD_ID).pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), From a154e8d18dc620b362a5cdda4e6f8734c28dab62 Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 09:53:25 +0200 Subject: [PATCH 4/7] fix(sync): preserve concurrent shell updates --- packages/client-runtime/src/state/shell.ts | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index f316ba40e07..865cb919500 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -124,18 +124,15 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") error: Option.none(), })); const setSynchronizing = invalidateMembership.pipe(Effect.andThen(setSynchronizingState)); - const setReady = Effect.gen(function* () { - const current = yield* SubscriptionRef.get(state); - if (current.status === "live") { - return; - } - yield* invalidateMembership; - yield* SubscriptionRef.set(state, { - ...current, - status: "synchronizing" as const, - error: Option.none(), - }); - }); + const setReady = SubscriptionRef.update(state, (current) => + current.status === "live" + ? current + : { + ...current, + status: "synchronizing" as const, + error: Option.none(), + }, + ); const setStreamError = (error: unknown) => Ref.set(awaitingCompletion, false).pipe( Effect.andThen(invalidateMembership), From 975caac90bfb3b2fbb45abafcf82b7082ffd599b Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 10:29:25 +0200 Subject: [PATCH 5/7] fix(sync): address missing-thread review races --- apps/server/src/server.test.ts | 27 +++ apps/server/src/ws.ts | 17 ++ .../client-runtime/src/rpc/client.test.ts | 50 +++++- packages/client-runtime/src/rpc/client.ts | 155 ++++++++++-------- .../src/state/shell-sync.test.ts | 35 ++-- packages/client-runtime/src/state/shell.ts | 44 ++--- 6 files changed, 228 insertions(+), 100 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 9b46c04e540..e9876bf829d 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5687,6 +5687,33 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("tags a missing resumed thread subscription as not found", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getThreadShellById: () => Effect.succeed(Option.none()), + getThreadDetailSnapshot: () => Effect.die("unused for resumed subscriptions"), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 7, + }).pipe(Stream.runCollect), + ).pipe(Effect.result), + ); + + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationThreadNotFoundError"); + assert.equal(result.failure.threadId, defaultThreadId); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("buffers shell events published while the fallback snapshot loads", () => Effect.gen(function* () { const liveEvents = yield* PubSub.unbounded(); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0a73f33125a..cbd6afbaa8d 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1388,6 +1388,23 @@ const makeWsRpcLayer = ( // snapshot sequence) and the per-thread filter runs after reading, // so a global cap could otherwise omit this thread's events. if (input.afterSequence !== undefined) { + const thread = yield* projectionSnapshotQuery + .getThreadShellById(input.threadId) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to load thread ${input.threadId}`, + cause, + }), + ), + ); + if (Option.isNone(thread)) { + return yield* new OrchestrationThreadNotFoundError({ + threadId: input.threadId, + }); + } + const afterSequence = input.afterSequence; const catchUpStream = orchestrationEngine .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) diff --git a/packages/client-runtime/src/rpc/client.test.ts b/packages/client-runtime/src/rpc/client.test.ts index 507d137cacc..5b014d5c379 100644 --- a/packages/client-runtime/src/rpc/client.test.ts +++ b/packages/client-runtime/src/rpc/client.test.ts @@ -25,7 +25,13 @@ import { import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; -import { EnvironmentRpcRequestObserver, request, runStream, subscribe } from "./client.ts"; +import { + EnvironmentRpcRequestObserver, + request, + runStream, + subscribe, + subscribeDynamicWithContext, +} from "./client.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -185,6 +191,48 @@ describe("environment RPC", () => { }), ); + it.effect("keeps dynamic subscription context bound to its originating session", () => + Effect.gen(function* () { + const firstEvents = yield* Queue.unbounded(); + const secondEvents = yield* Queue.unbounded(); + const firstClient = { + [WS_METHODS.subscribeTerminalEvents]: () => Stream.fromQueue(firstEvents), + } as unknown as WsRpcProtocolClient; + const secondClient = { + [WS_METHODS.subscribeTerminalEvents]: () => Stream.fromQueue(secondEvents), + } as unknown as WsRpcProtocolClient; + const { activeSession, supervisor } = yield* makeHarness(); + const observedContexts = yield* Queue.unbounded(); + const nextContext = yield* Ref.make(0); + + const subscriptionFiber = yield* subscribeDynamicWithContext( + WS_METHODS.subscribeTerminalEvents, + () => + Ref.updateAndGet(nextContext, (context) => context + 1).pipe( + Effect.map((context) => [{}, context] as const), + ), + ).pipe( + Stream.runForEach(([, context]) => Queue.offer(observedContexts, context)), + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.forkChild, + ); + + yield* SubscriptionRef.set(activeSession, Option.some(session(firstClient))); + for (let attempt = 0; attempt < 100 && (yield* Ref.get(nextContext)) < 1; attempt += 1) { + yield* Effect.yieldNow; + } + yield* Queue.offer(firstEvents, {} as never); + expect(yield* Queue.take(observedContexts)).toBe(1); + yield* SubscriptionRef.set(activeSession, Option.some(session(secondClient))); + for (let attempt = 0; attempt < 100 && (yield* Ref.get(nextContext)) < 2; attempt += 1) { + yield* Effect.yieldNow; + } + yield* Queue.offer(secondEvents, {} as never); + expect(yield* Queue.take(observedContexts)).toBe(2); + yield* Fiber.interrupt(subscriptionFiber); + }), + ); + it.effect("keeps durable subscriptions alive across a transport failure and new session", () => Effect.gen(function* () { const subscriptions: string[] = []; diff --git a/packages/client-runtime/src/rpc/client.ts b/packages/client-runtime/src/rpc/client.ts index c7c928b3c95..aa2bcdfebcc 100644 --- a/packages/client-runtime/src/rpc/client.ts +++ b/packages/client-runtime/src/rpc/client.ts @@ -156,15 +156,18 @@ interface SubscriptionOptions { readonly resubscribe?: Stream.Stream; } -export function subscribeDynamic( +type SubscriptionMethod = ( + input: EnvironmentRpcInput, +) => Stream.Stream, EnvironmentRpcStreamFailure>; + +function subscribeDynamicStream( tag: TTag, - makeInput: (session: RpcSession) => Effect.Effect>, + makeStream: ( + session: RpcSession, + method: SubscriptionMethod, + ) => Stream.Stream>, options?: SubscriptionOptions, -): Stream.Stream< - EnvironmentRpcStreamValue, - EnvironmentRpcStreamFailure, - EnvironmentSupervisor -> { +): Stream.Stream, EnvironmentSupervisor> { return Stream.unwrap( EnvironmentSupervisor.pipe( Effect.map((supervisor) => { @@ -183,67 +186,52 @@ export function subscribeDynamic( Option.match({ onNone: () => Stream.empty, onSome: (session) => { - const method = session.client[tag] as ( - input: EnvironmentRpcInput, - ) => Stream.Stream< - EnvironmentRpcStreamValue, - EnvironmentRpcStreamFailure - >; + const method = session.client[tag] as SubscriptionMethod; const subscribeToSession = (): Stream.Stream< - EnvironmentRpcStreamValue, + TValue, EnvironmentRpcStreamFailure > => Stream.suspend(() => - Stream.unwrap( - makeInput(session).pipe( - Effect.map((input) => - method(input).pipe( - Stream.catchCause((cause) => { - const hasOnlyExpectedFailures = - cause.reasons.length > 0 && - cause.reasons.every((reason) => reason._tag === "Fail"); - const isTransportFailure = - hasOnlyExpectedFailures && - cause.reasons.every( - (reason) => - reason._tag === "Fail" && isRpcClientError(reason.error), - ); - if (isTransportFailure) { - return Stream.fromEffect( - Effect.logWarning( - "Durable RPC subscription lost its transport; waiting for the next session.", - { - cause: Cause.pretty(cause), - method: tag, - environmentId: supervisor.target.environmentId, - }, - ), - ).pipe(Stream.drain); - } - if ( - hasOnlyExpectedFailures && - options?.onExpectedFailure !== undefined - ) { - const handled = Stream.fromEffect( - options.onExpectedFailure(cause), - ).pipe(Stream.drain); - if (options.retryExpectedFailureAfter === undefined) { - return handled; - } - return handled.pipe( - Stream.concat( - Stream.fromEffect( - Effect.sleep(options.retryExpectedFailureAfter), - ).pipe(Stream.drain), - ), - Stream.concat(subscribeToSession()), - ); - } - return Stream.failCause(cause); - }), - ), - ), - ), + makeStream(session, method).pipe( + Stream.catchCause((cause) => { + const hasOnlyExpectedFailures = + cause.reasons.length > 0 && + cause.reasons.every((reason) => reason._tag === "Fail"); + const isTransportFailure = + hasOnlyExpectedFailures && + cause.reasons.every( + (reason) => reason._tag === "Fail" && isRpcClientError(reason.error), + ); + if (isTransportFailure) { + return Stream.fromEffect( + Effect.logWarning( + "Durable RPC subscription lost its transport; waiting for the next session.", + { + cause: Cause.pretty(cause), + method: tag, + environmentId: supervisor.target.environmentId, + }, + ), + ).pipe(Stream.drain); + } + if (hasOnlyExpectedFailures && options?.onExpectedFailure !== undefined) { + const handled = Stream.fromEffect(options.onExpectedFailure(cause)).pipe( + Stream.drain, + ); + if (options.retryExpectedFailureAfter === undefined) { + return handled; + } + return handled.pipe( + Stream.concat( + Stream.fromEffect( + Effect.sleep(options.retryExpectedFailureAfter), + ).pipe(Stream.drain), + ), + Stream.concat(subscribeToSession()), + ); + } + return Stream.failCause(cause); + }), ), ); return subscribeToSession(); @@ -260,6 +248,45 @@ export function subscribeDynamic( ); } +export function subscribeDynamic( + tag: TTag, + makeInput: (session: RpcSession) => Effect.Effect>, + options?: SubscriptionOptions, +): Stream.Stream< + EnvironmentRpcStreamValue, + EnvironmentRpcStreamFailure, + EnvironmentSupervisor +> { + return subscribeDynamicStream( + tag, + (session, method) => Stream.unwrap(makeInput(session).pipe(Effect.map(method))), + options, + ); +} + +export function subscribeDynamicWithContext( + tag: TTag, + makeInput: (session: RpcSession) => Effect.Effect, TContext]>, + options?: SubscriptionOptions, +): Stream.Stream< + readonly [EnvironmentRpcStreamValue, TContext], + EnvironmentRpcStreamFailure, + EnvironmentSupervisor +> { + return subscribeDynamicStream( + tag, + (session, method) => + Stream.unwrap( + makeInput(session).pipe( + Effect.map(([input, context]) => + method(input).pipe(Stream.map((value) => [value, context] as const)), + ), + ), + ), + options, + ); +} + export function subscribe( tag: TTag, input: EnvironmentRpcInput, diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 1c763414c9d..949f59cf67b 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -1,10 +1,12 @@ import { EnvironmentId, ORCHESTRATION_WS_METHODS, + ThreadId, type OrchestrationShellSnapshot, type OrchestrationShellStreamItem, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; @@ -23,7 +25,7 @@ import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { makeEnvironmentShellState, ShellSnapshotLoader } from "./shell.ts"; -import { EnvironmentShellMembership } from "./shellMembership.ts"; +import { EnvironmentShellMembership, environmentShellMembershipLayer } from "./shellMembership.ts"; const TARGET = new PrimaryConnectionTarget({ environmentId: EnvironmentId.make("environment-1"), @@ -62,8 +64,14 @@ describe("environment shell synchronization", () => { it.effect("publishes live state before persistence and preserves it when ready", () => Effect.gen(function* () { const events = yield* Queue.unbounded(); + const subscriptionStarted = yield* Deferred.make(); const client = { - [ORCHESTRATION_WS_METHODS.subscribeShell]: () => Stream.fromQueue(events), + [ORCHESTRATION_WS_METHODS.subscribeShell]: () => + Stream.unwrap( + Deferred.succeed(subscriptionStarted, undefined).pipe( + Effect.as(Stream.fromQueue(events)), + ), + ), } as unknown as WsRpcProtocolClient; const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); const activeSession = yield* SubscriptionRef.make>( @@ -95,24 +103,14 @@ describe("environment shell synchronization", () => { const snapshotLoader = ShellSnapshotLoader.of({ load: () => Effect.succeed(Option.none()), }); - const shellMembership = yield* Ref.make<"unknown" | "authoritative">("unknown"); + const shellMembership = yield* EnvironmentShellMembership; const shellState = yield* makeEnvironmentShellState().pipe( Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), Effect.provideService(Persistence.EnvironmentCacheStore, cache), Effect.provideService(ShellSnapshotLoader, snapshotLoader), - Effect.provideService( - EnvironmentShellMembership, - EnvironmentShellMembership.of({ - getThreadMembership: () => - Ref.get(shellMembership).pipe( - Effect.map((status) => (status === "authoritative" ? "absent" : "unknown")), - ), - setAuthoritative: () => Ref.set(shellMembership, "authoritative"), - setUnknown: () => Ref.set(shellMembership, "unknown").pipe(Effect.as(1)), - }), - ), ); + yield* Deferred.await(subscriptionStarted); yield* SubscriptionRef.set(supervisorState, { desired: true, network: "online", @@ -158,8 +156,13 @@ describe("environment shell synchronization", () => { const state = yield* SubscriptionRef.get(shellState); expect(state.status).toBe("live"); expect(Option.getOrThrow(state.snapshot)).toEqual(LIVE_SHELL_SNAPSHOT); - expect(yield* Ref.get(shellMembership)).toBe("authoritative"); - }), + expect( + yield* shellMembership.getThreadMembership( + TARGET.environmentId, + ThreadId.make("missing-thread"), + ), + ).toBe("absent"); + }).pipe(Effect.provide(environmentShellMembershipLayer)), ); it.effect("replaces a warm shell cache with an authoritative HTTP snapshot", () => diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 865cb919500..102c768d11a 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -20,7 +20,7 @@ import { EnvironmentSupervisor } from "../connection/supervisor.ts"; import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import { EnvironmentCacheStore } from "../platform/persistence.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { subscribeDynamicWithContext } from "../rpc/client.ts"; import { ShellSnapshotLoader } from "./shellSnapshotHttp.ts"; import { applyShellStreamEvent } from "./shellReducer.ts"; import { EnvironmentShellMembership, type ShellMembershipRevision } from "./shellMembership.ts"; @@ -74,18 +74,17 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }); const awaitingCompletion = yield* Ref.make(false); const persistence = yield* Queue.sliding(1); - const activeMembershipRevision = yield* Ref.make(0); const invalidateMembership = Option.match(shellMembership, { onNone: () => Effect.succeed(0), onSome: (service) => service.setUnknown(environmentId), }); - const setMembershipAuthoritative = (snapshot: OrchestrationShellSnapshot) => + const setMembershipAuthoritative = ( + snapshot: OrchestrationShellSnapshot, + revision: ShellMembershipRevision, + ) => Option.match(shellMembership, { onNone: () => Effect.void, - onSome: (service) => - Ref.get(activeMembershipRevision).pipe( - Effect.flatMap((revision) => service.setAuthoritative(environmentId, snapshot, revision)), - ), + onSome: (service) => service.setAuthoritative(environmentId, snapshot, revision), }); const persist = Effect.fn("EnvironmentShellState.persist")(function* ( @@ -123,7 +122,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") status: "synchronizing" as const, error: Option.none(), })); - const setSynchronizing = invalidateMembership.pipe(Effect.andThen(setSynchronizingState)); + const setSynchronizing = setSynchronizingState; const setReady = SubscriptionRef.update(state, (current) => current.status === "live" ? current @@ -152,6 +151,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const applyItem = Effect.fn("EnvironmentShellState.applyItem")(function* ( item: OrchestrationShellStreamItem, + membershipRevision: ShellMembershipRevision, ) { if (item.kind === "synchronized") { yield* Ref.set(awaitingCompletion, false); @@ -162,7 +162,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const current = yield* SubscriptionRef.get(state); if (current.status === "live" && Option.isSome(current.snapshot)) { - yield* setMembershipAuthoritative(current.snapshot.value); + yield* setMembershipAuthoritative(current.snapshot.value, membershipRevision); } return; } @@ -189,7 +189,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") error: Option.none(), }); if (!waiting) { - yield* setMembershipAuthoritative(nextSnapshot); + yield* setMembershipAuthoritative(nextSnapshot, membershipRevision); } yield* Queue.offer(persistence, nextSnapshot); }); @@ -200,9 +200,10 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") service.changes.pipe(Stream.filter((reason) => reason === "application-active")), }); + yield* invalidateMembership; yield* setSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( + subscribeDynamicWithContext( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { const supportsCompletionMarker = yield* session.initialConfig.pipe( @@ -211,7 +212,6 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); yield* Ref.set(awaitingCompletion, supportsCompletionMarker); const membershipRevision = yield* invalidateMembership; - yield* Ref.set(activeMembershipRevision, membershipRevision); yield* setSynchronizingState; const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( @@ -230,21 +230,27 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ); const httpSnapshot = yield* snapshotLoader.load(prepared); if (Option.isSome(httpSnapshot)) { - yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }); - return { - afterSequence: httpSnapshot.value.snapshotSequence, - ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), - }; + yield* applyItem({ kind: "snapshot", snapshot: httpSnapshot.value }, membershipRevision); + return [ + { + afterSequence: httpSnapshot.value.snapshotSequence, + ...(supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}), + }, + membershipRevision, + ] as const; } - return supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}; + return [ + supportsCompletionMarker ? { requestCompletionMarker: true as const } : {}, + membershipRevision, + ] as const; }), { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEach(applyItem)), + ).pipe(Stream.runForEach(([item, membershipRevision]) => applyItem(item, membershipRevision))), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { From d479716cc8d98dcf21b05f7518a37d74d8a08136 Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 11:04:30 +0200 Subject: [PATCH 6/7] fix(sync): preserve archived thread resumes --- apps/server/src/server.test.ts | 38 ++++++++++++++++++++++++++++++++++ apps/server/src/ws.ts | 18 ++++++++++++++-- 2 files changed, 54 insertions(+), 2 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index e9876bf829d..ac8832606b6 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -5714,6 +5714,44 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + it.effect("keeps an archived resumed thread subscription alive", () => + Effect.gen(function* () { + yield* buildAppUnderTest({ + layers: { + projectionSnapshotQuery: { + getThreadShellById: () => Effect.succeed(Option.none()), + getArchivedShellSnapshot: () => + Effect.succeed({ + snapshotSequence: 7, + projects: [], + threads: [ + makeDefaultOrchestrationThreadShell({ + id: defaultThreadId, + archivedAt: "2026-01-01T00:00:00.000Z", + }), + ], + updatedAt: "2026-01-01T00:00:00.000Z", + }), + getThreadDetailSnapshot: () => Effect.die("unused for resumed subscriptions"), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const firstItem = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeThread]({ + threadId: defaultThreadId, + afterSequence: 7, + requestCompletionMarker: true, + }).pipe(Stream.runHead), + ), + ); + + assert.deepEqual(Option.getOrThrow(firstItem), { kind: "synchronized" }); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("buffers shell events published while the fallback snapshot loads", () => Effect.gen(function* () { const liveEvents = yield* PubSub.unbounded(); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index cbd6afbaa8d..fadda3df1e9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1388,7 +1388,7 @@ const makeWsRpcLayer = ( // snapshot sequence) and the per-thread filter runs after reading, // so a global cap could otherwise omit this thread's events. if (input.afterSequence !== undefined) { - const thread = yield* projectionSnapshotQuery + const activeThread = yield* projectionSnapshotQuery .getThreadShellById(input.threadId) .pipe( Effect.mapError( @@ -1399,7 +1399,21 @@ const makeWsRpcLayer = ( }), ), ); - if (Option.isNone(thread)) { + const archivedThreadExists = Option.isNone(activeThread) + ? yield* projectionSnapshotQuery.getArchivedShellSnapshot().pipe( + Effect.map((snapshot) => + snapshot.threads.some((thread) => thread.id === input.threadId), + ), + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: `Failed to load archived threads for ${input.threadId}`, + cause, + }), + ), + ) + : false; + if (Option.isNone(activeThread) && !archivedThreadExists) { return yield* new OrchestrationThreadNotFoundError({ threadId: input.threadId, }); From 1ce1c2501db27ec672e6a75e437e4a42a9950e64 Mon Sep 17 00:00:00 2001 From: Wout Stiens Date: Tue, 21 Jul 2026 11:12:58 +0200 Subject: [PATCH 7/7] fix(sync): query thread presence atomically --- .../checkpointing/CheckpointDiffQuery.test.ts | 5 ++++ .../Layers/OrchestrationEngine.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.test.ts | 4 +++ .../Layers/ProjectionSnapshotQuery.ts | 25 +++++++++++++++++++ .../Services/ProjectionSnapshotQuery.ts | 6 +++++ .../project/ProjectSetupScriptRunner.test.ts | 1 + .../Layers/ProviderSessionReaper.test.ts | 1 + apps/server/src/server.test.ts | 19 ++++---------- apps/server/src/serverRuntimeStartup.test.ts | 4 +++ apps/server/src/ws.ts | 20 +++------------ 10 files changed, 55 insertions(+), 31 deletions(-) diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts index 8e0e5fb74d5..93c33bc072b 100644 --- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts +++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts @@ -106,6 +106,7 @@ describe("CheckpointDiffQuery.layer", () => { }); }), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), @@ -199,6 +200,7 @@ describe("CheckpointDiffQuery.layer", () => { getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), @@ -282,6 +284,7 @@ describe("CheckpointDiffQuery.layer", () => { getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), @@ -350,6 +353,7 @@ describe("CheckpointDiffQuery.layer", () => { getThreadCheckpointContext: () => Effect.succeed(Option.some(threadCheckpointContext)), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), @@ -403,6 +407,7 @@ describe("CheckpointDiffQuery.layer", () => { getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts index b00c00e0d3f..c0bbd42c31f 100644 --- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts +++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts @@ -199,6 +199,7 @@ describe("OrchestrationEngine", () => { getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 9a136b06872..7a2e101164e 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -559,6 +559,10 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { [ThreadId.make("thread-archived")], ); assert.equal(archivedShellSnapshot.threads[0]?.archivedAt, "2026-04-06T00:00:06.000Z"); + + assert.isTrue(yield* snapshotQuery.hasThreadById(ThreadId.make("thread-active"))); + assert.isTrue(yield* snapshotQuery.hasThreadById(ThreadId.make("thread-archived"))); + assert.isFalse(yield* snapshotQuery.hasThreadById(ThreadId.make("thread-missing"))); }), ); diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 32210436e67..f493439cc08 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -767,6 +767,19 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { `, }); + const getNonDeletedThreadIdById = SqlSchema.findOneOption({ + Request: ThreadIdLookupInput, + Result: ProjectionThreadIdLookupRowSchema, + execute: ({ threadId }) => + sql` + SELECT thread_id AS "threadId" + FROM projection_threads + WHERE thread_id = ${threadId} + AND deleted_at IS NULL + LIMIT 1 + `, + }); + const listThreadMessageRowsByThread = SqlSchema.findAll({ Request: ThreadIdLookupInput, Result: ProjectionThreadMessageDbRowSchema, @@ -1895,6 +1908,17 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { } satisfies OrchestrationThreadShell); }); + const hasThreadById: ProjectionSnapshotQueryShape["hasThreadById"] = (threadId) => + getNonDeletedThreadIdById({ threadId }).pipe( + Effect.map(Option.isSome), + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProjectionSnapshotQuery.hasThreadById:query", + "ProjectionSnapshotQuery.hasThreadById:decodeRow", + ), + ), + ); + const getThreadDetailById: ProjectionSnapshotQueryShape["getThreadDetailById"] = (threadId) => Effect.gen(function* () { const [ @@ -2076,6 +2100,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { getThreadCheckpointContext, getFullThreadDiffContext, getThreadShellById, + hasThreadById, getThreadDetailById, getThreadDetailSnapshot, } satisfies ProjectionSnapshotQueryShape; diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts index 23b291d8778..8ba4fe1468e 100644 --- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts @@ -152,6 +152,12 @@ export interface ProjectionSnapshotQueryShape { threadId: ThreadId, ) => Effect.Effect, ProjectionRepositoryError>; + /** + * Check whether a non-deleted thread projection exists. Archived threads are + * included so lifecycle transitions cannot be mistaken for deletion. + */ + readonly hasThreadById: (threadId: ThreadId) => Effect.Effect; + /** * Read a single active thread detail snapshot by id. */ diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts index 15612908079..238ff1cdaaa 100644 --- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts +++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts @@ -42,6 +42,7 @@ const makeProjectionSnapshotQueryLayer = (project: OrchestrationProject) => getThreadCheckpointContext: () => Effect.die("unused"), getFullThreadDiffContext: () => Effect.die("unused"), getThreadShellById: () => Effect.die("unused"), + hasThreadById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), }); diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 166a88ef17b..66fd171cde2 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -208,6 +208,7 @@ describe("ProviderSessionReaper", () => { ? Option.some(input.readModel.threads.find((thread) => thread.id === threadId)!) : Option.none(), ), + hasThreadById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), }), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index ac8832606b6..99c61b981c9 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -703,6 +703,7 @@ const buildAppUnderTest = (options?: { getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 0 }), getProjectShellById: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), getCounts: () => Effect.succeed({ projectCount: 0, threadCount: 0 }), @@ -5692,7 +5693,7 @@ it.layer(NodeServices.layer)("server router seam", (it) => { yield* buildAppUnderTest({ layers: { projectionSnapshotQuery: { - getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailSnapshot: () => Effect.die("unused for resumed subscriptions"), }, }, @@ -5719,19 +5720,9 @@ it.layer(NodeServices.layer)("server router seam", (it) => { yield* buildAppUnderTest({ layers: { projectionSnapshotQuery: { - getThreadShellById: () => Effect.succeed(Option.none()), - getArchivedShellSnapshot: () => - Effect.succeed({ - snapshotSequence: 7, - projects: [], - threads: [ - makeDefaultOrchestrationThreadShell({ - id: defaultThreadId, - archivedAt: "2026-01-01T00:00:00.000Z", - }), - ], - updatedAt: "2026-01-01T00:00:00.000Z", - }), + hasThreadById: () => Effect.succeed(true), + getThreadShellById: () => + Effect.die("active shell lookup must not determine existence"), getThreadDetailSnapshot: () => Effect.die("unused for resumed subscriptions"), }, }, diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts index b8102bda9ad..e6be383c563 100644 --- a/apps/server/src/serverRuntimeStartup.test.ts +++ b/apps/server/src/serverRuntimeStartup.test.ts @@ -95,6 +95,7 @@ it.effect("launchStartupHeartbeat does not block the caller while counts are loa getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.succeed(Option.none()), + hasThreadById: () => Effect.succeed(false), getThreadDetailById: () => Effect.succeed(Option.none()), getThreadDetailSnapshot: () => Effect.succeed(Option.none()), }), @@ -158,6 +159,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets returns existing project and threa getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), + hasThreadById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), }), @@ -202,6 +204,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets creates a project and thread when getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), + hasThreadById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), }), @@ -252,6 +255,7 @@ it.effect("resolveAutoBootstrapWelcomeTargets preserves typed UUID generation fa getThreadCheckpointContext: () => Effect.succeed(Option.none()), getFullThreadDiffContext: () => Effect.succeed(Option.none()), getThreadShellById: () => Effect.die("unused"), + hasThreadById: () => Effect.die("unused"), getThreadDetailById: () => Effect.die("unused"), getThreadDetailSnapshot: () => Effect.die("unused"), }), diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index fadda3df1e9..24e132c3a12 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1388,8 +1388,8 @@ const makeWsRpcLayer = ( // snapshot sequence) and the per-thread filter runs after reading, // so a global cap could otherwise omit this thread's events. if (input.afterSequence !== undefined) { - const activeThread = yield* projectionSnapshotQuery - .getThreadShellById(input.threadId) + const threadExists = yield* projectionSnapshotQuery + .hasThreadById(input.threadId) .pipe( Effect.mapError( (cause) => @@ -1399,21 +1399,7 @@ const makeWsRpcLayer = ( }), ), ); - const archivedThreadExists = Option.isNone(activeThread) - ? yield* projectionSnapshotQuery.getArchivedShellSnapshot().pipe( - Effect.map((snapshot) => - snapshot.threads.some((thread) => thread.id === input.threadId), - ), - Effect.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: `Failed to load archived threads for ${input.threadId}`, - cause, - }), - ), - ) - : false; - if (Option.isNone(activeThread) && !archivedThreadExists) { + if (!threadExists) { return yield* new OrchestrationThreadNotFoundError({ threadId: input.threadId, });