diff --git a/docs/design/background-agent-runtime-generations.md b/docs/design/background-agent-runtime-generations.md new file mode 100644 index 00000000000..46e09186ccb --- /dev/null +++ b/docs/design/background-agent-runtime-generations.md @@ -0,0 +1,15 @@ +# Background Agent runtime generations + +## Problem + +A background Agent that does not settle after cancellation can leave its ACP child usable enough to answer transport probes but unsafe for fresh work. Replacing that child must not move its existing Sessions, create unbounded children, or route new work back to the draining generation. + +## Design + +Each ACP bridge channel has one of three states: `active`, `draining`, or `dying`. Existing Session entries continue to route through their recorded channel while it drains. Existing timeout retirement paths and explicit recycle requests mark only the affected generation as draining; fresh work then creates a new active generation. + +The bridge keeps at most two OS-live generations. If both slots are occupied and neither can accept fresh work, admission fails with `503 runtime_recycling` until an older generation exits. Dying generations remain tracked until process exit so synchronous shutdown can still reach them. + +After a logical watchdog abort, the Agent gets a fixed five-second cooperative exit window. If it still has not settled, its registry entry and sidecar become failed once while the underlying run keeps its concurrency slot. The terminal notification is recorded and displayed without starting another model turn, then the trusted child-to-daemon route requests recycle for the Session's owner generation. A late Agent settlement releases the physical slot but cannot replace the failed terminal state. + +This changes no persisted Session format and adds no public timeout configuration. diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 2ee3ce3fef6..e2763f99616 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -122,6 +122,7 @@ import { InvalidRewindTargetError, PromptDeadlineExceededError, BridgeChannelQuarantinedError, + BridgeRuntimeRecyclingError, McpAuthenticationInProgressError, StandaloneSessionSpawnError, } from './bridgeErrors.js'; @@ -1055,9 +1056,10 @@ interface ChannelInfo { * `killAllSync` must still find the channel during the SIGTERM * grace window to fire SIGKILL on `process.exit(1)`. `aliveChannels` * holds the dying entry until `channel.exited` fires (OS-level - * reap); `isDying` is the "available-for-new-spawns" half of the - * two-bit (alive, dying) state. + * reap). Draining generations retain their Session owners but cannot accept + * fresh work; dying generations are unavailable while the OS reaps them. */ + state: 'active' | 'draining' | 'dying'; isDying: boolean; /** Existing sessions stay usable, but no fresh session work may enter. */ isQuarantined: boolean; @@ -2749,7 +2751,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { | { channel: ChannelInfo; reason: BridgeChannelUnavailableReason } | undefined => { for (const ci of aliveChannels) { - if (ci.isDying) continue; + if (ci.state !== 'active') continue; if (ci.isQuarantined) { return { channel: ci, reason: 'restore_cleanup_failed' }; } @@ -3004,8 +3006,8 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // same-workspace attach under `single` scope reuses). Thread-scope // sessions add to `byId` but don't displace `defaultEntry`. let defaultEntry: SessionEntry | undefined; - // `channelInfo` is the SINGLE attach-available channel. Cleared - // ONLY by the `channel.exited` handler (see below) when the OS + // `channelInfo` is the newest generation. It is attach-available only while + // active, and is cleared ONLY by its `channel.exited` handler when the OS // reaps the underlying child process. Teardown initiators // (`killSession` last-session-leaving — via `startIdleTimer` -> // `killChannelWithLog` / `reapPendingEmptyChannel`, @@ -3694,17 +3696,36 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ci: ChannelInfo, context: string, ): Promise { - if (ci.isDying) return; + if (ci.state === 'dying') return; if (hasNoSessionWork(ci)) { await killChannelWithLog(ci, context); return; } + if (ci.state === 'draining') return; + ci.state = 'draining'; + if (channelInfo === ci) cancelIdleTimer(); ci.retireWhenSessionsDrain = true; writeStderrLine( `qwen serve: ${context}; deferring channel retirement until ${ci.sessionIds.size} active session(s) drain`, ); } + async function requestRuntimeRecycleForSession( + sessionId: string, + ): Promise { + const entry = byId.get(sessionId); + if (!entry) throw new SessionNotFoundError(sessionId); + const owner = channelInfoForEntry(entry); + if (!owner || owner.state === 'dying') { + throw new SessionNotFoundError(sessionId); + } + await retireChannelAfterSessionsDrain( + owner, + `runtime recycle requested by session ${JSON.stringify(sessionId)}`, + ); + if (!owner.isDying) await ensureChannel(); + } + async function retireChannelOnTimeout( ci: ChannelInfo, error: unknown, @@ -4493,8 +4514,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // would either hang or land the caller with a sessionId that // immediately 404s on every follow-up. cancelIdleTimer(); - if (channelInfo && !channelInfo.isDying) return channelInfo; + if (channelInfo?.state === 'active') return channelInfo; if (inFlightChannelSpawn) return await inFlightChannelSpawn; + if (aliveChannels.size >= 2) throw new BridgeRuntimeRecyclingError(); const promise = (async () => { const privateParentCapability = randomBytes(32).toString('base64url'); @@ -4554,7 +4576,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // instead of throwing. Surface that ambiguity loudly. (sessionId) => { if (sessionId) return byId.get(sessionId); - if (channelInfo && channelInfo.sessionIds.size > 1) { + if (sessionIds.size > 1) { throw new Error( 'BridgeClient: ACP call without sessionId on a ' + 'multi-session channel cannot be routed — workspace=' + @@ -4648,9 +4670,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { .catch(() => undefined); }, opts.onChannelDelivery, - () => - channelInfo?.sessionIds === sessionIds && - channelInfo.sessionSpawnsInFlight > 0, + () => (infoRef.current?.sessionSpawnsInFlight ?? 0) > 0, () => liveScreenContextCaptureHandler, () => liveTaskToolRequestHandler, () => liveSpeakToUserHandler, @@ -4667,6 +4687,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // nothing else would settle what its last drain missed. settleMidTurnQueueAfterGoalTurn, opts.onCreateCurrentSessionScheduledTask, + requestRuntimeRecycleForSession, ); const rawConnection = new ClientSideConnection( () => @@ -4738,7 +4759,13 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { newSessionCleanupFailed: false, transportFailed: false, transportFailureInitiatedTeardown: false, - isDying: false, + state: 'active', + get isDying() { + return this.state === 'dying'; + }, + set isDying(value) { + if (value) this.state = 'dying'; + }, isQuarantined: false, handshakeComplete: false, }; @@ -5128,10 +5155,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }), onFailure: failChannelLiveness, isActive: () => - channelInfo === info && - aliveChannels.has(info) && - !info.isDying && - !shuttingDown, + aliveChannels.has(info) && !info.isDying && !shuttingDown, }); } telemetry.metrics?.channelLifecycle('spawn'); @@ -5348,7 +5372,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // `ensureChannel`, never spawning a fresh one. Tear down the // empty channel so the next attempt gets a clean spawn. const channelPath = - channelInfo && !channelInfo.isDying + channelInfo?.state === 'active' ? 'reused' : inFlightChannelSpawn ? 'joined' @@ -6107,7 +6131,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }; const liveChannelInfo = (): ChannelInfo | undefined => { - if (!channelInfo || channelInfo.isDying) return undefined; + if (channelInfo?.state !== 'active') return undefined; return channelInfo; }; @@ -9430,7 +9454,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }, isChannelLive() { - return !!liveChannelInfo(); + return liveChannelInfo() !== undefined; }, getWorkspaceRuntimeLifecycleSnapshot() { @@ -9447,7 +9471,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } const starting = inFlightChannelSpawn !== undefined; const stopping = Array.from(aliveChannels).some( - (candidate) => candidate.isDying, + (candidate) => candidate.state !== 'active', ); const reservedWork = runtimeOperationReservations > 0 || @@ -9476,6 +9500,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }; }, + async requestRuntimeRecycle(sessionId) { + await requestRuntimeRecycleForSession(sessionId); + }, + get pendingPermissionCount() { return permissionMediator.pendingCount; }, diff --git a/packages/acp-bridge/src/bridgeClient.ts b/packages/acp-bridge/src/bridgeClient.ts index 8675b4d9d5d..5838f87f862 100644 --- a/packages/acp-bridge/src/bridgeClient.ts +++ b/packages/acp-bridge/src/bridgeClient.ts @@ -915,6 +915,8 @@ export class BridgeClient implements Client { */ private readonly onGoalTurnEnded?: (sessionId: string) => void, private readonly onCreateCurrentSessionScheduledTask?: CurrentSessionScheduledTaskCreateHandler, + /** Owner-scoped runtime recycle; wired only by the managed daemon bridge. */ + private readonly onRuntimeRecycle?: (sessionId: string) => Promise, ) {} async requestPermission( @@ -1314,7 +1316,9 @@ export class BridgeClient implements Client { * `qwen/control/client_mcp/message` (reverse tool channel), * `qwen/control/create-sub-session` (the `create_sub_session` tool → daemon * spawns a sub-session and, for `'first-turn'`, returns its first-turn - * result), and `craft/drainMidTurnQueue`: the ACP child calls the last one + * result), `qwen/control/session/runtime/recycle` (trusted owner-generation + * recycle after an Agent ignores abort), and `craft/drainMidTurnQueue`: the + * ACP child calls the last one * between tool batches to pull any messages the browser queued mid-turn. We splice the per-session * queue, return them to the child as the response, and — when non-empty — * publish a `mid_turn_message_injected` SSE frame so the browser can move @@ -1328,6 +1332,23 @@ export class BridgeClient implements Client { method: string, params: Record, ): Promise> { + if (method === SERVE_CONTROL_EXT_METHODS.sessionRuntimeRecycle) { + if (!this.onRuntimeRecycle) throw RequestError.methodNotFound(method); + const sessionId = params['sessionId']; + if ( + typeof sessionId !== 'string' || + !this.ownsSession(sessionId) || + !this.resolveEntry(sessionId) || + params['reason'] !== 'unresponsive_agent' + ) { + throw RequestError.invalidParams( + undefined, + 'Invalid unresponsive Agent runtime recycle request.', + ); + } + await this.onRuntimeRecycle(sessionId); + return { accepted: true }; + } // Reverse tool channel (issue #5626, Phase 2): the child's session // `McpClientManager` routes a client-hosted MCP server's // `sendSdkMcpMessage` UP to the parent through this method. We hand the diff --git a/packages/acp-bridge/src/bridgeErrors.ts b/packages/acp-bridge/src/bridgeErrors.ts index bb3aec22ef2..f9f84fe48c0 100644 --- a/packages/acp-bridge/src/bridgeErrors.ts +++ b/packages/acp-bridge/src/bridgeErrors.ts @@ -673,6 +673,17 @@ export class BridgeChannelQuarantinedError extends Error { } } +export class BridgeRuntimeRecyclingError extends Error { + readonly code = 'runtime_recycling'; + + constructor() { + super( + 'The ACP runtime is recycling; retry after an older generation exits', + ); + this.name = 'BridgeRuntimeRecyclingError'; + } +} + export class InvalidRewindTargetError extends Error { readonly sessionId: string; constructor(sessionId: string, message?: string) { diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 6b4ed18043e..0675868e8a8 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -2448,13 +2448,10 @@ export interface AcpSessionBridge extends WorkspaceEventBridge { readonly sessionCount: number; /** - * Whether an ACP channel is currently live (spawned and not dying). - * Distinct from `sessionCount > 0`: a channel can be live with zero - * attached sessions during the cold-spawn window, and conversely a - * killed channel may briefly retain sessions before reaping. Consumers - * that need true channel liveness (e.g. the workspace service's - * `acpChannelLive` envelope field) must use this rather than the - * session count. + * Whether an ACP channel is active and can accept fresh workspace work. + * Distinct from `sessionCount > 0`: a channel can be active with zero + * attached sessions during the cold-spawn window, while a draining + * generation still owns existing sessions but cannot accept new work. */ isChannelLive(): boolean; @@ -2465,6 +2462,12 @@ export interface AcpSessionBridge extends WorkspaceEventBridge { */ getWorkspaceRuntimeLifecycleSnapshot?(): BridgeWorkspaceRuntimeLifecycleSnapshot; + /** + * Stop admitting fresh work to the generation that owns `sessionId` and + * prepare a replacement without moving existing Sessions between children. + */ + requestRuntimeRecycle?(sessionId: string): Promise; + /** Number of sessions with an active prompt. */ readonly activePromptCount: number; diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index 2a5031a8b70..e01a0f28bd9 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -191,6 +191,8 @@ export const SERVE_CONTROL_EXT_METHODS = { sessionLiveConversation: 'qwen/control/session/live-conversation', sessionLiveTranscript: 'qwen/control/session/live-transcript', sessionBackgroundNotification: 'qwen/control/session/background_notification', + /** Private child→daemon request for an abort-ignoring background Agent. */ + sessionRuntimeRecycle: 'qwen/control/session/runtime/recycle', sessionArtifactsPersist: 'qwen/control/session/artifacts/persist', workspaceMcpRestart: 'qwen/control/workspace/mcp/restart', workspaceMcpManage: 'qwen/control/workspace/mcp/manage', diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index 88f6f96864e..b5edc79addf 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -9505,14 +9505,12 @@ export class Session implements SessionContext { (entry ? buildBackgroundEntryLabel(entry, { includePrefix: false }) : undefined); - this.#enqueueBackgroundNotification({ + const item: BackgroundNotificationQueueItem = { displayText, modelText, taskId: meta.agentId, status: meta.status, kind: 'agent', - continuesTodoStopGuardWorkChain: - this.#agentContinuesTodoStopGuardWorkChain(meta.agentId), toolUseId: meta.toolUseId, todoWorkChainId: meta.todoWorkChainId, label: label ? truncateNotificationLabel(label) : undefined, @@ -9523,6 +9521,15 @@ export class Session implements SessionContext { ), } : undefined, + }; + if (meta.recordOnly) { + void this.#recordUnresponsiveAgentNotification(item); + return; + } + this.#enqueueBackgroundNotification({ + ...item, + continuesTodoStopGuardWorkChain: + this.#agentContinuesTodoStopGuardWorkChain(meta.agentId), }); }, ); @@ -9710,6 +9717,42 @@ export class Session implements SessionContext { void this.#drainNotificationQueue(); } + async #recordUnresponsiveAgentNotification( + item: BackgroundNotificationQueueItem, + ): Promise { + this.activeNotificationAcceptances.add(item.taskId); + this.#activeWorkChanged(); + try { + const accepted = await this.#persistDaemonBackgroundNotification( + item, + false, + ); + if (accepted && !this.disposed && !this.closing) { + await this.#emitBackgroundNotificationDisplay(item); + } + } catch (error) { + debugLogger.warn( + `Unresponsive Agent notification failed [session ${this.sessionId}, task ${item.taskId}]: ${this.#formatError(error)}`, + ); + } finally { + try { + await this.client.extMethod( + SERVE_CONTROL_EXT_METHODS.sessionRuntimeRecycle, + { + sessionId: this.sessionId, + reason: 'unresponsive_agent', + }, + ); + } catch (error) { + debugLogger.warn( + `Unresponsive Agent runtime recycle failed [session ${this.sessionId}, task ${item.taskId}]: ${this.#formatError(error)}`, + ); + } + this.activeNotificationAcceptances.delete(item.taskId); + this.#activeWorkChanged(); + } + } + async enqueueBackgroundNotification( item: BackgroundNotificationQueueItem, ): Promise<{ accepted: boolean }> { @@ -9742,6 +9785,7 @@ export class Session implements SessionContext { async #persistDaemonBackgroundNotification( item: BackgroundNotificationQueueItem, + enqueue = true, ): Promise { if (this.disposed || this.closing) return false; const recording = this.config.getChatRecordingService(); @@ -9766,7 +9810,7 @@ export class Session implements SessionContext { } this.persistedBackgroundNotificationTaskIds.add(item.taskId); - if (!this.disposed && !this.closing) { + if (enqueue && !this.disposed && !this.closing) { this.#enqueueBackgroundNotification({ ...item, continuesTodoStopGuardWorkChain: diff --git a/packages/cli/src/nonInteractiveCli.ts b/packages/cli/src/nonInteractiveCli.ts index f53bf6f4b5b..2ec259e1a5e 100644 --- a/packages/cli/src/nonInteractiveCli.ts +++ b/packages/cli/src/nonInteractiveCli.ts @@ -1489,7 +1489,7 @@ export async function runNonInteractive( // tool-call chain can push completions onto the queue. const registry = config.getBackgroundTaskRegistry(); registry.setNotificationCallback((displayText, modelText, meta) => { - localQueue.push({ + const item: LocalQueueItem = { displayText, modelText, sendMessageType: SendMessageType.Notification, @@ -1506,7 +1506,12 @@ export async function runNonInteractive( } : undefined, }, - }); + }; + if (meta.recordOnly) { + emitNotificationToSdk(item); + return; + } + localQueue.push(item); }); registry.setRegisterCallback((entry) => { diff --git a/packages/cli/src/serve/acp-http/dispatch.ts b/packages/cli/src/serve/acp-http/dispatch.ts index 23a45f8abfa..9d8750dcbdc 100644 --- a/packages/cli/src/serve/acp-http/dispatch.ts +++ b/packages/cli/src/serve/acp-http/dispatch.ts @@ -41,6 +41,7 @@ import { } from '../acp-session-bridge.js'; import type { BridgeChannelQuarantinedError, + BridgeRuntimeRecyclingError, BridgeTimeoutError, RestoreInProgressError, SessionRestoreTimeoutError, @@ -910,6 +911,19 @@ export function toRpcError(err: unknown): { }, }; } + case 'BridgeRuntimeRecyclingError': { + const recyclingError = err as BridgeRuntimeRecyclingError; + return { + code: RPC.INTERNAL_ERROR, + message: recyclingError.message, + data: { + code: recyclingError.code, + errorKind: recyclingError.code, + httpStatus: 503, + retryable: true, + }, + }; + } case 'SessionArchivedError': return { code: RPC.INTERNAL_ERROR, diff --git a/packages/cli/src/serve/acp-session-bridge.ts b/packages/cli/src/serve/acp-session-bridge.ts index 3277a244be7..2b43604b46f 100644 --- a/packages/cli/src/serve/acp-session-bridge.ts +++ b/packages/cli/src/serve/acp-session-bridge.ts @@ -141,6 +141,7 @@ export { SessionBusyError, WorkspaceDrainingError, BridgeChannelQuarantinedError, + BridgeRuntimeRecyclingError, InvalidRewindTargetError, TotalSessionLimitExceededError, NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE, diff --git a/packages/cli/src/serve/server/error-response.ts b/packages/cli/src/serve/server/error-response.ts index 195b1abad5f..b772ec40f0d 100644 --- a/packages/cli/src/serve/server/error-response.ts +++ b/packages/cli/src/serve/server/error-response.ts @@ -23,6 +23,7 @@ import { writeStderrLine } from '../../utils/stdioHelpers.js'; import { BranchWhilePromptActiveError, BridgeChannelQuarantinedError, + BridgeRuntimeRecyclingError, BridgeTimeoutError, CancelSentinelCollisionError, CdWhilePromptActiveError, @@ -341,6 +342,16 @@ export function sendBridgeError( }); return; } + if (err instanceof BridgeRuntimeRecyclingError) { + recordExpectedBridgeError(err, ctx, daemonLog); + res.status(503).json({ + error: err.message, + code: err.code, + errorKind: err.code, + retryable: true, + }); + return; + } if (err instanceof DaemonDrainingError) { res.status(503).json({ error: err.message, diff --git a/packages/cli/src/ui/hooks/use-llm-stream.ts b/packages/cli/src/ui/hooks/use-llm-stream.ts index d0b91ce61c0..02ef9310687 100644 --- a/packages/cli/src/ui/hooks/use-llm-stream.ts +++ b/packages/cli/src/ui/hooks/use-llm-stream.ts @@ -6204,6 +6204,7 @@ export const useLlmStream = ( useEffect(() => { const registry = config.getBackgroundTaskRegistry(); registry.setNotificationCallback((displayText, modelText, meta) => { + if (meta.recordOnly) return; notificationQueueRef.current.push({ displayText, modelText, diff --git a/packages/core/src/agents/background-agent-resume.ts b/packages/core/src/agents/background-agent-resume.ts index 5323c89e471..6a27af51286 100644 --- a/packages/core/src/agents/background-agent-resume.ts +++ b/packages/core/src/agents/background-agent-resume.ts @@ -1294,6 +1294,7 @@ export class BackgroundAgentResumeService { stopHookWarning, ); const stats = getCompletionStats(subagent, liveToolCallCount); + if (registry.get(meta.agentId)?.retainsPhysicalSlot) break; if (terminateMode === AgentTerminateMode.GOAL) { const pending = registry.drainMessages(meta.agentId); if (pending.length > 0) { @@ -1359,6 +1360,7 @@ export class BackgroundAgentResumeService { debugLogger.error( `[BackgroundAgentResume] Background agent failed: ${errorMessage}`, ); + if (registry.get(meta.agentId)?.retainsPhysicalSlot) return; if (turnAbortController.signal.aborted && !progressTimeout) { const stats = getCompletionStats(subagent, liveToolCallCount); registry.finalizeCancelled(meta.agentId, errorMessage, stats); @@ -1403,6 +1405,7 @@ export class BackgroundAgentResumeService { bgEmitter, turnAbortController, () => monitorRegistry.hasRunningForOwner(meta.agentId), + (error) => registry.failUnresponsive(meta.agentId, error.message), ); // Restore the persisted launch depth so a resumed nested agent keeps // its original nesting level (and spawn eligibility) instead of @@ -1419,7 +1422,10 @@ export class BackgroundAgentResumeService { target.isFork ? runInForkContext(invocationRunBody) : invocationRunBody() - ).finally(disposeWatchdog); + ).finally(() => { + disposeWatchdog(); + registry.releaseRetainedPhysicalSlot(meta.agentId); + }); }; const reportUnexpectedBackgroundError = (error: unknown) => { diff --git a/packages/core/src/agents/background-tasks.ts b/packages/core/src/agents/background-tasks.ts index 2f1ac1633f0..ca997dccdcb 100644 --- a/packages/core/src/agents/background-tasks.ts +++ b/packages/core/src/agents/background-tasks.ts @@ -368,6 +368,8 @@ export interface AgentTask extends TaskBase { * `running` so `/resume` can recover the work later. */ persistedCancellationStatus?: Extract; + /** The underlying run ignored abort and still occupies a physical slot. */ + retainsPhysicalSlot?: true; } /** @@ -398,6 +400,7 @@ export interface NotificationMeta { toolUseId?: string; todoWorkChainId?: string; label?: string; + recordOnly?: true; } export type BackgroundNotificationCallback = ( @@ -914,6 +917,41 @@ export class BackgroundTaskRegistry { this.drainWaitQueue(); } + failUnresponsive(agentId: string, error: string): void { + const entry = this.agents.get(agentId); + if ( + !entry || + (entry.status !== 'running' && entry.status !== 'cancelled') || + entry.notified + ) + return; + + entry.status = 'failed'; + entry.endTime = Date.now(); + entry.error = error; + entry.retainsPhysicalSlot = true; + if (entry.metaPath) { + patchAgentMeta(entry.metaPath, { + status: 'failed', + lastUpdatedAt: new Date().toISOString(), + lastError: error, + }); + } + this.releaseFinishingWaiters(agentId, true); + this.rejectPendingApprovals(entry); + this.emitNotification(entry, true); + this.emitStatusChange(entry); + this.disposeResidentAgent(agentId); + } + + releaseRetainedPhysicalSlot(agentId: string): void { + const entry = this.agents.get(agentId); + if (!entry?.retainsPhysicalSlot) return; + delete entry.retainsPhysicalSlot; + this.emitStatusChange(entry); + this.drainWaitQueue(); + } + // Cancellation aborts the signal and marks the entry as cancelled, but // does *not* emit the terminal notification immediately. The natural // completion path (bgBody) fires complete()/fail()/finalizeCancelled() @@ -1286,16 +1324,17 @@ export class BackgroundTaskRegistry { return Array.from(this.agents.values()); } - // Counts backgrounded agents that still occupy a slot: running, or - // cancelled-but-not-yet-finalized. When `model` is given, only agents on - // that model are counted (per-model cap); otherwise all of them (global). + // Counts backgrounded agents that still occupy a slot: running, + // cancelled-but-not-yet-finalized, or watchdog-terminal but not physically + // settled. When `model` is given, only agents on that model are counted. private getRunningBackgroundCount(model?: string): number { let count = 0; for (const entry of this.agents.values()) { const occupiesSlot = entry.isBackgrounded && (entry.status === 'running' || - (entry.status === 'cancelled' && !entry.notified)); + (entry.status === 'cancelled' && !entry.notified) || + entry.retainsPhysicalSlot === true); if (!occupiesSlot) { continue; } @@ -1451,6 +1490,8 @@ export class BackgroundTaskRegistry { * registry right after passing the gate, which suppresses that very * notification, so blocking on it made the command silently no-op * when the user cleared immediately after cancelling (issue #5949). + * A watchdog-terminal run is excluded even while its physical slot remains + * reserved: runtime recycling, not Session work retention, owns its teardown. * Headless holdback loops must keep using `hasUnfinalizedTasks()` so * every task_started still pairs with a task_notification. */ @@ -1693,7 +1734,7 @@ export class BackgroundTaskRegistry { return buildBackgroundEntryLabel(entry); } - private emitNotification(entry: AgentTask): void { + private emitNotification(entry: AgentTask, recordOnly = false): void { // Mark notified *before* invoking the callback so that a re-entrant // terminal call inside the callback chain (cancel → complete race) // sees the flag and short-circuits, rather than firing twice. @@ -1772,6 +1813,7 @@ export class BackgroundTaskRegistry { stats: entry.stats, toolUseId: entry.toolUseId, todoWorkChainId: entry.todoWorkChainId, + ...(recordOnly ? { recordOnly: true } : {}), label: buildBackgroundEntryLabel(entry, { includePrefix: false }), }; @@ -1833,6 +1875,7 @@ export class BackgroundTaskRegistry { private pruneTerminalEntries(): void { const evictable = Array.from(this.agents.values()) .filter((entry) => entry.notified === true) + .filter((entry) => !entry.retainsPhysicalSlot) .sort( (a, b) => (a.endTime ?? a.startTime) - (b.endTime ?? b.startTime) || diff --git a/packages/core/src/agents/runtime/agent-progress-watchdog.ts b/packages/core/src/agents/runtime/agent-progress-watchdog.ts index 88c80681bc9..ce585796ac1 100644 --- a/packages/core/src/agents/runtime/agent-progress-watchdog.ts +++ b/packages/core/src/agents/runtime/agent-progress-watchdog.ts @@ -15,6 +15,7 @@ import type { const MODEL_CONTROL_PROGRESS_TIMEOUT_MS = 15 * 60_000; const TOOL_PROGRESS_TIMEOUT_MS = 10 * 60_000; +const UNRESPONSIVE_ABORT_GRACE_MS = 5_000; export class AgentProgressTimeoutError extends Error { constructor( @@ -49,17 +50,27 @@ export function attachAgentProgressWatchdog( emitter: AgentEventEmitter, controller: AbortController, isWaitingForExternalInput: () => boolean, + onUnresponsive: (error: AgentProgressTimeoutError) => void, ): () => void { let disposed = false; let waitingForExternalInput = false; let modelRetrying = false; let roundHadToolCalls = false; let modelTimer: ReturnType | undefined; + let escalationTimer: ReturnType | undefined; const tools = new Map(); const abort = (error: AgentProgressTimeoutError) => { if (disposed || controller.signal.aborted) return; controller.abort(error); + const armEscalation = () => { + escalationTimer = schedule( + UNRESPONSIVE_ABORT_GRACE_MS, + () => onUnresponsive(error), + armEscalation, + ); + }; + armEscalation(); }; const schedule = ( timeoutMs: number, @@ -187,6 +198,7 @@ export function attachAgentProgressWatchdog( if (disposed) return; disposed = true; clearModel(); + if (escalationTimer) clearTimeout(escalationTimer); for (const tool of tools.values()) { if (tool.timer) clearTimeout(tool.timer); } diff --git a/packages/core/src/tools/agent/agent.ts b/packages/core/src/tools/agent/agent.ts index 766074de4df..f528ef10c79 100644 --- a/packages/core/src/tools/agent/agent.ts +++ b/packages/core/src/tools/agent/agent.ts @@ -3595,6 +3595,8 @@ class AgentToolInvocation extends BaseToolInvocation { recordTerminalOutcome(); } + if (registry.get(hookOpts.agentId)?.retainsPhysicalSlot) break; + if (terminateMode === AgentTerminateMode.GOAL) { keepResident = residentRegistered && !needsAutoPermissionLease(); @@ -3706,6 +3708,8 @@ class AgentToolInvocation extends BaseToolInvocation { } const errorMsg = baseErrorMsg + wtSuffix; + if (registry.get(hookOpts.agentId)?.retainsPhysicalSlot) return; + // If the error came from a cancellation, preserve the cancelled // status so the model's notification matches what task_stop // requested rather than reporting it as a generic failure. @@ -3763,6 +3767,8 @@ class AgentToolInvocation extends BaseToolInvocation { this.config .getMonitorRegistry() .hasRunningForOwner(hookOpts.agentId), + (error) => + registry.failUnresponsive(hookOpts.agentId, error.message), ); const framedBgBody = () => this.runWithSubagentSpan( @@ -3787,7 +3793,10 @@ class AgentToolInvocation extends BaseToolInvocation { ); return ( isFork ? runInForkContext(framedBgBody) : framedBgBody() - ).finally(disposeWatchdog); + ).finally(() => { + disposeWatchdog(); + registry.releaseRetainedPhysicalSlot(hookOpts.agentId); + }); }; const reportUnexpectedBackgroundError = (err: unknown) => {