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..b542a7ed4d710 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,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 => + 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; + } + 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 = 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( 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..b1c892f4c90fc --- /dev/null +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.releasedSubagents.expbkt3.test.ts @@ -0,0 +1,473 @@ +// 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 }); + // 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 = { + 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", "unloaded-thread"].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 = []; + /** Threads whose full recovery read ran, in order. */ + const recoveryReads: 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: [], + 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" }), + ); + } + 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, recoveryReads, 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, 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" || + event.type === "node.updated" + ? [String(event.payload.id)] + : [], + ); + for (const name of [ + "started-later", + "other-session", + "delegated", + "other-driver", + "finished", + "unloaded-thread", + ]) { + 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 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* () { + 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..bb4ed0f661a61 --- /dev/null +++ b/apps/server/src/orchestration-v2/releasedSessionSubagents.expbkt3.ts @@ -0,0 +1,288 @@ +// 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, + ); + // 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 !== undefined && + (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.catchTags({ 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();