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/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<Option.Option<OrchestrationThreadShell>, 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<boolean, ProjectionRepositoryError>;
+
   /**
    * 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 6122c498145..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 }),
@@ -5662,6 +5663,86 @@ 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 === "OrchestrationThreadNotFoundError");
+      assert.equal(result.failure.threadId, defaultThreadId);
+    }).pipe(Effect.provide(NodeHttpServer.layerTest)),
+  );
+
+  it.effect("tags a missing resumed thread subscription as not found", () =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest({
+        layers: {
+          projectionSnapshotQuery: {
+            hasThreadById: () => Effect.succeed(false),
+            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("keeps an archived resumed thread subscription alive", () =>
+    Effect.gen(function* () {
+      yield* buildAppUnderTest({
+        layers: {
+          projectionSnapshotQuery: {
+            hasThreadById: () => Effect.succeed(true),
+            getThreadShellById: () =>
+              Effect.die("active shell lookup must not determine existence"),
+            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<OrchestrationEvent>();
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 08b1770a0a2..24e132c3a12 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,
@@ -1387,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 threadExists = yield* projectionSnapshotQuery
+                  .hasThreadById(input.threadId)
+                  .pipe(
+                    Effect.mapError(
+                      (cause) =>
+                        new OrchestrationGetSnapshotError({
+                          message: `Failed to load thread ${input.threadId}`,
+                          cause,
+                        }),
+                    ),
+                  );
+                if (!threadExists) {
+                  return yield* new OrchestrationThreadNotFoundError({
+                    threadId: input.threadId,
+                  });
+                }
+
                 const afterSequence = input.afterSequence;
                 const catchUpStream = orchestrationEngine
                   .readEvents(afterSequence, Number.MAX_SAFE_INTEGER)
@@ -1426,9 +1444,8 @@ const makeWsRpcLayer = (
                 );
 
               if (Option.isNone(snapshot)) {
-                return yield* new OrchestrationGetSnapshotError({
-                  message: `Thread ${input.threadId} was not found`,
-                  cause: input.threadId,
+                return yield* new OrchestrationThreadNotFoundError({
+                  threadId: input.threadId,
                 });
               }
 
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/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<never>();
+      const secondEvents = yield* Queue.unbounded<never>();
+      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<number>();
+      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<TTag extends EnvironmentSubscriptionRpcTag> {
   readonly resubscribe?: Stream.Stream<unknown, never, never>;
 }
 
-export function subscribeDynamic<TTag extends EnvironmentSubscriptionRpcTag>(
+type SubscriptionMethod<TTag extends EnvironmentSubscriptionRpcTag> = (
+  input: EnvironmentRpcInput<TTag>,
+) => Stream.Stream<EnvironmentRpcStreamValue<TTag>, EnvironmentRpcStreamFailure<TTag>>;
+
+function subscribeDynamicStream<TTag extends EnvironmentSubscriptionRpcTag, TValue>(
   tag: TTag,
-  makeInput: (session: RpcSession) => Effect.Effect<EnvironmentRpcInput<TTag>>,
+  makeStream: (
+    session: RpcSession,
+    method: SubscriptionMethod<TTag>,
+  ) => Stream.Stream<TValue, EnvironmentRpcStreamFailure<TTag>>,
   options?: SubscriptionOptions<TTag>,
-): Stream.Stream<
-  EnvironmentRpcStreamValue<TTag>,
-  EnvironmentRpcStreamFailure<TTag>,
-  EnvironmentSupervisor
-> {
+): Stream.Stream<TValue, EnvironmentRpcStreamFailure<TTag>, EnvironmentSupervisor> {
   return Stream.unwrap(
     EnvironmentSupervisor.pipe(
       Effect.map((supervisor) => {
@@ -183,67 +186,52 @@ export function subscribeDynamic<TTag extends EnvironmentSubscriptionRpcTag>(
             Option.match({
               onNone: () => Stream.empty,
               onSome: (session) => {
-                const method = session.client[tag] as (
-                  input: EnvironmentRpcInput<TTag>,
-                ) => Stream.Stream<
-                  EnvironmentRpcStreamValue<TTag>,
-                  EnvironmentRpcStreamFailure<TTag>
-                >;
+                const method = session.client[tag] as SubscriptionMethod<TTag>;
                 const subscribeToSession = (): Stream.Stream<
-                  EnvironmentRpcStreamValue<TTag>,
+                  TValue,
                   EnvironmentRpcStreamFailure<TTag>
                 > =>
                   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<TTag extends EnvironmentSubscriptionRpcTag>(
   );
 }
 
+export function subscribeDynamic<TTag extends EnvironmentSubscriptionRpcTag>(
+  tag: TTag,
+  makeInput: (session: RpcSession) => Effect.Effect<EnvironmentRpcInput<TTag>>,
+  options?: SubscriptionOptions<TTag>,
+): Stream.Stream<
+  EnvironmentRpcStreamValue<TTag>,
+  EnvironmentRpcStreamFailure<TTag>,
+  EnvironmentSupervisor
+> {
+  return subscribeDynamicStream(
+    tag,
+    (session, method) => Stream.unwrap(makeInput(session).pipe(Effect.map(method))),
+    options,
+  );
+}
+
+export function subscribeDynamicWithContext<TTag extends EnvironmentSubscriptionRpcTag, TContext>(
+  tag: TTag,
+  makeInput: (session: RpcSession) => Effect.Effect<readonly [EnvironmentRpcInput<TTag>, TContext]>,
+  options?: SubscriptionOptions<TTag>,
+): Stream.Stream<
+  readonly [EnvironmentRpcStreamValue<TTag>, TContext],
+  EnvironmentRpcStreamFailure<TTag>,
+  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<TTag extends EnvironmentSubscriptionRpcTag>(
   tag: TTag,
   input: EnvironmentRpcInput<TTag>,
diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts
index a4514d5f0e6..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,6 +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, environmentShellMembershipLayer } from "./shellMembership.ts";
 
 const TARGET = new PrimaryConnectionTarget({
   environmentId: EnvironmentId.make("environment-1"),
@@ -61,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<OrchestrationShellStreamItem>();
+      const subscriptionStarted = yield* Deferred.make<void>();
       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<Option.Option<RpcSession.RpcSession>>(
@@ -94,12 +103,14 @@ describe("environment shell synchronization", () => {
       const snapshotLoader = ShellSnapshotLoader.of({
         load: () => Effect.succeed(Option.none()),
       });
+      const shellMembership = yield* EnvironmentShellMembership;
       const shellState = yield* makeEnvironmentShellState().pipe(
         Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
         Effect.provideService(Persistence.EnvironmentCacheStore, cache),
         Effect.provideService(ShellSnapshotLoader, snapshotLoader),
       );
 
+      yield* Deferred.await(subscriptionStarted);
       yield* SubscriptionRef.set(supervisorState, {
         desired: true,
         network: "online",
@@ -145,7 +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* 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 6ccb11797f5..102c768d11a 100644
--- a/packages/client-runtime/src/state/shell.ts
+++ b/packages/client-runtime/src/state/shell.ts
@@ -20,9 +20,10 @@ 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";
 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,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
   });
   const awaitingCompletion = yield* Ref.make(false);
   const persistence = yield* Queue.sliding<OrchestrationShellSnapshot>(1);
+  const invalidateMembership = Option.match(shellMembership, {
+    onNone: () => Effect.succeed<ShellMembershipRevision>(0),
+    onSome: (service) => service.setUnknown(environmentId),
+  });
+  const setMembershipAuthoritative = (
+    snapshot: OrchestrationShellSnapshot,
+    revision: ShellMembershipRevision,
+  ) =>
+    Option.match(shellMembership, {
+      onNone: () => Effect.void,
+      onSome: (service) => service.setAuthoritative(environmentId, snapshot, revision),
+    });
 
   const persist = Effect.fn("EnvironmentShellState.persist")(function* (
     snapshot: OrchestrationShellSnapshot,
@@ -95,6 +109,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
   );
 
   const setDisconnected = Ref.set(awaitingCompletion, false).pipe(
+    Effect.andThen(invalidateMembership),
     Effect.andThen(
       SubscriptionRef.update(state, (current) => ({
         ...current,
@@ -102,11 +117,12 @@ 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(),
   }));
+  const setSynchronizing = setSynchronizingState;
   const setReady = SubscriptionRef.update(state, (current) =>
     current.status === "live"
       ? current
@@ -118,6 +134,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
   );
   const setStreamError = (error: unknown) =>
     Ref.set(awaitingCompletion, false).pipe(
+      Effect.andThen(invalidateMembership),
       Effect.andThen(Effect.logWarning("Could not synchronize the environment shell.")),
       Effect.annotateLogs({
         environmentId,
@@ -134,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);
@@ -142,6 +160,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, membershipRevision);
+      }
       return;
     }
 
@@ -166,6 +188,9 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
       status: waiting ? "synchronizing" : "live",
       error: Option.none(),
     });
+    if (!waiting) {
+      yield* setMembershipAuthoritative(nextSnapshot, membershipRevision);
+    }
     yield* Queue.offer(persistence, nextSnapshot);
   });
 
@@ -175,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(
@@ -185,7 +211,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
           Effect.orElseSucceed(() => false),
         );
         yield* Ref.set(awaitingCompletion, supportsCompletionMarker);
-        yield* setSynchronizing;
+        const membershipRevision = yield* invalidateMembership;
+        yield* setSynchronizingState;
 
         const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe(
           Effect.flatMap(
@@ -203,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) => {
@@ -233,6 +266,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
     Effect.forkScoped,
   );
 
+  yield* Effect.addFinalizer(() => invalidateMembership);
+
   return state;
 });
 
@@ -385,4 +420,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..6bb15ee905b
--- /dev/null
+++ b/packages/client-runtime/src/state/shellMembership.test.ts
@@ -0,0 +1,70 @@
+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");
+      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");
+
+      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
new file mode 100644
index 00000000000..561f19f4ce4
--- /dev/null
+++ b/packages/client-runtime/src/state/shellMembership.ts
@@ -0,0 +1,64 @@
+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 type ShellMembershipRevision = number;
+
+interface EnvironmentMembershipState {
+  readonly revision: ShellMembershipRevision;
+  readonly snapshot: OrchestrationShellSnapshot | undefined;
+}
+
+export class EnvironmentShellMembership extends Context.Service<
+  EnvironmentShellMembership,
+  {
+    readonly getThreadMembership: (
+      environmentId: EnvironmentId,
+      threadId: ThreadId,
+    ) => Effect.Effect<ShellThreadMembership>;
+    readonly setAuthoritative: (
+      environmentId: EnvironmentId,
+      snapshot: OrchestrationShellSnapshot,
+      revision: ShellMembershipRevision,
+    ) => Effect.Effect<void>;
+    readonly setUnknown: (environmentId: EnvironmentId) => Effect.Effect<ShellMembershipRevision>;
+  }
+>()("@t3tools/client-runtime/state/shellMembership/EnvironmentShellMembership") {}
+
+export const environmentShellMembershipLayer = Layer.effect(
+  EnvironmentShellMembership,
+  Effect.gen(function* () {
+    const states = yield* Ref.make<ReadonlyMap<EnvironmentId, EnvironmentMembershipState>>(
+      new Map(),
+    );
+
+    return EnvironmentShellMembership.of({
+      getThreadMembership: (environmentId, threadId) =>
+        Ref.get(states).pipe(
+          Effect.map((current) => {
+            const snapshot = current.get(environmentId)?.snapshot;
+            if (snapshot === undefined) {
+              return "unknown";
+            }
+            return snapshot.threads.some((thread) => thread.id === threadId) ? "present" : "absent";
+          }),
+        ),
+      setAuthoritative: (environmentId, snapshot, revision) =>
+        Ref.update(states, (current) => {
+          const state = current.get(environmentId);
+          if (state === undefined || state.revision !== revision) {
+            return current;
+          }
+          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-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..cbcb656905d 100644
--- a/packages/client-runtime/src/state/threads-sync.test.ts
+++ b/packages/client-runtime/src/state/threads-sync.test.ts
@@ -2,6 +2,8 @@ import {
   EnvironmentId,
   EventId,
   ORCHESTRATION_WS_METHODS,
+  OrchestrationGetSnapshotError,
+  OrchestrationThreadNotFoundError,
   ProjectId,
   ProviderInstanceId,
   ThreadId,
@@ -36,6 +38,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 +134,7 @@ const makeHarness = Effect.fn("TestEnvironmentThreads.makeHarness")(function* (o
   readonly cached?: OrchestrationThread;
   readonly httpSnapshot?: Option.Option<OrchestrationThreadDetailSnapshot>;
   readonly completionMarker?: boolean;
+  readonly shellMembership?: ShellThreadMembership;
 }) {
   const inputs = yield* Queue.unbounded<TestThreadInput>();
   const observed = yield* Queue.unbounded<EnvironmentThreadState>();
@@ -217,10 +221,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.succeed(0),
+  });
   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 +279,17 @@ const snapshot = (thread: OrchestrationThread): OrchestrationThreadStreamItem =>
 
 const synchronized = (): OrchestrationThreadStreamItem => ({ kind: "synchronized" });
 
+const missingThreadError = () =>
+  new OrchestrationThreadNotFoundError({
+    threadId: THREAD_ID,
+  });
+
+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 +488,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 +588,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 +626,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..c742eb675a9 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,
+  OrchestrationThreadNotFoundError,
   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,14 @@ function formatThreadError(cause: Cause.Cause<unknown>): string {
     : "Could not synchronize the thread.";
 }
 
+const isOrchestrationThreadNotFoundError = Schema.is(OrchestrationThreadNotFoundError);
+
+function isMissingThreadFailure(cause: Cause.Cause<unknown>): boolean {
+  return cause.reasons.some(
+    (reason) => reason._tag === "Fail" && isOrchestrationThreadNotFoundError(reason.error),
+  );
+}
+
 function shouldPersistThread(thread: OrchestrationThread): boolean {
   const status = thread.session?.status;
   return status !== "starting" && status !== "running";
@@ -54,6 +66,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 +94,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
   );
   const awaitingCompletion = yield* Ref.make(false);
   const persistence = yield* Queue.sliding<OrchestrationThreadDetailSnapshot>(1);
+  const deletedSignal = yield* Deferred.make<void>();
 
   const persist = Effect.fn("EnvironmentThreadState.persist")(function* (
     snapshot: OrchestrationThreadDetailSnapshot,
@@ -177,6 +191,20 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make
         ),
       ),
     );
+    yield* Deferred.succeed(deletedSignal, undefined);
+  });
+
+  const handleStreamFailure = Effect.fn("EnvironmentThreadState.handleStreamFailure")(function* (
+    cause: Cause.Cause<unknown>,
+  ) {
+    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 +319,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 +350,11 @@ export function threadStateChanges(environmentId: EnvironmentIdType, threadId: T
 
 export function createEnvironmentThreadStateAtoms<R, E>(
   runtime: Atom.AtomRuntime<
-    EnvironmentRegistry | EnvironmentCacheStore | ThreadSnapshotLoader | R,
+    | EnvironmentRegistry
+    | EnvironmentCacheStore
+    | EnvironmentShellMembership
+    | ThreadSnapshotLoader
+    | R,
     E
   >,
 ) {
@@ -347,6 +379,7 @@ export function createEnvironmentThreadStateAtoms<R, E>(
 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..9fe788c0dd3 100644
--- a/packages/contracts/src/orchestration.ts
+++ b/packages/contracts/src/orchestration.ts
@@ -1295,6 +1295,17 @@ export class OrchestrationGetSnapshotError extends Schema.TaggedErrorClass<Orche
   },
 ) {}
 
+export class OrchestrationThreadNotFoundError extends Schema.TaggedErrorClass<OrchestrationThreadNotFoundError>()(
+  "OrchestrationThreadNotFoundError",
+  {
+    threadId: ThreadId,
+  },
+) {
+  override get message(): string {
+    return `Thread ${this.threadId} was not found`;
+  }
+}
+
 export class OrchestrationDispatchCommandError extends Schema.TaggedErrorClass<OrchestrationDispatchCommandError>()(
   "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,
   },
 );