From fe9e0b9f305fbc8c474d8fb9f7433bc9a71fd083 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Thu, 13 Aug 2026 12:03:21 +0800 Subject: [PATCH] feat(daemon): track background shells in activeWork Co-authored-by: Qwen-Coder --- docs/design/2026-08-06-active-work-health.md | 20 +- ...2026-08-13-active-work-background-shell.md | 50 +++++ docs/developers/qwen-serve-protocol.md | 2 +- packages/acp-bridge/src/bridge.test.ts | 112 +++++++++- packages/acp-bridge/src/bridge.ts | 17 ++ packages/acp-bridge/src/bridgeTypes.ts | 21 +- .../cli/src/acp-integration/acpAgent.test.ts | 179 ++++++++++++++- packages/cli/src/acp-integration/acpAgent.ts | 40 +++- .../active-work-reporter.test.ts | 36 ++- .../acp-integration/active-work-reporter.ts | 8 +- .../session/Session.review-lease.test.ts | 3 + .../acp-integration/session/Session.test.ts | 205 +++++++++++++++++- .../src/acp-integration/session/Session.ts | 22 +- .../session/Session.worktree.test.ts | 3 + .../services/backgroundShellRegistry.test.ts | 25 +++ .../src/services/backgroundShellRegistry.ts | 35 +-- 16 files changed, 728 insertions(+), 50 deletions(-) create mode 100644 docs/design/2026-08-13-active-work-background-shell.md diff --git a/docs/design/2026-08-06-active-work-health.md b/docs/design/2026-08-06-active-work-health.md index 91939b8f12f..7c9672b9dc7 100644 --- a/docs/design/2026-08-06-active-work-health.md +++ b/docs/design/2026-08-06-active-work-health.md @@ -8,7 +8,7 @@ `GET /health?deep=1` gains three fields: `activeWork`, `activeWorkReporting`, and `activeWorkStaleMs`. -`activeWork` is true while any managed workspace has an accepted-but-unsettled prompt, a running background Agent, or an Agent terminal notification that is queued, awaiting acceptance, or being processed by its parent continuation. It deliberately does **not** cover background shells, Monitors, workflows, or cron. That exclusion is a scope decision, not an oversight: those categories have no equivalent signal today, and a controller that treats `activeWork: false` as "nothing at all is running" will be wrong about them. +`activeWork` is true while any managed workspace has an accepted-but-unsettled prompt, a running background Agent, an Agent terminal notification that is queued, awaiting acceptance, or being processed by its parent continuation, or Session-managed background shell work. Shell work covers a running registry entry and the terminal notification until its parent continuation settles. It deliberately does **not** cover Monitors, workflows, cron, or external processes the shell registry can no longer track. It is also **Session-scoped, not channel-scoped**. Channel-level work with no Session attached yet — a spawn in flight, a pending restore, MCP discovery or authentication — is not counted, so `activeWork` can read false while the daemon's own `hasNoChannelWork` is simultaneously refusing to reclaim that channel. The two answer different questions and are allowed to disagree: this field describes work owned by Sessions, and widening it to cover channel setup would change what the boolean means for every existing reader. A controller that needs "is this daemon reclaimable" must combine the three-term rule below with a graceful-shutdown handshake, not read more into this one field than it claims. @@ -16,12 +16,14 @@ Restart policy stays with the external controller. The daemon publishes facts; i ## Why holds, and why full snapshots -Each Session reports a set of named **holds**, each carrying a category (`agent`, `notification`). Two properties follow, and both are the point: +Each Session reports a set of named **holds**, each carrying a category (`agent`, `notification`, or `shell`). Two properties follow, and both are the point: -**Holds are derived, never maintained.** `Session.collectActiveWorkHolds()` reads the owners of the work — the background-task registry's unfinalized set, the notification queue, the in-flight acceptance and continuation state — on every call. There is no acquire/release ledger kept alongside the work, because a ledger can miss a release, and a leaked hold would pin its Session forever while every snapshot faithfully republished the leak. +**Holds are derived, never maintained.** `Session.collectActiveWorkHolds()` reads the owners of the work — the background-task registry's unfinalized set, the background-shell registry's running entries, the notification queue, and the in-flight acceptance and continuation state — on every call. There is no acquire/release ledger kept alongside the work, because a ledger can miss a release, and a leaked hold would pin its Session forever while every snapshot faithfully republished the leak. The agent category uses `BackgroundTaskRegistry.hasUnfinalizedTasks()`'s predicate rather than `hasRunningTasks()`'. A cancelled agent still owes its terminal task-notification: `cancel()` flips status and emits a status change, but the notification arrives later from `finalizeCancelled()` or the 5s grace timer. Keying on "running" would make the Session look idle inside that window, and a detached Session would be closed with the notification still owed. +Shells use one aggregate hold, `{ "category": "shell", "id": "background-shells" }`, regardless of the number of running shells. The task registry and `/tasks` surface remain the detailed roster; active-work only needs the bounded retention fact. The aggregate also prevents an unbounded shell roster from exceeding the protocol's per-Session hold limit. + **Reports are complete snapshots at channel scope, not per-Session transitions.** One message per ACP channel carries every Session the child owns and every hold it holds: ```json @@ -40,19 +42,20 @@ Prompts are absent from the child's report on purpose. The daemon accepts, queue ## Ordering -A snapshot is flushed ahead of the prompt response on the same stream. The daemon drops its pending-prompt count the instant that response lands, so a hold the prompt left behind — a background Agent it started — must already be on the wire, or the daemon briefly sees neither fact. +A snapshot is flushed ahead of the prompt response on the same stream. The daemon drops its pending-prompt count the instant that response lands, so a hold the prompt left behind — a background Agent or shell it started — must already be on the wire, or the daemon briefly sees neither fact. -## Three states, and closing atomically +## Reporting states, and closing atomically Per Session the daemon holds one of: - **unsupported** — the channel never negotiated. Contributes nothing; pre-existing cleanup behavior applies unchanged. Treating this as "unknown" would make every legacy Session permanently unreapable. +- **incomplete** — the channel negotiated but does not report every category the daemon currently requires. Health is graded `partial`, and ordinary automatic cleanup is disabled for that Session. Unlike unknown freshness, another round trip cannot make an older child understand a category it did not negotiate. - **unknown** — negotiated, not yet heard from _recently enough_. Reads as busy on the health surface, but is not a state the daemon sits in: it asks. - **known** — a fresh snapshot has been applied. Never-reported and gone-quiet are the same state on purpose. A snapshot older than the grading window (`intervalMs × 3`) is not a report that the Session is idle, it is the absence of one — a background Agent could have started at any point since — so it stops counting as evidence. -**Unknown is a reason to ask, not a reason to skip.** The two consumers read it differently, and they have to: the health surface reports unknown as busy (a controller must never mistake "nobody told me" for "nothing is running"), while automatic cleanup treats it as a candidate and goes on to the conditional close below. Only _known_ work — daemon-owned, or a fresh report of held work — blocks the attempt outright. Skipping on unknown instead would look safe and in fact be the worse failure: nothing would ever resolve it, so a Session on a channel that went quiet would be retained forever with no path out. Asking costs one bounded round trip and still retains on any non-answer, and the child can answer authoritatively under its close gate whether or not its snapshots are arriving. +**Unknown is a reason to ask, not a reason to skip.** The two consumers read it differently, and they have to: the health surface reports unknown as busy (a controller must never mistake "nobody told me" for "nothing is running"), while automatic cleanup treats it as a candidate and goes on to the conditional close below. Only _known_ work — daemon-owned, or a fresh report of held work — blocks the attempt outright. Skipping on unknown instead would look safe and in fact be the worse failure: nothing would ever resolve it, so a Session on a channel that went quiet would be retained forever with no path out. Asking costs one bounded round trip and still retains on any non-answer, and the child can answer authoritatively under its close gate whether or not its snapshots are arriving. Incomplete coverage is different and does skip: a negotiated child that omits `shell` can truthfully answer according to its older predicate while missing a running shell, so its answer cannot authorize automatic destruction. Reclaiming a channel that has stopped answering entirely is still not this mechanism's job; see below. @@ -63,7 +66,7 @@ qwen/control/session/close { sessionId, onlyIfUnheld: true } → { closed: true, holds: [] } | { closed: false, holds: [...] } ``` -The child evaluates it under its own close gate, before anything destructive runs. With the gate held the Session admits no new prompt and starts no new automatic turn, so a hold cannot appear between the check and the teardown **on the child side**. If holds exist, the gate is released and they are handed back; the daemon adopts them and backs off. +The child evaluates it under its own close gate, before anything destructive runs. It rejects known holds immediately, drains any turn that was already active when the gate closed, then evaluates the unfiltered collector again. The second read matters because an already-running out-of-scope turn such as cron can register a background shell while it drains. With the gate still held no new turn can start after that final read, so a hold cannot appear between final authorization and teardown **on the child side**. If either read finds holds, the gate is released and they are handed back; the daemon adopts them and backs off. The daemon side needs its own cover, because the round trip is an await of up to ten seconds. A Session with a conditional close outstanding is marked in-flight, and every admission path — attach, prompt, rewind — refuses it exactly as it refuses one that is already closing. Without that, a prompt accepted during the round trip is lost when the teardown it raced completes; the previous synchronous guard-then-teardown sequence got this for free, and splitting it is what created the need to say so explicitly. @@ -80,6 +83,7 @@ Four things can decide it is time to look at a Session: the last client detachin | not already closing or close-in-flight | two paths racing the same teardown duplicate the round trip and race each other's guards | | no SSE subscriber | someone is watching this Session's stream | | nothing daemon-owned in flight | queued and dispatched prompts and notifications the daemon is pushing; never depends on the child reporting anything | +| negotiated reporting covers every category | an older predicate must not authorize teardown while work in a newer category exists | | no fresh child report of held work | only _known_ work blocks; unknown is a candidate that goes on to ask | | the child confirms under its own close gate | the cache says what _was_ true; only the child can say what is true now | @@ -105,7 +109,7 @@ Killing a whole multiplexed channel is reasonable when the channel is _actually_ | `activeWorkReporting` | `full` / `partial` / `none` — how much of that boolean is vouched for | | `activeWorkStaleMs` | Age of the oldest snapshot it rests on; `0` when nothing is covered | -Freshness is graded by the daemon, not the controller: the reporting cadence is negotiated per channel (the child proposes, the daemon clamps into an agreed range), so only the daemon can judge it. A stale snapshot or a child that omits a category degrades the grade to `partial` rather than silently narrowing what the boolean covers. `activeWorkStaleMs` is diagnostic, and it measures only the _covered_ Sessions — an uncovered one already shows up in the grade, so letting it also drag the age down would double-count it and produce a positive staleness next to a grade saying nothing is covered. +Freshness is graded by the daemon, not the controller: the reporting cadence is negotiated per channel (the daemon requests a cadence and category set; the child echoes the clamped cadence and the supported intersection), so only the daemon can judge it. A stale snapshot or a child that omits a category degrades the grade to `partial` rather than silently narrowing what the boolean covers. A v1 request without `categories` means the legacy `agent`/`notification` baseline, which lets a new child keep its wire report readable by an old daemon while its local collector still sees shell work for conditional close. `activeWorkStaleMs` is diagnostic, and it measures only the _covered_ Sessions — an uncovered one already shows up in the grade, so letting it also drag the age down would double-count it and produce a positive staleness next to a grade saying nothing is covered. The grade is computed once over the whole daemon rather than per runtime and then combined, because grades do not compose: a runtime with no Sessions vouches for everything it has, and folding that vacuous `full` in as evidence let an empty workspace vouch for another workspace's unreported Sessions. Each runtime therefore exposes coverage counts and the route sums them before grading. diff --git a/docs/design/2026-08-13-active-work-background-shell.md b/docs/design/2026-08-13-active-work-background-shell.md new file mode 100644 index 00000000000..51924d52b94 --- /dev/null +++ b/docs/design/2026-08-13-active-work-background-shell.md @@ -0,0 +1,50 @@ +# Background shell active-work coverage + +## Problem + +A Prompt can start a long-running background shell and finish immediately. Before this change the daemon then observed `activePrompts: 0` and `activeWork: false` even though `GET /session/:id/tasks` still reported a running shell. A restart controller could therefore treat the daemon as idle and terminate the Session before the shell's terminal notification reached the parent continuation. + +## Decision + +Session-managed background shells join the existing active-work snapshot protocol as category `shell`. A Session publishes one aggregate hold while its shell registry has a running entry, a shell terminal notification is queued, or that notification is driving the parent continuation: + +```json +{ "category": "shell", "id": "background-shells" } +``` + +The hold is deliberately aggregate. The shell registry and task-status surfaces remain the detailed roster, while the retention protocol stays bounded even if a Session owns more than 1024 shells. + +The Session collector remains an unfiltered statement of local truth. Category negotiation is applied only when the reporter serializes a wire snapshot. This distinction is required for compatibility: a new child talking to an old v1 daemon filters `shell` from the wire, but its conditional-close check still sees the running shell locally and answers `closed: false`. + +## Negotiation and compatibility + +The protocol version remains v1. The daemon initialize request advertises `agent`, `notification`, and `shell`; the child answers with the intersection it supports. A request with no `categories` is the pre-negotiation v1 baseline, `agent` and `notification`. + +| Peers | Reporting result | Ordinary automatic cleanup | +| ------------------------------------------ | ---------------------------------------- | ----------------------------------------------------- | +| new daemon + new child | `full`; shell hold crosses the wire | existing conditional-close flow | +| new daemon + old v1 child | `partial`; `shell` is missing | disabled for that Session | +| old v1 daemon + new child | wire contains only the legacy categories | local conditional close still rejects a running shell | +| daemon + child with no active-work support | `none` | historical legacy cleanup | + +Negotiated-but-incomplete and unsupported are intentionally different. An unsupported historical child keeps the behavior it had before active-work existed. A child that negotiated the protocol but omitted a currently required category has explicitly disclosed that its predicate is incomplete, so it cannot authorize an ordinary teardown. Explicit close, kill, daemon shutdown, channel exit, and condemned restore cleanup keep their force semantics. + +## Lifecycle and ordering + +The shell registry synchronously reports registration and terminal transitions. Session installs an identity-safe status callback that triggers the existing change-coalesced reporter and removes exactly that callback on dispose. + +At shell completion, the registry invokes the notification callback before publishing the terminal status change. The notification is therefore already queued when the running entry becomes terminal. When the drain removes the queue item it marks the shell continuation active before yielding. These transitions ensure the derived aggregate hold has no false gap between running, queued, and executing states. Prompt teardown also retains the existing reporter flush-before-response ordering, so a shell started by the Prompt is visible before the daemon decrements its own prompt count. + +`Session.isIdle()` consumes the same unfiltered collector. Workspace reload therefore skips a Session while a background shell or its terminal continuation is active. + +Conditional close reads the unfiltered collector once before disturbing active turns and again after those turns drain, while the Session close gate remains held. The final read closes the window where an already-running, otherwise out-of-scope cron or automatic turn registers a shell during drain; the new shell refuses ordinary teardown without adding cron itself to `activeWork`. + +## Boundaries + +This change tracks the logical lifecycle owned by `BackgroundShellRegistry`; it does not use PID probes or sidecars to reconstruct process liveness. `task_stop` follows the registry's terminal status and does not promise an additional OS-level exit confirmation. A promoted or externally detached process that the registry no longer tracks is outside the signal. + +Long-running development servers consequently keep `activeWork: true`. This is the intended retention fact, not shell-stall detection or a restart lease. Monitor, workflow, cron, and follow-up work remain out of scope, and the public health shape, persistence formats, shell admission policy, heartbeat behavior, and watchdog behavior do not change. + +## Verification + +Unit coverage pins aggregate cardinality, running-to-notification handoff, reporter filtering, legacy negotiation, bridge parsing, incomplete-child retention, post-drain conditional-close authorization, explicit force close, callback cleanup, and unchanged unsupported-child behavior. The E2E plan reproduces the released baseline with a running `sleep` shell and compares it with the local build through shell completion and parent continuation settlement. diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 9835f423235..0f63cf5f178 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -516,7 +516,7 @@ Pass `?deep=1` (also accepts `?deep=true` or bare `?deep`) for a daemon-wide pro } ``` -`sessions`, `pendingPermissions`, and `activePrompts` are sums. `activeWork` **does not count background shells, Monitors, workflows, cron jobs, or follow-up suggestions** — it is true when any runtime has an accepted but unsettled prompt (including a FIFO-waiting prompt), a running background Agent, or a queued/in-progress Agent terminal notification, and nothing else. It is session-scoped: channel-level work with no session attached yet — a spawn in flight, a pending restore, MCP discovery or authentication — is not counted, so `activeWork` may read false while the daemon still declines to reclaim that channel. Do not read this field as "the daemon is reclaimable"; it describes session-owned work only. `activeWorkReporting` says how much of that boolean is actually vouched for: `full` when every live session is covered by a fresh report from a child that reports all categories, `none` when no session is, `partial` for anything between — including a stale snapshot or an older child that never acknowledged the capability. A snapshot older than three report intervals stops counting as coverage: it is not a report that the session is idle, so the session goes back to reading as retained, exactly as if the child had never reported. `activeWorkStaleMs` is the age of the oldest snapshot the boolean rests on **among the covered sessions**, and is `0` when no session is covered; it is diagnostic, because freshness is already graded into `activeWorkReporting` by the daemon (only the daemon knows each channel's negotiated cadence). The grade is computed once over every managed runtime rather than per runtime and then combined — a runtime with no sessions is vacuously complete, and treating that as evidence would let an empty workspace vouch for another workspace's unreported sessions. `lastActivityAt` is the latest non-null workspace activity time and `idleSinceMs` is derived from that same snapshot. `channelAlive` means at least one managed workspace channel is live; it does not mean every workspace is healthy. `connectedClients` and the optional `rateLimitHits` remain daemon-wide counters rather than per-workspace sums. +`sessions`, `pendingPermissions`, and `activePrompts` are sums. `activeWork` is true when any runtime has an accepted but unsettled prompt (including a FIFO-waiting prompt), a running background Agent, a queued/in-progress Agent terminal notification, or Session-managed background shell work. Shell work remains active while the shell registry reports a running entry and while its terminal notification is queued or driving the parent continuation; any number of shells contributes one bounded aggregate hold. Monitors, workflows, cron jobs, follow-up suggestions, and external processes the shell registry can no longer track remain outside the field. It is session-scoped: channel-level work with no session attached yet — a spawn in flight, a pending restore, MCP discovery or authentication — is not counted, so `activeWork` may read false while the daemon still declines to reclaim that channel. Do not read this field as "the daemon is reclaimable"; it describes session-owned work only. `activeWorkReporting` says how much of that boolean is actually vouched for: `full` when every live session is covered by a fresh report from a child that reports all required categories, `none` when no session negotiated reporting, and `partial` for anything between — including a stale snapshot or a negotiated child that omits a required category. A snapshot older than three report intervals stops counting as coverage: it is not a report that the session is idle, so the session goes back to reading as retained, exactly as if the child had never reported. Ordinary automatic cleanup is also disabled for a negotiated-but-incomplete child; a child that does not understand `shell` cannot safely authorize conditional close according to the complete current predicate. Completely unsupported historical children retain legacy cleanup behavior, and explicit close, kill, shutdown, and channel exit remain force operations. `activeWorkStaleMs` is the age of the oldest snapshot the boolean rests on **among the covered sessions**, and is `0` when no session is covered; it is diagnostic, because freshness is already graded into `activeWorkReporting` by the daemon (only the daemon knows each channel's negotiated cadence). The grade is computed once over every managed runtime rather than per runtime and then combined — a runtime with no sessions is vacuously complete, and treating that as evidence would let an empty workspace vouch for another workspace's unreported sessions. `lastActivityAt` is the latest non-null workspace activity time and `idleSinceMs` is derived from that same snapshot. `channelAlive` means at least one managed workspace channel is live; it does not mean every workspace is healthy. `connectedClients` and the optional `rateLimitHits` remain daemon-wide counters rather than per-workspace sums. Restart controllers should treat the daemon as busy when: diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index be0de381cb7..299b31534f4 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -75,6 +75,7 @@ import { ACTIVE_WORK_HEARTBEAT_META_KEY, ACTIVE_WORK_HEARTBEAT_VERSION, ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_LEGACY_HOLD_CATEGORIES, ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM, ACTIVE_WORK_MAX_SESSION_HOLDS, ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS, @@ -218,6 +219,10 @@ function agentHold(id: string) { return { category: 'agent' as const, id }; } +function shellHold() { + return { category: 'shell' as const, id: 'background-shells' }; +} + /** * The grade `/health?deep=1` would report for a single-runtime daemon. The * bridge exposes counts rather than a grade (an empty runtime must not vouch @@ -251,6 +256,7 @@ describe('createAcpSessionBridge', () => { [ACTIVE_WORK_HEARTBEAT_META_KEY]: { v: ACTIVE_WORK_HEARTBEAT_VERSION, intervalMs: ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], }, [CHANNEL_STARTUP_PROFILE_META_KEY]: { v: CHANNEL_STARTUP_PROFILE_VERSION, @@ -299,12 +305,29 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); - it('grades a child that omits a category as partial', async () => { + it('retains sessions reported by a negotiated but incomplete child', async () => { + let conditionalCloseCalls = 0; + let forcedCloseCalls = 0; const handle = makeChannel({ initializeImpl: () => - activeWorkInitializeResponse({ categories: ['agent'] }), + activeWorkInitializeResponse({ + categories: [...ACTIVE_WORK_LEGACY_HOLD_CATEGORIES], + }), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + if (params[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] === true) { + conditionalCloseCalls++; + } else { + forcedCloseCalls++; + } + return { closed: true, holds: [] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 10, + sessionIdleTimeoutMs: 10, }); - const bridge = makeBridge({ channelFactory: async () => handle.channel }); const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); await sendActiveWorkSnapshot(handle, 1, [ @@ -313,6 +336,81 @@ describe('createAcpSessionBridge', () => { expect(bridge.activeWork).toBe(false); expect(reportingGrade(bridge)).toBe('partial'); + await bridge.detachClient(session.sessionId, session.clientId); + // Wait through multiple reaper ticks too. Both detach and reaper must + // stop at the shared incomplete-reporting candidate guard. + await new Promise((resolve) => setTimeout(resolve, 40)); + expect(conditionalCloseCalls).toBe(0); + expect(bridge.sessionCount).toBe(1); + + // Direct close remains a force operation even when reporting is + // incomplete; only ordinary automatic cleanup is fail-closed. + await bridge.closeSession(session.sessionId); + expect(forcedCloseCalls).toBe(1); + expect(bridge.sessionCount).toBe(0); + + await bridge.shutdown(); + }); + + it('accepts the aggregate shell hold as active work', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [shellHold()] }, + ]); + expect(bridge.activeWork).toBe(true); + expect(reportingGrade(bridge)).toBe('full'); + + await sendActiveWorkSnapshot(handle, 2, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(false); + + await bridge.shutdown(); + }); + + it('keeps explicit close and kill forceful while shell holds are reported', async () => { + const closeParams: Array> = []; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method === SERVE_CONTROL_EXT_METHODS.sessionClose) { + closeParams.push(params); + return { closed: true, holds: [] }; + } + return {}; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const first = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sessionScope: 'thread', + }); + const second = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sessionScope: 'thread', + }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: first.sessionId, holds: [shellHold()] }, + { sessionId: second.sessionId, holds: [shellHold()] }, + ]); + + await bridge.closeSession(first.sessionId); + await expect(bridge.killSession(second.sessionId)).resolves.toBe(true); + expect(closeParams).toHaveLength(2); + expect( + closeParams.some( + (params) => params[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] === true, + ), + ).toBe(false); + expect(bridge.sessionCount).toBe(0); + await bridge.shutdown(); }); @@ -6613,7 +6711,13 @@ describe('createAcpSessionBridge', () => { vi.useFakeTimers(); const lateRestore = deferred(); const handle = makeChannel({ - initializeImpl: () => activeWorkInitializeResponse(), + // An old v1 child cannot report the newer shell category. Ordinary + // cleanup must retain it, but once the restore lifecycle condemns this + // channel the existing bounded force-recovery path still has to run. + initializeImpl: () => + activeWorkInitializeResponse({ + categories: [...ACTIVE_WORK_LEGACY_HOLD_CATEGORIES], + }), loadSessionImpl: () => lateRestore.promise, // The wedged child answers NO close at all — neither the hold probe nor // the plain agent close. An earlier version of this test let the plain diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 6a715418b40..3615b7d921f 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -2231,6 +2231,22 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // can elapse inside a slow restore too. const owner = channelInfoForEntry(entry); if (owner?.pendingRestoreIds.has(entry.sessionId)) return false; + // A child that negotiated active-work but cannot report every category + // the daemon currently relies on must not authorize ordinary teardown. + // This differs from a legacy child that never negotiated at all: legacy + // cleanup keeps its historical behavior, while an incomplete negotiated + // answer is explicitly known not to cover the full retention predicate. + const capability = owner?.activeWork; + if ( + capability && + !owner.isQuarantined && + !owner.restoreSettlementOverdue && + ACTIVE_WORK_HOLD_CATEGORIES.some( + (category) => !capability.categories.includes(category), + ) + ) { + return false; + } return !childReportsHeldWork(entry); } @@ -3517,6 +3533,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { [ACTIVE_WORK_HEARTBEAT_META_KEY]: { v: ACTIVE_WORK_HEARTBEAT_VERSION, intervalMs: ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], }, [CHANNEL_STARTUP_PROFILE_META_KEY]: { v: CHANNEL_STARTUP_PROFILE_VERSION, diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index a1b41316bc7..aedd2938d2f 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -218,16 +218,21 @@ export const ACTIVE_WORK_MAX_SESSION_HOLDS = 1024; export const WORKTREE_MCP_DEFER_META_KEY = 'qwen.session.deferMcpDiscovery'; /** - * Work categories a child reports holds for. Deliberately excludes background - * shells, Monitors, workflows, and cron: those are out of `activeWork`'s - * declared scope. The category travels on every hold so widening the scope - * later adds data rather than changing what the `activeWork` boolean means. + * Work categories a child reports holds for. Monitors, workflows, and cron + * remain outside `activeWork`'s declared scope. The category travels on every + * hold so peers can negotiate coverage explicitly when the scope widens. */ -export type ActiveWorkHoldCategory = 'agent' | 'notification'; +export type ActiveWorkHoldCategory = 'agent' | 'notification' | 'shell'; + +/** Categories understood by active-work v1 before category negotiation was + * added to the daemon's initialize request. */ +export const ACTIVE_WORK_LEGACY_HOLD_CATEGORIES: readonly ActiveWorkHoldCategory[] = + ['agent', 'notification']; export const ACTIVE_WORK_HOLD_CATEGORIES: readonly ActiveWorkHoldCategory[] = [ 'agent', 'notification', + 'shell', ]; export interface ActiveWorkHeartbeatCapabilityV1 { @@ -1854,9 +1859,9 @@ export interface AcpSessionBridge { readonly activePromptCount: number; /** - * Whether an accepted prompt, a running background Agent, or an Agent - * terminal notification is unsettled. Background shells, Monitors, - * workflows, and cron are deliberately outside this. + * Whether an accepted prompt, a running background Agent, an Agent terminal + * notification, or Session-managed background shell work is unsettled. + * Monitors, workflows, and cron are deliberately outside this. */ readonly activeWork: boolean; diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index c7f5e80e201..74cbf610c60 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -931,6 +931,8 @@ import { ACTIVE_WORK_HEARTBEAT_MIN_INTERVAL_MS, ACTIVE_WORK_HEARTBEAT_VERSION, ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_LEGACY_HOLD_CATEGORIES, + ACTIVE_WORK_NOTIFICATION_METHOD, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, PROMPT_CANCEL_METHOD, @@ -1912,7 +1914,13 @@ describe('QwenAgent MCP SSE/HTTP support', () => { | ((conn: AgentSideConnectionLike) => AgentLike) | undefined; - type AgentSideConnectionLike = { closed: Promise }; + type AgentSideConnectionLike = { + closed: Promise; + extNotification?: ( + method: string, + params: Record, + ) => Promise; + }; type AgentLike = { initialize: (args: Record) => Promise; newSession: (args: Record) => Promise; @@ -1940,12 +1948,15 @@ describe('QwenAgent MCP SSE/HTTP support', () => { clearActiveTodoPlanRevision: ReturnType; clearTodoStopGuardTrust: ReturnType; hardSuspendTodoStopGuard: ReturnType; + beginClose: ReturnType; beginCloseIfAvailable: ReturnType; waitForActiveTurnsToSettle: ReturnType; cancelPendingPrompt: ReturnType; enqueueBackgroundNotification: ReturnType; enableLiveScreenContext: ReturnType; appendLiveConversationTranscript: ReturnType; + collectActiveWorkHolds: ReturnType; + dispose: ReturnType; prompt: ReturnType; releaseTodoStopGuardQueuedPromptWait: ReturnType; } @@ -2569,6 +2580,26 @@ describe('QwenAgent MCP SSE/HTTP support', () => { it('merges active-work negotiation and enables Session reporting', async () => { await setupSessionMocks('active-work-session'); initializeAcpStartupProfiler(); + let blockActiveWorkSnapshot = false; + let markActiveWorkSnapshotStarted!: () => void; + const activeWorkSnapshotStarted = new Promise((resolve) => { + markActiveWorkSnapshotStarted = resolve; + }); + let releaseActiveWorkSnapshot!: () => void; + const activeWorkSnapshotGate = new Promise((resolve) => { + releaseActiveWorkSnapshot = resolve; + }); + const extNotification = vi.fn( + async (method: string, _params: Record) => { + if ( + blockActiveWorkSnapshot && + method === ACTIVE_WORK_NOTIFICATION_METHOD + ) { + markActiveWorkSnapshotStarted(); + await activeWorkSnapshotGate; + } + }, + ); const agentPromise = runAcpAgent( mockConfig, makeSessionSettings(), @@ -2579,6 +2610,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { get closed() { return mockConnectionState.promise; }, + extNotification, }) as AgentLike; const response = (await agent.initialize({ @@ -2592,6 +2624,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { // Absurd cadence: the child must answer with the clamped value it // will actually use, not echo this back. intervalMs: 1, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], }, }, })) as { _meta?: Record }; @@ -2611,6 +2644,148 @@ describe('QwenAgent MCP SSE/HTTP support', () => { // channel owns the timing. expect(typeof vi.mocked(Session).mock.calls.at(-1)?.[4]).toBe('function'); + lastSessionMock?.collectActiveWorkHolds.mockReturnValue([ + { category: 'shell', id: 'background-shells' }, + ]); + blockActiveWorkSnapshot = true; + let promptSettled = false; + const prompt = agent + .prompt({ + sessionId: 'active-work-session', + prompt: [{ type: 'text', text: 'start a shell' }], + }) + .finally(() => { + promptSettled = true; + }); + await activeWorkSnapshotStarted; + expect(promptSettled).toBe(false); + releaseActiveWorkSnapshot(); + await expect(prompt).resolves.toEqual({ stopReason: 'end_turn' }); + const snapshots = extNotification.mock.calls.filter( + ([method]) => method === ACTIVE_WORK_NOTIFICATION_METHOD, + ); + expect(snapshots.at(-1)?.[1]).toMatchObject({ + sessions: [ + { + sessionId: 'active-work-session', + holds: [{ category: 'shell', id: 'background-shells' }], + }, + ], + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('uses the pre-category v1 baseline when initialize omits categories', async () => { + await setupSessionMocks('legacy-active-work-session'); + const extNotification = vi.fn().mockResolvedValue(undefined); + const agentPromise = runAcpAgent( + mockConfig, + makeSessionSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + extNotification, + }) as AgentLike; + + const response = (await agent.initialize({ + clientCapabilities: {}, + _meta: { + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: 15_000, + }, + }, + })) as { _meta?: Record }; + + expect(response._meta).toMatchObject({ + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + categories: [...ACTIVE_WORK_LEGACY_HOLD_CATEGORIES], + }, + }); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + lastSessionMock?.collectActiveWorkHolds.mockReturnValue([ + { category: 'shell', id: 'background-shells' }, + ]); + + await agent.prompt({ + sessionId: 'legacy-active-work-session', + prompt: [{ type: 'text', text: 'start a shell' }], + }); + const snapshots = extNotification.mock.calls.filter( + ([method]) => method === ACTIVE_WORK_NOTIFICATION_METHOD, + ); + expect(snapshots.at(-1)?.[1]).toMatchObject({ + sessions: [{ sessionId: 'legacy-active-work-session', holds: [] }], + }); + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionClose, { + sessionId: 'legacy-active-work-session', + onlyIfUnheld: true, + }), + ).resolves.toEqual({ + sessionId: 'legacy-active-work-session', + closed: false, + holds: [{ category: 'shell', id: 'background-shells' }], + }); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('rechecks local holds after active turns drain during conditional close', async () => { + await setupSessionMocks('active-work-close-race'); + const agentPromise = runAcpAgent( + mockConfig, + makeSessionSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + }) as AgentLike; + + await agent.initialize({ clientCapabilities: {} }); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + let closeGateHeld = false; + let turnsSettled = false; + lastSessionMock?.beginClose.mockImplementation(() => { + closeGateHeld = true; + return () => { + closeGateHeld = false; + }; + }); + lastSessionMock?.waitForActiveTurnsToSettle.mockImplementation(async () => { + turnsSettled = true; + }); + lastSessionMock?.collectActiveWorkHolds.mockImplementation(() => + closeGateHeld && turnsSettled + ? [{ category: 'shell', id: 'background-shells' }] + : [], + ); + + await expect( + agent.extMethod(SERVE_CONTROL_EXT_METHODS.sessionClose, { + sessionId: 'active-work-close-race', + onlyIfUnheld: true, + }), + ).resolves.toEqual({ + sessionId: 'active-work-close-race', + closed: false, + holds: [{ category: 'shell', id: 'background-shells' }], + }); + expect(lastSessionMock?.waitForActiveTurnsToSettle).toHaveBeenCalledOnce(); + expect(lastSessionMock?.dispose).not.toHaveBeenCalled(); + expect(closeGateHeld).toBe(false); + mockConnectionState.resolve(); await agentPromise; }); @@ -3653,6 +3828,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { ); vi.mocked(Session).mockImplementation((createdSessionId, createdConfig) => { const sessionMock = { + sessionId: createdSessionId, getId: vi.fn().mockReturnValue(createdSessionId), getConfig: vi.fn().mockReturnValue(createdConfig), sendAvailableCommandsUpdate: vi.fn().mockResolvedValue(undefined), @@ -3670,6 +3846,7 @@ describe('QwenAgent MCP SSE/HTTP support', () => { .mockResolvedValue({ accepted: true }), enableLiveScreenContext: vi.fn().mockResolvedValue(undefined), appendLiveConversationTranscript: vi.fn().mockResolvedValue(undefined), + collectActiveWorkHolds: vi.fn().mockReturnValue([]), assertCanStartTurn: vi.fn().mockResolvedValue(undefined), dispose: vi.fn(), emitGoalStatus: vi.fn(), diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index b56d9e9b883..1ef4ea54505 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -325,6 +325,7 @@ import { ACTIVE_WORK_HEARTBEAT_META_KEY, ACTIVE_WORK_HEARTBEAT_VERSION, ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_LEGACY_HOLD_CATEGORIES, clampActiveWorkIntervalMs, type ActiveWorkHoldV1, CHANNEL_STARTUP_PROFILE_META_KEY, @@ -4346,12 +4347,10 @@ class QwenAgent implements Agent { const cancelClose = opts?.waitForCloseGate ? await beginSessionCloseAfterCurrentGate(session, drainTimeoutMs) : session.beginClose(); - // Checked under the close gate and before anything destructive runs. The - // gate is what makes this atomic: with it held the Session admits no new - // prompt and starts no new automatic turn, so a hold cannot appear between - // this read and the teardown below. Without it, a caller that asked - // "anything running?" and then closed would race exactly the work it was - // trying to protect. + // Reject known work before disturbing any active turn. The close gate + // prevents new turns, but a turn that was already running can still settle + // into a new hold while the drain below is in progress, so this early read + // is an optimization rather than the final authorization. if (opts?.onlyIfUnheld) { const holds = session.collectActiveWorkHolds(); if (holds.length > 0) { @@ -4383,6 +4382,17 @@ class QwenAgent implements Agent { 'close', ); + // Existing out-of-scope work such as a cron turn may have registered a + // background shell while it drained. Re-check after every active turn + // has settled and while the close gate still blocks new ones; only this + // read can authorize the destructive recorder/session cleanup below. + if (opts?.onlyIfUnheld) { + const holds = session.collectActiveWorkHolds(); + if (holds.length > 0) { + return { closed: false, holds }; + } + } + recorder?.finalize(); let flushError: unknown; try { @@ -4671,12 +4681,23 @@ class QwenAgent implements Agent { (requestedActiveWork as Record)['intervalMs'], ) : undefined; + const requestedActiveWorkCategories = activeWorkRequested + ? (requestedActiveWork as Record)['categories'] + : undefined; + const activeWorkCategories = activeWorkRequested + ? Array.isArray(requestedActiveWorkCategories) + ? ACTIVE_WORK_HOLD_CATEGORIES.filter((category) => + requestedActiveWorkCategories.includes(category), + ) + : ACTIVE_WORK_LEGACY_HOLD_CATEGORIES + : undefined; if (activeWorkIntervalMs !== undefined) { this.activeWorkReporter?.dispose(); this.activeWorkReporter = new ActiveWorkReporter( (method, params) => this.connection.extNotification(method, params), () => this.sessions.values(), activeWorkIntervalMs, + activeWorkCategories ?? [], ); } @@ -4695,7 +4716,7 @@ class QwenAgent implements Agent { [ACTIVE_WORK_HEARTBEAT_META_KEY]: { v: ACTIVE_WORK_HEARTBEAT_VERSION, intervalMs: activeWorkIntervalMs, - categories: [...ACTIVE_WORK_HOLD_CATEGORIES], + categories: [...(activeWorkCategories ?? [])], }, } : {}), @@ -5531,8 +5552,9 @@ class QwenAgent implements Agent { // Order a fresh snapshot ahead of this response on the same stream. The // daemon drops its own pending-prompt count the instant the response // lands, so any hold this prompt left behind — a background agent it - // started, its terminal notification — has to already be on the wire or - // the daemon sees an idle Session for as long as the next report takes. + // started, a background shell, or a terminal notification — has to + // already be on the wire or the daemon sees an idle Session for as long + // as the next report takes. await this.activeWorkReporter?.flush(); } } diff --git a/packages/cli/src/acp-integration/active-work-reporter.test.ts b/packages/cli/src/acp-integration/active-work-reporter.test.ts index 1b36207def3..b5a2304ea57 100644 --- a/packages/cli/src/acp-integration/active-work-reporter.test.ts +++ b/packages/cli/src/acp-integration/active-work-reporter.test.ts @@ -7,6 +7,7 @@ import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; import { ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, ACTIVE_WORK_NOTIFICATION_METHOD, type ActiveWorkHoldV1, type ActiveWorkSnapshotV1, @@ -63,6 +64,7 @@ describe('ActiveWorkReporter', () => { source('s2', () => []), ], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); await reporter.flush(); @@ -76,7 +78,12 @@ describe('ActiveWorkReporter', () => { }); it('increases seq monotonically across reports', async () => { - const reporter = new ActiveWorkReporter(send, () => [], INTERVAL_MS); + const reporter = new ActiveWorkReporter( + send, + () => [], + INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, + ); await reporter.flush(); await reporter.flush(); @@ -86,6 +93,26 @@ describe('ActiveWorkReporter', () => { reporter.dispose(); }); + it('filters only the wire snapshot to negotiated categories', async () => { + const collect = vi.fn().mockReturnValue([ + { category: 'agent', id: 'a1' }, + { category: 'shell', id: 'background-shells' }, + ] satisfies ActiveWorkHoldV1[]); + const reporter = new ActiveWorkReporter( + send, + () => [source('s1', collect)], + INTERVAL_MS, + ['agent', 'notification'], + ); + await reporter.flush(); + + expect(collect).toHaveBeenCalled(); + expect(snapshots().at(-1)?.sessions).toEqual([ + { sessionId: 's1', holds: [{ category: 'agent', id: 'a1' }] }, + ]); + reporter.dispose(); + }); + describe('when a source throws while collecting', () => { it('does not let the failure escape the interval timer', async () => { vi.useFakeTimers(); @@ -96,6 +123,7 @@ describe('ActiveWorkReporter', () => { send, () => [throwingSource('s1')], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); // The constructor already published once; drive several more ticks. await vi.advanceTimersByTimeAsync(INTERVAL_MS * 3); @@ -111,6 +139,7 @@ describe('ActiveWorkReporter', () => { send, () => [throwingSource('s1')], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); reporter.notifyChanged(); // The coalesced publish runs in a microtask; if it threw, this await @@ -125,6 +154,7 @@ describe('ActiveWorkReporter', () => { send, () => [throwingSource('s1')], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); await expect(reporter.flush()).resolves.toBeUndefined(); reporter.dispose(); @@ -141,6 +171,7 @@ describe('ActiveWorkReporter', () => { throwingSource('broken'), ], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); await reporter.flush(); @@ -158,6 +189,7 @@ describe('ActiveWorkReporter', () => { : source('s1', () => [{ category: 'agent', id: 'a1' }]), ], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); await reporter.flush(); expect(snapshots()).toHaveLength(0); @@ -180,6 +212,7 @@ describe('ActiveWorkReporter', () => { }, () => [source('s1', () => [])], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); await expect(reporter.flush()).resolves.toBeUndefined(); await expect(reporter.flush()).resolves.toBeUndefined(); @@ -192,6 +225,7 @@ describe('ActiveWorkReporter', () => { send, () => [source('s1', () => [])], INTERVAL_MS, + ACTIVE_WORK_HOLD_CATEGORIES, ); const before = snapshots().length; reporter.dispose(); diff --git a/packages/cli/src/acp-integration/active-work-reporter.ts b/packages/cli/src/acp-integration/active-work-reporter.ts index 086da22f800..8f182edf12a 100644 --- a/packages/cli/src/acp-integration/active-work-reporter.ts +++ b/packages/cli/src/acp-integration/active-work-reporter.ts @@ -8,6 +8,7 @@ import { createDebugLogger } from '@qwen-code/qwen-code-core'; import { ACTIVE_WORK_HEARTBEAT_VERSION, ACTIVE_WORK_NOTIFICATION_METHOD, + type ActiveWorkHoldCategory, type ActiveWorkHoldV1, type ActiveWorkSnapshotV1, } from '@qwen-code/acp-bridge/bridgeTypes'; @@ -51,12 +52,15 @@ export class ActiveWorkReporter { #coalescing = false; #timer: ReturnType | undefined; #disposed = false; + readonly #enabledCategories: ReadonlySet; constructor( private readonly send: SendNotification, private readonly listSources: () => Iterable, readonly intervalMs: number, + enabledCategories: readonly ActiveWorkHoldCategory[], ) { + this.#enabledCategories = new Set(enabledCategories); this.#timer = setInterval(() => { this.#publish(); }, intervalMs); @@ -119,7 +123,9 @@ export class ActiveWorkReporter { for (const source of this.listSources()) { sessions.push({ sessionId: source.sessionId, - holds: source.collectActiveWorkHolds(), + holds: source + .collectActiveWorkHolds() + .filter((hold) => this.#enabledCategories.has(hold.category)), }); } } catch (error) { diff --git a/packages/cli/src/acp-integration/session/Session.review-lease.test.ts b/packages/cli/src/acp-integration/session/Session.review-lease.test.ts index 8369b20e641..06ac046192a 100644 --- a/packages/cli/src/acp-integration/session/Session.review-lease.test.ts +++ b/packages/cli/src/acp-integration/session/Session.review-lease.test.ts @@ -153,6 +153,9 @@ describe('Session review-worktree lease sweep', () => { }), getBackgroundShellRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), + hasRunningEntries: vi.fn().mockReturnValue(false), }), setSubSessionSpawner: vi.fn(), getSubSessionSpawner: vi.fn(), diff --git a/packages/cli/src/acp-integration/session/Session.test.ts b/packages/cli/src/acp-integration/session/Session.test.ts index 45106c16344..6eb5aa0e1a0 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -436,6 +436,9 @@ describe('Session', () => { }; let mockBackgroundShellRegistry: { setNotificationCallback: ReturnType; + setStatusChangeCallback: ReturnType; + clearStatusChangeCallback: ReturnType; + hasRunningEntries: ReturnType; getAll: ReturnType; }; let mockToolRegistry: { @@ -627,6 +630,9 @@ describe('Session', () => { }; mockBackgroundShellRegistry = { setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), + hasRunningEntries: vi.fn().mockReturnValue(false), getAll: vi.fn().mockReturnValue([]), }; mockWorkflowRunRegistry = { @@ -924,7 +930,7 @@ describe('Session', () => { ); } - function holdIds(category: 'agent' | 'notification'): string[] { + function holdIds(category: 'agent' | 'notification' | 'shell'): string[] { return session .collectActiveWorkHolds() .filter((hold) => hold.category === category) @@ -975,6 +981,80 @@ describe('Session', () => { ).not.toHaveBeenCalledWith(undefined); }); + it('represents any number of running shells with one aggregate hold', () => { + createReportingSession(); + mockBackgroundShellRegistry.hasRunningEntries.mockImplementation(() => + ( + mockBackgroundShellRegistry.getAll() as Array<{ status: string }> + ).some((entry) => entry.status === 'running'), + ); + mockBackgroundShellRegistry.getAll.mockReturnValue( + Array.from({ length: 2_000 }, (_unused, index) => ({ + id: `shell-${index}`, + status: 'running', + })), + ); + + expect(holdIds('shell')).toEqual(['background-shells']); + expect(session.collectActiveWorkHolds()).toHaveLength(1); + expect(session.isIdle()).toBe(false); + + mockBackgroundShellRegistry.getAll.mockReturnValue([]); + expect(session.collectActiveWorkHolds()).toEqual([]); + expect(session.isIdle()).toBe(true); + session.dispose(); + }); + + it('tracks shell status changes and retracts only its callback', () => { + createReportingSession(); + const statusChanged = + mockBackgroundShellRegistry.setStatusChangeCallback.mock.calls.at( + -1, + )?.[0] as (() => void) | undefined; + const before = changes; + statusChanged?.(); + expect(changes).toBe(before + 1); + + session.dispose(); + expect( + mockBackgroundShellRegistry.clearStatusChangeCallback, + ).toHaveBeenCalledWith(statusChanged); + expect( + mockBackgroundShellRegistry.setStatusChangeCallback, + ).not.toHaveBeenCalledWith(undefined); + }); + + it('holds a queued shell notification before its continuation can start', async () => { + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + createReportingSession(); + const releaseCloseGate = session.beginClose(); + const notify = + mockBackgroundShellRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { shellId: string; status: string }, + ) => void; + + notify('Shell completed.', '', { + shellId: 'shell-queued', + status: 'completed', + }); + + expect(mockChat.sendMessageStream).not.toHaveBeenCalled(); + expect(holdIds('shell')).toEqual(['background-shells']); + expect(session.isIdle()).toBe(false); + + releaseCloseGate(); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + session.dispose(); + }); + it('holds an Agent terminal notification from persistence to continuation', async () => { let finishPersistence!: () => void; mockChatRecordingService.recordNotificationStrict.mockImplementationOnce( @@ -1093,6 +1173,129 @@ describe('Session', () => { expect(session.isIdle()).toBe(true); session.dispose(); }); + + it('keeps a shell hold across queue-to-continuation handoff', async () => { + let releaseNotification!: () => void; + const notificationGate = new Promise((resolve) => { + releaseNotification = resolve; + }); + async function* notificationStream() { + yield { + type: core.StreamEventType.CHUNK, + value: { + candidates: [{ content: { parts: [{ text: 'working' }] } }], + }, + }; + await notificationGate; + } + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(notificationStream()); + createReportingSession(); + const notify = + mockBackgroundShellRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { shellId: string; status: string }, + ) => void; + + notify('Shell completed.', '', { + shellId: 'shell-1', + status: 'completed', + }); + await vi.waitFor(() => + expect(holdIds('shell')).toEqual(['background-shells']), + ); + expect(session.isIdle()).toBe(false); + + releaseNotification(); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + expect(session.isIdle()).toBe(true); + session.dispose(); + }); + + it('releases the shell hold after a continuation failure', async () => { + let rejectNotification!: (reason: Error) => void; + mockChat.sendMessageStream = vi.fn().mockReturnValue( + new Promise((_resolve, reject) => { + rejectNotification = reject; + }), + ); + createReportingSession(); + const notify = + mockBackgroundShellRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { shellId: string; status: string }, + ) => void; + + notify('Shell failed.', '', { + shellId: 'shell-1', + status: 'failed', + }); + await vi.waitFor(() => + expect(holdIds('shell')).toEqual(['background-shells']), + ); + + rejectNotification(new Error('continuation failed')); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + session.dispose(); + }); + + it('releases the shell hold after a cancelled continuation exits', async () => { + let releaseNotification!: () => void; + const notificationGate = new Promise((resolve) => { + releaseNotification = resolve; + }); + async function* notificationStream() { + yield { + type: core.StreamEventType.CHUNK, + value: { + candidates: [{ content: { parts: [{ text: 'working' }] } }], + }, + }; + await notificationGate; + } + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(notificationStream()); + createReportingSession(); + const notify = + mockBackgroundShellRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { shellId: string; status: string }, + ) => void; + + notify('Shell completed.', '', { + shellId: 'shell-1', + status: 'completed', + }); + await vi.waitFor(() => + expect(holdIds('shell')).toEqual(['background-shells']), + ); + await vi.waitFor(() => + expect(mockChat.sendMessageStream).toHaveBeenCalledOnce(), + ); + + await session.cancelPendingPrompt(); + expect(holdIds('shell')).toEqual(['background-shells']); + releaseNotification(); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + session.dispose(); + }); }); it('bridges workflow approvals through ACP permission requests', async () => { diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index f068c355e6d..87223fdfa3b 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -1610,6 +1610,7 @@ export class Session implements SessionContext { private notificationAbortController: AbortController | null = null; private notificationCompletion: Promise | null = null; private currentAgentNotificationTaskId: string | null = null; + private currentShellNotificationActive = false; private readonly persistedBackgroundNotificationTaskIds = new Set(); private readonly backgroundNotificationAcceptances = new Map< string, @@ -1639,6 +1640,7 @@ export class Session implements SessionContext { /** The exact status-change callback this Session installed, so dispose can * retract its own and nobody else's. */ #statusChangeCallback: (() => void) | undefined; + #shellStatusChangeCallback: (() => void) | undefined; private readonly workflowApprovalAbortController = new AbortController(); private activeTodoPlanRevision?: { planId: string; @@ -2952,6 +2954,13 @@ export class Session implements SessionContext { for (const taskId of notificationIds) { holds.push({ category: 'notification', id: taskId }); } + const shellActive = + this.config.getBackgroundShellRegistry().hasRunningEntries() || + this.notificationQueue.some((item) => item.kind === 'shell') || + this.currentShellNotificationActive; + if (shellActive) { + holds.push({ category: 'shell', id: 'background-shells' }); + } return holds; } @@ -3107,7 +3116,12 @@ export class Session implements SessionContext { this.#statusChangeCallback = undefined; } this.config.getMonitorRegistry().setNotificationCallback(undefined); - this.config.getBackgroundShellRegistry().setNotificationCallback(undefined); + const shellRegistry = this.config.getBackgroundShellRegistry(); + shellRegistry.setNotificationCallback(undefined); + if (this.#shellStatusChangeCallback) { + shellRegistry.clearStatusChangeCallback(this.#shellStatusChangeCallback); + this.#shellStatusChangeCallback = undefined; + } this.config.getChatRecordingService()?.setTitleRecordedCallback(undefined); this.unsubscribeChatRecordingFailure?.(); this.unsubscribeChatRecordingFailure = undefined; @@ -7163,6 +7177,10 @@ export class Session implements SessionContext { }); const shellRegistry = this.config.getBackgroundShellRegistry(); + this.#shellStatusChangeCallback = () => { + this.#activeWorkChanged(); + }; + shellRegistry.setStatusChangeCallback(this.#shellStatusChangeCallback); shellRegistry.setNotificationCallback((displayText, modelText, meta) => { this.#enqueueBackgroundNotification({ displayText, @@ -7382,6 +7400,7 @@ export class Session implements SessionContext { if (!item) break; this.currentAgentNotificationTaskId = item.kind === 'agent' ? item.taskId : null; + this.currentShellNotificationActive = item.kind === 'shell'; this.#activeWorkChanged(); try { await runWithInvocationContext(undefined, () => @@ -7391,6 +7410,7 @@ export class Session implements SessionContext { ); } finally { this.currentAgentNotificationTaskId = null; + this.currentShellNotificationActive = false; this.#activeWorkChanged(); } } diff --git a/packages/cli/src/acp-integration/session/Session.worktree.test.ts b/packages/cli/src/acp-integration/session/Session.worktree.test.ts index dfb74ac0897..9600e3a4c4f 100644 --- a/packages/cli/src/acp-integration/session/Session.worktree.test.ts +++ b/packages/cli/src/acp-integration/session/Session.worktree.test.ts @@ -173,6 +173,9 @@ describe('Session.pendingWorktreeNotice', () => { }), getBackgroundShellRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), + hasRunningEntries: vi.fn().mockReturnValue(false), }), setSubSessionSpawner: vi.fn(), getSubSessionSpawner: vi.fn(), diff --git a/packages/core/src/services/backgroundShellRegistry.test.ts b/packages/core/src/services/backgroundShellRegistry.test.ts index 730a958ad63..2c6fe32df0b 100644 --- a/packages/core/src/services/backgroundShellRegistry.test.ts +++ b/packages/core/src/services/backgroundShellRegistry.test.ts @@ -185,6 +185,26 @@ describe('BackgroundShellRegistry', () => { }); describe('callbacks', () => { + it('clears the status callback only when identities match', () => { + const reg = new BackgroundShellRegistry(); + const installed = vi.fn(); + const replacement = vi.fn(); + + reg.setStatusChangeCallback(installed); + reg.clearStatusChangeCallback(replacement); + reg.register(makeEntry({ shellId: 'a' })); + expect(installed).toHaveBeenCalledOnce(); + + reg.setStatusChangeCallback(replacement); + reg.clearStatusChangeCallback(installed); + reg.register(makeEntry({ shellId: 'b' })); + expect(replacement).toHaveBeenCalledOnce(); + + reg.clearStatusChangeCallback(replacement); + reg.register(makeEntry({ shellId: 'c' })); + expect(replacement).toHaveBeenCalledOnce(); + }); + it('fires register callback synchronously when an entry is added', () => { const reg = new BackgroundShellRegistry(); const seen: string[] = []; @@ -673,6 +693,11 @@ describe('BackgroundShellRegistry', () => { expect(e.status).toBe('running'); expect(e.endTime).toBeUndefined(); expect(ac.signal.aborted).toBe(true); + expect(reg.hasRunningEntries()).toBe(true); + + reg.cancel('a', 2000); + + expect(reg.hasRunningEntries()).toBe(false); }); it('is a no-op on a terminal entry', () => { diff --git a/packages/core/src/services/backgroundShellRegistry.ts b/packages/core/src/services/backgroundShellRegistry.ts index 58ed52f3c62..1f48efdc5aa 100644 --- a/packages/core/src/services/backgroundShellRegistry.ts +++ b/packages/core/src/services/backgroundShellRegistry.ts @@ -241,9 +241,9 @@ export type BackgroundShellNotificationCallback = ( ) => void; /** - * Fires on every status transition (running → terminal). Symmetric with - * `BackgroundTaskRegistry.setStatusChangeCallback` so the same UI hook can - * subscribe to both registries. + * Fires after registration and every status transition (running → + * terminal). Symmetric with `BackgroundTaskRegistry.setStatusChangeCallback` + * so the same UI hook can subscribe to both registries. */ export type BackgroundShellStatusChangeCallback = (entry?: ShellTask) => void; @@ -257,8 +257,8 @@ export class BackgroundShellRegistry { /** * Subscribe to new-entry events. Called synchronously inside `register()`. * Setting `undefined` clears the existing subscriber. Single-subscriber on - * purpose — the UI hook is the only consumer in the codebase, and a list - * would invite drift in error-handling. + * purpose — each runtime installs one owner callback, and a list would + * invite drift in error-handling. */ setRegisterCallback(cb: BackgroundShellRegisterCallback | undefined): void { this.registerCallback = cb; @@ -271,10 +271,9 @@ export class BackgroundShellRegistry { } /** - * Subscribe to status transitions (running → terminal). Called - * synchronously inside `complete()` / `fail()` / `cancel()` after the - * entry has been mutated. Same single-subscriber rationale as - * `setRegisterCallback`. + * Subscribe to registration and status transitions (running → terminal). + * Called synchronously after the registry has been mutated. Same + * single-subscriber rationale as `setRegisterCallback`. */ setStatusChangeCallback( cb: BackgroundShellStatusChangeCallback | undefined, @@ -282,6 +281,13 @@ export class BackgroundShellRegistry { this.statusChangeCallback = cb; } + /** Retract `cb` only if it is still the installed callback. */ + clearStatusChangeCallback(cb: BackgroundShellStatusChangeCallback): void { + if (this.statusChangeCallback === cb) { + this.statusChangeCallback = undefined; + } + } + register(registration: ShellTaskRegistration): ShellTask { // Mutate the registration in place to graduate it to a `ShellTask`. // Returning the same reference keeps the existing call sites that @@ -597,8 +603,8 @@ export class BackgroundShellRegistry { * statusChange callback exactly once after the loop. The per-entry * `cancel()` path would have triggered both side channels for every * running shell — wasteful on shutdown / `/clear` where the only - * subscriber (`useBackgroundTaskView`) just re-pulls `getAll()` - * regardless of the entry argument. + * current subscriber just re-pulls the registry regardless of the entry + * argument. */ abortAll(): void { const endTime = Date.now(); @@ -610,10 +616,9 @@ export class BackgroundShellRegistry { } if (!lastCancelled) return; this.pruneTerminalEntries(); - // The single subscriber (`useBackgroundTaskView`) ignores the entry - // arg and re-pulls `getAll()`, so passing the last cancelled entry - // here is informational only — any of the just-cancelled entries - // would be equally valid as the "what changed" signal. + // The current subscriber re-pulls the registry, so passing the last + // cancelled entry here is informational only — any of the just-cancelled + // entries would be equally valid as the "what changed" signal. this.fireStatusChange(lastCancelled); } }