From e96c98662b86e7c064c6d2bd6ac4ee60d62f04ab Mon Sep 17 00:00:00 2001 From: Carl Gabel Date: Sun, 19 Jul 2026 22:51:05 +1000 Subject: [PATCH] fix(server): harden shell catch-up replay against silent event drops --- apps/server/src/server.test.ts | 459 ++++++++++++++++++ apps/server/src/ws.ts | 167 +++++-- .../src/state/shell-sync.test.ts | 226 +++++++++ packages/client-runtime/src/state/shell.ts | 11 + plans/2026-07-19-shell-catchup-silent-drop.md | 360 ++++++++++++++ 5 files changed, 1171 insertions(+), 52 deletions(-) create mode 100644 plans/2026-07-19-shell-catchup-silent-drop.md diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index e109de2ecc33..5b7291ba5b5f 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -271,6 +271,26 @@ const makeShellSnapshotWithThreads = (threads: ReadonlyArray => ({ + sequence, + eventId: EventId.make(`event-${sequence}`), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: "2026-07-19T00:08:00.000Z", + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.unarchived", + payload: { threadId, updatedAt: "2026-07-19T00:08:00.000Z" }, +}); + const browserOtlpTracingLayer = Layer.mergeAll( FetchHttpClient.layer, OtlpSerialization.layerJson, @@ -5794,6 +5814,445 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); + // loom: shell catch-up silent-drop fix (see plans/2026-07-19-shell-catchup-silent-drop.md). + // These two tests exercise the fix's bounded retry, whose exponential backoff + // uses real Effect.sleep server-side. That never advances under it.effect's + // TestClock (and it.live is unavailable inside this it.layer block), so the + // whole body runs under TestClock.withLive — the live clock lets the ~75ms + // backoff elapse. The socket/RPC machinery already works under real time. + it.effect( + "fails the shell subscription instead of silently dropping an event whose projection lookup fails", + () => + Effect.gen(function* () { + // The canonical dropped thread from the incident: created via the Slack + // bridge while a client was offline, then dropped by a transient lookup + // failure during that client's catch-up. + const gapThreadId = ThreadId.make("2238e38b-6d72-491d-b85b-caf239a366c2"); + const okThreadId = ThreadId.make("thread-ok"); + const lookupError = new PersistenceSqlError({ + operation: "ProjectionSnapshotQuery.getThreadShellById:test", + detail: "transient contention on the busy cockpit DB", + }); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => + Stream.make( + makeThreadUnarchivedEvent(1, gapThreadId), + makeThreadUnarchivedEvent(2, okThreadId), + ), + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 2 }), + getThreadShellById: (threadId) => + threadId === gapThreadId + ? Effect.fail(lookupError) + : Effect.succeed( + Option.some(makeDefaultOrchestrationThreadShell({ id: threadId })), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.result), + ); + + // Today's code SUCCEEDS yielding only the second event — the silent + // omission whose sequence advance seals the gap. The fix must fail loudly + // (before yielding either) so the client's round-1 self-heal fires. + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); + }).pipe(TestClock.withLive, Effect.provide(NodeHttpServer.layerTest)), + ); + + // Runs under TestClock.withLive so the retry's real backoff elapses (see note above). + it.effect("absorbs a transient projection-lookup failure via retry during shell catch-up", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-transient"); + let attempts = 0; + const lookupError = new PersistenceSqlError({ + operation: "ProjectionSnapshotQuery.getThreadShellById:test", + detail: "transient blip", + }); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => Stream.make(makeThreadUnarchivedEvent(1, threadId)), + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + // Fails on the first two attempts, succeeds on the third — within the + // 3-attempt retry budget, so the client sees no failure. + getThreadShellById: (id) => + Effect.suspend(() => { + attempts += 1; + return attempts <= 2 + ? Effect.fail(lookupError) + : Effect.succeed(Option.some(makeDefaultOrchestrationThreadShell({ id }))); + }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + + assert.equal(items.length, 1); + assert.equal(items[0]?.kind, "thread-upserted"); + assert.equal(attempts, 3); + }).pipe(TestClock.withLive, Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("keeps a genuinely-absent projection row silent during shell catch-up", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-absent"); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => Stream.make(makeThreadUnarchivedEvent(1, threadId)), + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 1 }), + // A *successful* Option.none is the legitimate row-absent signal and + // must stay silent (the other half of the lookup-failed/row-absent + // taxonomy) — guards against over-correcting the fix into failing here. + getThreadShellById: () => Effect.succeed(Option.none()), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + + assert.deepEqual(items, []); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect( + "serves a fresh snapshot instead of replaying when the shell catch-up gap exceeds the cap", + () => + Effect.gen(function* () { + let readEventsCalled = false; + const threadId = ThreadId.make("thread-snap"); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => { + readEventsCalled = true; + return Stream.empty; + }, + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 5000 }), + getShellSnapshot: () => + Effect.succeed( + makeShellSnapshotWithThreads([ + makeDefaultOrchestrationThreadShell({ id: threadId }), + ]), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + // gap = 5000 - 10 = 4990 > SHELL_CATCHUP_MAX_EVENTS (500). + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 10 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + + assert.equal(items.length, 1); + assert.equal(items[0]?.kind, "snapshot"); + assertTrue(!readEventsCalled); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("replays exactly the sampled gap window (limit == gap) when within the cap", () => + Effect.gen(function* () { + const readLimits: Array = []; + const threadId = ThreadId.make("thread-replay"); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: (from, limit) => { + readLimits.push(limit); + return Stream.make(makeThreadUnarchivedEvent(from + 1, threadId)); + }, + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 3 }), + getThreadShellById: (id) => + Effect.succeed(Option.some(makeDefaultOrchestrationThreadShell({ id }))), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + + // The cap must be enforced at the read, not just at the branch: the limit + // passed to readEvents equals the sampled gap (3 - 0), so a permitted + // replay covers precisely (afterSequence, snapshotSequence]. + assert.deepEqual(readLimits, [3]); + assert.equal(items.length, 1); + assert.equal(items[0]?.kind, "thread-upserted"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect("serves a snapshot when the client's afterSequence is ahead of the server", () => + Effect.gen(function* () { + let readEventsCalled = false; + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + readEvents: () => { + readEventsCalled = true; + return Stream.empty; + }, + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 5 }), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + // afterSequence (42) > snapshotSequence (5): restored backup / reset. + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 42 }).pipe( + Stream.runCollect, + ), + ).pipe(Effect.map((chunk) => Array.from(chunk))), + ); + + assert.equal(items.length, 1); + assert.equal(items[0]?.kind, "snapshot"); + assertTrue(!readEventsCalled); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + // loom: connect-gap regression (design §5 test 5). Pins the eager + // subscribeDomainEvents ordering: the live PubSub subscription must attach + // BEFORE the snapshot read so an event committed during the snapshot load + // window buffers in the subscription queue and drains AFTER the snapshot item, + // instead of falling into a silent connect-gap. A Deferred gates the snapshot + // stub so the ordering is deterministic (no timing races): the coordinator + // fiber waits until the snapshot load is in flight — which, in the fixed code, + // proves the eager subscribe already ran — then publishes the live event and + // releases the gate. Reverting ws.ts to a lazy Stream.fromPubSub attach makes + // this fail (the event lands on neither leg). + it.effect( + "delivers an event published during snapshot load after the snapshot (no-afterSequence flow)", + () => + Effect.gen(function* () { + const eventPubSub = yield* PubSub.unbounded(); + const snapshotInFlight = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const threadId = ThreadId.make("thread-connect-gap"); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), (subscription) => + Stream.fromSubscription(subscription), + ), + }, + projectionSnapshotQuery: { + // Signal that the snapshot load has started (eager subscribe is + // already done in the fixed code), then block until released. + getShellSnapshot: () => + Deferred.succeed(snapshotInFlight, undefined).pipe( + Effect.andThen(Deferred.await(releaseSnapshot)), + Effect.as(makeShellSnapshotWithThreads([])), + ), + getThreadShellById: (id) => + Effect.succeed(Option.some(makeDefaultOrchestrationThreadShell({ id }))), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + yield* Deferred.await(snapshotInFlight).pipe( + Effect.andThen( + PubSub.publish(eventPubSub, makeThreadUnarchivedEvent(100, threadId)), + ), + Effect.andThen(Deferred.succeed(releaseSnapshot, undefined)), + Effect.forkChild, + ); + return yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({}).pipe( + Stream.take(2), + Stream.runCollect, + Effect.map((chunk) => Array.from(chunk)), + ); + }), + ), + ); + + assert.equal(items.length, 2); + assert.equal(items[0]?.kind, "snapshot"); + const delivered = items[1]; + assert.equal(delivered?.kind, "thread-upserted"); + if (delivered?.kind === "thread-upserted") { + assert.equal(delivered.thread.id, threadId); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + // loom: same connect-gap invariant on the capped fallback (gap > cap) — the + // snapshot-serving afterSequence branch must inherit the eager-attach ordering + // too, not just the ordinary flow. + it.effect( + "delivers an event published during snapshot load after the snapshot (capped fallback)", + () => + Effect.gen(function* () { + const eventPubSub = yield* PubSub.unbounded(); + const snapshotInFlight = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const threadId = ThreadId.make("thread-connect-gap-capped"); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), (subscription) => + Stream.fromSubscription(subscription), + ), + // Should never be read: gap (5000 - 10) exceeds the cap, so the + // fallback serves a snapshot rather than replaying. + readEvents: () => Stream.empty, + }, + projectionSnapshotQuery: { + getSnapshotSequence: () => Effect.succeed({ snapshotSequence: 5000 }), + getShellSnapshot: () => + Deferred.succeed(snapshotInFlight, undefined).pipe( + Effect.andThen(Deferred.await(releaseSnapshot)), + Effect.as(makeShellSnapshotWithThreads([])), + ), + getThreadShellById: (id) => + Effect.succeed(Option.some(makeDefaultOrchestrationThreadShell({ id }))), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + yield* Deferred.await(snapshotInFlight).pipe( + Effect.andThen( + PubSub.publish(eventPubSub, makeThreadUnarchivedEvent(6000, threadId)), + ), + Effect.andThen(Deferred.succeed(releaseSnapshot, undefined)), + Effect.forkChild, + ); + return yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({ + afterSequence: 10, + }).pipe( + Stream.take(2), + Stream.runCollect, + Effect.map((chunk) => Array.from(chunk)), + ); + }), + ), + ); + + assert.equal(items.length, 2); + assert.equal(items[0]?.kind, "snapshot"); + const delivered = items[1]; + assert.equal(delivered?.kind, "thread-upserted"); + if (delivered?.kind === "thread-upserted") { + assert.equal(delivered.thread.id, threadId); + } + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + // loom: error-preserving buffering shape (design §5 test 5 variant). A buffered + // live event whose projection lookup fails persistently must FAIL the + // subscription with OrchestrationGetSnapshotError — not hang and not silently + // vanish. This is the guard against a fork-into-value-queue shape, which would + // amputate the mapper's error channel (killing a detached producer fibre while + // the value-only queue just stops). Runs under TestClock.withLive because the + // failing lookup exhausts the real-backoff retry before propagating. + it.effect("fails the subscription when a buffered live event's lookup fails persistently", () => + Effect.gen(function* () { + const eventPubSub = yield* PubSub.unbounded(); + const snapshotInFlight = yield* Deferred.make(); + const releaseSnapshot = yield* Deferred.make(); + const threadId = ThreadId.make("thread-connect-gap-poison"); + const lookupError = new PersistenceSqlError({ + operation: "ProjectionSnapshotQuery.getThreadShellById:test", + detail: "persistent contention on the buffered live event", + }); + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), (subscription) => + Stream.fromSubscription(subscription), + ), + }, + projectionSnapshotQuery: { + getShellSnapshot: () => + Deferred.succeed(snapshotInFlight, undefined).pipe( + Effect.andThen(Deferred.await(releaseSnapshot)), + Effect.as(makeShellSnapshotWithThreads([])), + ), + getThreadShellById: () => Effect.fail(lookupError), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const result = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + Effect.gen(function* () { + yield* Deferred.await(snapshotInFlight).pipe( + Effect.andThen(PubSub.publish(eventPubSub, makeThreadUnarchivedEvent(100, threadId))), + Effect.andThen(Deferred.succeed(releaseSnapshot, undefined)), + Effect.forkChild, + ); + return yield* client[ORCHESTRATION_WS_METHODS.subscribeShell]({}).pipe( + Stream.take(2), + Stream.runCollect, + ); + }).pipe(Effect.result), + ), + ); + + // The buffered live event's error channel survives the buffering window: + // the subscription fails loudly rather than hanging or omitting the event. + assertTrue(result._tag === "Failure"); + assertTrue(result.failure._tag === "OrchestrationGetSnapshotError"); + }).pipe(TestClock.withLive, Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("enriches replayed project events with repository identity metadata", () => Effect.gen(function* () { const repositoryIdentity = { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 44607f0a4711..525bdcfbef20 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -8,6 +8,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; import * as Ref from "effect/Ref"; +import * as Schedule from "effect/Schedule"; import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import { @@ -32,7 +33,6 @@ import { OrchestrationDispatchCommandError, type OrchestrationEvent, type OrchestrationShellStreamEvent, - type OrchestrationShellStreamItem, type OrchestrationThreadStreamItem, OrchestrationGetFullThreadDiffError, OrchestrationGetSnapshotError, @@ -78,6 +78,7 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import { normalizeDispatchCommand } from "./orchestration/Normalizer.ts"; import * as OrchestrationEngine from "./orchestration/Services/OrchestrationEngine.ts"; import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { ProjectionRepositoryError } from "./persistence/Errors.ts"; import * as UsageBreakdownQuery from "./orchestration/Services/UsageBreakdownQuery.ts"; import * as ReasoningStreamBus from "./orchestration/Services/ReasoningStreamBus.ts"; import { LOOM_RPC_SCOPES, makeLoomWsHandlers } from "./loom/wsMethods.ts"; // loom: @@ -135,6 +136,16 @@ import * as RelayClient from "@t3tools/shared/relayClient"; import { subtreeOf } from "@t3tools/shared/workstreamGraph"; const isOrchestrationDispatchCommandError = Schema.is(OrchestrationDispatchCommandError); +// loom: cap catch-up replay on the shell subscription's afterSequence path. +// Beyond this many events a fresh snapshot is strictly cheaper than an event +// tail (≥5 projection queries per event + a full thread-shell payload repeated +// per touched thread, vs each aggregate sent once) and closes the silent-drop +// window that lives in the per-event lookup. 500 == one event-store read page +// (READ_PAGE_SIZE), so a permitted replay is always a single page; it also +// covers the common resume cases the resume path exists for (tab refocus, brief +// blips, short sleep). A large overnight gap — the incident habitat — snapshots. +const SHELL_CATCHUP_MAX_EVENTS = 500; + const nowIso = Effect.map(DateTime.now, DateTime.formatIso); function unexpectedCompatibilityError(error: never): never { @@ -600,9 +611,27 @@ const makeWsRpcLayer = ( const enrichOrchestrationEvents = (events: ReadonlyArray) => Effect.forEach(events, enrichProjectEvent, { concurrency: 4 }); + // loom: silent-drop fix. Diverges deliberately from upstream #2968 + // ("Refactor recoverable Effect fallbacks to orElseSucceed"), which + // swallowed these projection-lookup failures as Option.none() — a *failed* + // lookup then became indistinguishable from a genuinely-absent row and was + // silently dropped from catch-up replay, permanently wedging the client's + // cache past the gap. A future upstream sync must NOT re-collapse the two: + // the error channel (ProjectionRepositoryError) means "lookup failed, state + // unknown" and must stay loud so the client self-heals via a fresh + // snapshot; a *successful* Option.none means "row genuinely absent" and + // stays silent (the goal branch depends on it to emit goal-removed). The + // bounded retry absorbs a transient SQLite contention blip (~75ms across 3 + // attempts) without tearing down every connected subscription. + const shellLookupRetry = Schedule.exponential("25 millis").pipe(Schedule.take(2)); + const toShellStreamEvent = ( event: OrchestrationEvent, - ): Effect.Effect, never, never> => { + ): Effect.Effect< + Option.Option, + ProjectionRepositoryError, + never + > => { switch (event.type) { case "project.created": case "project.meta-updated": @@ -614,7 +643,7 @@ const makeWsRpcLayer = ( project: nextProject, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.retry(shellLookupRetry), // loom: fail loud, don't swallow ); case "project.deleted": return Effect.succeed( @@ -642,7 +671,7 @@ const makeWsRpcLayer = ( thread: nextThread, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.retry(shellLookupRetry), // loom: fail loud, don't swallow ); default: // loom: goal aggregate → goal-upserted/goal-removed shell-stream events. @@ -665,7 +694,10 @@ const makeWsRpcLayer = ( }), }), ), - Effect.orElseSucceed(() => Option.none()), + // loom: fail loud, don't swallow. A *successful* Option.none here + // is load-bearing (emits goal-removed); folding a lookup failure + // into it would fabricate a goal-removed for a live goal. + Effect.retry(shellLookupRetry), ); } if (event.aggregateKind !== "thread") { @@ -681,7 +713,7 @@ const makeWsRpcLayer = ( thread: nextThread, })), ), - Effect.orElseSucceed(() => Option.none()), + Effect.retry(shellLookupRetry), // loom: fail loud, don't swallow ); } }; @@ -1019,51 +1051,34 @@ const makeWsRpcLayer = ( observeRpcStreamEffect( ORCHESTRATION_WS_METHODS.subscribeShell, Effect.gen(function* () { - const liveStream = orchestrationEngine.streamDomainEvents.pipe( + // loom: eager PubSub attach BEFORE any cursor/snapshot read + // (subscribeDomainEvents, a fork-added engine facility) so events + // published during that window buffer in the subscription queue + // instead of falling into a connect-gap — the ordinary + // no-afterSequence flow previously used a lazy Stream.fromPubSub + // that only attached after the snapshot element, leaving a silent + // gap for events committed between the snapshot query and the first + // pull. toShellStreamEvent (now fallible — see the silent-drop fix) + // is mapped on THIS consuming stream, never forked into a value-only + // queue: a mapper failure must land in the stream's own error + // channel (→ OrchestrationGetSnapshotError → client self-heal), not + // kill a detached producer fibre while a bare queue silently stops. + const rawLive = yield* orchestrationEngine.subscribeDomainEvents; + const liveLeg = rawLive.pipe( Stream.mapEffect(toShellStreamEvent), Stream.flatMap((event) => Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, ), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to project live orchestration shell event", + cause, + }), + ), ); - // When the client already holds a shell snapshot (cached, or loaded - // over HTTP) it passes that snapshot's sequence, and we resume by - // replaying shell events after it instead of re-sending the whole - // projects/threads list over the socket. As in the thread path, the - // live subscription is attached (into a scope-bound buffer) before - // draining the catch-up replay so no event published during the - // replay window is lost; overlapping events are deduped by sequence - // on the client. The full range is read (not the store's default - // page limit) since the shell filter runs after reading. - if (input.afterSequence !== undefined) { - const afterSequence = input.afterSequence; - return Stream.unwrap( - Effect.gen(function* () { - const liveBuffer = yield* Queue.unbounded(); - yield* Effect.forkScoped( - liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))), - ); - const catchUpStream = orchestrationEngine - .readEvents(afterSequence, Number.MAX_SAFE_INTEGER) - .pipe( - Stream.mapEffect(toShellStreamEvent), - Stream.flatMap((event) => - Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, - ), - Stream.mapError( - (cause) => - new OrchestrationGetSnapshotError({ - message: "Failed to replay orchestration shell events", - cause, - }), - ), - ); - return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)); - }), - ); - } - - const snapshot = yield* projectionSnapshotQuery.getShellSnapshot().pipe( + const loadSnapshotItem = projectionSnapshotQuery.getShellSnapshot().pipe( Effect.tapError((cause) => Effect.logError("orchestration shell snapshot load failed", { cause }), ), @@ -1074,15 +1089,63 @@ const makeWsRpcLayer = ( cause, }), ), + Effect.map((snapshot) => ({ kind: "snapshot" as const, snapshot })), ); - return Stream.concat( - Stream.make({ - kind: "snapshot" as const, - snapshot, - }), - liveStream, - ); + // When the client already holds a shell snapshot (cached, or loaded + // over HTTP) it passes that snapshot's sequence, and we resume by + // replaying shell events after it instead of re-sending the whole + // projects/threads list over the socket. Overlapping events are + // deduped by sequence on the client. + if (input.afterSequence !== undefined) { + const afterSequence = input.afterSequence; + // loom: sample the cursor AFTER the eager live attach. An event + // publishes to the PubSub only after its projection update commits + // (same txn), so any event missing from the already-attached + // subscription is ≤ this cursor and therefore inside the read + // interval below — the catch-up/live seam is gap-free. + const { snapshotSequence } = yield* projectionSnapshotQuery + .getSnapshotSequence() + .pipe( + Effect.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to read orchestration projection cursor", + cause, + }), + ), + ); + const gap = snapshotSequence - afterSequence; + // loom: cap the replay, and handle client-ahead-of-server + // (afterSequence > snapshotSequence: restored DB backup / + // projection reset, which previously left the client confidently + // stale with phantom threads). Beyond the cap or when ahead, a + // snapshot is cheaper and correct; the client applies a mid-stream + // snapshot as a wholesale replace via its existing reducer path. + if (afterSequence > snapshotSequence || gap > SHELL_CATCHUP_MAX_EVENTS) { + return Stream.concat(Stream.make(yield* loadSnapshotItem), liveLeg); + } + // loom: read EXACTLY the sampled interval (limit = gap), not + // Number.MAX_SAFE_INTEGER — otherwise events committed after the + // sample would extend the read past the cap, making it advisory. + // Everything later arrives via the already-attached live leg. + const catchUpStream = orchestrationEngine.readEvents(afterSequence, gap).pipe( + Stream.mapEffect(toShellStreamEvent), + Stream.flatMap((event) => + Option.isSome(event) ? Stream.succeed(event.value) : Stream.empty, + ), + Stream.mapError( + (cause) => + new OrchestrationGetSnapshotError({ + message: "Failed to replay orchestration shell events", + cause, + }), + ), + ); + return Stream.concat(catchUpStream, liveLeg); + } + + return Stream.concat(Stream.make(yield* loadSnapshotItem), liveLeg); }), { "rpc.aggregate": "orchestration" }, ), diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 05edc774913f..e8498b8356d8 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -491,4 +491,230 @@ describe("environment shell synchronization", () => { expect((yield* SubscriptionRef.get(savedSnapshots)).at(-1)).toEqual(freshSnapshot); }), ); + + // loom: shell catch-up silent-drop fix, client half of the gap-cap path. When + // the offline gap exceeds the server's cap (or the client is ahead of the + // server) the afterSequence resume answers with a fresh `snapshot` item + // instead of an event tail; the warm leg must apply it as a wholesale replace + // (and re-persist it) via the existing applyItems path — no cold-path fallback. + it.effect( + "wholesale-replaces state and cache when the resume responds with a mid-stream snapshot", + () => + Effect.gen(function* () { + const cachedSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 5, + goals: [], + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + const freshSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 42, + goals: [STUB_GOAL], + projects: [], + threads: [STUB_THREAD], + updatedAt: "2026-06-07T00:00:00.000Z", + }; + const events = yield* Queue.unbounded(); + const subscribeSequences = yield* SubscriptionRef.make>( + [], + ); + const loaderCalls = yield* SubscriptionRef.make(0); + const savedSnapshots = yield* SubscriptionRef.make< + ReadonlyArray + >([]); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => + Stream.unwrap( + SubscriptionRef.update(subscribeSequences, (calls) => [ + ...calls, + input.afterSequence, + ]).pipe(Effect.as(Stream.fromQueue(events))), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(Option.some(cachedSnapshot)), + saveShell: (_environmentId, snapshot) => + SubscriptionRef.update(savedSnapshots, (saved) => [...saved, snapshot]), + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe( + Effect.as(Option.none()), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + // Let the warm resume subscribe from the cached sequence (5). + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + // The server's gap-cap path answers the afterSequence resume with a + // fresh snapshot item instead of an event tail. + yield* Queue.offer(events, { kind: "snapshot", snapshot: freshSnapshot }); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + yield* TestClock.adjust("20 millis"); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + yield* TestClock.adjust("500 millis"); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + + // No cold HTTP fetch and no resubscribe: the warm leg replaced in place. + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(0); + expect(yield* SubscriptionRef.get(subscribeSequences)).toEqual([5]); + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("live"); + expect(Option.getOrThrow(state.snapshot)).toEqual(freshSnapshot); + expect((yield* SubscriptionRef.get(savedSnapshots)).at(-1)).toEqual(freshSnapshot); + }), + ); + + // loom: shell catch-up silent-drop fix, client half of the cold-leg + // resilience completion. Server lookup failures are now loud, so the cold leg + // carries retryExpectedFailureAfter: after an expected failure on an + // established connection it resubscribes from the SAME afterSequence (5s + // later) rather than parking on the error banner until the next session + // change; the replay re-covers the interval and recovers to live. + it.effect( + "retries the cold-leg subscription after an expected failure and recovers to live", + () => + Effect.gen(function* () { + const coldSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 7, + goals: [], + projects: [], + threads: [], + updatedAt: "2026-06-06T00:00:00.000Z", + }; + // The payload the retry receives once it resubscribes. + const recoveredSnapshot: OrchestrationShellSnapshot = { + snapshotSequence: 8, + goals: [STUB_GOAL], + projects: [], + threads: [STUB_THREAD], + updatedAt: "2026-06-07T00:00:00.000Z", + }; + const recoveredEvents = yield* Queue.unbounded(); + const subscribeSequences = yield* SubscriptionRef.make>( + [], + ); + const loaderCalls = yield* SubscriptionRef.make(0); + const client = { + [ORCHESTRATION_WS_METHODS.subscribeShell]: (input: { readonly afterSequence?: number }) => + Stream.unwrap( + SubscriptionRef.get(subscribeSequences).pipe( + Effect.tap(() => + SubscriptionRef.update(subscribeSequences, (calls) => [ + ...calls, + input.afterSequence, + ]), + ), + Effect.map((priorCalls) => + // First subscribe fails with an expected (non-transport) failure, + // standing in for a now-loud transient server lookup failure; the + // retry resubscribes from the same afterSequence and recovers. + priorCalls.length === 0 + ? Stream.fail(new Error("transient shell lookup failure")) + : Stream.fromQueue(recoveredEvents), + ), + ), + ), + } as unknown as WsRpcProtocolClient; + const supervisorState = yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE); + const activeSession = yield* SubscriptionRef.make>( + Option.some(session(client)), + ); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: supervisorState, + session: activeSession, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + // Cold start: no warm cache, so the base loads over HTTP. + loadShell: () => Effect.succeed(Option.none()), + saveShell: () => Effect.void, + loadThread: () => Effect.succeed(Option.none()), + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeed(Option.none()), + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeed(Option.none()), + saveVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.of({ + load: () => + SubscriptionRef.update(loaderCalls, (count) => count + 1).pipe( + Effect.as(Option.some(coldSnapshot)), + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader, snapshotLoader), + ); + + // Cold base loads and the first cold-leg subscribe fails (expected failure). + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + yield* TestClock.adjust("20 millis"); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + // Buffer the recovery payload the retry will receive, then advance past + // the 5s retry gap so the cold leg resubscribes. + yield* Queue.offer(recoveredEvents, { kind: "snapshot", snapshot: recoveredSnapshot }); + yield* TestClock.adjust("5 seconds"); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + yield* TestClock.adjust("20 millis"); + for (let index = 0; index < 20; index += 1) { + yield* Effect.yieldNow; + } + + // Only one cold HTTP load; the retry resubscribed from the SAME + // afterSequence (7) and recovered to live with the fresh payload. + expect(yield* SubscriptionRef.get(loaderCalls)).toBe(1); + expect(yield* SubscriptionRef.get(subscribeSequences)).toEqual([7, 7]); + const state = yield* SubscriptionRef.get(shellState); + expect(state.status).toBe("live"); + expect(Option.getOrThrow(state.snapshot)).toEqual(recoveredSnapshot); + }), + ); }); diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index 8a08acb57a38..bdf7c65bc08b 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -187,6 +187,17 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") }); yield* subscribe(ORCHESTRATION_WS_METHODS.subscribeShell, subscribeInput, { onExpectedFailure: (cause) => setStreamError(Cause.squash(cause)), + // loom: cold-leg resilience completion for the silent-drop fix. Server + // lookup failures are now loud (they fail the stream instead of silently + // dropping an event), so a transient failure on an established + // connection must not park the cold leg on the error banner until the + // next session change. Resubscribe after 5s reusing the same + // afterSequence; the replay re-covers the interval and applyItems dedupes + // by sequence, so the retry is idempotent. onExpectedFailure still fires, + // so the sync warning shows during the retry window. (The WARM leg keeps + // no-retry: its failure self-heals to this cold path with a fresh + // snapshot, and retrying its identical replay was round 1's wedge.) + retryExpectedFailureAfter: "5 seconds", }).pipe(Stream.groupedWithin(64, "20 millis"), Stream.runForEach(applyItems)); }); diff --git a/plans/2026-07-19-shell-catchup-silent-drop.md b/plans/2026-07-19-shell-catchup-silent-drop.md new file mode 100644 index 000000000000..d124e8fa2a95 --- /dev/null +++ b/plans/2026-07-19-shell-catchup-silent-drop.md @@ -0,0 +1,360 @@ +--- +manager_sessions: + - id: b99631c2-2432-4ce0-b23e-d71132f2020b + role: plan + authored_at: 2026-07-19T12:00:57.005Z +--- + +# Eliminating silent shell-event drops in catch-up replay + +**Status:** design — revised after review rounds 1–2 (connect-gap-safe snapshot fallback, enforced read limit, deterministic core test, error-preserving live buffering) +**Date:** 2026-07-19 +**Scope:** `apps/server/src/ws.ts` (subscribeShell), `packages/client-runtime/src/state/shell.ts`, tests. No wire-contract changes. + +## 1. Problem + +A client that reconnects with a warm shell cache resumes via +`subscribeShell({ afterSequence })`. The server replays every persisted event +after that sequence through `toShellStreamEvent` (`apps/server/src/ws.ts:601`), +which performs a per-event projection lookup (`getThreadShellById`, +`getGoalShellById`, `getProjectShellById`). **Every lookup branch swallows +failures**: + +```ts +Effect.orElseSucceed(() => Option.none()) +``` + +`Option.none()` is indistinguishable from the legitimate "row absent" signal, so +a transient DB failure (the ~830 MB cockpit DB logs `orchestration command +slow` under contention) silently drops that event from the replay. The stream +then *succeeds*, so the round-1 client self-heal (PR #88 — discard cache on +replay **failure**) never fires. + +The drop is permanent by construction: the client reducer +(`packages/client-runtime/src/state/shellReducer.ts:16`) skips any event with +`sequence <= snapshotSequence`, and the very next replayed event advances +`snapshotSequence` past the dropped one. The advanced sequence is persisted to +the cache, so every future reconnect resumes from *beyond* the gap. A thread +whose only shell events fell in the gap (e.g. created from another device or +the Slack bridge while this client was offline) stays invisible until the cache +is cleared or the thread is touched again. + +**Confirmed instance:** thread `2238e38b-6d72-491d-b85b-caf239a366c2` +("Claude Code work-splitting prompt research"), created 2026-07-19T00:08Z via +the Slack bridge, healthy in `projection_threads`, absent from a reconnecting +Windows client's sidebar. Round-1 fix was live. Nothing errored. + +Two aggravating factors, both verified in code: + +1. **Unbounded replay window.** The catch-up reads + `readEvents(afterSequence, Number.MAX_SAFE_INTEGER)` + (`apps/server/src/ws.ts:1045`). An offline gap of 15k+ events means 15k+ + sequential per-event lookups — `getThreadShellById` alone runs **five** + queries per event (`ProjectionSnapshotQuery.ts:2872`). This maximises both + the drop probability under contention and the replay cost (each + `thread-upserted` carries a *full* `OrchestrationThreadShell`, so N events + touching the same thread cost N × the shell size — usually more bytes than a + snapshot that sends each thread once). +2. **The same swallow exists on the live leg.** `liveStream` pipes + `streamDomainEvents` through the same `toShellStreamEvent` + (`ws.ts:1020`), so a contention blip during normal operation can also drop a + live event with identical permanence. + +## 2. Root-cause taxonomy: lookup-failed vs row-absent + +`Option.none()` carries two meanings today; the fix must separate them. + +| Signal | Meaning | Correct handling | +|---|---|---| +| Lookup **succeeds**, returns `Option.none` | Projection row genuinely absent (thread deleted/archived since the event; goal removed) | Keep current behaviour: skip the event (thread branches) or emit `goal-removed` (goal branch). Legitimately silent. | +| Lookup **fails** (`ProjectionRepositoryError`: SQL error, decode error, contention) | We don't know the row's state | Must be **loud**: fail the stream so the client's round-1 self-heal path runs. Never skip. | + +The distinction is already present in the Effect type — the projection queries +fail with `ProjectionRepositoryError` and succeed with +`Option`. `orElseSucceed` is the only thing collapsing them. Removing it +lets the type system enforce the taxonomy: the error channel *is* lookup-failed; +`Option.none` in the success channel *is* row-absent. + +## 3. Decision + +Three coordinated, individually small changes. No contract changes — the +existing `OrchestrationShellStreamItem` union (which already includes the +`snapshot` kind) and the existing `OrchestrationGetSnapshotError` cover +everything. + +### 3.1 Server: make lookup failures fail the stream (the correctness fix) + +In `toShellStreamEvent`, replace each +`Effect.orElseSucceed(() => Option.none())` with a small bounded retry, then +let the error propagate: + +```ts +// loom: silent-drop fix — a projection lookup FAILURE must not masquerade as +// row-absent. Retry absorbs transient DB contention; a persistent failure +// fails the stream so the client self-heals via a fresh snapshot. +Effect.retry(Schedule.intersect(Schedule.exponential("25 millis"), Schedule.recurs(2))) +``` + +- **Catch-up leg** (`ws.ts` afterSequence path): the existing + `Stream.mapError → OrchestrationGetSnapshotError` already turns the failure + into the wire error. The client's warm-cache resume sees a failed stream → + round-1 self-heal → cache discarded → fresh HTTP snapshot. The silent drop + becomes a loud, self-healing resync. +- **Live leg**: the live stream gains the same error type; map it to + `OrchestrationGetSnapshotError` on the consuming stream in every flow. The + error channel must survive end-to-end — §3.2 specifies the buffering shape + that guarantees this (no fork-into-value-queue). A live-phase persistent + failure now + terminates the subscription with an error instead of dropping the event; the + client recovers per §3.3. +- Retry justification: the observed failure mode is transient contention on a + busy SQLite DB. Three attempts spanning ~75 ms absorb a lock blip without + tearing down every connected client's subscription; anything that survives + three attempts is not a blip and *should* surface. (Single-user cockpit + server — no thundering-herd concern.) + +The goal branch deserves a comment in code: for goals, a *successful* none is +load-bearing (it emits `goal-removed`), which is precisely why a failure must +not be folded into it — folding could otherwise be "upgraded" someday to +fabricate a `goal-removed` for a live goal. + +### 3.2 Server: cap the catch-up window (the bounding fix) + +In the `subscribeShell` afterSequence path, **first** acquire the live leg +eagerly, **then** sample the projection cursor, then branch: + +``` +rawLive ← subscribeDomainEvents (eager PubSub attach; the subscription queue IS the buffer) +snapshotSequence ← projectionSnapshotQuery.getSnapshotSequence() +gap = snapshotSequence - afterSequence +liveLeg = rawLive → toShellStreamEvent → filter Option → mapError(OrchestrationGetSnapshotError) +if (afterSequence > snapshotSequence || gap > SHELL_CATCHUP_MAX_EVENTS) + → Stream.concat(snapshot item from getShellSnapshot(), liveLeg) +else + → Stream.concat(replay readEvents(afterSequence, gap) → toShellStreamEvent …, liveLeg) +``` + +**No intermediate queue.** The stream returned by `subscribeDomainEvents` is +retained *raw* (unmapped) while the cursor/snapshot work runs — the eagerly +attached PubSub subscription buffers events during that window all by itself. +`toShellStreamEvent` is applied lazily, on pull, after the concat's first leg +drains. This matters for §3.1: the mapper is now fallible, and a +fork-into-value-queue shape (today's `Effect.forkScoped(liveStream.pipe( +Stream.runForEach(offer)))`, `ws.ts:1041`) would let a mapper failure kill the +detached producer fibre while the value-only queue just stops — an error +channel amputation that recreates the silent live-event drop §3.1 exists to +kill. Mapping on the consuming stream keeps the failure in the stream's own +error channel, where it terminates the RPC subscription loudly. (If an +implementation ever does need an explicit buffer here, it must carry exits, +not bare items — but the raw-stream shape makes that unnecessary.) + +Two ordering rules make this race-free, and both were review findings against +the first draft: + +- **Eager live subscription before any cursor/snapshot read.** The ordinary + no-afterSequence flow builds `Stream.concat(Stream.make(snapshot), + liveStream)` where `liveStream` is a lazy `Stream.fromPubSub` — it only + attaches when first pulled, *after* the snapshot element, so an event + committed between the snapshot query and that first pull is on neither side + and stays invisible for the connection's lifetime. That is a second silent + connect-gap, and the capped fallback must not inherit it. The engine already + exposes `subscribeDomainEvents` + (`apps/server/src/orchestration/Services/OrchestrationEngine.ts:62`) — + an effect that attaches the PubSub subscription the moment it runs — + precisely to close this gap (added by `38289a138`, mirroring the + thread-detail reasoning-bus pre-subscribe). Both the capped fallback **and + the existing ordinary snapshot flow** switch to: acquire the raw + `subscribeDomainEvents` stream first, then read, then concatenate + snapshot/replay + the mapped live leg (shape above). Overlap is deduped by + sequence on the client, as today. The existing afterSequence path's + `Effect.forkScoped(liveStream.pipe(Stream.runForEach(offer)))` has the same + lazy-attach hazard in miniature (the forked fibre subscribes asynchronously) + *and* the error-amputation hazard once the mapper is fallible; it moves to + the same raw-stream shape. +- **The replay reads exactly the sampled interval.** The first draft kept + `readEvents(afterSequence, Number.MAX_SAFE_INTEGER)`, which makes the cap + advisory: events committed between the cursor sample and the read extend the + query beyond 500. Pass `gap` as the read limit — the replay covers precisely + `(afterSequence, snapshotSequence]` and everything later arrives via the + already-attached live subscription. Sampling the cursor *after* the live + attach makes the seam gap-free: an event publishes to the PubSub only after + its projection update commits (same transaction, `OrchestrationEngine.ts` + `processEnvelope`), so any event missing from the subscription (committed + before attach) is ≤ the cursor sampled after attach and therefore inside + the read interval. + +- **`SHELL_CATCHUP_MAX_EVENTS = 500`.** Justification: (a) one event-store read + page (`READ_PAGE_SIZE = 500` in `OrchestrationEventStore.ts:72`), so a + permitted replay — now genuinely bounded by `limit = gap` — is always a + single page; (b) it comfortably covers the + common resume cases this optimisation exists for — tab refocus, brief network + blips, laptop sleep of minutes (a busy turn emits a few events per second); + (c) beyond it, replay is strictly worse than a snapshot on both axes: ≥5 + queries/event vs a fixed set of aggregate queries, and repeated full + thread-shell payloads vs each thread once. A 15k-event overnight gap — the + incident habitat — goes straight to snapshot. +- **`afterSequence > snapshotSequence`** (client ahead of server — restored DB + backup, or projection reset) currently replays nothing and leaves the client + confidently stale with phantom threads. Falling into the snapshot path fixes + this adjacent wedge for free. +- **Restart spanning:** not detected separately. A restart long enough to + matter shows up as a large gap; a quick restart with a small gap replays + correctly. Keeping one numeric criterion avoids a second detection mechanism. +- The client needs **no change** for this: `applyItems` + (`shell.ts:135`) has always handled a mid-stream `kind: "snapshot"` item as a + wholesale replace, and re-persists it. Old clients handle the new server + behaviour natively. + +Note the thread-detail subscription (`subscribeThread`, `ws.ts:1173`) shares +the unbounded read but replays raw events with **no** per-event lookup, so it +has no silent-drop hole; capping it is out of scope. + +### 3.3 Client: retry the cold-path subscription (resilience completion) + +Round 1 made the *warm* path self-heal (discard cache → cold path). But the +cold path's own subscription (`runShellSyncLeg`, `shell.ts:183`) has no +`retryExpectedFailureAfter`: a now-loud stream failure sets the error banner +and then waits for the next *session change* to resubscribe (the +`subscribe` helper in `rpc/client.ts` only re-invokes on transport loss or +session replacement). With server failures becoming loud, a transient failure +on an established connection would otherwise leave the client parked on the +error banner until the next reconnect. + +Change: pass `retryExpectedFailureAfter: "5 seconds"` in the cold-leg +`subscribe` options. Resubscribing reuses the same `afterSequence` (the cold +base's sequence); the replay re-covers the interval and `applyItems` dedupes by +sequence, so the retry is idempotent. The existing `onExpectedFailure` → +`setStreamError` still fires, so the user sees the sync warning during the +retry window rather than nothing. + +The warm path deliberately keeps its **no-retry** semantics: its failure +handler falls through to the cold path (fresh snapshot), which is the correct +recovery for a possibly-poisoned cache, and retrying the identical replay was +exactly round 1's wedge. + +### What the client sees, case by case + +| Scenario | Server behaviour | Client behaviour | +|---|---|---| +| Small gap, healthy DB | Event replay (unchanged) | Events applied (unchanged) | +| Small gap, lookup blip | Retry absorbs it; replay succeeds | Unchanged, no drop | +| Small gap, persistent lookup failure | Stream fails with `OrchestrationGetSnapshotError` | Warm cache discarded → cold HTTP snapshot (round-1 path) | +| Gap > 500 events (the incident) | Live subscription attached, then `snapshot` item + buffered live | Wholesale replace via existing `applyItems` path; missing threads appear | +| Client ahead of server | `snapshot` item + live stream | Phantom state replaced | +| Live-phase persistent failure | Subscription errors | Warm: self-heal; cold: banner + 5 s retry from same base | + +## 4. Compatibility + +Deploys ship both sides together but tabs stay open across deploys, so the +mixed matrix matters: + +- **Old client (round-1 era) + new server:** the two new server behaviours are + (a) an error where silence used to be — caught by the shipped round-1 + self-heal — and (b) a `snapshot` item on the afterSequence path — handled by + the shipped `applyItems`/reducer, which has accepted mid-stream snapshots + since the item union's inception. Fully compatible; old clients get most of + the benefit. +- **Pre-round-1 client + new server:** a persistent replay failure shows the + sync-error banner instead of silently wedging; a transient one recovers on + the next reconnect. Strictly better than the status quo. +- **New client + old server:** the client change is only a retry duration on + the cold leg — harmless no-op against old server behaviour. + +No schema, RPC, or event changes; `OrchestrationSubscribeShellInput` is +untouched. + +## 5. Test plan + +**Server — `apps/server/src/server.test.ts`** (modelled on the existing +"routes websocket rpc orchestration shell snapshot errors" test at :5760, +using `buildAppUnderTest` layer stubs): + +1. **Silent-drop regression (the core test):** stub + `orchestrationEngine.readEvents` to emit **two** thread events, and + `projectionSnapshotQuery.getThreadShellById` to fail persistently for the + first thread and succeed for the second. Take stream elements until the + first event or error (the subscription is durable — catch-up concatenates + into the live leg and never completes on its own, so a bare `runCollect` + would hang; use `Stream.take`/timeout-bounded collection). Today's code + *succeeds* and yields only the second event — the exact silent omission + whose sequence advance seals the gap; fixed code must **fail** with + `OrchestrationGetSnapshotError` before yielding either. This pins "a + successful stream must not be missing an event", not merely the new error + path. +2. **Transient absorption:** `getThreadShellById` fails twice then succeeds; + assert the subscription yields the `thread-upserted` event (retry works, + no client-visible failure). +3. **Row-absent stays silent:** `getThreadShellById` succeeds with + `Option.none`; assert the event is skipped without error (taxonomy's other + half — guards against over-correction). +4. **Gap cap:** stub `getSnapshotSequence` far ahead of `afterSequence`; assert + the first stream item is `kind: "snapshot"` and `readEvents` is never + called. Companions: (a) gap ≤ 500 replays, asserting the **limit argument + passed to `readEvents` equals the sampled gap** (the cap must be enforced at + the read, not just at the branch); (b) `afterSequence > snapshotSequence` + also snapshots. +5. **Connect-gap during snapshot load:** publish a domain event while the + stubbed `getShellSnapshot` is in flight (gate the stub on a deferred); + assert the event is delivered after the snapshot item rather than lost. + Run it for both the no-afterSequence flow and the capped fallback — this + pins the eager `subscribeDomainEvents` ordering. Variant (pins the + error-preserving buffering shape from §3.2): the buffered event's + projection lookup fails persistently; after the snapshot gate releases, + the subscription must **fail** with `OrchestrationGetSnapshotError` — + not hang, not silently omit the event — proving the live leg's error + channel survives the buffering window. + +**Client — `packages/client-runtime/src/state/shell-sync.test.ts`** (alongside +the round-1 "self-heals to the cold path" test at :384): + +6. **Mid-stream snapshot replace:** warm-cache resume where the stubbed + `subscribeShell` responds to `afterSequence` with a fresh `snapshot` item; + assert the state and persisted cache are wholesale-replaced (client half of + §3.2, and a guard that future reducer changes keep this path). +7. **Cold-leg retry:** cold-path subscription fails once with an expected + failure, then succeeds; advance `TestClock` past 5 s; assert resubscribe + with the same `afterSequence` and eventual `live` status (client half of + §3.3). + +The end-to-end incident shape (thread created during the gap + one flaky +lookup → thread ultimately visible) is covered compositionally by 1 + the +existing round-1 self-heal test: 1 proves the server turns the drop into the +exact failure that test already proves the client heals from. + +Tests 4(a) and 5 rely on observing the `readEvents` limit and the eager +subscription; the `buildAppUnderTest` layer stubs already permit substituting +`orchestrationEngine`, so both are recordable there. + +## 6. Rejected alternatives + +- **Client belt-and-braces verification** (fetch a fresh snapshot after every + catch-up and diff counts): pays a full snapshot per reconnect, which defeats + `afterSequence`'s purpose; and count-comparison can false-negative (a drop + paired with a legitimate delete). The server-side fix closes the hole at its + source instead of detecting it downstream. +- **Always snapshot on afterSequence** (drop replay entirely): simplest + possible fix and closes the hole, but regresses the common tab-refocus case + the resume path exists for (tiny gaps, zero-byte reconnects). The 500-event + cap keeps that win. +- **Making every `Option.none` loud:** wrong — row-absent is a legitimate, + meaningful outcome (goal branch *depends* on it for `goal-removed`). +- **Periodic background resync / resync framework:** disproportionate; this is + a targeted reliability fix, and a correct-by-construction stream plus + self-heal makes scheduled reconciliation redundant. + +## 7. Implementation notes (fork conventions) + +- `toShellStreamEvent` and the `subscribeShell` handler are upstream-owned code + with existing `// loom:` splices (goal branch). The `orElseSucceed` removals, + the gap-cap branch, and the switch to eager `subscribeDomainEvents` (a + fork-added engine facility) are small in-place edits, each tagged `// loom:` + with a one-line rationale; the retry schedule and `SHELL_CATCHUP_MAX_EVENTS` + constant live next to the handler. Upstream introduced the swallow pattern in + #2968 ("Refactor recoverable Effect fallbacks to orElseSucceed") — the loom + comments should note the deliberate divergence so a future upstream sync + doesn't "fix" it back. +- `packages/client-runtime/src/state/shell.ts` cold-leg change extends the + existing round-1 `// loom:` block. +- Related but out of scope: `enrichProjectEvent`'s `orElseSucceed(() => event)` + (`ws.ts:592`) degrades enrichment rather than dropping events — acceptable; + and `subscribeThread`'s unbounded replay has no lookup hole (§3.2 note).