From 70be50aaf444202f1d26a711d2418bfd56aea20c Mon Sep 17 00:00:00 2001
From: Adolanium <94890352+Adolanium@users.noreply.github.com>
Date: Thu, 27 Aug 2026 19:57:13 +0300
Subject: [PATCH] fix(server): start thread live subscribe immediately

subscribeShell already starts its live buffer fork now. subscribeThread
did not. Effect 4 runs forkScoped on a later tick unless startImmediately
is set, so an event published while the thread snapshot loads can miss
the buffer.

The existing buffer test published after a 25ms sleep, which hid that
gap. It now publishes during the snapshot load, same as the shell test.
---
 apps/server/src/server.test.ts | 5 ++++-
 apps/server/src/ws.ts          | 1 +
 2 files changed, 5 insertions(+), 1 deletion(-)

diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts
index a9a2c3fa10d6..5e536b4aa69e 100644
--- a/apps/server/src/server.test.ts
+++ b/apps/server/src/server.test.ts
@@ -6398,7 +6398,10 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
           projectionSnapshotQuery: {
             getThreadDetailSnapshot: () =>
               Effect.gen(function* () {
-                yield* Effect.sleep("25 millis");
+                // Publish during the snapshot load, with no sleep. Effect 4
+                // schedules forkScoped work on a later tick unless
+                // startImmediately is set, so a delay here would hide a lost
+                // event.
                 yield* PubSub.publish(liveEvents, messageEvent);
                 return Option.some({ snapshotSequence: 1, thread });
               }),
diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts
index 226c82cdb1ac..b8c088767b6c 100644
--- a/apps/server/src/ws.ts
+++ b/apps/server/src/ws.ts
@@ -1465,6 +1465,7 @@ const makeWsRpcLayer = (
               const liveBuffer = yield* Queue.unbounded<OrchestrationThreadStreamItem>();
               yield* Effect.forkScoped(
                 liveStream.pipe(Stream.runForEach((item) => Queue.offer(liveBuffer, item))),
+                { startImmediately: true },
               );
               const bufferedLiveStream = Stream.fromQueue(liveBuffer);