From 168bd3cd4e305a268434392fc6eb78b14cb95ad3 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Tue, 20 Oct 2026 12:58:27 +0000 Subject: [PATCH 1/3] fix(server): subagents no longer stay Running after an account restart stops their session The Claude account service stops a thread's provider session between turns (rebalance, mode change, hard limit). Its idle check looked only at the parent turn, so it also stopped sessions whose background subagents were still working, and the session release recorded nothing for them: the subagent rows, the parent's subagent cards and the child's root turn stayed running until the parent got another message (84 minutes on 2026-10-20). - A released provider session now settles the provider-native subagents it leaves open (cancelled), for every release reason except server shutdown, which startup recovery already handles. A cheap subagent-row check first keeps idle releases from paying for the full recovery read. - Rebalance and mode-change restarts now wait for the session's background work, re-checked on the 15 s status poll and capped at 60 minutes. Access-revoked and hard-limit stops are unchanged. Co-Authored-By: Claude Opus 5.5 --- .../ClaudeAccountsService.test.ts | 98 ++++ .../claudeAccounts/ClaudeAccountsService.ts | 156 ++++++- ...nManager.releasedSubagents.expbkt3.test.ts | 430 ++++++++++++++++++ .../ProviderSessionManager.ts | 24 + .../releasedSessionSubagents.expbkt3.ts | 285 ++++++++++++ .../provider/ForkProviderSessions.expbkt3.ts | 34 ++ 6 files changed, 1009 insertions(+), 18 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts create mode 100644 apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts index 6a59576d32be2..da2e70c1dfd40 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.test.ts +++ b/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>([]); + /** Threads whose live session still runs background work (a subagent) between turns. */ + const backgroundWork = yield* Ref.make>(new Set()); const shellSession = yield* Ref.make(null); /** The thread shell's owner; `undefined` means no shell (a draft). */ const shellOwner = yield* Ref.make(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( diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts index 78c11b3fad3aa..267ce91dd9b09 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts +++ b/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/.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(); const notices = new Map(); const pendingRestart = new Set(); + /** + * 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(); + /** Deferred restarts that wait for the turn only, never for background work (access revoked). */ + const restartIgnoresBackgroundWork = new Set(); + /** 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(); /** @@ -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 => + const turnIsIdle = (threadId: ThreadId, settled?: OrchestrationSession): Effect.Effect => Effect.gen(function* () { if (activeTurns.has(threadId)) return false; // A `thread.session-set` event carries the session as just applied; the @@ -1381,39 +1398,140 @@ 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 => + 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 => + 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; } - pendingRestart.delete(threadId); + if (yield* waitsForBackgroundWork(threadId)) { + yield* Effect.logInfo("claude.account.restart-still-deferred", { + threadId, + waitingFor: "background-work", + }); + return; + } + // 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 = 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 +1689,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 +2267,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 +2321,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( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts new file mode 100644 index 0000000000000..bce2ff2fdaf93 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -0,0 +1,430 @@ +// T3-CUSTOM(expbkt3): a released provider session settles the subagents it leaves open. +import { assert, it } from "@effect/vitest"; +import { + NodeId, + type OrchestrationV2DomainEvent, + type OrchestrationV2Subagent, + type OrchestrationV2ThreadProjection, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ProviderThreadId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts"; +import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts"; +import * as EventSink from "./EventSink.ts"; +import * as IdAllocator from "./IdAllocator.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import { + ProviderAdapterProtocolError, + type ProviderAdapterV2Event, + type ProviderAdapterV2Shape, +} from "./ProviderAdapter.ts"; +import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; +import * as ProviderEventIngestor from "./ProviderEventIngestor.ts"; +import * as ProviderSessionManager from "./ProviderSessionManager.ts"; + +const driver = ProviderDriverKind.make("claudeAgent"); +const instanceId = ProviderInstanceId.make("claudeAgent"); +const parentThreadId = ThreadId.make("released-parent-thread"); +const childThreadId = (name: string) => ThreadId.make(`released-child-${name}`); +const releasedSessionId = ProviderSessionId.make("released-session"); +const otherSessionId = ProviderSessionId.make("other-session"); +const otherProviderThreadId = ProviderThreadId.make("other-session-provider-thread"); +const epoch = DateTime.makeUnsafe("1970-01-01T00:00:00Z"); +/** The release happens ten minutes in (TestClock), after `epoch`, before `later`. */ +const later = DateTime.makeUnsafe("1970-01-01T00:20:00Z"); + +type Projection = ProjectionStore.ProjectionRuntimeRecoveryState; + +function subagent( + name: string, + overrides: Partial = {}, +): OrchestrationV2Subagent { + return { + id: NodeId.make(`subagent-${name}`), + threadId: parentThreadId, + runId: null, + parentNodeId: NodeId.make("parent-root"), + origin: "provider_native", + createdBy: "agent", + driver, + providerInstanceId: instanceId, + providerThreadId: null, + childThreadId: childThreadId(name), + nativeTaskRef: null, + prompt: `review ${name}`, + title: null, + model: null, + status: "running", + result: null, + startedAt: epoch, + completedAt: null, + updatedAt: epoch, + ...overrides, + }; +} + +/** The parent's card for a subagent. */ +function subagentItem(task: OrchestrationV2Subagent) { + return { + id: TurnItemId.make(`card-${task.id}`), + threadId: task.threadId, + runId: task.runId, + nodeId: task.id, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: task.status, + title: null, + startedAt: task.startedAt, + completedAt: null, + updatedAt: task.updatedAt, + type: "subagent" as const, + subagentId: task.id, + origin: task.origin, + driver: task.driver, + providerInstanceId: task.providerInstanceId, + childThreadId: task.childThreadId, + prompt: task.prompt, + result: null, + }; +} + +function node( + id: string, + threadId: ThreadId, + kind: "subagent" | "root_turn", + startedAt: DateTime.Utc = epoch, +) { + return { + id: NodeId.make(id), + threadId, + runId: null, + parentNodeId: null, + rootNodeId: NodeId.make(id), + kind, + status: "running" as const, + countsForRun: false, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + runtimeRequestId: null, + checkpointScopeId: null, + startedAt, + completedAt: null, + }; +} + +function projection(threadId: ThreadId, records: Partial): Projection { + return { + thread: { id: threadId }, + runs: [], + attempts: [], + nodes: [], + subagents: [], + providerSessions: [], + providerThreads: [], + providerTurns: [], + runtimeRequests: [], + messages: [], + turnItems: [], + ...records, + } as unknown as Projection; +} + +/** + * The incident's shape: a parent whose turn completed while its background + * subagent kept working in a child thread (a runless root turn with a running + * command), next to subagents the release must leave alone. + */ +function incidentProjections(): ReadonlyMap { + const running = subagent("running"); + const startedLater = subagent("started-later", { startedAt: later, updatedAt: later }); + const otherSession = subagent("other-session", { providerThreadId: otherProviderThreadId }); + const delegated = subagent("delegated", { origin: "app_owned" }); + const otherDriver = subagent("other-driver", { driver: ProviderDriverKind.make("codex") }); + const finished = subagent("finished", { status: "completed", completedAt: epoch }); + const tasks = [running, startedLater, otherSession, delegated, otherDriver, finished]; + const childRoot = node("child-root-running", childThreadId("running"), "root_turn"); + // The last thing the stopped subagent ran, still open under its root turn. + const childCommand = { + id: TurnItemId.make("child-command"), + threadId: childThreadId("running"), + runId: null, + nodeId: childRoot.id, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 2, + status: "running" as const, + title: null, + startedAt: epoch, + completedAt: null, + updatedAt: epoch, + type: "command_execution" as const, + input: "cat notes.md", + }; + return new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: tasks, + turnItems: tasks.map(subagentItem) as never, + nodes: tasks.map((task) => node(task.id, parentThreadId, "subagent")), + providerThreads: [ + { id: otherProviderThreadId, providerSessionId: otherSessionId } as never, + ], + }), + ], + [ + childThreadId("running"), + projection(childThreadId("running"), { + nodes: [childRoot], + turnItems: [childCommand] as never, + }), + ], + // Reached only if the release wrongly settled their subagents. + ...["started-later", "other-session", "delegated", "other-driver"].map( + (name) => + [ + childThreadId(name), + projection(childThreadId(name), { + nodes: [node(`child-root-${name}`, childThreadId(name), "root_turn")], + }), + ] as const, + ), + ]); +} + +function fixture( + options: { + readonly projections?: ReadonlyMap; + readonly failRecoveryRead?: boolean; + } = {}, +) { + const projections = options.projections ?? incidentProjections(); + const written: Array = []; + const adapter: ProviderAdapterV2Shape = { + instanceId, + driver, + getCapabilities: () => Effect.succeed(CodexProviderCapabilitiesV2), + planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }), + openSession: (input) => + Effect.gen(function* () { + const events = yield* Queue.unbounded(); + const unused = () => + Effect.fail(new ProviderAdapterProtocolError({ driver, detail: "unused" })); + return { + instanceId, + driver, + providerSessionId: input.providerSessionId, + providerSession: { + id: input.providerSessionId, + driver, + providerInstanceId: instanceId, + status: "ready" as const, + cwd: "/tmp", + model: "test", + capabilities: CodexProviderCapabilitiesV2, + createdAt: epoch, + updatedAt: epoch, + lastError: null, + }, + events: Stream.fromQueue(events), + ensureThread: unused, + resumeThread: (request) => Effect.succeed(request.providerThread), + forkThread: unused, + readThreadSnapshot: unused, + rollbackThread: unused, + startTurn: () => Effect.void, + steerTurn: () => Effect.void, + interruptTurn: () => Effect.void, + respondToRuntimeRequest: () => Effect.void, + }; + }), + }; + const layer = ProviderSessionManager.layerWithOptions({ + // The clock moves in these tests; idle release must not run on its own. + idleTimeoutMs: 24 * 60 * 60 * 1000, + configureMcp: false, + }).pipe( + Layer.provide( + Layer.mergeAll( + IdAllocator.layer, + FileSystem.layerNoop({}), + ProviderAdapterRegistry.layerSingle(adapter), + Layer.mock(EventSink.EventSinkV2)({ + write: (input) => + Effect.sync(() => { + written.push(...input.events); + return []; + }), + }), + Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({}), + Layer.mock(ProjectionStore.ProjectionStoreV2)({ + getThreadRecords: (threadId) => + Effect.succeed({ + thread: { id: threadId }, + messages: [], + runtimeRequests: [], + nodes: [], + turnItems: [], + providerThreads: [], + providerTurns: [], + } as unknown as OrchestrationV2ThreadProjection), + getRuntimeRecoveryProjection: (threadId) => { + if (options.failRecoveryRead === true) { + return Effect.fail( + new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "boom" }), + ); + } + const found = projections.get(threadId); + return found === undefined + ? Effect.fail(new ProjectionStore.ProjectionStoreThreadNotFoundError({ threadId })) + : Effect.succeed(found); + }, + }), + Layer.mock(McpSessionRegistry.McpSessionRegistry)({ + revokeThread: () => Effect.void, + revokeProviderSession: () => Effect.void, + }), + ), + ), + ); + const open = (threadId: ThreadId, providerSessionId = releasedSessionId) => ({ + threadId, + providerSessionId, + modelSelection: { instanceId, model: "test" }, + runtimePolicy: { cwd: null, runtimeMode: "full-access", interactionMode: "default" } as const, + }); + /** The cancellations the release wrote, by event type. */ + const cancelled = (type: "subagent.updated" | "turn-item.updated" | "node.updated") => + written.flatMap((event) => { + if ( + event.type !== type || + (event.type !== "subagent.updated" && + event.type !== "turn-item.updated" && + event.type !== "node.updated") + ) { + return []; + } + return event.payload.status === "cancelled" ? [String(event.payload.id)] : []; + }); + return { layer, written, open, cancelled }; +} + +it.effect( + "a released session cancels the subagent it left running, its card and its child thread", + () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "manual_shutdown" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + 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"]); + // Started after the release, on another session, a delegate_task child, + // another provider, already finished: none of them is touched. + 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 [ + "started-later", + "other-session", + "delegated", + "other-driver", + "finished", + ]) { + 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 detach that leaves the shared session open settles nothing", () => { + const f = fixture(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.open(f.open(ThreadId.make("released-sibling-thread"))); + yield* TestClock.adjust("10 minutes"); + yield* manager.detach({ providerSessionId: releasedSessionId, threadId: parentThreadId }); + + assert.isTrue(Option.isSome(yield* manager.get(releasedSessionId))); + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("turn-item.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + }).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" }); + + 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(); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "server_shutdown" }); + + assert.deepEqual(f.cancelled("subagent.updated"), []); + assert.deepEqual(f.cancelled("node.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); + +it.effect("a failed subagent read never fails the release", () => { + const f = fixture({ failRecoveryRead: true }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "runtime_error" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + assert.isTrue( + f.written.some( + (event) => event.type === "provider-session.updated" && event.payload.status === "error", + ), + ); + assert.deepEqual(f.cancelled("subagent.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 30f86c8c2f378..49e10783bd416 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -69,6 +69,8 @@ import { import { SourceControlProfileService } from "../sourceControl/SourceControlProfileService.ts"; import type { ProviderSessionExecutionOptions } from "../provider/sessionExecutionOptions.expbkt3.ts"; import { makeBackgroundTurnGuard } from "./backgroundTurnGrant.expbkt3.ts"; +// T3-CUSTOM(expbkt3): a released session settles the subagents it leaves open. +import { settleReleasedSessionSubagents } from "./releasedSessionSubagents.expbkt3.ts"; const DEFAULT_IDLE_TIMEOUT_MS = 30 * 60 * 1000; const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; @@ -925,6 +927,28 @@ export const layerWithOptions = ( input.entry.requestEventPermit.withPermits(1), ), ), + // T3-CUSTOM(expbkt3): BEGIN the released process can no longer end its subagents. + // Never fails, so it never fails or retries the release. + Effect.exit( + settleReleasedSessionSubagents({ + projectionStore, + idAllocator, + eventSink, + providerSessionId: input.entry.runtime.providerSessionId, + reason: input.reason, + driver: input.entry.runtime.driver, + providerInstanceId: input.entry.runtime.instanceId, + threadIds: input.entry.attachedThreadIds, + liveThreadIds: Ref.get(sessions).pipe( + Effect.map( + (current) => + new Set([...current.values()].flatMap((live) => [...live.attachedThreadIds])), + ), + ), + releasedAt: input.releasedAt, + }), + ), + // T3-CUSTOM(expbkt3): END ], { concurrency: 1 }, ).pipe(Effect.flatMap(Exit.asVoidAll)); diff --git a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts new file mode 100644 index 0000000000000..90480ecee502f --- /dev/null +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -0,0 +1,285 @@ +// T3-CUSTOM(expbkt3): a released provider session settles the subagents it leaves open. +import type { + NodeId, + OrchestrationV2DomainEvent, + OrchestrationV2Subagent, + OrchestrationV2ThreadProjection, + ProviderDriverKind, + ProviderInstanceId, + ProviderSessionId, + ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; + +import type { EventSinkV2Shape } from "./EventSink.ts"; +import type { IdAllocatorV2Shape } from "./IdAllocator.ts"; +import type { ProjectionRuntimeRecoveryState, ProjectionStoreV2Shape } from "./ProjectionStore.ts"; +import type { ProviderSessionReleaseReason } from "./ProviderSessionManager.ts"; + +type TurnItem = OrchestrationV2ThreadProjection["turnItems"][number]; +type ExecutionNode = OrchestrationV2ThreadProjection["nodes"][number]; + +export interface ReleasedSessionSubagentsInput { + readonly projectionStore: ProjectionStoreV2Shape; + readonly idAllocator: IdAllocatorV2Shape; + readonly eventSink: EventSinkV2Shape; + readonly providerSessionId: ProviderSessionId; + readonly reason: ProviderSessionReleaseReason; + readonly driver: ProviderDriverKind; + readonly providerInstanceId: ProviderInstanceId; + /** The app threads the released process served. */ + readonly threadIds: Iterable; + /** App threads a live session still serves. That session owns their subagents. */ + readonly liveThreadIds: Effect.Effect>; + /** Work that started after the release belongs to a later session. */ + readonly releasedAt: DateTime.Utc; +} + +const isOpenStatus = (status: string): boolean => + status === "pending" || status === "running" || status === "waiting"; + +/** Same as startup recovery: a closed card also closes its form and stops streaming. */ +const cancelledItem = (item: TurnItem, now: DateTime.Utc): TurnItem => + item.type === "secret_request" && item.secretStatus === "pending" + ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now } + : { + ...item, + status: "cancelled", + completedAt: now, + updatedAt: now, + ...(item.type === "reasoning" || item.type === "assistant_message" + ? { streaming: false } + : {}), + }; + +/** + * Settles the provider-native subagents a released provider session leaves open. + * + * A subagent runs inside its parent's provider process. When that process goes + * away between turns (an account rebalance or mode change, a hard limit, a + * crash), nothing reports the subagent's end: its record, the parent's card and + * its child thread stay running until a server restart. Startup recovery + * (ProviderRuntimeRecoveryService) settles them after a restart; this settles + * them for one released session, with the same events and the same reads. + * + * Only open `provider_native` subagents of the released driver and instance, + * started at or before the release, are settled. A delegate_task child has its + * own runs and is never touched, and neither is a thread a live session still + * serves. A later SendMessage resume reopens a settled subagent: its new events + * overwrite the cancelled status. + * + * A server shutdown is left to startup recovery, which also wakes a settled + * run whose background work the restart cancelled; settled here, that work + * would no longer look lost to it. + * + * Never fails: the release must complete either way, and startup recovery is + * the fallback for a write that did not land. + */ +export const settleReleasedSessionSubagents = ( + input: ReleasedSessionSubagentsInput, +): Effect.Effect => + Effect.gen(function* () { + if (input.reason === "server_shutdown") return; + const liveThreadIds = yield* input.liveThreadIds; + const now = yield* DateTime.now; + const events: Array = []; + const settledSubagentIds: Array = []; + const allocateEventId = (threadId: ThreadId) => + input.idAllocator.allocate.event({ threadId, providerSessionId: input.providerSessionId }); + const startedByRelease = (startedAt: DateTime.Utc | null, fallback: DateTime.Utc | null) => { + const at = startedAt ?? fallback; + return at === null || DateTime.isLessThanOrEqualTo(at, input.releasedAt); + }; + const ranInReleasedSession = ( + subagent: OrchestrationV2Subagent, + projection: ProjectionRuntimeRecoveryState, + ) => { + if ( + subagent.origin !== "provider_native" || + !isOpenStatus(subagent.status) || + subagent.driver !== input.driver || + subagent.providerInstanceId !== input.providerInstanceId || + !startedByRelease(subagent.startedAt, subagent.updatedAt) + ) { + return false; + } + // Claude subagents carry no provider thread; their parent thread's + // attachment is the link. One that does must not name another session. + if (subagent.providerThreadId === null) return true; + const providerThread = projection.providerThreads.find( + (candidate) => candidate.id === subagent.providerThreadId, + ); + return ( + providerThread?.providerSessionId == null || + providerThread.providerSessionId === input.providerSessionId + ); + }; + + const settleNode = Effect.fnUntraced(function* (node: ExecutionNode) { + events.push({ + id: yield* allocateEventId(node.threadId), + type: "node.updated", + threadId: node.threadId, + ...(node.runId === null ? {} : { runId: node.runId }), + nodeId: node.id, + driver: input.driver, + providerInstanceId: input.providerInstanceId, + occurredAt: now, + payload: { ...node, status: "cancelled", completedAt: now }, + }); + }); + const settleItem = Effect.fnUntraced(function* (item: TurnItem) { + events.push({ + id: yield* allocateEventId(item.threadId), + type: "turn-item.updated", + threadId: item.threadId, + ...(item.runId === null ? {} : { runId: item.runId }), + ...(item.nodeId === null ? {} : { nodeId: item.nodeId }), + driver: input.driver, + providerInstanceId: input.providerInstanceId, + occurredAt: now, + payload: cancelledItem(item, now), + }); + }); + + /** + * Settles one thread's subagents, and for a provider-native child thread its + * runless root turn and the items under it. Returns the child threads of the + * subagents it settled, whose own work died with the same process. + */ + const settleThread = Effect.fnUntraced(function* (threadId: ThreadId, isChild: boolean) { + // Most releases (idle ones) leave nothing open. The subagent rows alone + // are cheap; the recovery read below can take seconds on a cold cache + // for a thread with thousands of items, on the event loop. A failed + // check falls through to the full read. + if (!isChild) { + const records = yield* input.projectionStore + .getThreadRecords(threadId, ["subagents"]) + .pipe(Effect.option); + if ( + records._tag === "Some" && + !records.value.subagents.some( + (subagent) => + subagent.origin === "provider_native" && + isOpenStatus(subagent.status) && + subagent.driver === input.driver && + subagent.providerInstanceId === input.providerInstanceId, + ) + ) { + return []; + } + } + // The thread's open subagents, nodes and items, as startup recovery reads them. + const projection = yield* input.projectionStore + .getRuntimeRecoveryProjection(threadId) + .pipe(Effect.catchTag("ProjectionStoreThreadNotFoundError", () => Effect.succeed(null))); + // A deleted thread has nothing left to show as running. + if (projection === null) return []; + const settledNodeIds = new Set(); + const settledItemIds = new Set(); + const childThreadIds: Array = []; + for (const subagent of projection.subagents) { + if (!ranInReleasedSession(subagent, projection)) continue; + settledSubagentIds.push(subagent.id); + events.push({ + id: yield* allocateEventId(subagent.threadId), + type: "subagent.updated", + threadId: subagent.threadId, + ...(subagent.runId === null ? {} : { runId: subagent.runId }), + nodeId: subagent.id, + driver: subagent.driver, + providerInstanceId: subagent.providerInstanceId, + occurredAt: now, + payload: { ...subagent, status: "cancelled", completedAt: now, updatedAt: now }, + }); + // The parent's card for this subagent. + for (const item of projection.turnItems) { + if ( + item.type !== "subagent" || + item.subagentId !== subagent.id || + !isOpenStatus(item.status) || + settledItemIds.has(item.id) + ) { + continue; + } + settledItemIds.add(item.id); + yield* settleItem(item); + } + const node = projection.nodes.find( + (candidate) => candidate.id === subagent.id && isOpenStatus(candidate.status), + ); + if (node !== undefined && !settledNodeIds.has(node.id)) { + settledNodeIds.add(node.id); + yield* settleNode(node); + } + if (subagent.childThreadId !== null) childThreadIds.push(subagent.childThreadId); + } + if (isChild) { + // A provider-native child thread has no runs: its work is a runless root + // turn (shown as Working) and the items under it. + for (const node of projection.nodes) { + if ( + node.kind !== "root_turn" || + node.runId !== null || + !isOpenStatus(node.status) || + settledNodeIds.has(node.id) || + !startedByRelease(node.startedAt, null) + ) { + continue; + } + settledNodeIds.add(node.id); + yield* settleNode(node); + for (const item of projection.turnItems) { + if ( + item.nodeId !== node.id || + item.runId !== null || + !isOpenStatus(item.status) || + settledItemIds.has(item.id) + ) { + continue; + } + settledItemIds.add(item.id); + yield* settleItem(item); + } + } + } + return childThreadIds; + }); + + const visited = new Set(); + let pending: Array<{ readonly threadId: ThreadId; readonly isChild: boolean }> = []; + for (const threadId of input.threadIds) { + if (!liveThreadIds.has(threadId)) pending.push({ threadId, isChild: false }); + } + while (pending.length > 0) { + const next: typeof pending = []; + for (const { threadId, isChild } of pending) { + if (visited.has(threadId)) continue; + visited.add(threadId); + const childThreadIds = yield* settleThread(threadId, isChild); + for (const childThreadId of childThreadIds) { + if (!liveThreadIds.has(childThreadId)) + next.push({ threadId: childThreadId, isChild: true }); + } + } + pending = next; + } + + if (events.length === 0) return; + yield* input.eventSink.write({ events }); + yield* Effect.logInfo("orchestration-v2.provider-session-release-subagents-cancelled", { + providerSessionId: input.providerSessionId, + subagentIds: settledSubagentIds, + }); + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("orchestration-v2.provider-session-release-subagents-failed", { + providerSessionId: input.providerSessionId, + cause: Cause.pretty(cause), + }), + ), + ); diff --git a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts index aca78e36042aa..dfae2b42b5cbd 100644 --- a/apps/server/src/provider/ForkProviderSessions.expbkt3.ts +++ b/apps/server/src/provider/ForkProviderSessions.expbkt3.ts @@ -6,6 +6,7 @@ import { type ProviderSession, type ThreadId, } from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; @@ -37,6 +38,19 @@ export class ForkProviderSessions extends Context.Service< ForkProviderSessions, { readonly stopSession: (input: { readonly threadId: ThreadId }) => Effect.Effect; + /** + * Whether a live session of the thread still runs background work between + * turns: a subagent, a background command or monitor, or a wake not yet + * delivered. Stopping that session ends the work. A failed check answers + * true, as the session manager's own configuration refresh does: a restart + * can wait, but killed work does not come back. + * + * Each check counts as activity on the thread's own sessions (their idle + * release starts again), so poll it only for threads that wait on it. + */ + readonly hasLiveBackgroundWork: (input: { + readonly threadId: ThreadId; + }) => Effect.Effect; readonly listSessions: () => Effect.Effect>; readonly streamEvents: Stream.Stream; } @@ -70,6 +84,26 @@ export const layer = Layer.effect( yield* sessions.release({ providerSessionId, reason: "manual_shutdown" }); } }).pipe(Effect.orDie), + hasLiveBackgroundWork: ({ threadId }) => + Effect.gen(function* () { + const { providerThreads } = yield* projection.getThreadRecords(threadId, [ + "providerThreads", + ]); + for (const providerSessionId of new Set( + providerThreads.flatMap((thread) => + thread.providerSessionId === null ? [] : [thread.providerSessionId], + ), + )) { + const live = yield* sessions.get(providerSessionId); + if (live._tag === "None" || live.value.hasPendingBackgroundWork === undefined) continue; + if (yield* live.value.hasPendingBackgroundWork) return true; + } + return false; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) ? Effect.interrupt : Effect.succeed(true), + ), + ), listSessions: () => Effect.gen(function* () { const shell = yield* projection.getShellSnapshot(); From c2876b338bd86bded84f6dbf3c51dca37d753ca3 Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Tue, 20 Oct 2026 13:06:42 +0000 Subject: [PATCH 2/3] fix(server): lint and test fake for the released-session subagent check Use Effect.catchTags (t3code/prefer-catch-tags), give the test store's getThreadRecords the projection's subagents, and cover an idle release with no open subagent skipping the full recovery read. Co-Authored-By: Claude Opus 5.5 --- ...nManager.releasedSubagents.expbkt3.test.ts | 31 ++++++++++++++++++- .../releasedSessionSubagents.expbkt3.ts | 2 +- 2 files changed, 31 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts index bce2ff2fdaf93..f2d6203054941 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -219,6 +219,8 @@ function fixture( ) { const projections = options.projections ?? incidentProjections(); const written: Array = []; + /** Threads whose full recovery read ran, in order. */ + const recoveryReads: Array = []; const adapter: ProviderAdapterV2Shape = { instanceId, driver, @@ -286,8 +288,10 @@ function fixture( turnItems: [], providerThreads: [], providerTurns: [], + subagents: projections.get(threadId)?.subagents ?? [], } as unknown as OrchestrationV2ThreadProjection), getRuntimeRecoveryProjection: (threadId) => { + recoveryReads.push(threadId); if (options.failRecoveryRead === true) { return Effect.fail( new ProjectionStore.ProjectionStoreReadError({ threadId, cause: "boom" }), @@ -325,7 +329,7 @@ function fixture( } return event.payload.status === "cancelled" ? [String(event.payload.id)] : []; }); - return { layer, written, open, cancelled }; + return { layer, written, recoveryReads, open, cancelled }; } it.effect( @@ -412,6 +416,31 @@ it.effect("a server shutdown leaves the subagents to startup recovery and its wa }).pipe(Effect.provide(f.layer)); }); +it.effect("a release with no open subagent skips the full recovery read", () => { + const finished = subagent("finished", { status: "completed", completedAt: epoch }); + const f = fixture({ + projections: new Map([ + [ + parentThreadId, + projection(parentThreadId, { + subagents: [finished], + turnItems: [subagentItem(finished)] as never, + }), + ], + ]), + }); + return Effect.gen(function* () { + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + yield* manager.open(f.open(parentThreadId)); + yield* TestClock.adjust("10 minutes"); + yield* manager.release({ providerSessionId: releasedSessionId, reason: "idle_timeout" }); + + assert.isTrue(Option.isNone(yield* manager.get(releasedSessionId))); + assert.deepEqual(f.recoveryReads, []); + assert.deepEqual(f.cancelled("subagent.updated"), []); + }).pipe(Effect.provide(f.layer)); +}); + it.effect("a failed subagent read never fails the release", () => { const f = fixture({ failRecoveryRead: true }); return Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts index 90480ecee502f..5fbe0f31cc9f3 100644 --- a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -174,7 +174,7 @@ export const settleReleasedSessionSubagents = ( // The thread's open subagents, nodes and items, as startup recovery reads them. const projection = yield* input.projectionStore .getRuntimeRecoveryProjection(threadId) - .pipe(Effect.catchTag("ProjectionStoreThreadNotFoundError", () => Effect.succeed(null))); + .pipe(Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) })); // A deleted thread has nothing left to show as running. if (projection === null) return []; const settledNodeIds = new Set(); From ffa176f511d0700f664dacaf4a6898b3f4c4ffce Mon Sep 17 00:00:00 2001 From: bk-agent-01 Date: Tue, 20 Oct 2026 13:09:44 +0000 Subject: [PATCH 3/3] fix(server): review follow-ups for released-session subagents and restart waits - Skip a subagent whose provider thread the recovery read did not load: it may still run in another shared session. - A completePendingRestart fiber that lost the race no longer leaves a stale background-wait start behind. Co-Authored-By: Claude Opus 5.5 --- .../claudeAccounts/ClaudeAccountsService.ts | 6 ++++++ ...nManager.releasedSubagents.expbkt3.test.ts | 20 ++++++++++++++++--- .../releasedSessionSubagents.expbkt3.ts | 7 +++++-- 3 files changed, 28 insertions(+), 5 deletions(-) diff --git a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts index 267ce91dd9b09..b542a7ed4d710 100644 --- a/apps/server/src/claudeAccounts/ClaudeAccountsService.ts +++ b/apps/server/src/claudeAccounts/ClaudeAccountsService.ts @@ -1485,6 +1485,12 @@ export const make = Effect.gen(function* () { 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", diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts index f2d6203054941..b1c892f4c90fc 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -158,7 +158,19 @@ function incidentProjections(): ReadonlyMap { const delegated = subagent("delegated", { origin: "app_owned" }); const otherDriver = subagent("other-driver", { driver: ProviderDriverKind.make("codex") }); const finished = subagent("finished", { status: "completed", completedAt: epoch }); - const tasks = [running, startedLater, otherSession, delegated, otherDriver, finished]; + // Its provider thread was not loaded: it may run in another shared session. + const unloadedThread = subagent("unloaded-thread", { + providerThreadId: ProviderThreadId.make("unloaded-provider-thread"), + }); + const tasks = [ + running, + startedLater, + otherSession, + delegated, + otherDriver, + finished, + unloadedThread, + ]; const childRoot = node("child-root-running", childThreadId("running"), "root_turn"); // The last thing the stopped subagent ran, still open under its root turn. const childCommand = { @@ -199,7 +211,7 @@ function incidentProjections(): ReadonlyMap { }), ], // Reached only if the release wrongly settled their subagents. - ...["started-later", "other-session", "delegated", "other-driver"].map( + ...["started-later", "other-session", "delegated", "other-driver", "unloaded-thread"].map( (name) => [ childThreadId(name), @@ -350,7 +362,8 @@ it.effect( ]); assert.sameMembers(f.cancelled("node.updated"), ["subagent-running", "child-root-running"]); // Started after the release, on another session, a delegate_task child, - // another provider, already finished: none of them is touched. + // another provider, already finished, on an unloaded provider thread: + // none of them is touched. const touched = f.written.flatMap((event) => event.type === "subagent.updated" || event.type === "turn-item.updated" || @@ -364,6 +377,7 @@ it.effect( "delegated", "other-driver", "finished", + "unloaded-thread", ]) { assert.notInclude(touched, `subagent-${name}`); assert.notInclude(touched, `card-subagent-${name}`); diff --git a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts index 5fbe0f31cc9f3..bb4ed0f661a61 100644 --- a/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -111,9 +111,12 @@ export const settleReleasedSessionSubagents = ( const providerThread = projection.providerThreads.find( (candidate) => candidate.id === subagent.providerThreadId, ); + // The recovery read loads only live-looking provider threads: one it did + // not load may belong to another shared session that still runs it. return ( - providerThread?.providerSessionId == null || - providerThread.providerSessionId === input.providerSessionId + providerThread !== undefined && + (providerThread.providerSessionId == null || + providerThread.providerSessionId === input.providerSessionId) ); };