fix(server): account policies no longer freeze the server reading the shell

apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts

@@ -201,6 +201,8 @@ const makeHarness = (options: HarnessOptions = {}) =>
     const place = options.place ?? spreadPlace(status);
     const enabled = yield* Ref.make(options.enabled ?? true);
     const sessions = yield* Ref.make<ReadonlyArray<ProviderSession>>([]);
+    /** How often the service listed the provider sessions, each a projection read in production. */
+    const listSessionsCalls = yield* Ref.make(0);
     /** Threads whose live session still runs background work (a subagent) between turns. */
     const backgroundWork = yield* Ref.make<ReadonlySet<ThreadId>>(new Set());
     const shellSession = yield* Ref.make<OrchestrationSession | null>(null);
@@ -289,7 +291,10 @@ const makeHarness = (options: HarnessOptions = {}) =>
             }),
           hasLiveBackgroundWork: ({ threadId }) =>
             Ref.get(backgroundWork).pipe(Effect.map((threads) => threads.has(threadId))),
-          listSessions: () => Ref.get(sessions),
+          listSessions: () =>
+            Ref.update(listSessionsCalls, (count) => count + 1).pipe(
+              Effect.andThen(Ref.get(sessions)),
+            ),
           streamEvents: Stream.fromSubscription(providerSubscription),
         }),
       ),
@@ -333,6 +338,7 @@ const makeHarness = (options: HarnessOptions = {}) =>
       dispatched,
       enabled,
       sessions,
+      listSessionsCalls,
       backgroundWork,
       shellSession,
       shellOwner,
@@ -2792,6 +2798,68 @@ describe("ClaudeAccountsService rebalancing at turn boundaries", () => {
     );
   }
 
+  {
+    const gate = gatedPlace();
+    // tushar takes no new session, and every other account is at its fair share.
+    const fullStatus = (dir: DirFor) => ({
+      ...switcherStatus([
+        switcherProfile(dir, "tushar", { sessions: 3, fair_share: null, guard_ok: false, load: 3 }),
+        switcherProfile(dir, "agent", {
+          rank: 2,
+          elected: false,
+          sessions: 2,
+          fair_share: 2,
+          load: 2,
+        }),
+        switcherProfile(dir, "barsha", {
+          rank: 3,
+          elected: false,
+          sessions: 2,
+          fair_share: 2,
+          load: 2,
+        }),
+      ]),
+      place_rule: "space-share-v2",
+    });
+    scenario(
+      "reads no sessions at a turn end on a full account while no other account has room",
+      { place: gate.place, status: fullStatus },
+      (harness, service) =>
+        Effect.gen(function* () {
+          yield* service.refreshStatus();
+          const threadId = ThreadId.make("thread-rebalance-no-room");
+          gate.state.open = ["tushar"];
+          yield* service.resolveForSession(threadId);
+          gate.state.open = ["agent"];
+          yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]);
+          const reads = yield* Ref.get(harness.listSessionsCalls);
+          const placements = harness.placeCalls.length;
+
+          yield* endTurn(harness, threadId);
+          yield* drainDomainEvents(harness, service);
+          assert.equal(yield* Ref.get(harness.listSessionsCalls), reads);
+          assert.equal(harness.placeCalls.length, placements);
+          assert.deepEqual(harness.stoppedThreads, []);
+          assert.equal((yield* service.getThread(threadId)).resolvedProfile, "tushar");
+
+          // Once agent has room, the next turn end moves the thread as before.
+          yield* Ref.update(harness.status, (current) => ({
+            ...current,
+            profiles: current.profiles.map((profile) =>
+              profile.name === "agent" ? { ...profile, sessions: 0, load: 0 } : profile,
+            ),
+          }));
+          yield* service.refreshStatus();
+          yield* endTurn(harness, threadId);
+          yield* drainDomainEvents(harness, service);
+          assert.isAbove(yield* Ref.get(harness.listSessionsCalls), reads);
+          assert.include(harness.placeCalls.at(-1)!.avoid, "tushar");
+          assert.deepEqual(harness.stoppedThreads, [threadId]);
+          assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent");
+        }),
+    );
+  }
+
   {
     const gate = gatedPlace();
     scenario(

apps/server/src/claudeAccounts/ClaudeAccountsService.ts

@@ -2088,11 +2088,9 @@ export const make = Effect.gen(function* () {
       if (!(yield* readSettings).enabled) return;
       const row = yield* repository.get(threadId);
       if (row.mode.kind !== "auto" || row.resolvedProfile !== current) return;
-      const live = (yield* providers.listSessions()).some(
-        (session) => session.threadId === threadId && session.status !== "closed",
-      );
-      if (!live || !(yield* threadIsIdle(threadId, settled))) return;
-      // Without a subprocess: is there anywhere better at all?
+      // Without a subprocess or a session read: is there anywhere better at
+      // all? While every account is full, each turn end on a full one would
+      // otherwise read the sessions for nothing.
       yield* pruneExhausted;
       const { disallowed } = yield* disallowedFor(threadId);
       const pending = yield* pendingCounts;
@@ -2108,6 +2106,10 @@ export const make = Effect.gen(function* () {
               (pending[profile.name] ?? 0),
       );
       if (!roomElsewhere) return;
+      const live = (yield* providers.listSessions()).some(
+        (session) => session.threadId === threadId && session.status !== "closed",
+      );
+      if (!live || !(yield* threadIsIdle(threadId, settled))) return;
 
       const target = yield* Semaphore.withPermit(placementLock)(
         Effect.gen(function* () {

apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts

@@ -229,29 +229,34 @@ function fixture(
   options: {
     readonly projections?: ReadonlyMap<ThreadId, Projection>;
     readonly failRecoveryRead?: boolean;
+    /** The released session's driver; Claude by default. */
+    readonly driver?: ProviderDriverKind;
   } = {},
 ) {
   const projections = options.projections ?? incidentProjections();
+  const sessionDriver = options.driver ?? driver;
   const written: Array<OrchestrationV2DomainEvent> = [];
   /** Threads whose full recovery read ran, in order. */
   const recoveryReads: Array<ThreadId> = [];
   const adapter: ProviderAdapterV2["Service"] = {
     instanceId,
-    driver,
+    driver: sessionDriver,
     getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2),
     planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
     openSession: (input) =>
       Effect.gen(function* () {
         const events = yield* Queue.unbounded<ProviderAdapterV2Event, Cause.Done>();
         const unused = () =>
-          Effect.fail(new ProviderAdapterProtocolError({ driver, detail: "unused" }));
+          Effect.fail(
+            new ProviderAdapterProtocolError({ driver: sessionDriver, detail: "unused" }),
+          );
         return {
           instanceId,
-          driver,
+          driver: sessionDriver,
           providerSessionId: input.providerSessionId,
           providerSession: {
             id: input.providerSessionId,
-            driver,
+            driver: sessionDriver,
             providerInstanceId: instanceId,
             status: "ready" as const,
             cwd: "/tmp",
@@ -413,19 +418,108 @@ it.effect("a detach that leaves the shared session open settles nothing", () =>
   }).pipe(Effect.provide(f.layer));
 });
 
-it.effect("a thread another live session still serves keeps its subagents", () => {
-  const f = fixture();
-  return Effect.gen(function* () {
-    const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
-    yield* manager.open(f.open(parentThreadId));
-    yield* manager.open(f.open(parentThreadId, otherSessionId));
-    yield* TestClock.adjust("10 minutes");
-    yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" });
+it.effect(
+  "a replacement Claude session on the thread does not keep the released process's subagents open",
+  () => {
+    const f = fixture();
+    return Effect.gen(function* () {
+      const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+      yield* manager.open(f.open(parentThreadId));
+      // The incident: a hard limit stopped the session, and the thread was
+      // placed on another account before that release finished.
+      yield* manager.open(f.open(parentThreadId, otherSessionId));
+      yield* TestClock.adjust("10 minutes");
+      yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" });
 
-    assert.deepEqual(f.cancelled("subagent.updated"), []);
-    assert.deepEqual(f.cancelled("node.updated"), []);
-  }).pipe(Effect.provide(f.layer));
-});
+      assert.isTrue(Option.isSome(yield* manager.get(otherSessionId)));
+      // The process-bound subagent died with the released process.
+      assert.deepEqual(f.cancelled("subagent.updated"), ["subagent-running"]);
+      assert.sameMembers(f.cancelled("turn-item.updated"), [
+        "card-subagent-running",
+        "child-command",
+      ]);
+      assert.sameMembers(f.cancelled("node.updated"), ["subagent-running", "child-root-running"]);
+      // One bound by its provider thread may run in the live session, and one
+      // started after the release is the replacement's own.
+      const touched = f.written.flatMap((event) =>
+        event.type === "subagent.updated" ||
+        event.type === "turn-item.updated" ||
+        event.type === "node.updated"
+          ? [String(event.payload.id)]
+          : [],
+      );
+      for (const name of ["other-session", "unloaded-thread", "started-later"]) {
+        assert.notInclude(touched, `subagent-${name}`);
+        assert.notInclude(touched, `card-subagent-${name}`);
+        assert.notInclude(touched, `child-root-${name}`);
+      }
+    }).pipe(Effect.provide(f.layer));
+  },
+);
+
+it.effect(
+  "a subagent started after the release stays open while a replacement serves the thread",
+  () => {
+    const replacement = subagent("replacement", { startedAt: later, updatedAt: later });
+    const f = fixture({
+      projections: new Map([
+        [
+          parentThreadId,
+          projection(parentThreadId, {
+            subagents: [replacement],
+            turnItems: [subagentItem(replacement)] as never,
+            nodes: [node(replacement.id, parentThreadId, "subagent", later)],
+          }),
+        ],
+      ]),
+    });
+    return Effect.gen(function* () {
+      const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+      yield* manager.open(f.open(parentThreadId));
+      yield* manager.open(f.open(parentThreadId, otherSessionId));
+      yield* TestClock.adjust("10 minutes");
+      yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" });
+
+      assert.deepEqual(f.cancelled("subagent.updated"), []);
+      assert.deepEqual(f.cancelled("turn-item.updated"), []);
+      assert.deepEqual(f.cancelled("node.updated"), []);
+      // The subagent rows alone showed nothing the released process ran.
+      assert.deepEqual(f.recoveryReads, []);
+    }).pipe(Effect.provide(f.layer));
+  },
+);
+
+it.effect(
+  "a thread another live session still serves keeps its subagents for other drivers",
+  () => {
+    const codex = ProviderDriverKind.make("codex");
+    const running = subagent("codex-running", { driver: codex });
+    const f = fixture({
+      driver: codex,
+      projections: new Map([
+        [
+          parentThreadId,
+          projection(parentThreadId, {
+            subagents: [running],
+            turnItems: [subagentItem(running)] as never,
+            nodes: [node(running.id, parentThreadId, "subagent")],
+          }),
+        ],
+      ]),
+    });
+    return Effect.gen(function* () {
+      const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
+      yield* manager.open(f.open(parentThreadId));
+      yield* manager.open(f.open(parentThreadId, otherSessionId));
+      yield* TestClock.adjust("10 minutes");
+      yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" });
+
+      assert.deepEqual(f.recoveryReads, []);
+      assert.deepEqual(f.cancelled("subagent.updated"), []);
+      assert.deepEqual(f.cancelled("node.updated"), []);
+    }).pipe(Effect.provide(f.layer));
+  },
+);
 
 it.effect("a server shutdown leaves the subagents to startup recovery and its wake", () => {
   const f = fixture();

apps/server/src/orchestration-v2/ProviderSessionManager.ts

@@ -179,6 +179,18 @@ export interface ProviderSessionManagerV2Shape {
     ProviderSessionManagerV2Error
   >;
   readonly isLive: (providerSessionId: ProviderSessionId) => Effect.Effect<boolean>;
+  // T3-CUSTOM(expbkt3): BEGIN the live sessions and the threads each serves, from memory.
+  /**
+   * Reads no projection and counts as no activity. Optional so test layers
+   * that build the full shape need not supply it.
+   */
+  readonly listLive?: () => Effect.Effect<
+    ReadonlyArray<{
+      readonly providerSessionId: ProviderSessionId;
+      readonly threadIds: ReadonlyArray<ThreadId>;
+    }>
+  >;
+  // T3-CUSTOM(expbkt3): END
   readonly close: (
     providerSessionId: ProviderSessionId,
   ) => Effect.Effect<void, ProviderSessionManagerV2Error>;
@@ -2561,6 +2573,17 @@ export const layerWithOptions = (
           Ref.get(sessions).pipe(
             Effect.map((current) => current.has(sessionKey(providerSessionId))),
           ),
+        // T3-CUSTOM(expbkt3): BEGIN account policies list live sessions without a shell read.
+        listLive: () =>
+          Ref.get(sessions).pipe(
+            Effect.map((current) =>
+              Array.from(current.values(), (entry) => ({
+                providerSessionId: entry.runtime.providerSessionId,
+                threadIds: [...entry.attachedThreadIds],
+              })),
+            ),
+          ),
+        // T3-CUSTOM(expbkt3): END
         get: Effect.fn("ProviderSessionManager.get")(
           function* (providerSessionId) {
             const entry = (yield* Ref.get(sessions)).get(sessionKey(providerSessionId));