Confirm resumed thread subscriptions

apps/server/src/server.test.ts

@@ -5607,6 +5607,30 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
     }).pipe(Effect.provide(NodeHttpServer.layerTest)),
   );
 
+  it.effect("confirms a resumed thread subscription when no replay events exist", () =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest({
+        layers: {
+          orchestrationEngine: {
+            readEvents: () => Stream.empty,
+          },
+        },
+      });
+
+      const wsUrl = yield* getWsServerUrl("/ws");
+      const firstItem = yield* Effect.scoped(
+        withWsRpcClient(wsUrl, (client) =>
+          client[ORCHESTRATION_WS_METHODS.subscribeThread]({
+            threadId: defaultThreadId,
+            afterSequence: 0,
+          }).pipe(Stream.runHead),
+        ),
+      );
+
+      assert.deepEqual(Option.getOrThrow(firstItem), { kind: "synchronized" });
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
   it.effect("enriches replayed project events with repository identity metadata", () =>
     Effect.gen(function* () {
       const repositoryIdentity = {

apps/server/src/ws.ts

@@ -1104,7 +1104,14 @@ const makeWsRpcLayer = (
                             }),
                         ),
                       );
-                    return Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer));
+                    // A warm client already has the full snapshot and might have no
+                    // replay events to apply. Emit a lightweight marker once its live
+                    // subscription is attached so it can leave the syncing state
+                    // without retransmitting the full thread body.
+                    return Stream.concat(
+                      Stream.make({ kind: "synchronized" as const }),
+                      Stream.concat(catchUpStream, Stream.fromQueue(liveBuffer)),
+                    );
                   }),
                 );
               }

packages/client-runtime/src/state/threads-sync.test.ts

@@ -77,6 +77,10 @@ const BASE_THREAD: OrchestrationThread = {
 
 type TestThreadInput = OrchestrationThreadStreamItem | Error;
 
+function synchronized(): OrchestrationThreadStreamItem {
+  return { kind: "synchronized" };
+}
+
 function testSession(client: WsRpcProtocolClient): RpcSession.RpcSession {
   return {
     client,
@@ -484,38 +488,47 @@ describe("EnvironmentThreads", () => {
     }),
   );
 
-  it.effect("restores live status after reconnect when no new thread events arrive", () =>
-    Effect.gen(function* () {
-      const harness = yield* makeHarness({ cached: BASE_THREAD });
-      yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+  it.effect(
+    "waits for stream synchronization after reconnect when no new thread events arrive",
+    () =>
+      Effect.gen(function* () {
+        const harness = yield* makeHarness({ cached: BASE_THREAD });
+        yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+
+        yield* SubscriptionRef.set(harness.supervisorState, {
+          desired: true,
+          network: "online",
+          phase: "connecting",
+          stage: "synchronizing",
+          attempt: 2,
+          generation: 1,
+          lastFailure: null,
+          retryAt: null,
+        });
+        yield* awaitThreadState(harness.observed, (value) => value.status === "synchronizing");
+
+        yield* SubscriptionRef.set(harness.supervisorState, {
+          desired: true,
+          network: "online",
+          phase: "connected",
+          stage: null,
+          attempt: 2,
+          generation: 2,
+          lastFailure: null,
+          retryAt: null,
+        });
+        for (let index = 0; index < 10; index += 1) {
+          yield* Effect.yieldNow;
+        }
 
-      yield* SubscriptionRef.set(harness.supervisorState, {
-        desired: true,
-        network: "online",
-        phase: "connecting",
-        stage: "synchronizing",
-        attempt: 2,
-        generation: 1,
-        lastFailure: null,
-        retryAt: null,
-      });
-      yield* awaitThreadState(harness.observed, (value) => value.status === "synchronizing");
+        expect((yield* Ref.get(harness.latest)).status).toBe("synchronizing");
 
-      yield* SubscriptionRef.set(harness.supervisorState, {
-        desired: true,
-        network: "online",
-        phase: "connected",
-        stage: null,
-        attempt: 2,
-        generation: 2,
-        lastFailure: null,
-        retryAt: null,
-      });
-      yield* awaitThreadState(harness.observed, (value) => value.status === "live");
+        yield* Queue.offer(harness.inputs, synchronized());
+        yield* awaitThreadState(harness.observed, (value) => value.status === "live");
 
-      const latest = yield* Ref.get(harness.latest);
-      expect(latest.status).toBe("live");
-      expect(Option.getOrThrow(latest.data)).toEqual(BASE_THREAD);
-    }),
+        const latest = yield* Ref.get(harness.latest);
+        expect(latest.status).toBe("live");
+        expect(Option.getOrThrow(latest.data)).toEqual(BASE_THREAD);
+      }),
   );
 });

packages/client-runtime/src/state/threads.ts

@@ -99,11 +99,11 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
     error: Option.none(),
   }));
   const setReady = SubscriptionRef.update(state, (current) =>
-    current.status === "deleted"
+    current.status === "live" || current.status === "deleted"
       ? current
       : {
           ...current,
-          status: Option.isSome(current.data) ? ("live" as const) : ("synchronizing" as const),
+          status: "synchronizing" as const,
           error: Option.none(),
         },
   );
@@ -154,6 +154,15 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
   const applyItem = Effect.fn("EnvironmentThreadState.applyItem")(function* (
     item: OrchestrationThreadStreamItem,
   ) {
+    if (item.kind === "synchronized") {
+      yield* SubscriptionRef.update(state, (current) =>
+        Option.isSome(current.data) && current.status !== "deleted"
+          ? { ...current, status: "live" as const, error: Option.none() }
+          : current,
+      );
+      return;
+    }
+
     if (item.kind === "snapshot") {
       yield* SubscriptionRef.set(lastSequence, item.snapshot.snapshotSequence);
       yield* setThread(item.snapshot.thread);

packages/contracts/src/orchestration.ts

@@ -1135,6 +1135,11 @@ export const OrchestrationEvent = Schema.Union([
 export type OrchestrationEvent = typeof OrchestrationEvent.Type;
 
 export const OrchestrationThreadStreamItem = Schema.Union([
+  // Confirms that a resumed detail subscription is live even when no events
+  // were emitted after the client's cached snapshot sequence.
+  Schema.Struct({
+    kind: Schema.Literal("synchronized"),
+  }),
   Schema.Struct({
     kind: Schema.Literal("snapshot"),
     snapshot: OrchestrationThreadDetailSnapshot,