fix(server): promote shell reader worker, subagent archive and stuck-subagent fixes to bkt3

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>>([]);
+    /** 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);
     /** The thread shell's owner; `undefined` means no shell (a draft). */
     const shellOwner = yield* Ref.make<UserId | null | undefined>(owner);
@@ -285,6 +287,8 @@ const makeHarness = (options: HarnessOptions = {}) =>
             Effect.sync(() => {
               stoppedThreads.push(input.threadId);
             }),
+          hasLiveBackgroundWork: ({ threadId }) =>
+            Ref.get(backgroundWork).pipe(Effect.map((threads) => threads.has(threadId))),
           listSessions: () => Ref.get(sessions),
           streamEvents: Stream.fromSubscription(providerSubscription),
         }),
@@ -329,6 +333,7 @@ const makeHarness = (options: HarnessOptions = {}) =>
       dispatched,
       enabled,
       sessions,
+      backgroundWork,
       shellSession,
       shellOwner,
       shellFails,
@@ -985,6 +990,63 @@ describe("ClaudeAccountsService mode changes on a live session", () => {
       }),
   );
 
+  scenario(
+    "waits for a running subagent between turns and restarts on the status poll once it ends",
+    {},
+    (harness, service) =>
+      Effect.gen(function* () {
+        const threadId = ThreadId.make("thread-background-switch");
+        yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]);
+        yield* Ref.set(harness.backgroundWork, new Set([threadId]));
+        yield* service.refreshStatus();
+        yield* service.resolveForSession(threadId);
+        const view = yield* service.setThreadMode({
+          threadId,
+          mode: { kind: "profile", profile: "agent" },
+        });
+        // No turn runs, but a stop now would kill the subagent.
+        assert.equal(view.pendingRestart, true);
+        assert.deepEqual(harness.stoppedThreads, []);
+
+        yield* service.refreshStatus();
+        assert.deepEqual(harness.stoppedThreads, []);
+
+        // The subagent ends between turns: no turn end follows, the poll restarts.
+        yield* Ref.set(harness.backgroundWork, new Set());
+        yield* service.refreshStatus();
+        assert.deepEqual(harness.stoppedThreads, [threadId]);
+        assert.notEqual((yield* service.getThread(threadId)).pendingRestart, true);
+      }),
+  );
+
+  scenario(
+    "restarts anyway once background work has held a deferred restart for an hour",
+    {},
+    (harness, service) =>
+      Effect.gen(function* () {
+        const threadId = ThreadId.make("thread-background-cap");
+        yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]);
+        yield* Ref.set(harness.backgroundWork, new Set([threadId]));
+        yield* service.refreshStatus();
+        yield* service.resolveForSession(threadId);
+        yield* service.setThreadMode({ threadId, mode: { kind: "profile", profile: "agent" } });
+        assert.deepEqual(harness.stoppedThreads, []);
+        // The background poll stays idle while the clock moves; each status
+        // read below is the one the poll would make.
+        yield* Ref.set(harness.enabled, false);
+
+        yield* TestClock.adjust("59 minutes");
+        yield* service.refreshStatus();
+        assert.deepEqual(harness.stoppedThreads, []);
+        assert.equal((yield* service.getThread(threadId)).pendingRestart, true);
+
+        yield* TestClock.adjust("2 minutes");
+        yield* service.refreshStatus();
+        assert.deepEqual(harness.stoppedThreads, [threadId]);
+        assert.notEqual((yield* service.getThread(threadId)).pendingRestart, true);
+      }),
+  );
+
   scenario(
     "keeps the current account when Auto is chosen from a pin and it is still usable",
     {},
@@ -2052,6 +2114,8 @@ describe("ClaudeAccountsService access per user: owners, probing, hard limits, r
           liveSession(busy, "running"),
           liveSession(untouched, "ready"),
         ]);
+        // Background work never holds a session on an account its owner lost.
+        yield* Ref.set(harness.backgroundWork, new Set([idle]));
 
         yield* service.setAccess({ profile: "tushar", userIds: [colleague], actorUserId: null });
         assert.deepEqual(harness.stoppedThreads, [idle]);
@@ -2661,6 +2725,40 @@ describe("ClaudeAccountsService rebalancing at turn boundaries", () => {
     );
   }
 
+  {
+    const gate = gatedPlace();
+    scenario(
+      "leaves an idle thread alone while its session still runs a background subagent",
+      { place: gate.place, status: sharedStatus },
+      (harness, service) =>
+        Effect.gen(function* () {
+          yield* service.refreshStatus();
+          const threadId = ThreadId.make("thread-rebalance-background");
+          gate.state.open = ["tushar"];
+          yield* service.resolveForSession(threadId);
+          gate.state.open = ["agent"];
+          yield* Ref.set(harness.sessions, [liveSession(threadId, "ready")]);
+          yield* Ref.set(harness.backgroundWork, new Set([threadId]));
+          const placements = harness.placeCalls.length;
+
+          yield* endTurn(harness, threadId);
+          yield* drainDomainEvents(harness, service);
+          assert.equal(harness.placeCalls.length, placements);
+          assert.deepEqual(harness.stoppedThreads, []);
+          const view = yield* service.getThread(threadId);
+          assert.equal(view.resolvedProfile, "tushar");
+          assert.notEqual(view.pendingRestart, true);
+
+          // Once the work ended, the next turn boundary moves the thread as usual.
+          yield* Ref.set(harness.backgroundWork, new Set());
+          yield* endTurn(harness, threadId);
+          yield* drainDomainEvents(harness, service);
+          assert.deepEqual(harness.stoppedThreads, [threadId]);
+          assert.equal((yield* service.getThread(threadId)).resolvedProfile, "agent");
+        }),
+    );
+  }
+
   {
     const gate = gatedPlace();
     scenario(

apps/server/src/claudeAccounts/ClaudeAccountsService.ts

@@ -114,6 +114,11 @@ export const REBALANCE_PROBE_BACKOFF_MS = 5 * 60 * 1000;
 export const RESET_WAIT_THRESHOLD_MS = 5 * 60 * 1000;
 /** Margin after the reset so the status poll sees the account free before the thread continues. */
 export const RESET_WAIT_BUFFER_MS = 60 * 1000;
+/**
+ * How long a deferred account restart waits for the session's background work
+ * (a subagent, a background command) before it restarts the session anyway.
+ */
+export const BACKGROUND_WORK_RESTART_CAP_MS = 60 * 60 * 1000;
 const SNAPSHOT_CHANGE = "\u0000snapshot";
 /** The fields the CLI's `sessions/<pid>.json` carries that the live-session index needs. */
 const decodeSessionFile = Schema.decodeUnknownEffect(
@@ -358,6 +363,19 @@ export const make = Effect.gen(function* () {
   const threadProfiles = new Map<ThreadId, string>();
   const notices = new Map<ThreadId, Notice>();
   const pendingRestart = new Set<ThreadId>();
+  /**
+   * When each deferred restart first found the session's background work
+   * running. Its wait is capped at BACKGROUND_WORK_RESTART_CAP_MS from here.
+   */
+  const backgroundWaitSince = new Map<ThreadId, number>();
+  /** Deferred restarts that wait for the turn only, never for background work (access revoked). */
+  const restartIgnoresBackgroundWork = new Set<ThreadId>();
+  /** Forgets a deferred restart and its wait. True when one was pending. */
+  const forgetPendingRestart = (threadId: ThreadId): boolean => {
+    backgroundWaitSince.delete(threadId);
+    restartIgnoresBackgroundWork.delete(threadId);
+    return pendingRestart.delete(threadId);
+  };
   /** Threads with a turn in flight, as seen on the provider event bus. */
   const activeTurns = new Set<ThreadId>();
   /**
@@ -852,6 +870,8 @@ export const make = Effect.gen(function* () {
       yield* expireNotices;
       // Accounts may have room now: continue the threads waiting for one.
       yield* revivePass;
+      // Restarts deferred for background work have no turn end to wake them.
+      yield* recheckBackgroundWaits;
     });
 
   // Poll only while the setting is on; a disabled feature costs nothing but
@@ -1241,7 +1261,7 @@ export const make = Effect.gen(function* () {
       const row = rowResult.success;
       yield* pruneExhausted;
       // A spawn is the restart a deferred mode change was waiting for.
-      if (pendingRestart.delete(threadId)) yield* announce(threadId);
+      if (forgetPendingRestart(threadId)) yield* announce(threadId);
 
       if (row.mode.kind === "profile") {
         const pinned = row.mode.profile;
@@ -1351,10 +1371,7 @@ export const make = Effect.gen(function* () {
    * shell (which marks a turn active as soon as its command is accepted, before
    * the provider has started it) and the provider session itself.
    */
-  const threadIsIdle = (
-    threadId: ThreadId,
-    settled?: OrchestrationSession,
-  ): Effect.Effect<boolean> =>
+  const turnIsIdle = (threadId: ThreadId, settled?: OrchestrationSession): Effect.Effect<boolean> =>
     Effect.gen(function* () {
       if (activeTurns.has(threadId)) return false;
       // A `thread.session-set` event carries the session as just applied; the
@@ -1381,39 +1398,146 @@ export const make = Effect.gen(function* () {
       return live === undefined || (live.status !== "running" && live.activeTurnId === undefined);
     });
 
-  /** Restarts the thread's provider session so its next turn spawns on the new account. */
-  const restartForNewAccount = (threadId: ThreadId, why = "mode-changed") =>
+  /**
+   * Whether the thread can lose its provider session now: between turns, and
+   * with no background work (a subagent, a background command) still running
+   * in that session, which a stop would kill.
+   */
+  const threadIsIdle = (
+    threadId: ThreadId,
+    settled?: OrchestrationSession,
+  ): Effect.Effect<boolean> =>
+    Effect.gen(function* () {
+      if (!(yield* turnIsIdle(threadId, settled))) return false;
+      return !(yield* providers.hasLiveBackgroundWork({ threadId }));
+    });
+
+  /**
+   * Whether a deferred restart keeps waiting for the session's background
+   * work. The wait counts from the first check that finds work. Past
+   * BACKGROUND_WORK_RESTART_CAP_MS the restart runs anyway, and the released
+   * session marks the subagents it leaves open as cancelled.
+   */
+  const waitsForBackgroundWork = (threadId: ThreadId): Effect.Effect<boolean> =>
+    Effect.gen(function* () {
+      if (restartIgnoresBackgroundWork.has(threadId)) return false;
+      if (!(yield* providers.hasLiveBackgroundWork({ threadId }))) return false;
+      const nowMs = yield* Clock.currentTimeMillis;
+      const since = backgroundWaitSince.get(threadId);
+      if (since === undefined) {
+        backgroundWaitSince.set(threadId, nowMs);
+        return true;
+      }
+      if (nowMs - since < BACKGROUND_WORK_RESTART_CAP_MS) return true;
+      yield* Effect.logWarning("claude.account.restart-background-wait-expired", {
+        threadId,
+        waitedMs: nowMs - since,
+      });
+      return false;
+    });
+
+  /**
+   * Restarts the thread's provider session so its next turn spawns on the new
+   * account. A running turn defers the restart to its end, and so does the
+   * session's background work, unless `waitForBackgroundWork` is false.
+   */
+  const restartForNewAccount = (
+    threadId: ThreadId,
+    why = "mode-changed",
+    options: { readonly waitForBackgroundWork?: boolean } = {},
+  ) =>
     Effect.gen(function* () {
       const sessions = yield* providers.listSessions();
       const live = sessions.find(
         (session) => session.threadId === threadId && session.status !== "closed",
       );
       if (live === undefined) return;
-      if (!(yield* threadIsIdle(threadId))) {
+      const ignoresBackgroundWork = options.waitForBackgroundWork === false;
+      const turnIdle = yield* turnIsIdle(threadId);
+      const waitingForBackgroundWork =
+        turnIdle && !ignoresBackgroundWork && (yield* waitsForBackgroundWork(threadId));
+      if (!turnIdle || waitingForBackgroundWork) {
         pendingRestart.add(threadId);
-        yield* Effect.logInfo("claude.account.restart-deferred", { threadId });
+        if (ignoresBackgroundWork) restartIgnoresBackgroundWork.add(threadId);
+        yield* Effect.logInfo("claude.account.restart-deferred", {
+          threadId,
+          waitingFor: turnIdle ? "background-work" : "turn",
+        });
         return;
       }
+      backgroundWaitSince.delete(threadId);
       yield* stopSessionQuietly(threadId, why);
     });
 
   /**
-   * The turn a mode change waited for has ended. Stop only if nothing new has
-   * started since (a queued message may already be on its way); otherwise
-   * keep waiting for the next completion.
+   * The turn or the background work a mode change waited for has ended. Stop
+   * only if nothing new has started since (a queued message may already be on
+   * its way); otherwise keep waiting for the next completion.
    */
   const completePendingRestart = (threadId: ThreadId, settled?: OrchestrationSession) =>
     Effect.gen(function* () {
       if (!pendingRestart.has(threadId)) return;
-      if (!(yield* threadIsIdle(threadId, settled))) {
-        yield* Effect.logInfo("claude.account.restart-still-deferred", { threadId });
+      if (!(yield* turnIsIdle(threadId, settled))) {
+        yield* Effect.logInfo("claude.account.restart-still-deferred", {
+          threadId,
+          waitingFor: "turn",
+        });
+        return;
+      }
+      if (yield* waitsForBackgroundWork(threadId)) {
+        // Another fiber completed the restart while this check ran: its wait
+        // start must not outlive it.
+        if (!pendingRestart.has(threadId)) {
+          backgroundWaitSince.delete(threadId);
+          return;
+        }
+        yield* Effect.logInfo("claude.account.restart-still-deferred", {
+          threadId,
+          waitingFor: "background-work",
+        });
         return;
       }
-      pendingRestart.delete(threadId);
+      // Another fiber may have completed this restart while the checks ran.
+      if (!forgetPendingRestart(threadId)) return;
       yield* stopSessionQuietly(threadId, "mode-changed-after-turn");
       yield* announce(threadId);
     });
 
+  /**
+   * A restart deferred for background work has no turn end to wake it when
+   * that work ends between turns, so the status poll checks each such thread.
+   * The cheap check reads only the thread's own sessions: the full idle check
+   * reads every session, and each read starts that session's idle release
+   * again.
+   */
+  const recheckBackgroundWaits: Effect.Effect<void> = Effect.suspend(() =>
+    Effect.forEach(
+      [...backgroundWaitSince],
+      ([threadId, since]) =>
+        Effect.gen(function* () {
+          if (!pendingRestart.has(threadId)) {
+            backgroundWaitSince.delete(threadId);
+            return;
+          }
+          // A running turn re-checks the restart itself when it ends.
+          if (activeTurns.has(threadId)) return;
+          const nowMs = yield* Clock.currentTimeMillis;
+          if (
+            nowMs - since < BACKGROUND_WORK_RESTART_CAP_MS &&
+            (yield* providers.hasLiveBackgroundWork({ threadId }))
+          ) {
+            return;
+          }
+          yield* completePendingRestart(threadId);
+        }),
+      { discard: true },
+    ),
+  ).pipe(
+    Effect.catchCause((cause) =>
+      Effect.logWarning("claude.account.background-wait-recheck-failed", { cause }),
+    ),
+  );
+
   // ------------------------------------------------------------ revive queue
 
   /** Forgets the rejections already handled for a thread, so the next one is judged afresh. */
@@ -1571,7 +1695,7 @@ export const make = Effect.gen(function* () {
       if (wouldChange) {
         yield* restartForNewAccount(threadId);
       } else {
-        pendingRestart.delete(threadId);
+        forgetPendingRestart(threadId);
       }
       // Choosing an account for a thread that stopped mid-turn to wait is
       // "continue now on that account": its session already stopped.
@@ -2149,7 +2273,7 @@ export const make = Effect.gen(function* () {
       projectedRunning.delete(threadId);
       threadProfiles.delete(threadId);
       notices.delete(threadId);
-      pendingRestart.delete(threadId);
+      forgetPendingRestart(threadId);
       activeTurns.delete(threadId);
       movedAwaitingTurn.delete(threadId);
       yield* repository
@@ -2203,7 +2327,9 @@ export const make = Effect.gen(function* () {
       if (mayUse(access, owner, profile)) continue;
       if (row.mode.kind === "profile") setNotice(threadId, noAccessDetail(profile));
       yield* Effect.logInfo("claude.account.access-revoked", { threadId, profile });
-      yield* restartForNewAccount(threadId);
+      // The session must leave an account its owner may no longer use, even
+      // while its background work runs.
+      yield* restartForNewAccount(threadId, "mode-changed", { waitForBackgroundWork: false });
       yield* announce(threadId);
     }
   }).pipe(