@@ -26,7 +26,8 @@ import * as ConnectionWakeups from "../connection/wakeups.ts";
 import { safeErrorLogAttributes } from "../errors/safeLog.ts";
 import * as Persistence from "../platform/persistence.ts";
 import { runCachePersistence } from "./cachePersistence.ts";
-import { subscribeDynamic } from "../rpc/client.ts";
+import { subscribeDynamicWithSession } from "../rpc/client.ts";
+import type { RpcSession } from "../rpc/session.ts";
 import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts";
 import type { DeferredShellSnapshot } from "./shellPullRequests.ts";
 import {
@@ -81,6 +82,16 @@ function withoutDeferredPullRequests(value: DeferredShellSnapshot): Orchestratio
 
 const SHELL_SYNCHRONIZATION_ERROR_MESSAGE = "Could not synchronize environment data.";
 
+function groupBySession<A>(entries: ReadonlyArray<readonly [RpcSession, A]>) {
+  const groups: Array<[RpcSession, Array<A>]> = [];
+  for (const [session, value] of entries) {
+    const last = groups.at(-1);
+    if (last?.[0] === session) last[1].push(value);
+    else groups.push([session, [value]]);
+  }
+  return groups;
+}
+
 export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")(function* () {
   const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
   const cache = yield* Persistence.EnvironmentCacheStore;
@@ -108,7 +119,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
   // The omissions and server build of the subscription whose snapshot the
   // in-memory shell last came from; null while it holds only cached data.
   const authoritativeOmission = yield* Ref.make<string | null>(null);
-  const activeSubscriptionOmission = yield* Ref.make<string | null>(null);
+  // Items buffered from a previous session keep that session's key.
+  const sessionOmissions = new WeakMap<RpcSession, string | null>();
   const latestLiveSnapshot = yield* Ref.make<Option.Option<OrchestrationV2ShellSnapshot>>(
     Option.none(),
   );
@@ -209,12 +221,13 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
 
   const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* (
     items: ReadonlyArray<OrchestrationV2ShellStreamItem>,
+    omission: string | null,
     pendingLinkIds?: ReadonlySet<ThreadId>,
   ) {
     const initial = yield* SubscriptionRef.get(state);
     let waiting = yield* Ref.get(awaitingCompletion);
     let next = initial;
-    let receivedSnapshot = false;
+    let receivedAuthoritativeSnapshot = false;
     let deltas: Array<ShellStreamEvent> = [];
     const flushDeltas = () => {
       if (deltas.length > 0 && Option.isSome(next.snapshot)) {
@@ -268,7 +281,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
           ? undefined
           : { resolvedRepositoryIdentityRoots: item.resolvedRepositoryIdentityRoots },
       );
-      receivedSnapshot = true;
+      if (authoritative) receivedAuthoritativeSnapshot = true;
       next = {
         snapshot: Option.some(nextSnapshot),
         status: waiting ? "synchronizing" : "live",
@@ -289,11 +302,8 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
       yield* Ref.set(latestLiveSnapshot, next.snapshot);
     }
     yield* SubscriptionRef.set(state, next);
-    if (receivedSnapshot) {
-      const omission = yield* Ref.get(activeSubscriptionOmission);
-      if (omission !== null) {
-        yield* Ref.set(authoritativeOmission, omission);
-      }
+    if (receivedAuthoritativeSnapshot && omission !== null) {
+      yield* Ref.set(authoritativeOmission, omission);
     }
     if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) {
       yield* Queue.offer(persistence, next.snapshot.value);
@@ -375,7 +385,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
 
   yield* setSynchronizing;
   yield* Effect.forkScoped(
-    subscribeDynamic(
+    subscribeDynamicWithSession(
       ORCHESTRATION_V2_WS_METHODS.subscribeShell,
       Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) {
         const { omitIdleSubagents, omitOlderSettled, omissionKey } =
@@ -402,7 +412,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
               omissionKey: null,
             })),
           );
-        yield* Ref.set(activeSubscriptionOmission, omissionKey);
+        sessionOmissions.set(session, omissionKey);
         yield* Ref.set(awaitingCompletion, true);
         yield* setSynchronizing;
 
@@ -445,7 +455,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
                 ? []
                 : httpSnapshot.value.threads.map((thread) => thread.id),
             );
-            yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds);
+            yield* applyItems([{ kind: "snapshot", snapshot }], omissionKey, pendingLinkIds);
             canResume = true;
             current = yield* SubscriptionRef.get(state);
             if (Option.isSome(current.snapshot)) {
@@ -472,7 +482,15 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make")
         retryExpectedFailureAfter: "250 millis",
         resubscribe: foregroundResubscriptions,
       },
-    ).pipe(Stream.runForEachArray((items) => applyItems(items))),
+    ).pipe(
+      Stream.runForEachArray((entries) =>
+        Effect.forEach(
+          groupBySession(entries),
+          ([session, items]) => applyItems(items, sessionOmissions.get(session) ?? null),
+          { discard: true },
+        ),
+      ),
+    ),
   );
   yield* SubscriptionRef.changes(supervisor.state).pipe(
     Stream.runForEach((connectionState) => {