From c1ca02c6dd4e36abb75e09ad819ef5dc0b728b51 Mon Sep 17 00:00:00 2001 From: Theo Browne Date: Thu, 3 Sep 2026 17:08:54 -0700 Subject: [PATCH 1/4] fix(server): keep events during thread subscription startup (#9521) (cherry picked from commit 9c9ae3dc0e94a957d9c4a61bb211caf914828054) --- apps/server/src/server.test.ts | 12 ++++++++---- apps/server/src/ws.ts | 4 +++- 2 files changed, 11 insertions(+), 5 deletions(-) diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 549f84d8d..98fd66001 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7068,7 +7068,6 @@ it.layer(NodeServices.layer)("server router seam", (it) => { projectionSnapshotQuery: { getThreadDetailSnapshot: () => Effect.gen(function* () { - yield* Effect.sleep("25 millis"); yield* PubSub.publish(liveEvents, messageEvent); return Option.some({ snapshotSequence: 1, thread }); }), @@ -7081,14 +7080,19 @@ it.layer(NodeServices.layer)("server router seam", (it) => { withWsRpcClient(wsUrl, (client) => client[ORCHESTRATION_WS_METHODS.subscribeThread]({ threadId: defaultThreadId, - }).pipe(Stream.take(2), Stream.runCollect), + requestCompletionMarker: true, + }).pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), ), - ).pipe(Effect.timeout("2 seconds")); + ); assert.equal(items[0]?.kind, "snapshot"); assert.equal(items[1]?.kind, "event"); assert.equal(items[1]?.kind === "event" ? items[1].event.sequence : null, 2); - }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), + assert.equal(items[2]?.kind, "synchronized"); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), ); it.effect("coalesces buffered live tool updates to the latest state", () => diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 8ed4900fa..df07314ed 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1571,7 +1571,9 @@ const makeWsRpcLayer = ( // Attach live delivery before reading either replay or snapshot state. // Otherwise an event published while the snapshot is loading is lost. const liveBuffer = yield* makeThreadLiveEventCoalescer(); - yield* Effect.forkScoped(liveStream.pipe(Stream.runForEach(liveBuffer.offer))); + yield* Effect.forkScoped(liveStream.pipe(Stream.runForEach(liveBuffer.offer)), { + startImmediately: true, + }); const bufferedLiveStream = liveBuffer.stream; // When the client already loaded the snapshot over HTTP it passes From a73427b75b6047b5d2bfe1def5dda4847136f958 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 5 Sep 2026 19:43:07 -0600 Subject: [PATCH 2/4] docs(upstream): record thread subscription startup fix --- .agents/upstream-review.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index 5ba4d4963..490c11a8b 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -15,6 +15,23 @@ Two standing sections outlive any single batch and must be read on every review: ## Review batches +## 2026-09-05 — thread subscription startup (A2, partial) + +The maintainer approved A2 against upstream `f12d39359f0f76a64ff2d77959c5baf821df15be`. +Adopted #9521 (`9c9ae3dc0e94a957d9c4a61bb211caf914828054`) as `10a7fc4113` +on `upstream/2026-09-05-thread-subscription`, awaiting merge. Clean port: +attach the live consumer immediately before reading the snapshot, preserving +events emitted during snapshot loading. Applies to all providers and clients, +including local, remote, relay, and tunnel WebSocket connections. + +Eight focused subscription/buffering tests pass; server typecheck and targeted +lint pass. The new regression fails deterministically with the old consumer +startup and passes with the fix, using the synchronized marker rather than sleeps. +No client contract or Prime admission/recovery change. + +The cursor is unchanged for this partial selection. Deferred/watch results +are recorded with #268; no register or watch row changes in this PR. + ## 2026-09-02 — `9b2d04317c68233782e0630464ac86d77d0686f3..beae2147a9487ec47ac992319f2216914b4cb62d` The maintainer's standing instruction for this batch was to stop escalating From 2aa927dd88e75ed335dead83ae342e1719ce8570 Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 5 Sep 2026 19:56:58 -0600 Subject: [PATCH 3/4] test(server): synchronize the legacy live-event fixture --- .agents/upstream-review.md | 5 +++-- apps/server/src/server.test.ts | 15 +++++++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index 490c11a8b..318806422 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -24,8 +24,9 @@ attach the live consumer immediately before reading the snapshot, preserving events emitted during snapshot loading. Applies to all providers and clients, including local, remote, relay, and tunnel WebSocket connections. -Eight focused subscription/buffering tests pass; server typecheck and targeted -lint pass. The new regression fails deterministically with the old consumer +All 154 server-router tests pass; server typecheck and targeted lint pass. +The legacy rollback-filter test now waits for the client synchronization receipt +before publishing live events, so it does not depend on fiber scheduling. The new regression fails deterministically with the old consumer startup and passes with the fix, using the synchronized marker rather than sleeps. No client contract or Prime admission/recovery change. diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 98fd66001..2349dfd8a 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7507,12 +7507,15 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }, } satisfies Extract; + const readyForLiveEvents = yield* Deferred.make(); yield* buildAppUnderTest({ layers: { orchestrationEngine: { latestSequence: Effect.succeed(3), readEvents: () => Stream.empty, - streamDomainEvents: Stream.make(rollbackEvent, laterEvent), + streamDomainEvents: Stream.fromEffect(Deferred.await(readyForLiveEvents)).pipe( + Stream.flatMap(() => Stream.make(rollbackEvent, laterEvent)), + ), }, }, }); @@ -7523,7 +7526,15 @@ it.layer(NodeServices.layer)("server router seam", (it) => { threadId: defaultThreadId, afterSequence: 3, requestCompletionMarker: true, - }).pipe(Stream.take(2), Stream.runCollect), + }).pipe( + Stream.tap((item) => + item.kind === "synchronized" + ? Deferred.succeed(readyForLiveEvents, undefined) + : Effect.void, + ), + Stream.take(2), + Stream.runCollect, + ), ), ); assert.deepEqual( From 9f6b1170038f82c59aec27c1e7ec4f14c0ca09ec Mon Sep 17 00:00:00 2001 From: Trevor Walker Date: Sat, 5 Sep 2026 20:32:55 -0600 Subject: [PATCH 4/4] docs(upstream): consolidate batch records in PR #279 --- .agents/upstream-review.md | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/.agents/upstream-review.md b/.agents/upstream-review.md index 318806422..5ba4d4963 100644 --- a/.agents/upstream-review.md +++ b/.agents/upstream-review.md @@ -15,24 +15,6 @@ Two standing sections outlive any single batch and must be read on every review: ## Review batches -## 2026-09-05 — thread subscription startup (A2, partial) - -The maintainer approved A2 against upstream `f12d39359f0f76a64ff2d77959c5baf821df15be`. -Adopted #9521 (`9c9ae3dc0e94a957d9c4a61bb211caf914828054`) as `10a7fc4113` -on `upstream/2026-09-05-thread-subscription`, awaiting merge. Clean port: -attach the live consumer immediately before reading the snapshot, preserving -events emitted during snapshot loading. Applies to all providers and clients, -including local, remote, relay, and tunnel WebSocket connections. - -All 154 server-router tests pass; server typecheck and targeted lint pass. -The legacy rollback-filter test now waits for the client synchronization receipt -before publishing live events, so it does not depend on fiber scheduling. The new regression fails deterministically with the old consumer -startup and passes with the fix, using the synchronized marker rather than sleeps. -No client contract or Prime admission/recovery change. - -The cursor is unchanged for this partial selection. Deferred/watch results -are recorded with #268; no register or watch row changes in this PR. - ## 2026-09-02 — `9b2d04317c68233782e0630464ac86d77d0686f3..beae2147a9487ec47ac992319f2216914b4cb62d` The maintainer's standing instruction for this batch was to stop escalating