diff --git a/docs/design/2026-08-06-active-work-health.md b/docs/design/2026-08-06-active-work-health.md new file mode 100644 index 00000000000..91939b8f12f --- /dev/null +++ b/docs/design/2026-08-06-active-work-health.md @@ -0,0 +1,125 @@ +# Active-work health signal + +## Problem + +`activePrompts` counts prompts currently dispatched to an ACP child. A prompt can finish after starting background Agents, leaving `activePrompts` at zero while session-owned work is still running. A restart controller that reads zero active prompts as idle can restart the daemon before those Agents finish and before their terminal notifications reach the parent session. + +## Scope + +`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. + +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. + +Restart policy stays with the external controller. The daemon publishes facts; it does not publish `restartSafe`. + +## 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: + +**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. + +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. + +**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 +{ + "v": 1, + "seq": 12, + "sessions": [ + { "sessionId": "…", "holds": [{ "category": "agent", "id": "a1b2" }] } + ] +} +``` + +A dropped report therefore costs one interval of staleness and needs no retransmit, ack, or "last reported" state to diff against — the next snapshot is the whole truth again. `seq` guards against reordering only; a gap is not an error. Channel scope is what keeps an always-on cadence affordable (one small message per interval regardless of Session count) and it gives the daemon a second fact for free: because the report is complete, a Session **absent** from a fresh snapshot holds nothing on the child side. Absence and reported-with-no-holds are therefore the same fact and take the same path — one that ends in asking the child, never in assuming. + +Prompts are absent from the child's report on purpose. The daemon accepts, queues, dispatches, and settles them, so its own `pendingPromptCount` is authoritative and strictly wider — it covers prompts still waiting in the FIFO, which the child cannot see. Reporting them from both sides would create two sources of truth for one fact with nothing to reconcile them. + +## 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. + +## Three 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. +- **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. + +Reclaiming a channel that has stopped answering entirely is still not this mechanism's job; see below. + +The cache decides _when_ it is worth asking. It never authorizes destruction, because a fresh empty snapshot only describes the moment it was built and work can start in the gap. So automatic cleanup closes through a conditional RPC: + +``` +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 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. + +On timeout the daemon cannot tell whether the child closed. It does not retry in place and does not assume: it leaves the Session alone and lets the next snapshot settle it. Absence from that snapshot is not consent to destroy — it makes the Session a candidate, and the candidate still has to clear every ordinary guard (no SSE subscriber, no registered client, nothing daemon-owned in flight) before the daemon asks the child once more. A child that already closed the Session answers `closed` for a Session it no longer has, which is how a lost close response is recovered without ever guessing. + +Explicit close, kill, shutdown, and channel exit keep their force semantics and do not go through this path. + +## One guard model, four triggers + +Four things can decide it is time to look at a Session: the last client detaching, a prompt settling, a terminal notification settling, and the idle reaper's TTL. Each brings its own policy, and none of them may weaken the shared part: + +| Guard | Why it is shared | +| ------------------------------------------- | -------------------------------------------------------------------------------------------------------------------- | +| 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 | +| 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 | + +The reaper deliberately ignores registered client ids — it exists for the crash path where a detach never arrived — but that is the only difference, and it still has to ask the child before destroying anything. + +## What this deliberately does not do + +There is no heartbeat watchdog and no channel kill driven by work state. Inferring "this channel is dead" from "one Session stopped reporting" kills every Session on that process, and a suspend, a long event-loop stall, or a single dropped notification all look identical to a stalled child. Three separate concerns, three separate mechanisms: + +| Concern | Mechanism | +| ------------------------------------------------------- | ----------------------------------------- | +| Transport / process liveness | channel ping-pong (separate change) | +| Agent logic stalling while the process stays responsive | progress-based watchdog (separate change) | +| Session work retention | this document | + +Killing a whole multiplexed channel is reasonable when the channel is _actually_ dead — every Session on it is unreachable anyway. It is not reasonable as an inference from one Session's reporting. + +## Health surface + +| Field | Meaning | +| --------------------- | --------------------------------------------------------------------- | +| `activeWork` | OR across runtimes of daemon-owned work and reported holds | +| `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. + +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. + +Controllers should treat the daemon as busy when: + +```ts +const busy = + health.activePrompts > 0 || + health.activeWork || + health.activeWorkReporting !== 'full'; +``` + +`activePrompts` keeps its exact previous meaning as an independent compatibility signal. + +## Limits + +This is an observation cache, not a restart lease. Even a fresh, empty, fully-graded snapshot describes the moment it was taken; new work can begin immediately afterwards. The rule above substantially lowers the risk of a wrong restart — it does not eliminate it. Strict safety needs a prepare-restart fence that stops new work admission, confirms the drain, and only then shuts down. That is graceful shutdown, and it is out of scope here. diff --git a/docs/design/daemon-global-deep-health.md b/docs/design/daemon-global-deep-health.md index f69d587b0b4..42a552a5d6d 100644 --- a/docs/design/daemon-global-deep-health.md +++ b/docs/design/daemon-global-deep-health.md @@ -23,6 +23,9 @@ that are draining but have not completed bridge cleanup. | `sessions` | Sum | | `pendingPermissions` | Sum | | `activePrompts` | Sum | +| `activeWork` | True when any managed runtime reports active work | +| `activeWorkReporting`| Worst grade across runtimes (`full`/`partial`/`none`) | +| `activeWorkStaleMs` | Age of the oldest snapshot behind it; `0` when uncovered | | `connectedClients` | Existing daemon-wide REST SSE count | | `channelAlive` | True when any managed runtime channel is live | | `lastActivityAt` | Latest non-null bridge activity time | diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 57426ead602..fc996bd1c6f 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -486,6 +486,9 @@ Pass `?deep=1` (also accepts `?deep=true` or bare `?deep`) for a daemon-wide pro "sessions": 3, "pendingPermissions": 1, "activePrompts": 1, + "activeWork": true, + "activeWorkReporting": "full", + "activeWorkStaleMs": 4200, "connectedClients": 2, "channelAlive": true, "lastActivityAt": "2026-07-15T08:30:00.000Z", @@ -493,9 +496,22 @@ Pass `?deep=1` (also accepts `?deep=true` or bare `?deep`) for a daemon-wide pro } ``` -`sessions`, `pendingPermissions`, and `activePrompts` are sums. `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` **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. -> ⚠️ The deep probe is **informational**, not a real liveness verification or an atomic reclaim lease. It reads counter accessors which don't ping individual child processes / channels and so won't detect a wedged-but-still-counted session. `connectedClients` counts REST SSE connections, not every ACP transport. Use repeated samples and graceful shutdown for idle reclamation; use authenticated `/daemon/status` for transport and per-workspace diagnostics. If any managed runtime getter throws, deep health fails closed with `503 {"status":"degraded","reason":"aggregation_failed"}` rather than returning partial totals, and the daemon log identifies the failing workspace runtime. During bootstrap, before the runtime registry is ready, it returns `503 {"status":"degraded","reason":"bootstrap"}` with `Retry-After: 1`. For listener liveness, use the default `/health` without `?deep`. +Restart controllers should treat the daemon as busy when: + +```ts +const busy = + health.activePrompts > 0 || + health.activeWork || + health.activeWorkReporting !== 'full'; +``` + +Dropping the third term makes `activeWork === false` indistinguishable from "no child told me anything", which is the one case where acting on it is unsafe. Unknown responses and failed probes must also prevent restart. `activePrompts` remains an independent compatibility signal. + +These fields are an observation cache, not a restart lease: even a fresh, fully-graded, empty answer describes the moment it was sampled, and work can start immediately afterwards. The rule above lowers the risk of a wrong restart substantially but does not eliminate it — strict safety needs a prepare-restart fence that stops new work admission, confirms the drain, and only then shuts down. + +> ⚠️ The deep probe is **informational**, not a real liveness verification or an atomic reclaim lease. Negotiated ACP children publish channel-wide active-work snapshots on a negotiated cadence, and the daemon grades their freshness into `activeWorkReporting` — but it never kills a channel over a missing report, because one session's silence is not evidence the process died. Transport liveness and stalled-Agent detection are separate mechanisms. `connectedClients` counts REST SSE connections, not every ACP transport. Use repeated samples and graceful shutdown for idle reclamation; use authenticated `/daemon/status` for transport and per-workspace diagnostics. If any managed runtime getter throws, deep health fails closed with `503 {"status":"degraded","reason":"aggregation_failed"}` rather than returning partial totals, and the daemon log identifies the failing workspace runtime. During bootstrap, before the runtime registry is ready, it returns `503 {"status":"degraded","reason":"bootstrap"}` with `Retry-After: 1`. For listener liveness, use the default `/health` without `?deep`. **Auth:** required **only on non-loopback binds**. On loopback (`127.0.0.1`, `::1`, `[::1]`) `/health` is registered before the bearer middleware so k8s/Compose probes inside the pod don't need to carry the token. On non-loopback (`--hostname 0.0.0.0` etc.) the route is registered after the bearer middleware and returns 401 without a valid token — otherwise an unauthenticated caller could probe arbitrary addresses to confirm a `qwen serve` exists, a low-severity info leak that combines poorly with port scanning. CORS deny + Host allowlist still apply on the loopback exemption. diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index db841e57216..bdede567a52 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -64,6 +64,17 @@ import { createInMemoryChannel } from './inMemoryChannel.js'; import { EventBus, type BridgeEvent } from './eventBus.js'; import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; import { + ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + ACTIVE_WORK_CLOSE_TIMEOUT_MS, + ACTIVE_WORK_HEARTBEAT_META_KEY, + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM, + ACTIVE_WORK_MAX_SESSION_HOLDS, + ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS, + ACTIVE_WORK_NOTIFICATION_METHOD, + ACTIVE_WORK_STALE_INTERVALS, + gradeActiveWorkCoverage, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, DAEMON_MODEL_PROMPT_META_KEY, @@ -109,7 +120,690 @@ function deferred(): { return { promise, resolve, reject }; } +async function sendActiveWorkSnapshot( + handle: { agentConnection: { extNotification: ExtNotificationFn } }, + seq: number, + sessions: Array<{ + sessionId: string; + holds: Array<{ category: string; id: string }>; + }>, +): Promise { + await handle.agentConnection.extNotification( + ACTIVE_WORK_NOTIFICATION_METHOD, + { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + seq, + sessions, + }, + ); +} + +type ExtNotificationFn = ( + method: string, + params: Record, +) => Promise; + +/** A cooperative child: conditional closes succeed, everything else no-ops. */ +const activeWorkCloseImpl = async (method: string) => + method === SERVE_CONTROL_EXT_METHODS.sessionClose + ? { closed: true, holds: [] } + : {}; + +function activeWorkInitializeResponse( + overrides: Record = {}, +): InitializeResponse { + return { + protocolVersion: PROTOCOL_VERSION, + agentInfo: { name: 'active-work-agent', version: '0' }, + authMethods: [], + agentCapabilities: {}, + _meta: { + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], + ...overrides, + }, + }, + }; +} + +function agentHold(id: string) { + return { category: 'agent' as const, id }; +} + +/** + * 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 + * for another one's unreported sessions), so tests grade through the same + * function the route uses instead of reimplementing the collapse. + */ +function reportingGrade(bridge: { + activeWorkCoverage: { + total: number; + covered: number; + onNegotiatedChannel: number; + }; +}): 'full' | 'partial' | 'none' { + return gradeActiveWorkCoverage(bridge.activeWorkCoverage); +} + describe('createAcpSessionBridge', () => { + describe('active work', () => { + it('negotiates the capability and counts accepted prompts locally', async () => { + const prompt = deferred(); + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + promptImpl: () => prompt.promise, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + expect(handle.agent.initializeCalls[0]?._meta).toMatchObject({ + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + }, + [CHANNEL_STARTUP_PROFILE_META_KEY]: { + v: CHANNEL_STARTUP_PROFILE_VERSION, + }, + }); + + // No snapshot has arrived yet, so the session is "unknown": busy rather + // than idle, and graded `partial` — the channel did negotiate, it just + // has not spoken yet, which is not the same as `none`. + expect(bridge.activeWork).toBe(true); + expect(reportingGrade(bridge)).toBe('partial'); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(false); + expect(reportingGrade(bridge)).toBe('full'); + + // Prompts are a daemon-owned fact: no child report is involved. + const running = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'start background work' }], + }); + expect(bridge.activeWork).toBe(true); + prompt.resolve({ stopReason: 'end_turn' }); + await running; + expect(bridge.activeWork).toBe(false); + + await bridge.shutdown(); + }); + + it('grades a child that never acknowledges the capability as none', async () => { + const handle = makeChannel({}); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + // Unsupported must not behave like unknown: an older child would + // otherwise pin every session as permanently busy and unreapable. + expect(bridge.activeWork).toBe(false); + expect(reportingGrade(bridge)).toBe('none'); + expect(bridge.activeWorkCoverage.oldestCoveredReportAt).toBeNull(); + + await bridge.detachClient(session.sessionId, session.clientId); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + + await bridge.shutdown(); + }); + + it('grades a child that omits a category as partial', async () => { + const handle = makeChannel({ + initializeImpl: () => + activeWorkInitializeResponse({ categories: ['agent'] }), + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(false); + expect(reportingGrade(bridge)).toBe('partial'); + + await bridge.shutdown(); + }); + + it('ignores reordered snapshots and foreign sessions', 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, 5, [ + { sessionId: session.sessionId, holds: [agentHold('a1')] }, + { sessionId: 'foreign-session', holds: [agentHold('a2')] }, + ]); + expect(bridge.activeWork).toBe(true); + + // Lower sequence: reordered, must not roll the state back. + await sendActiveWorkSnapshot(handle, 4, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(true); + + await sendActiveWorkSnapshot(handle, 6, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(false); + + await bridge.shutdown(); + }); + + it('rejects a malformed snapshot outright instead of applying part of it', 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: [agentHold('a1')] }, + ]); + expect(bridge.activeWork).toBe(true); + + await handle.agentConnection.extNotification( + ACTIVE_WORK_NOTIFICATION_METHOD, + { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + seq: 2, + sessions: [ + { sessionId: session.sessionId, holds: [{ category: 'bogus' }] }, + ], + }, + ); + // Partially applying it would have cleared the hold. + expect(bridge.activeWork).toBe(true); + + await bridge.shutdown(); + }); + + it('keeps a detached session until the child reports it unheld', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [agentHold('a1')] }, + ]); + await bridge.detachClient(session.sessionId, session.clientId); + expect(bridge.sessionCount).toBe(1); + + await sendActiveWorkSnapshot(handle, 2, [ + { sessionId: session.sessionId, holds: [] }, + ]); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + + await bridge.shutdown(); + }); + + it('does not close when the child refuses because work appeared', async () => { + let closeCalls = 0; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + closeCalls++; + // Work started between the snapshot and the close request — exactly + // the race a read-only "are you idle?" query could not close. + if (params['onlyIfUnheld'] === true) { + return { closed: false, holds: [agentHold('late-agent')] }; + } + return { closed: true, holds: [] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + await bridge.detachClient(session.sessionId, session.clientId); + + expect(closeCalls).toBeGreaterThan(0); + expect(bridge.sessionCount).toBe(1); + // The refusal's hold set is adopted, so the daemon now agrees it is busy. + expect(bridge.activeWork).toBe(true); + + await bridge.shutdown(); + }); + + it('leaves the session in place when the close request never resolves', async () => { + let closeAttempts = 0; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + closeAttempts++; + return new Promise(() => {}); + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + vi.useFakeTimers(); + try { + const detached = bridge + .detachClient(session.sessionId, session.clientId) + .catch(() => undefined); + await vi.advanceTimersByTimeAsync(ACTIVE_WORK_CLOSE_TIMEOUT_MS + 1_000); + await detached; + } finally { + vi.useRealTimers(); + } + + // Ambiguous outcome: the daemon neither retried nor assumed the child + // closed it. The session waits for the next snapshot to settle it. + expect(closeAttempts).toBe(1); + expect(bridge.sessionCount).toBe(1); + + await bridge.shutdown(); + }); + + it('keeps a session the child drops while a client is still registered', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.sessionCount).toBe(1); + + // Absence makes the session a teardown *candidate*, not a casualty. The + // spawn owner's client id is still registered, and that guard binds here + // exactly as it binds every other automatic close — otherwise one + // snapshot could destroy a session someone is holding. + await sendActiveWorkSnapshot(handle, 2, []); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.sessionCount).toBe(1); + + await bridge.shutdown(); + }); + + it('does not tear down a dropped session while an SSE subscriber watches', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const abort = new AbortController(); + const iter = bridge.subscribeEvents(session.sessionId, { + signal: abort.signal, + }); + const pending = iter[Symbol.asyncIterator]().next(); + await vi.waitFor(() => + expect( + bridge.getDaemonStatusSnapshot().sessions[0]?.subscriberCount, + ).toBe(1), + ); + await bridge.detachClient(session.sessionId, session.clientId); + expect(bridge.sessionCount).toBe(1); + + await sendActiveWorkSnapshot(handle, 1, []); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.sessionCount).toBe(1); + + abort.abort(); + await pending.catch(() => undefined); + await bridge.shutdown(); + }); + + it('recovers a lost close response once nothing local holds the session', async () => { + let closeAttempts = 0; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + // Count only conditional closes: the unconditional close that local + // teardown sends afterwards is a different request. + if (params?.[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] !== true) { + return { closed: true, holds: [] }; + } + closeAttempts++; + // First request's response is lost; the retry succeeds. + return closeAttempts === 1 + ? new Promise(() => {}) + : { closed: true, holds: [] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + + vi.useFakeTimers(); + try { + const detached = bridge + .detachClient(session.sessionId, session.clientId) + .catch(() => undefined); + await vi.advanceTimersByTimeAsync(ACTIVE_WORK_CLOSE_TIMEOUT_MS + 1_000); + await detached; + } finally { + vi.useRealTimers(); + } + expect(bridge.sessionCount).toBe(1); + + // The child omits the session it already closed. With no client left, + // the daemon asks once more rather than assuming, and this time gets an + // answer it can act on. + await sendActiveWorkSnapshot(handle, 2, []); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + expect(closeAttempts).toBe(2); + + await bridge.shutdown(); + }); + + it('asks the child about a session it has never reported on', async () => { + const closeCalls: Array> = []; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + if (params?.[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] === true) { + closeCalls.push(params); + } + return { closed: true, holds: [] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + // No snapshot has arrived, so the child's side is unknown. Health reads + // that as busy... + expect(bridge.activeWork).toBe(true); + // ...but unknown must not make the Session unreapable. Retaining without + // ever asking leaves nothing that can resolve it; the ask is bounded and + // still retains on any non-answer. + await bridge.detachClient(session.sessionId, session.clientId); + + await vi.waitFor(() => expect(closeCalls.length).toBe(1)); + expect(closeCalls[0]?.['sessionId']).toBe(session.sessionId); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + + await bridge.shutdown(); + }); + + it('keeps an unreported session when the child says it holds work', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + if (params?.[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] !== true) { + return { closed: true, holds: [] }; + } + return { closed: false, holds: [agentHold('never-reported')] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + // Asking is what resolves the unknown, and a refusal resolves it toward + // retention with the reason attached rather than leaving it a guess. + await bridge + .detachClient(session.sessionId, session.clientId) + .catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(bridge.sessionCount).toBe(1); + expect(bridge.activeWork).toBe(true); + expect(reportingGrade(bridge)).toBe('full'); + + await bridge.shutdown(); + }); + + it('treats a snapshot aged past the freshness window as no evidence', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + expect(bridge.activeWork).toBe(false); + expect(reportingGrade(bridge)).toBe('full'); + expect(bridge.activeWorkCoverage.oldestCoveredReportAt).not.toBeNull(); + + vi.useFakeTimers(); + try { + vi.setSystemTime( + Date.now() + + ACTIVE_WORK_HEARTBEAT_INTERVAL_MS * ACTIVE_WORK_STALE_INTERVALS + + 1_000, + ); + // A snapshot older than the grading window is the absence of evidence, + // not evidence of idleness — a background agent could have started at + // any point since. It must read as busy, exactly like never-reported, + // and it must not drag the reported staleness along with it. + expect(bridge.activeWork).toBe(true); + expect(reportingGrade(bridge)).toBe('partial'); + expect(bridge.activeWorkCoverage.oldestCoveredReportAt).toBeNull(); + } finally { + vi.useRealTimers(); + } + + await bridge.shutdown(); + }); + + it('refuses new prompts while a conditional close is in flight', async () => { + const closeRequested = deferred(); + const closeResponse = deferred>(); + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + closeRequested.resolve(); + return closeResponse.promise; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + + const detached = bridge + .detachClient(session.sessionId, session.clientId) + .catch(() => undefined); + await closeRequested.promise; + + // The teardown is authorized but not yet done. Admitting a prompt here + // would lose it: the close it raced is about to complete. `closing` alone + // does not cover this span, which is the whole reason the in-flight flag + // is checked alongside it. + await expect( + bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'race the teardown' }], + }), + ).rejects.toThrow(/closing/); + await expect( + bridge.rewindSession(session.sessionId, { promptId: 'whatever' }), + ).rejects.toThrow(/closing/); + + closeResponse.resolve({ closed: true, holds: [] }); + await detached; + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + + await bridge.shutdown(); + }); + + it('refuses a restore-path attach while a conditional close is in flight', async () => { + const closeRequested = deferred(); + const closeResponse = deferred>(); + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + closeRequested.resolve(); + return closeResponse.promise; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + + const detached = bridge + .detachClient(session.sessionId, session.clientId) + .catch(() => undefined); + await closeRequested.promise; + + // `session/load` is an admission path too. Guarding it on bare `closing` + // let a client attach inside the round trip and have the teardown it + // raced destroy the session under it. + await expect( + bridge.loadSession({ + sessionId: session.sessionId, + workspaceCwd: WS_A, + }), + ).rejects.toThrow(/closing/); + + closeResponse.resolve({ closed: true, holds: [] }); + await detached; + + await bridge.shutdown(); + }); + + it('discards an oversized snapshot whole rather than applying part of it', async () => { + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: activeWorkCloseImpl, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 0, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [agentHold('a1')] }, + ]); + expect(bridge.activeWork).toBe(true); + + // Over the per-session hold bound. Rejecting the packet outright is the + // safe direction: truncating it would look like work being released. + await sendActiveWorkSnapshot(handle, 2, [ + { + sessionId: session.sessionId, + holds: Array.from( + { length: ACTIVE_WORK_MAX_SESSION_HOLDS + 1 }, + (_unused, index) => agentHold(`h${index}`), + ), + }, + ]); + expect(bridge.activeWork).toBe(true); + + // Over the per-snapshot session bound, and claiming this session is idle. + await sendActiveWorkSnapshot(handle, 3, [ + { sessionId: session.sessionId, holds: [] }, + ...Array.from( + { length: ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS }, + (_unused, index) => ({ sessionId: `filler-${index}`, holds: [] }), + ), + ]); + expect(bridge.activeWork).toBe(true); + + // A well-formed snapshot still lands, so the bound rejects packets rather + // than wedging the channel. + await sendActiveWorkSnapshot(handle, 4, [ + { sessionId: session.sessionId, holds: [] }, + ]); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(1)); + expect(bridge.activeWork).toBe(false); + + await bridge.shutdown(); + }); + + it('asks the child before reaping an idle session', async () => { + const closeCalls: string[] = []; + const handle = makeChannel({ + initializeImpl: () => activeWorkInitializeResponse(), + extMethodImpl: async (method, params) => { + if (method !== SERVE_CONTROL_EXT_METHODS.sessionClose) return {}; + closeCalls.push(String(params?.['sessionId'])); + // A hold appeared after the cached empty snapshot — precisely the + // stale-empty race the reaper used to lose. + return { closed: false, holds: [agentHold('late-agent')] }; + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionReapIntervalMs: 20, + sessionIdleTimeoutMs: 20, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + await sendActiveWorkSnapshot(handle, 1, [ + { sessionId: session.sessionId, holds: [] }, + ]); + + // The reaper deliberately ignores `clientIds` (it exists for the crash + // path), so the cached empty snapshot is all it had to go on. + await vi.waitFor(() => expect(closeCalls.length).toBeGreaterThan(0)); + expect(closeCalls[0]).toBe(session.sessionId); + expect(bridge.sessionCount).toBe(1); + // The refusal is adopted, so the session now reads busy rather than + // being re-attempted from the same stale cache every tick. + expect(bridge.activeWork).toBe(true); + + await bridge.shutdown(); + }); + }); + it('streams workspace content without requiring a session', async () => { const completion = deferred>(); const handle = makeChannel({ @@ -19194,6 +19888,39 @@ describe('activePromptCount and lastActivityAt', () => { }); describe('createAcpSessionBridge — background notifications', () => { + it('counts a notification while the child is accepting it', async () => { + const acceptance = deferred>(); + const handle = makeChannel({ + extMethodImpl: async (method) => + method === SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification + ? acceptance.promise + : {}, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const notification = bridge.enqueueBackgroundNotification( + session.sessionId, + { + displayText: 'Worker completed.', + modelText: '', + taskId: 'worker-persisting', + status: 'completed', + kind: 'agent', + }, + ); + expect(bridge.activeWork).toBe(true); + await bridge.detachClient(session.sessionId, session.clientId); + expect(bridge.sessionCount).toBe(1); + + acceptance.resolve({ sessionId: session.sessionId, accepted: false }); + await notification; + expect(bridge.activeWork).toBe(false); + await vi.waitFor(() => expect(bridge.sessionCount).toBe(0)); + + await bridge.shutdown(); + }); + it('forwards a daemon-owned worker completion to the live parent session', async () => { const handle = makeChannel({ extMethodImpl: async (method, params) => diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 14e84d2f53f..bb34333027e 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -103,6 +103,17 @@ import { SESSION_SOURCE_META_KEY, } from './session-source.js'; import { + ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM, + ACTIVE_WORK_CLOSE_TIMEOUT_MS, + ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + ACTIVE_WORK_HEARTBEAT_META_KEY, + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_STALE_INTERVALS, + clampActiveWorkIntervalMs, + type ActiveWorkHeartbeatCapabilityV1, + type ActiveWorkHoldCategory, + type ActiveWorkSnapshotV1, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, DAEMON_CHANNEL_DELIVERY_META_KEY, @@ -476,6 +487,20 @@ interface ChannelInfo { * two-bit (alive, dying) state. */ isDying: boolean; + /** + * Negotiated active-work reporting for this channel, or `undefined` when the + * child never acknowledged the capability. `undefined` is *not* "idle": it + * means this channel contributes no active-work facts at all, so the + * daemon's reporting grade degrades and pre-existing cleanup behavior + * applies unchanged. Conflating the two would let an older child either + * pin every Session forever or look permanently idle. + */ + activeWork?: { + intervalMs: number; + categories: readonly ActiveWorkHoldCategory[]; + /** Highest snapshot sequence applied; guards against reordering only. */ + seq: number; + }; handshakeComplete: boolean; } @@ -518,6 +543,26 @@ interface SessionEntry { promptQueue: Promise; /** Accepted prompts that have not settled yet (queued + active). */ pendingPromptCount: number; + pendingAgentNotificationCount: number; + /** + * Last hold set the owning child reported for this Session, or `null` while + * the channel has negotiated reporting but has not yet been heard from. + * + * `null` (unknown) reads as *retained*, never as idle — but it is also the + * state that makes the daemon go ask, rather than a state it sits in + * forever. A channel that never negotiated leaves this `null` too; the + * `ChannelInfo.activeWork` presence check is what separates the two. + */ + childHolds: Map | null; + /** `Date.now()` of the snapshot behind `childHolds`; null while unknown. */ + childHoldsAt: number | null; + /** + * A close-if-unheld request is on the wire. Only one may be outstanding: on + * timeout the daemon cannot tell whether the child already closed, so it + * neither retries nor assumes — it clears this flag and lets the next + * snapshot settle the question. + */ + activeWorkCloseInFlight: boolean; /** * Detailed list of prompts accepted into the FIFO queue. Each entry * carries its `promptId`, summary, and an `abortController` so the @@ -1616,6 +1661,332 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { lastActivityTimestamp = Date.now(); } + /** + * Daemon-owned facts only: prompts this daemon accepted (queued *or* + * dispatched) and notifications it is currently pushing into the child. + * These never depend on the child reporting anything, which is why they + * stay authoritative for a channel that never negotiated. + */ + function entryHasLocalWork(entry: SessionEntry): boolean { + return ( + entry.pendingPromptCount > 0 || entry.pendingAgentNotificationCount > 0 + ); + } + + /** + * Whether the child's cached hold set is recent enough to be evidence of + * anything. Anything older than the grading window is not a report that the + * Session is idle, it is the absence of a report. + */ + function childHoldsAreFresh( + entry: SessionEntry, + capability: Pick, + ): boolean { + if (entry.childHoldsAt === null) return false; + return ( + Date.now() - entry.childHoldsAt <= + capability.intervalMs * ACTIVE_WORK_STALE_INTERVALS + ); + } + + /** + * Whether the child has told us, recently enough to count, that it is + * holding work for this Session. Positive knowledge only — a channel that + * never negotiated and one that has gone quiet both answer `false` here, + * because neither is a report *of work*. + */ + function childReportsHeldWork(entry: SessionEntry): boolean { + const owner = channelInfoForEntry(entry); + if (!owner?.activeWork) return false; + if (!childHoldsAreFresh(entry, owner.activeWork)) return false; + return entry.childHolds !== null && entry.childHolds.size > 0; + } + + /** + * Whether the child's side of this Session's state is currently unknown: + * the channel negotiated reporting, but no snapshot recent enough to grade + * has arrived. Never-reported and gone-quiet are the same state on purpose — + * a snapshot from ten minutes ago says nothing about whether a background + * agent started since. + * + * A channel that never negotiated is not "unknown", it is out of scope: + * treating it as unknown would make every legacy Session unreapable. + */ + function childWorkIsUnknown(entry: SessionEntry): boolean { + const owner = channelInfoForEntry(entry); + if (!owner?.activeWork) return false; + return !childHoldsAreFresh(entry, owner.activeWork); + } + + /** + * Whether this Session counts as busy for the health surface. + * + * Fails closed on ignorance: unknown reads the same as busy, because a + * controller must not be able to mistake "nobody told me" for "nothing is + * running". The reporting grade published alongside is what lets a caller + * tell those two apart when it needs to. + */ + function entryHasActiveWork(entry: SessionEntry): boolean { + return ( + entryHasLocalWork(entry) || + childReportsHeldWork(entry) || + childWorkIsUnknown(entry) + ); + } + + /** + * The guards every automatic teardown shares, whichever policy decided it + * was time to look. Each caller adds its own policy on top (the reaper its + * TTL, the detach path its client bookkeeping) but none of them may skip + * these. + * + * Note what is deliberately *not* here: `childWorkIsUnknown`. Unknown is not + * a reason to skip, it is a reason to ask — the candidate goes on to + * `confirmChildUnheld`, and the child answers authoritatively under its own + * close gate whether or not its snapshots are arriving. Skipping on unknown + * instead would retain such a Session forever, with no path that ever + * resolves it; asking costs one bounded round trip and still retains on any + * non-answer. Only *known* work — daemon-owned, or a fresh report of held + * work — blocks the attempt outright. + * + * `activeWorkCloseInFlight` is in here because a conditional close is a + * multi-step, awaited sequence: while one is outstanding this Session is + * already a teardown candidate under consideration, and a second path + * evaluating it concurrently would either duplicate the round trip or race + * its own guards against the first one's outcome. + */ + function entryIsAutoCloseCandidate(entry: SessionEntry): boolean { + if (byId.get(entry.sessionId) !== entry) return false; + if (isClosingOrAuthorizingClose(entry)) return false; + if (entry.events.subscriberCount > 0) return false; + if (entryHasLocalWork(entry)) return false; + // A restore in flight looks exactly like an abandoned Session and is the + // opposite of one. `session/load` registers the entry before it awaits + // `artifacts.restore()` and `seedSessionUpdates()`, and its first client + // is registered only after those resolve — so for that whole window there + // are no clients, no subscribers, and nothing held, and the child answers + // the conditional close truthfully. Excluded here rather than at the + // snapshot trigger so every automatic path is covered: the reaper's TTL + // can elapse inside a slow restore too. + const owner = channelInfoForEntry(entry); + if (owner?.pendingRestoreIds.has(entry.sessionId)) return false; + return !childReportsHeldWork(entry); + } + + /** + * Whether this Session is off-limits to new work. + * + * Two states, one meaning. `closing` is teardown already under way; + * `activeWorkCloseInFlight` is teardown authorized and being confirmed. Both + * must refuse admission, or a prompt accepted during the confirmation round + * trip is lost when the teardown it raced completes. + */ + function isClosingOrAuthorizingClose(entry: SessionEntry): boolean { + return entry.closing || entry.activeWorkCloseInFlight; + } + + /** + * Single decision point for "this Session is detached and has nothing left + * to do — let it go". + * + * Every automatic cleanup path funnels through here (last-client detach, + * prompt settle, notification settle, a child reporting itself idle) so the + * preservation rule lives in exactly one place. Explicit close, kill, and + * shutdown deliberately do NOT come through here: they keep their force + * semantics. + */ + async function maybeCloseIdleSession( + entry: SessionEntry, + reason: string, + ): Promise { + if (!entryIsAutoCloseCandidate(entry)) return; + // Note the asymmetry, preserved from the call sites this replaces: the + // kill path keys off `attachCount`, the close path off `clientIds`. A + // spawn owner that asked for a kill gets one once nothing is attached, + // even if some client id is still registered. + if (entry.spawnOwnerWantedKill && entry.attachCount === 0) { + await bridgeApi.killSession(entry.sessionId).catch(() => { + /* best-effort; channel.exited will eventually reap anyway */ + }); + return; + } + if (entry.clientIds.size > 0) return; + await closeIfChildUnheld(entry, { + trigger: reason, + closeReason: 'last_client_detached', + }); + } + + /** + * Confirm with the child, then tear down locally — holding the in-flight flag + * across both steps. + * + * The span matters. `closeSessionImpl` sets `entry.closing` synchronously, so + * once teardown starts the ordinary close gate covers the rest; but the + * conditional-close round trip in front of it is an await of up to + * `ACTIVE_WORK_CLOSE_TIMEOUT_MS`. Leaving that span unmarked is what would + * let a client attach, prompt, or rewind into a Session that has already been + * authorized for destruction. Every admission path therefore checks this flag + * alongside `closing`, which is what restores the atomicity a single + * synchronous guard-then-teardown sequence used to give for free. + */ + async function closeIfChildUnheld( + entry: SessionEntry, + opts: { trigger: string; closeReason: string }, + ): Promise { + entry.activeWorkCloseInFlight = true; + try { + if (!(await confirmChildUnheld(entry))) return; + // Re-check identity, not just liveness. `closeSessionImpl` re-resolves + // the target by raw id, and the id can be re-registered to a *different* + // entry during the round trip — an explicit kill removes this one (kill + // deliberately ignores the in-flight flag, keeping its force semantics) + // and a `session/load` for the same persisted id registers a fresh one. + // Without this, the stale continuation tears down the newly restored + // Session under its just-attached client. + if (byId.get(entry.sessionId) !== entry) return; + await closeSessionImpl(entry.sessionId, undefined, { + reason: opts.closeReason, + }).catch((err) => { + writeStderrLine( + `qwen serve: deferred close (${opts.trigger}) failed for ` + + `${JSON.stringify(entry.sessionId)}: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`, + ); + }); + } finally { + entry.activeWorkCloseInFlight = false; + } + } + + /** + * Ask the owning child to close this Session only if it holds nothing, and + * report whether the daemon may now finish its own teardown. + * + * The cached hold set says what was true when the last snapshot was built, + * which is not the same as what is true now — new work can start in the gap. + * So the authorization to destroy comes from the child, under its close + * gate, not from the cache. The cache's job is only to decide *when* it is + * worth asking. + * + * The child's gate makes the check atomic **on the child side**: with it held + * the Session admits no prompt and starts no automatic turn, so a hold cannot + * appear between the child's read and its teardown. It says nothing about the + * daemon side — the round trip below is an await, and covering that span is + * `closeIfChildUnheld`'s job, not this function's. + * + * Returns false on every uncertainty: a channel that never negotiated is + * handled by the pre-existing path, a refusal means work appeared, and a + * timeout means we cannot tell whether the child closed. None of those are + * retried here — the next snapshot resolves it, and a Session that is truly + * gone will be reported with no holds in that snapshot. + */ + async function confirmChildUnheld(entry: SessionEntry): Promise { + const info = channelInfoForEntry(entry); + if (!info?.activeWork) return true; + if (info.isDying) return false; + try { + const response = await withTimeout( + entry.connection.extMethod(SERVE_CONTROL_EXT_METHODS.sessionClose, { + sessionId: entry.sessionId, + [ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM]: true, + }), + ACTIVE_WORK_CLOSE_TIMEOUT_MS, + SERVE_CONTROL_EXT_METHODS.sessionClose, + ); + if (response['closed'] === true) { + // The child is done with it; only local teardown remains. + return true; + } + // Refused: adopt the hold set it handed back so the cache reflects the + // reason we are backing off rather than the stale set that sent us here. + const holds = response['holds']; + if (Array.isArray(holds)) { + const adopted = new Map(); + for (const hold of holds) { + if (typeof hold !== 'object' || hold === null) continue; + const record = hold as Record; + const id = record['id']; + const category = record['category']; + if ( + typeof id === 'string' && + typeof category === 'string' && + ACTIVE_WORK_HOLD_CATEGORIES.includes( + category as ActiveWorkHoldCategory, + ) + ) { + adopted.set(id, category as ActiveWorkHoldCategory); + } + } + entry.childHolds = adopted; + entry.childHoldsAt = Date.now(); + } + return false; + } catch (err) { + writeStderrLine( + `qwen serve: close-if-unheld for session ${JSON.stringify(entry.sessionId)} ` + + `did not resolve (${err instanceof Error ? err.message : String(err)}); ` + + `leaving it in place for the next snapshot to settle`, + ); + return false; + } + } + + /** Applies a validated channel-wide snapshot to every Session it names. */ + function applyActiveWorkSnapshot( + info: ChannelInfo, + snapshot: ActiveWorkSnapshotV1, + ): void { + if (!info.activeWork || info.isDying) return; + // Reordering guard only. A gap is not an error: each snapshot is complete, + // so the newest one that arrives is the whole truth regardless of what was + // lost before it. + if (snapshot.seq <= info.activeWork.seq) return; + info.activeWork.seq = snapshot.seq; + const now = Date.now(); + const reported = new Map>(); + for (const session of snapshot.sessions) { + const holds = new Map(); + for (const hold of session.holds) holds.set(hold.id, hold.category); + reported.set(session.sessionId, holds); + } + // Iterate what the channel owns rather than what the snapshot named: a + // Session the child did not mention holds nothing on the child side. + // Because reports are complete, silence about a Session this channel owns + // is a statement about that Session, not a gap in the report — so absence + // and reported-with-no-holds are the same fact and take the same path. + // That is also how the daemon recovers from a close whose response never + // made it back: the next snapshot omits the Session, the daemon asks the + // child once more, and the child answers `closed` for a Session it no + // longer has. + // + // Crucially, absence does NOT authorize local teardown by itself. It only + // makes the Session a candidate, and every candidate still has to clear + // the shared guards — a live SSE subscriber or a registered client keeps it + // exactly as it keeps any other idle Session. + for (const sessionId of Array.from(info.sessionIds)) { + const entry = byId.get(sessionId); + if (!entry || entry.channel !== info.channel) continue; + const holds = reported.get(sessionId) ?? new Map(); + const previouslyHeld = entry.childHolds + ? entry.childHolds.size > 0 + : undefined; + entry.childHolds = holds; + entry.childHoldsAt = now; + // Only a change in whether the Session holds anything counts as + // activity. Cadence reports must not keep `lastActivityAt` warm, or a + // long-running agent would defeat every idle-based reclaim downstream. + if (previouslyHeld !== undefined && previouslyHeld !== holds.size > 0) { + touchActivity(); + } + if (holds.size === 0) { + void maybeCloseIdleSession( + entry, + reported.has(sessionId) ? 'child_idle' : 'child_dropped', + ); + } + } + } + /** * Idempotently clear a session's active-prompt bookkeeping, but only if * `promptId` still owns it. The ownership gate matters: after a deadline @@ -1794,10 +2165,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (shuttingDown) return; const now = Date.now(); for (const [id, entry] of byId) { - // `pendingPromptCount` (not `promptActive`): queued prompts and the - // FIFO hand-off gap between two prompts must also block the reap. - if (entry.pendingPromptCount > 0) continue; - if (entry.events.subscriberCount > 0) continue; + // Shared guards first (`pendingPromptCount` rather than `promptActive`, + // so queued prompts and the FIFO hand-off gap between two prompts also + // block the reap), then the reaper's own TTL policy on top. + if (!entryIsAutoCloseCandidate(entry)) continue; // Note: clientIds.size is NOT checked here. Close-on-last-detach // handles the normal path (client sends detach → immediate close). // The reaper covers the crash path where detach was never sent — @@ -1812,14 +2183,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { `(idle for ${Math.round(idle / 1000)}s, ` + `threshold ${Math.round(sessionIdleTimeoutMs / 1000)}s)`, ); - void closeSessionImpl(id, undefined, { reason: 'idle_timeout' }).catch( - (err) => { - writeStderrLine( - `qwen serve: session reaper failed to close ` + - `${JSON.stringify(id)}: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`, - ); - }, - ); + // The TTL says the *client* stopped caring, which is not the same as + // the child having nothing left to run. Ask before destroying, on the + // same terms as every other automatic path: an idle-looking cache is + // never enough on its own. + void closeIfChildUnheld(entry, { + trigger: 'idle_timeout', + closeReason: 'idle_timeout', + }); } }, sessionReapIntervalMs); sessionReaper.unref(); @@ -2159,28 +2530,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { entry.attachCount - (released + (attachCountDelta - 1)), ); unregisterClient(entry, clientId); - if ( - entry.spawnOwnerWantedKill && - entry.attachCount === 0 && - entry.events.subscriberCount === 0 - ) { - await bridgeApi.killSession(entry.sessionId).catch(() => { - /* best-effort; channel.exited will eventually reap anyway */ - }); - } else if ( - entry.clientIds.size === 0 && - entry.events.subscriberCount === 0 && - entry.pendingPromptCount === 0 - ) { - await closeSessionImpl(entry.sessionId, undefined, { - reason: 'last_client_detached', - }).catch((err) => { - writeStderrLine( - `qwen serve: close-on-attach-rollback failed for ` + - `${JSON.stringify(entry.sessionId)}: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`, - ); - }); - } + await maybeCloseIdleSession(entry, 'attach_rollback'); }; const resolveTrustedClientId = ( @@ -2228,6 +2578,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }), ); const sessionIds = new Set(); + const infoRef: { current?: ChannelInfo } = {}; let client: BridgeClient; let connection: ClientSideConnection; try { @@ -2341,6 +2692,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { () => liveTaskToolRequestHandler, () => liveSpeakToUserHandler, opts.externalToolGuard, + (snapshot) => { + const currentInfo = infoRef.current; + if (!currentInfo) return; + applyActiveWorkSnapshot(currentInfo, snapshot); + }, ); connection = new ClientSideConnection(() => client, channel.stream); } catch (error) { @@ -2392,6 +2748,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { isDying: false, handshakeComplete: false, }; + infoRef.current = info; aliveChannels.add(info); // Belt-and-suspenders leak detection. The set is intentionally // multi-entry to cover the `killSession`-then-`spawnOrAttach` @@ -2541,6 +2898,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { connection.initialize({ protocolVersion: PROTOCOL_VERSION, _meta: { + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: ACTIVE_WORK_HEARTBEAT_INTERVAL_MS, + }, [CHANNEL_STARTUP_PROFILE_META_KEY]: { v: CHANNEL_STARTUP_PROFILE_VERSION, }, @@ -2566,6 +2927,29 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); } } + const activeWorkCapability = isRecord(response._meta) + ? response._meta[ACTIVE_WORK_HEARTBEAT_META_KEY] + : undefined; + if ( + isRecord(activeWorkCapability) && + activeWorkCapability['v'] === ACTIVE_WORK_HEARTBEAT_VERSION + ) { + const advertised = activeWorkCapability['categories']; + // Take the child's cadence rather than demanding it match ours, + // but clamp it: an out-of-range value would either flood the + // transport or make the freshness grade meaningless. + info.activeWork = { + intervalMs: clampActiveWorkIntervalMs( + activeWorkCapability['intervalMs'], + ), + categories: Array.isArray(advertised) + ? ACTIVE_WORK_HOLD_CATEGORIES.filter((category) => + advertised.includes(category), + ) + : [], + seq: 0, + }; + } try { const attributes = getChannelStartupProfileAttributes( response, @@ -3969,6 +4353,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { cwdChangeQueue: Promise.resolve(), promptQueue: Promise.resolve(), pendingPromptCount: 0, + pendingAgentNotificationCount: 0, pendingPromptList: [], midTurnMessageQueue: [], modelChangeQueue: Promise.resolve(), @@ -3983,6 +4368,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { attachRefs: new Map(), spawnOwnerWantedKill: false, promptActive: false, + childHolds: null, + childHoldsAt: null, + activeWorkCloseInFlight: false, retryAllowed: false, }; ci.sessionIds.add(entry.sessionId); @@ -4479,7 +4867,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { const existing = byId.get(req.sessionId); if (existing) { - if (existing.closing) { + // `isClosingOrAuthorizingClose`, not bare `closing`: this is an admission + // path like attach/prompt/rewind, so a conditional close being confirmed + // must refuse it too. Otherwise a client attaches inside the round trip + // and the teardown it raced destroys the Session under it. + if (isClosingOrAuthorizingClose(existing)) { throw new SessionNotFoundError( req.sessionId, 'The session is closing; retry after close completes', @@ -4497,7 +4889,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { action === 'load' ? await resolveHistoryAnchorRecordId(existing, replayFields) : undefined; - if (byId.get(req.sessionId) !== existing || existing.closing) { + if ( + byId.get(req.sessionId) !== existing || + isClosingOrAuthorizingClose(existing) + ) { throw new SessionNotFoundError(req.sessionId); } existing.attachCount++; @@ -4765,6 +5160,16 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { const racedEntry = byId.get(req.sessionId); if (racedEntry) { restoreEvents.close(); + // Same admission rule as the two checks above. This branch had no + // closing guard at all, so it could also attach to a Session already + // tearing down — narrower than the conditional-close window this PR + // introduced, but the same defect, and the fix is the same predicate. + if (isClosingOrAuthorizingClose(racedEntry)) { + throw new SessionNotFoundError( + req.sessionId, + 'The session is closing; retry after close completes', + ); + } // Self + any coalescers we accumulated while the restore was // in flight. Coalescers must not bump attachCount themselves // (they read it off the registered entry on the next tick). @@ -5191,6 +5596,63 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { return activePromptCounter; }, + get activeWork() { + for (const entry of byId.values()) { + if (entryHasActiveWork(entry)) return true; + } + return false; + }, + + /** + * Raw coverage counts rather than a pre-collapsed grade. + * + * The grade has to be computed over the whole daemon, not per runtime and + * then combined: a runtime with zero Sessions is vacuously `full`, and + * folding that in as evidence made a deployment whose only real Sessions + * were unreported aggregate to `partial`. Counts compose; grades do not. + */ + get activeWorkCoverage() { + let covered = 0; + let onNegotiatedChannel = 0; + let total = 0; + let oldestCoveredReportAt: number | null = null; + for (const entry of byId.values()) { + total++; + const owner = channelInfoForEntry(entry); + const capability = owner?.activeWork; + if (!capability) continue; + // A child that negotiated but reports late or omits a category still + // tells us *something*; only a channel that never negotiated at all + // leaves us with nothing, which is what `none` is reserved for. + onNegotiatedChannel++; + // Missing categories and a stale snapshot are the same kind of defect + // from a controller's point of view: the boolean does not cover what + // it claims to. Both land in `partial` rather than being invisible. + if ( + ACTIVE_WORK_HOLD_CATEGORIES.some( + (category) => !capability.categories.includes(category), + ) + ) { + continue; + } + if (!childHoldsAreFresh(entry, capability)) continue; + covered++; + // Deliberately the oldest *covered* report, not the oldest report of + // any kind. An uncovered Session already shows up as a downgraded + // grade; letting it also drag the age down would double-count it, and + // it would make a positive staleness coexist with a grade saying + // nothing is covered. Bounded by the stale window by construction. + if ( + entry.childHoldsAt !== null && + (oldestCoveredReportAt === null || + entry.childHoldsAt < oldestCoveredReportAt) + ) { + oldestCoveredReportAt = entry.childHoldsAt; + } + } + return { total, covered, onNegotiatedChannel, oldestCoveredReportAt }; + }, + get lastActivityAt() { return lastActivityTimestamp; }, @@ -5270,7 +5732,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (effectiveScope === 'single') { const existing = defaultEntry; if (existing) { - if (existing.closing) { + if (isClosingOrAuthorizingClose(existing)) { throw new SessionNotFoundError( existing.sessionId, 'The session is closing; retry after close completes', @@ -5479,7 +5941,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { const queuedAt = Date.now(); const entry = byId.get(sessionId); if (!entry) return Promise.reject(new SessionNotFoundError(sessionId)); - if (entry.closing) { + if (isClosingOrAuthorizingClose(entry)) { return Promise.reject( new SessionNotFoundError( sessionId, @@ -6047,21 +6509,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // exact entry is still registered — after killSession's eager // delete the same persisted id can be re-registered as a NEW // entry by `session/load`, which a late settle must not close. - if ( - entry.clientIds.size === 0 && - entry.events.subscriberCount === 0 && - entry.pendingPromptCount === 0 && - byId.get(sessionId) === entry - ) { - void closeSessionImpl(sessionId, undefined, { - reason: 'last_client_detached', - }).catch((err) => { - writeStderrLine( - `qwen serve: deferred close-on-prompt-complete failed for ` + - `${JSON.stringify(sessionId)}: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`, - ); - }); - } + void maybeCloseIdleSession(entry, 'prompt_settled'); }) .catch(() => {}); return result; @@ -7940,18 +8388,27 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (!entry) throw new SessionNotFoundError(sessionId); const info = channelInfoForEntry(entry); if (!info || info.isDying) throw new SessionNotFoundError(sessionId); - const response = await Promise.race([ - withTimeout( - entry.connection.extMethod( + entry.pendingAgentNotificationCount++; + try { + const response = await Promise.race([ + withTimeout( + entry.connection.extMethod( + SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification, + { sessionId, ...notification }, + ), + initTimeoutMs, SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification, - { sessionId, ...notification }, ), - initTimeoutMs, - SERVE_CONTROL_EXT_METHODS.sessionBackgroundNotification, - ), - getTransportClosedReject(entry), - ]); - return { sessionId, accepted: response['accepted'] === true }; + getTransportClosedReject(entry), + ]); + return { sessionId, accepted: response['accepted'] === true }; + } finally { + entry.pendingAgentNotificationCount = Math.max( + 0, + entry.pendingAgentNotificationCount - 1, + ); + void maybeCloseIdleSession(entry, 'agent_notification_settled'); + } }, async generateSessionBtw(sessionId, question, signal, _context) { @@ -8242,7 +8699,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { async rewindSession(sessionId, req, context) { const entry = byId.get(sessionId); if (!entry) throw new SessionNotFoundError(sessionId); - if (entry.closing) { + if (isClosingOrAuthorizingClose(entry)) { throw new SessionNotFoundError(sessionId, 'The session is closing'); } const info = channelInfoForEntry(entry); @@ -8901,40 +9358,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (entry.attachCount > 0) entry.attachCount--; } unregisterClient(entry, clientId); - if ( - entry.spawnOwnerWantedKill && - entry.attachCount === 0 && - entry.events.subscriberCount === 0 - ) { - // Defer-completed reap. Re-use killSession's logic; pass - // `requireZeroAttaches: false` (default) because we've - // already validated all the conditions ourselves. - await this.killSession(sessionId).catch(() => { - /* best-effort; channel.exited will eventually reap anyway */ - }); - } else if ( - entry.clientIds.size === 0 && - entry.events.subscriberCount === 0 && - entry.pendingPromptCount === 0 - ) { - // Last registered client left, no SSE subscribers remain, and - // no prompt is pending (active OR queued — `pendingPromptCount` - // covers the FIFO hand-off gap where `promptActive` is briefly - // false between two prompts). Close the session immediately so - // it doesn't linger in memory. The JSONL transcript on disk is - // preserved — session/load or session/resume can restore it - // later. When prompts ARE pending, skip the close: the deferred - // close in `sendPrompt`'s result.finally fires after the last - // one settles (and publishes its terminal). - await closeSessionImpl(sessionId, undefined, { - reason: 'last_client_detached', - }).catch((err) => { - writeStderrLine( - `qwen serve: close-on-last-detach failed for ` + - `${JSON.stringify(sessionId)}: ${err instanceof Error ? (err.stack ?? err.message) : String(err)}`, - ); - }); - } + // Last registered client left. Whether that means close, kill, or + // nothing at all lives in one place now: a pending prompt (active OR + // queued — `pendingPromptCount` covers the FIFO hand-off gap), an + // unsettled Agent, or a child that has not confirmed it is unheld all + // hold the session open, and the deferred close fires from whichever + // path settles last. The JSONL transcript on disk survives either way, + // so session/load can restore it later. + await maybeCloseIdleSession(entry, 'last_client_detached'); }, killAllSync() { diff --git a/packages/acp-bridge/src/bridgeClient.ts b/packages/acp-bridge/src/bridgeClient.ts index ca52d83f732..4e73ff4fd18 100644 --- a/packages/acp-bridge/src/bridgeClient.ts +++ b/packages/acp-bridge/src/bridgeClient.ts @@ -25,8 +25,15 @@ import type { BridgeEvent, EventBus } from './eventBus.js'; // so a rename can't silently break the protocol. import { MID_TURN_MESSAGE_INJECTED_EVENT } from './daemonEventTypes.js'; import { + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, + ACTIVE_WORK_MAX_SESSION_HOLDS, + ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS, + ACTIVE_WORK_NOTIFICATION_METHOD, MID_TURN_QUEUE_DRAIN_METHOD, TODO_STOP_GUARD_CONTINUATION_CLAIM_METHOD, + type ActiveWorkHoldV1, + type ActiveWorkSnapshotV1, } from './bridgeTypes.js'; import type { BridgeWorkspaceGenerationNotificationEvent, @@ -76,6 +83,68 @@ import type { SessionArtifactStore, } from './sessionArtifacts.js'; +/** + * Validate a channel-wide active-work snapshot off the wire. + * + * Returns `undefined` for anything malformed so a bad report is ignored + * outright: the daemon's cached copy then simply ages, which its freshness + * grading already treats as untrustworthy. Partially applying a half-parsed + * snapshot would be worse than applying none, because full-snapshot semantics + * are what let a Session's absence mean "released". + */ +function parseActiveWorkSnapshot( + params: Record, +): ActiveWorkSnapshotV1 | undefined { + const seq = params['seq']; + const sessions = params['sessions']; + if ( + params['v'] !== ACTIVE_WORK_HEARTBEAT_VERSION || + typeof seq !== 'number' || + !Number.isSafeInteger(seq) || + seq <= 0 || + !Array.isArray(sessions) || + sessions.length > ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS + ) { + return undefined; + } + const parsed: ActiveWorkSnapshotV1['sessions'] = []; + for (const raw of sessions) { + if (typeof raw !== 'object' || raw === null) return undefined; + const entry = raw as Record; + const sessionId = entry['sessionId']; + const holds = entry['holds']; + if ( + typeof sessionId !== 'string' || + !Array.isArray(holds) || + holds.length > ACTIVE_WORK_MAX_SESSION_HOLDS + ) { + return undefined; + } + const parsedHolds: ActiveWorkHoldV1[] = []; + for (const rawHold of holds) { + if (typeof rawHold !== 'object' || rawHold === null) return undefined; + const hold = rawHold as Record; + const category = hold['category']; + const id = hold['id']; + if ( + typeof id !== 'string' || + typeof category !== 'string' || + !ACTIVE_WORK_HOLD_CATEGORIES.includes( + category as ActiveWorkHoldV1['category'], + ) + ) { + return undefined; + } + parsedHolds.push({ + category: category as ActiveWorkHoldV1['category'], + id, + }); + } + parsed.push({ sessionId, holds: parsedHolds }); + } + return { v: ACTIVE_WORK_HEARTBEAT_VERSION, seq, sessions: parsed }; +} + // Keep in sync with core `ToolNames.ARTIFACT`; acp-bridge avoids a runtime // import from core for this hot demux path. const PUBLISH_ARTIFACT_TOOL_NAME = 'artifact'; @@ -733,6 +802,7 @@ export class BridgeClient implements Client { * existing direct BridgeClient constructors remain source-compatible. */ private readonly externalToolGuard?: ExternalToolGuardHandler, + private readonly onActiveWork?: (snapshot: ActiveWorkSnapshotV1) => void, ) {} async requestPermission( @@ -1726,6 +1796,22 @@ export class BridgeClient implements Client { method: string, params: Record, ): Promise { + if (method === ACTIVE_WORK_NOTIFICATION_METHOD) { + const snapshot = parseActiveWorkSnapshot(params); + if (snapshot) { + // Sessions the child claims but this channel does not own are dropped + // rather than rejecting the whole snapshot: the rest of it is still + // usable, and a channel must never influence another channel's state. + this.onActiveWork?.({ + v: ACTIVE_WORK_HEARTBEAT_VERSION, + seq: snapshot.seq, + sessions: snapshot.sessions.filter((session) => + this.ownsSession(session.sessionId), + ), + }); + } + return; + } if (method === '_qwencode/end_turn') { const sessionId = params['sessionId']; const reason = params['reason']; diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 7893eb8ae04..3aa0d1cae6f 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -191,8 +191,116 @@ export const REQUESTED_SESSION_ID_META_KEY = 'qwen-code/sessionId'; export const CHANNEL_STARTUP_PROFILE_META_KEY = 'qwen.daemon.channelStartupProfile'; export const CHANNEL_STARTUP_PROFILE_VERSION = 1 as const; +export const ACTIVE_WORK_HEARTBEAT_META_KEY = 'qwen.daemon.activeWorkHeartbeat'; +export const ACTIVE_WORK_HEARTBEAT_VERSION = 1 as const; +/** Reporting cadence the daemon asks for; the child may choose another value + * inside [MIN, MAX] and the daemon clamps whatever comes back. */ +export const ACTIVE_WORK_HEARTBEAT_INTERVAL_MS = 15_000; +export const ACTIVE_WORK_HEARTBEAT_MIN_INTERVAL_MS = 5_000; +export const ACTIVE_WORK_HEARTBEAT_MAX_INTERVAL_MS = 60_000; +/** A channel's cached snapshot goes stale after this many report intervals. */ +export const ACTIVE_WORK_STALE_INTERVALS = 3; +export const ACTIVE_WORK_NOTIFICATION_METHOD = + 'qwen/notify/channel/active-work'; +export const ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM = 'onlyIfUnheld'; +/** Bound on the conditional-close round trip. Its own constant rather than the + * handshake timeout: this runs on the automatic-cleanup path, where waiting + * longer buys nothing — an unanswered request is simply left for the next + * snapshot to settle. */ +export const ACTIVE_WORK_CLOSE_TIMEOUT_MS = 10_000; +/** Bounds on a single snapshot. Generous next to any real deployment — they + * exist so a version-skewed or buggy child cannot make the daemon walk an + * unbounded structure per report, not to constrain legitimate use. A packet + * over either bound is discarded whole, like any other malformed one. */ +export const ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS = 1024; +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. + */ +export type ActiveWorkHoldCategory = 'agent' | 'notification'; + +export const ACTIVE_WORK_HOLD_CATEGORIES: readonly ActiveWorkHoldCategory[] = [ + 'agent', + 'notification', +]; + +export interface ActiveWorkHeartbeatCapabilityV1 { + v: typeof ACTIVE_WORK_HEARTBEAT_VERSION; + intervalMs: number; + /** Which categories this child actually reports. A daemon that cares about + * a category the child omits degrades its reporting grade rather than + * silently treating the gap as "no work". */ + categories: ActiveWorkHoldCategory[]; +} + +/** + * Coerce a peer-supplied reporting cadence into the agreed range. + * + * Both sides call this on whatever the other side sent. Neither is treated as + * hostile, but a version-skewed or buggy peer proposing 1ms would flood the + * transport and one proposing hours would make the daemon's freshness grade + * meaningless, so the value is never used raw. Anything unusable falls back to + * the default cadence rather than disabling reporting. + */ +export function clampActiveWorkIntervalMs(raw: unknown): number { + const value = typeof raw === 'number' && Number.isFinite(raw) ? raw : NaN; + if (Number.isNaN(value)) return ACTIVE_WORK_HEARTBEAT_INTERVAL_MS; + return Math.min( + ACTIVE_WORK_HEARTBEAT_MAX_INTERVAL_MS, + Math.max(ACTIVE_WORK_HEARTBEAT_MIN_INTERVAL_MS, Math.round(value)), + ); +} + +/** + * Collapse coverage counts into the grade `/health?deep=1` reports. + * + * Deliberately a function over summed counts rather than a per-runtime getter: + * grades do not compose. A runtime with no Sessions vouches for everything it + * has, so folding its vacuous `full` in as evidence let an empty workspace + * vouch for another workspace's unreported Sessions. Callers sum the counts + * across every runtime first, then grade once. + */ +export function gradeActiveWorkCoverage(totals: { + total: number; + covered: number; + onNegotiatedChannel: number; +}): 'full' | 'partial' | 'none' { + // No Sessions means nothing is unreported, so the picture is complete. + if (totals.total === 0 || totals.covered === totals.total) return 'full'; + // `none` is reserved for "not one Session sits on a channel that negotiated + // reporting" — the case where acting on `activeWork` is unsafe rather than + // merely degraded. + return totals.onNegotiatedChannel === 0 ? 'none' : 'partial'; +} + +export interface ActiveWorkHoldV1 { + category: ActiveWorkHoldCategory; + id: string; +} + +export interface ActiveWorkSessionSnapshotV1 { + sessionId: string; + holds: ActiveWorkHoldV1[]; +} + +/** + * A full, channel-wide snapshot: every Session the child currently owns, with + * every hold it currently holds. Full-snapshot (rather than incremental + * transition) semantics are what make a dropped report self-correcting in + * both directions, and a Session's *absence* from a fresh snapshot is + * positive evidence the child no longer owns it. + */ +export interface ActiveWorkSnapshotV1 { + v: typeof ACTIVE_WORK_HEARTBEAT_VERSION; + seq: number; + sessions: ActiveWorkSessionSnapshotV1[]; +} + export interface ChannelStartupProfileV1 { v: typeof CHANNEL_STARTUP_PROFILE_VERSION; complete: boolean; @@ -1695,6 +1803,44 @@ export interface AcpSessionBridge { /** Number of sessions with an active prompt. */ 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. + */ + readonly activeWork: boolean; + + /** + * How much of `activeWork` this runtime can vouch for, as counts rather than + * a grade. + * + * Counts, because the daemon-wide grade cannot be assembled from per-runtime + * grades: a runtime with zero Sessions vouches for everything it has and is + * therefore vacuously complete, which must not count as evidence that some + * *other* runtime's unreported Sessions are covered. Summing counts and + * grading once at the end is the only composition that gets that right. + * + * A Session counts as covered only when its owning channel negotiated + * reporting, reports every category, and its last snapshot is still inside + * the freshness window. Without this a controller cannot tell "nothing is + * running" from "nobody told me what is running", and those must not lead to + * the same decision. + */ + readonly activeWorkCoverage: { + /** Live Sessions in this runtime. */ + total: number; + /** Of those, how many `activeWork` actually speaks for. */ + covered: number; + /** Of those, how many sit on a channel that negotiated reporting at all. + * Zero is what distinguishes `none` from `partial`. */ + onNegotiatedChannel: number; + /** Epoch ms of the oldest snapshot among the *covered* Sessions, or null + * when none are covered. Diagnostic: the freshness decision is already + * folded into `covered`, because only the daemon knows each channel's + * negotiated cadence. */ + oldestCoveredReportAt: number | null; + }; + /** Queued prompts across all sessions — accepted but not yet dispatched, * excluding the one running per session — i.e. the queue-depth gauge for the * Daemon Status charts (distinct from `activePromptCount`). Optional: a diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 28f0c5a40f7..64dce116fce 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -766,13 +766,19 @@ vi.mock('../ui/commands/contextCommand.js', () => ({ .fn() .mockReturnValue('## Context Usage\nformatted'), })); -vi.mock('./session/Session.js', () => ({ - Session: vi.fn(), - buildAvailableCommandsSnapshot: vi.fn().mockResolvedValue({ - availableCommands: [], - availableSkills: [], - }), -})); +vi.mock('./session/Session.js', () => { + const SessionMock = vi.fn(); + // The agent's active-work reporter walks every live Session on a timer, so + // even tests that never look at reporting need this to exist on instances. + SessionMock.prototype.collectActiveWorkHolds = () => []; + return { + Session: SessionMock, + buildAvailableCommandsSnapshot: vi.fn().mockResolvedValue({ + availableCommands: [], + availableSkills: [], + }), + }; +}); vi.mock('../utils/languageUtils.js', () => ({ updateOutputLanguageFile: vi.fn(), writeOutputLanguageAndRegisterPath: vi.fn( @@ -890,6 +896,10 @@ import { } from '../utils/languageUtils.js'; import { buildAuthMethods } from './authMethods.js'; import { + ACTIVE_WORK_HEARTBEAT_META_KEY, + ACTIVE_WORK_HEARTBEAT_MIN_INTERVAL_MS, + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, PROMPT_CANCEL_METHOD, @@ -2498,6 +2508,55 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('merges active-work negotiation and enables Session reporting', async () => { + await setupSessionMocks('active-work-session'); + initializeAcpStartupProfiler(); + const agentPromise = runAcpAgent( + mockConfig, + makeSessionSettings(), + mockArgv, + ); + await vi.waitFor(() => expect(capturedAgentFactory).toBeDefined()); + const agent = capturedAgentFactory!({ + get closed() { + return mockConnectionState.promise; + }, + }) as AgentLike; + + const response = (await agent.initialize({ + clientCapabilities: {}, + _meta: { + [CHANNEL_STARTUP_PROFILE_META_KEY]: { + v: CHANNEL_STARTUP_PROFILE_VERSION, + }, + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + // Absurd cadence: the child must answer with the clamped value it + // will actually use, not echo this back. + intervalMs: 1, + }, + }, + })) as { _meta?: Record }; + + expect(response._meta).toMatchObject({ + [CHANNEL_STARTUP_PROFILE_META_KEY]: { + v: CHANNEL_STARTUP_PROFILE_VERSION, + }, + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: ACTIVE_WORK_HEARTBEAT_MIN_INTERVAL_MS, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], + }, + }); + await agent.newSession({ cwd: '/tmp', mcpServers: [] }); + // The Session gets a change callback, not a cadence: one reporter per + // channel owns the timing. + expect(typeof vi.mocked(Session).mock.calls.at(-1)?.[4]).toBe('function'); + + mockConnectionState.resolve(); + await agentPromise; + }); + it('runs text workspace generation through the shared transport', async () => { mockExecuteGeneration.mockImplementation( async ( diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index e5be8851751..5e2f7d00560 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -221,6 +221,7 @@ import { isInactiveExtensionSkill, } from './extension-skills.js'; import { Session, buildAvailableCommandsSnapshot } from './session/Session.js'; +import { ActiveWorkReporter } from './active-work-reporter.js'; import { buildSessionTasksStatus } from './session/tasksSnapshot.js'; import { collectHistoryReplayUpdates, @@ -310,6 +311,12 @@ import { SESSION_SOURCE_META_KEY, } from '@qwen-code/acp-bridge'; import { + ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM, + ACTIVE_WORK_HEARTBEAT_META_KEY, + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_HOLD_CATEGORIES, + clampActiveWorkIntervalMs, + type ActiveWorkHoldV1, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, CLIENT_MCP_OVER_WS_CONFIG_FLAG, @@ -3596,6 +3603,8 @@ class QwenAgent implements Agent { private readonly initializingConfigs = new Set(); private managedShuttingDown = false; private clientCapabilities: ClientCapabilities | undefined; + /** Set once the daemon negotiates active-work reporting; one per channel. */ + private activeWorkReporter: ActiveWorkReporter | undefined; private privateParentState: | 'uninitialized' | 'trusted' @@ -4152,6 +4161,9 @@ class QwenAgent implements Agent { cleanupErrors.push(error); } this.sessions.delete(sessionId); + // A Session missing from the next snapshot is how the daemon learns the + // child released it — including when it never saw our close response. + this.activeWorkReporter?.notifyChanged(); if (cleanupErrors.length > 0) { debugLogger.warn( `Session ${sessionId} closed after ${cleanupErrors.length} cleanup failure(s): ${cleanupErrors @@ -4271,12 +4283,19 @@ class QwenAgent implements Agent { drainTimeoutMs?: number; shutdownConfig?: boolean; waitForCloseGate?: boolean; + /** + * Close only if the Session holds no active work — the daemon's + * automatic-cleanup path. Explicit close, kill, and shutdown leave this + * unset and keep their force semantics. + */ + onlyIfUnheld?: boolean; }, - ): Promise { + ): Promise<{ closed: boolean; holds: ActiveWorkHoldV1[] }> { const session = this.sessions.get(sessionId); if (!session) { this.mcpPool?.releaseSession(sessionId); - return; + // Already gone is the outcome the caller wanted, not a refusal. + return { closed: true, holds: [] }; } const recorder = session.getConfig().getChatRecordingService(); @@ -4289,6 +4308,19 @@ 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. + if (opts?.onlyIfUnheld) { + const holds = session.collectActiveWorkHolds(); + if (holds.length > 0) { + cancelClose(); + return { closed: false, holds }; + } + } for (const [requestId, generation] of this.generationControllers) { if (generation.sessionId !== sessionId) continue; generation.controller.abort(); @@ -4344,6 +4376,7 @@ class QwenAgent implements Agent { } finally { if (!removedFromStore) cancelClose(); } + return { closed: true, holds: [] }; } private async discardStoredSessionIfCurrent( @@ -4363,6 +4396,8 @@ class QwenAgent implements Agent { } async disposeSessions(): Promise { + this.activeWorkReporter?.dispose(); + this.activeWorkReporter = undefined; for (const generation of this.generationControllers.values()) { generation.controller.abort(); } @@ -4582,6 +4617,30 @@ class QwenAgent implements Agent { !Array.isArray(requestedProfile) && (requestedProfile as Record)['v'] === CHANNEL_STARTUP_PROFILE_VERSION; + const requestedActiveWork = args._meta?.[ACTIVE_WORK_HEARTBEAT_META_KEY]; + const activeWorkRequested = + requestedActiveWork !== null && + typeof requestedActiveWork === 'object' && + !Array.isArray(requestedActiveWork) && + (requestedActiveWork as Record)['v'] === + ACTIVE_WORK_HEARTBEAT_VERSION; + // The daemon proposes a cadence; we answer with the one we will actually + // use, clamped into the range both sides agree on. The daemon clamps the + // echo again — neither side trusts the other to pick a sane number, and a + // flood or a multi-hour interval would each break freshness in its own way. + const activeWorkIntervalMs = activeWorkRequested + ? clampActiveWorkIntervalMs( + (requestedActiveWork as Record)['intervalMs'], + ) + : undefined; + if (activeWorkIntervalMs !== undefined) { + this.activeWorkReporter?.dispose(); + this.activeWorkReporter = new ActiveWorkReporter( + (method, params) => this.connection.extNotification(method, params), + () => this.sessions.values(), + activeWorkIntervalMs, + ); + } const responseMeta: Record = { ...(this.managedToolInvocationGuard @@ -4593,6 +4652,15 @@ class QwenAgent implements Agent { ...(profileRequested && startupProfile ? { [CHANNEL_STARTUP_PROFILE_META_KEY]: startupProfile } : {}), + ...(activeWorkIntervalMs !== undefined + ? { + [ACTIVE_WORK_HEARTBEAT_META_KEY]: { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + intervalMs: activeWorkIntervalMs, + categories: [...ACTIVE_WORK_HOLD_CATEGORIES], + }, + } + : {}), }; return Object.keys(responseMeta).length > 0 ? { ...response, _meta: responseMeta } @@ -5262,6 +5330,12 @@ class QwenAgent implements Agent { this.activePromptCalls.delete(params.sessionId); } settleCall(); + // 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. + await this.activeWorkReporter?.flush(); } } @@ -9386,13 +9460,18 @@ class QwenAgent implements Agent { 'Invalid session close drain timeout', ); } - await this.closeStoredSession(sessionId, { + const outcome = await this.closeStoredSession(sessionId, { requireFlush: params['requireFlush'] === true, + onlyIfUnheld: params[ACTIVE_WORK_CLOSE_IF_UNHELD_PARAM] === true, ...(typeof rawDrainTimeoutMs === 'number' ? { drainTimeoutMs: rawDrainTimeoutMs } : {}), }); - return { sessionId, closed: true }; + // `holds` rides along only on a refusal, so the response every existing + // caller already parses keeps its exact shape. + return outcome.closed + ? { sessionId, closed: true } + : { sessionId, closed: false, holds: outcome.holds }; } case SERVE_CONTROL_EXT_METHODS.sessionCd: { const sessionId = params['sessionId']; @@ -11816,8 +11895,17 @@ class QwenAgent implements Agent { throw new Error(`Session ${sessionId} is already active.`); } - const session = new Session(sessionId, config, this.connection, settings); + const session = new Session( + sessionId, + config, + this.connection, + settings, + () => this.activeWorkReporter?.notifyChanged(), + ); this.sessions.set(sessionId, session); + // The Session set itself is part of the snapshot: publish so the daemon + // learns about this Session from a report rather than inferring it. + this.activeWorkReporter?.notifyChanged(); this.initializingConfigs.delete(config); try { if (options.enableLiveScreenContext) { diff --git a/packages/cli/src/acp-integration/active-work-reporter.test.ts b/packages/cli/src/acp-integration/active-work-reporter.test.ts new file mode 100644 index 00000000000..1b36207def3 --- /dev/null +++ b/packages/cli/src/acp-integration/active-work-reporter.test.ts @@ -0,0 +1,201 @@ +/** + * @license + * Copyright 2026 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, expect, it, vi, beforeEach, afterEach } from 'vitest'; +import { + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_NOTIFICATION_METHOD, + type ActiveWorkHoldV1, + type ActiveWorkSnapshotV1, +} from '@qwen-code/acp-bridge/bridgeTypes'; +import { + ActiveWorkReporter, + type ActiveWorkSource, +} from './active-work-reporter.js'; + +const INTERVAL_MS = 5_000; + +function source( + sessionId: string, + collect: () => ActiveWorkHoldV1[], +): ActiveWorkSource { + return { sessionId, collectActiveWorkHolds: collect }; +} + +function throwingSource(sessionId: string): ActiveWorkSource { + return { + sessionId, + collectActiveWorkHolds: () => { + throw new Error('registry exploded'); + }, + }; +} + +describe('ActiveWorkReporter', () => { + let sent: Array<{ method: string; params: Record }>; + let send: (method: string, params: Record) => Promise; + + beforeEach(() => { + sent = []; + send = async (method, params) => { + sent.push({ method, params }); + }; + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + function snapshots(): ActiveWorkSnapshotV1[] { + return sent + .filter((call) => call.method === ACTIVE_WORK_NOTIFICATION_METHOD) + .map((call) => call.params as unknown as ActiveWorkSnapshotV1); + } + + it('publishes a full snapshot of every source immediately', async () => { + const reporter = new ActiveWorkReporter( + send, + () => [ + source('s1', () => [{ category: 'agent', id: 'a1' }]), + source('s2', () => []), + ], + INTERVAL_MS, + ); + await reporter.flush(); + + const last = snapshots().at(-1)!; + expect(last.v).toBe(ACTIVE_WORK_HEARTBEAT_VERSION); + expect(last.sessions).toEqual([ + { sessionId: 's1', holds: [{ category: 'agent', id: 'a1' }] }, + { sessionId: 's2', holds: [] }, + ]); + reporter.dispose(); + }); + + it('increases seq monotonically across reports', async () => { + const reporter = new ActiveWorkReporter(send, () => [], INTERVAL_MS); + await reporter.flush(); + await reporter.flush(); + + const seqs = snapshots().map((s) => s.seq); + expect(seqs).toEqual([...seqs].sort((a, b) => a - b)); + expect(new Set(seqs).size).toBe(seqs.length); + reporter.dispose(); + }); + + describe('when a source throws while collecting', () => { + it('does not let the failure escape the interval timer', async () => { + vi.useFakeTimers(); + const onUncaught = vi.fn(); + process.on('uncaughtException', onUncaught); + try { + const reporter = new ActiveWorkReporter( + send, + () => [throwingSource('s1')], + INTERVAL_MS, + ); + // The constructor already published once; drive several more ticks. + await vi.advanceTimersByTimeAsync(INTERVAL_MS * 3); + expect(onUncaught).not.toHaveBeenCalled(); + reporter.dispose(); + } finally { + process.off('uncaughtException', onUncaught); + } + }); + + it('does not let the failure escape notifyChanged', async () => { + const reporter = new ActiveWorkReporter( + send, + () => [throwingSource('s1')], + INTERVAL_MS, + ); + reporter.notifyChanged(); + // The coalesced publish runs in a microtask; if it threw, this await + // would surface it as an unhandled rejection rather than resolving. + await Promise.resolve(); + await Promise.resolve(); + reporter.dispose(); + }); + + it('does not reject flush, so a prompt never fails over reporting', async () => { + const reporter = new ActiveWorkReporter( + send, + () => [throwingSource('s1')], + INTERVAL_MS, + ); + await expect(reporter.flush()).resolves.toBeUndefined(); + reporter.dispose(); + }); + + it('abandons the whole snapshot rather than sending a partial one', async () => { + // A session omitted from a report reads as "released" and one reported + // with no holds reads as "safe to close", so a partial snapshot would + // invite the daemon to destroy the very work it could not enumerate. + const reporter = new ActiveWorkReporter( + send, + () => [ + source('healthy', () => [{ category: 'agent', id: 'a1' }]), + throwingSource('broken'), + ], + INTERVAL_MS, + ); + await reporter.flush(); + + expect(snapshots()).toHaveLength(0); + reporter.dispose(); + }); + + it('resumes reporting once collection recovers', async () => { + let broken = true; + const reporter = new ActiveWorkReporter( + send, + () => [ + broken + ? throwingSource('s1') + : source('s1', () => [{ category: 'agent', id: 'a1' }]), + ], + INTERVAL_MS, + ); + await reporter.flush(); + expect(snapshots()).toHaveLength(0); + + broken = false; + await reporter.flush(); + + expect(snapshots()).toHaveLength(1); + expect(snapshots()[0]?.sessions).toEqual([ + { sessionId: 's1', holds: [{ category: 'agent', id: 'a1' }] }, + ]); + reporter.dispose(); + }); + }); + + it('does not let a failing transport escape either', async () => { + const reporter = new ActiveWorkReporter( + async () => { + throw new Error('stream closed'); + }, + () => [source('s1', () => [])], + INTERVAL_MS, + ); + await expect(reporter.flush()).resolves.toBeUndefined(); + await expect(reporter.flush()).resolves.toBeUndefined(); + reporter.dispose(); + }); + + it('stops publishing after dispose', async () => { + vi.useFakeTimers(); + const reporter = new ActiveWorkReporter( + send, + () => [source('s1', () => [])], + INTERVAL_MS, + ); + const before = snapshots().length; + reporter.dispose(); + await vi.advanceTimersByTimeAsync(INTERVAL_MS * 3); + expect(snapshots()).toHaveLength(before); + }); +}); diff --git a/packages/cli/src/acp-integration/active-work-reporter.ts b/packages/cli/src/acp-integration/active-work-reporter.ts new file mode 100644 index 00000000000..086da22f800 --- /dev/null +++ b/packages/cli/src/acp-integration/active-work-reporter.ts @@ -0,0 +1,159 @@ +/** + * @license + * Copyright 2026 Qwen + * SPDX-License-Identifier: Apache-2.0 + */ + +import { createDebugLogger } from '@qwen-code/qwen-code-core'; +import { + ACTIVE_WORK_HEARTBEAT_VERSION, + ACTIVE_WORK_NOTIFICATION_METHOD, + type ActiveWorkHoldV1, + type ActiveWorkSnapshotV1, +} from '@qwen-code/acp-bridge/bridgeTypes'; + +const debugLogger = createDebugLogger('ACTIVE_WORK'); + +/** + * A Session, as far as active-work reporting is concerned. `collectHolds()` + * must *derive* its result from whatever subsystem actually owns the work + * (the background-task registry, the notification queue, ...) rather than + * read a ledger maintained alongside it: a ledger can miss a release and + * then pin the Session forever, and a full snapshot would faithfully + * republish that leak on every report. + */ +export interface ActiveWorkSource { + readonly sessionId: string; + collectActiveWorkHolds(): ActiveWorkHoldV1[]; +} + +type SendNotification = ( + method: string, + params: Record, +) => Promise; + +/** + * Publishes channel-wide active-work snapshots to the daemon. + * + * One reporter per ACP connection, not per Session. Reporting at channel + * scope is what keeps the always-on cadence affordable (one small message + * per interval regardless of Session count) and it lets the daemon treat a + * Session missing from a fresh snapshot as proof the child released it. + * + * Every message is a complete snapshot with a monotonic `seq`. The sequence + * exists only to discard reordered messages — never to detect gaps — so a + * dropped report costs at most one interval of staleness and needs no + * retransmit, ack, or local "last reported" state to diff against. + */ +export class ActiveWorkReporter { + #seq = 0; + #tail: Promise = Promise.resolve(); + #coalescing = false; + #timer: ReturnType | undefined; + #disposed = false; + + constructor( + private readonly send: SendNotification, + private readonly listSources: () => Iterable, + readonly intervalMs: number, + ) { + this.#timer = setInterval(() => { + this.#publish(); + }, intervalMs); + this.#timer.unref?.(); + // Publish immediately so the daemon leaves the "negotiated but never + // reported" state as early as possible instead of holding every Session + // as unknown until the first interval elapses. + this.#publish(); + } + + /** + * Note that some Session's derived state may have changed. Coalesced to + * one snapshot per microtask so a burst of transitions (an agent finishing + * and its terminal notification enqueuing in the same tick) produces a + * single message that already reflects the settled state. + */ + notifyChanged(): void { + if (this.#disposed || this.#coalescing) return; + this.#coalescing = true; + queueMicrotask(() => { + if (!this.#coalescing) return; + this.#coalescing = false; + this.#publish(); + }); + } + + /** + * Publish now and resolve once the snapshot has been handed to the + * transport. + * + * Callers use this to order a snapshot ahead of an RPC response on the same + * stream. The prompt path needs it: the daemon drops its own + * `pendingPromptCount` the moment the prompt response lands, so a hold + * taken during that prompt (a background agent it started) has to be on the + * wire *first* or the daemon briefly sees neither fact and may reap the + * Session. + */ + async flush(): Promise { + this.#coalescing = false; + this.#publish(); + // Never reject: this is awaited on the prompt path, and a reporting + // problem must not turn into a failed prompt for the user. + await this.#tail.catch(() => undefined); + } + + dispose(): void { + this.#disposed = true; + this.#coalescing = false; + if (this.#timer) { + clearInterval(this.#timer); + this.#timer = undefined; + } + } + + #publish(): void { + if (this.#disposed) return; + let sessions: ActiveWorkSnapshotV1['sessions']; + try { + sessions = []; + for (const source of this.listSources()) { + sessions.push({ + sessionId: source.sessionId, + holds: source.collectActiveWorkHolds(), + }); + } + } catch (error) { + // Abandon the whole snapshot rather than send a partial one. A Session + // missing from a report means "the child released it", and a Session + // reported with no holds means "safe to close" — so publishing whatever + // we managed to collect before the failure would actively invite the + // daemon to destroy live work. Sending nothing instead just lets the + // daemon's copy age, which its freshness grading already treats as + // untrustworthy and retains. + debugLogger.warn( + `active-work snapshot collection failed; skipping this report: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + return; + } + const snapshot: ActiveWorkSnapshotV1 = { + v: ACTIVE_WORK_HEARTBEAT_VERSION, + seq: ++this.#seq, + sessions, + }; + this.#tail = this.#tail + .then(() => + this.#disposed + ? undefined + : this.send( + ACTIVE_WORK_NOTIFICATION_METHOD, + snapshot as unknown as Record, + ), + ) + // A failed send needs no recovery path: the next snapshot carries the + // whole truth again. Swallowing here is only safe *because* reports are + // full snapshots on a fixed cadence. + .catch(() => undefined); + } +} 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 8edd8ca882f..c1d4486b20c 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 @@ -143,6 +143,9 @@ describe('Session review-worktree lease sweep', () => { getStopHookBlockingCap: vi.fn().mockReturnValue(0), getBackgroundTaskRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), + listUnfinalizedBackgroundAgentIds: vi.fn().mockReturnValue([]), }), getMonitorRegistry: vi.fn().mockReturnValue({ setNotificationCallback: 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 bf2aff1c7c0..bb3499fe65b 100644 --- a/packages/cli/src/acp-integration/session/Session.test.ts +++ b/packages/cli/src/acp-integration/session/Session.test.ts @@ -408,7 +408,11 @@ describe('Session', () => { let mockBackgroundTaskRegistry: { abortAll: ReturnType; setNotificationCallback: ReturnType; + setStatusChangeCallback: ReturnType; + clearStatusChangeCallback: ReturnType; hasUnfinalizedTasks: ReturnType; + hasRunningTasks: ReturnType; + listUnfinalizedBackgroundAgentIds: ReturnType; getAll: ReturnType; get: ReturnType; }; @@ -571,7 +575,11 @@ describe('Session', () => { mockBackgroundTaskRegistry = { abortAll: vi.fn(), setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), hasUnfinalizedTasks: vi.fn().mockReturnValue(false), + hasRunningTasks: vi.fn().mockReturnValue(false), + listUnfinalizedBackgroundAgentIds: vi.fn().mockReturnValue([]), getAll: vi.fn().mockReturnValue([]), get: vi.fn().mockImplementation((taskId: string) => ( @@ -827,6 +835,194 @@ describe('Session', () => { expect(replayDelivered).toBe(replayUpdate); }); + describe('active work holds', () => { + let changes: number; + + function createReportingSession(): void { + session.dispose(); + changes = 0; + session = new Session( + 'test-session-id', + mockConfig, + mockClient, + mockSettings, + () => { + changes++; + }, + ); + } + + function holdIds(category: 'agent' | 'notification'): string[] { + return session + .collectActiveWorkHolds() + .filter((hold) => hold.category === category) + .map((hold) => hold.id); + } + + it('derives agent holds from the registry, covering the cancel window', async () => { + createReportingSession(); + expect(session.collectActiveWorkHolds()).toEqual([]); + expect(session.isIdle()).toBe(true); + + // A cancelled-but-unfinalized agent still owes its terminal + // notification. hasRunningTasks() would already call this idle, which is + // exactly how a detached session got reaped inside the cancel window. + mockBackgroundTaskRegistry.listUnfinalizedBackgroundAgentIds.mockReturnValue( + ['agent-cancelled'], + ); + expect(holdIds('agent')).toEqual(['agent-cancelled']); + expect(session.isIdle()).toBe(false); + + mockBackgroundTaskRegistry.listUnfinalizedBackgroundAgentIds.mockReturnValue( + [], + ); + expect(session.collectActiveWorkHolds()).toEqual([]); + expect(session.isIdle()).toBe(true); + session.dispose(); + }); + + it('notifies the owner when the registry reports a status change', async () => { + createReportingSession(); + const statusChanged = + mockBackgroundTaskRegistry.setStatusChangeCallback.mock.calls.at( + -1, + )?.[0] as (() => void) | undefined; + const before = changes; + statusChanged?.(); + expect(changes).toBe(before + 1); + + session.dispose(); + // Retracts the exact callback it installed rather than blanking the slot. + // The registry holds one callback and the TUI uses the same registry, so + // an unconditional clear on dispose would unhook whoever owns it now. + expect( + mockBackgroundTaskRegistry.clearStatusChangeCallback, + ).toHaveBeenCalledWith(statusChanged); + expect( + mockBackgroundTaskRegistry.setStatusChangeCallback, + ).not.toHaveBeenCalledWith(undefined); + }); + + it('holds an Agent terminal notification from persistence to continuation', async () => { + let finishPersistence!: () => void; + mockChatRecordingService.recordNotificationStrict.mockImplementationOnce( + () => + new Promise((resolve) => { + finishPersistence = resolve; + }), + ); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + createReportingSession(); + + const notification = session.enqueueBackgroundNotification({ + displayText: 'Agent completed.', + modelText: '', + taskId: 'agent-persisting', + status: 'completed', + kind: 'agent', + }); + await vi.waitFor(() => + expect(holdIds('notification')).toEqual(['agent-persisting']), + ); + expect(session.isIdle()).toBe(false); + + finishPersistence(); + await expect(notification).resolves.toEqual({ accepted: true }); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + expect(session.isIdle()).toBe(true); + session.dispose(); + }); + + it('does not hold for a Monitor notification', async () => { + let finishPersistence!: () => void; + mockChatRecordingService.recordNotificationStrict.mockImplementationOnce( + () => + new Promise((resolve) => { + finishPersistence = resolve; + }), + ); + mockChat.sendMessageStream = vi + .fn() + .mockResolvedValue(createEmptyStream()); + createReportingSession(); + + const notification = session.enqueueBackgroundNotification({ + displayText: 'Monitor fired.', + modelText: '', + taskId: 'monitor-persisting', + status: 'completed', + kind: 'monitor', + }); + await vi.waitFor(() => + expect( + mockChatRecordingService.recordNotificationStrict, + ).toHaveBeenCalledOnce(), + ); + // Monitors are outside activeWork's declared scope. + expect(session.collectActiveWorkHolds()).toEqual([]); + + finishPersistence(); + await expect(notification).resolves.toEqual({ accepted: true }); + session.dispose(); + }); + + it('keeps holding while the parent continuation runs', 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 = + mockBackgroundTaskRegistry.setNotificationCallback.mock.calls.at( + -1, + )?.[0] as ( + displayText: string, + modelText: string, + meta: { agentId: string; status: string }, + ) => void; + + notify('Agent completed.', '', { + agentId: 'agent-1', + status: 'completed', + }); + await vi.waitFor(() => + expect(holdIds('notification')).toEqual(['agent-1']), + ); + expect(session.isIdle()).toBe(false); + + // The agent itself is finished; the hold now belongs to the terminal + // notification and its continuation turn, so the session stays held. + mockBackgroundTaskRegistry.listUnfinalizedBackgroundAgentIds.mockReturnValue( + [], + ); + await Promise.resolve(); + expect(holdIds('notification')).toEqual(['agent-1']); + + releaseNotification(); + await vi.waitFor(() => + expect(session.collectActiveWorkHolds()).toEqual([]), + ); + expect(session.isIdle()).toBe(true); + session.dispose(); + }); + }); + it('bridges workflow approvals through ACP permission requests', async () => { mockToolRegistry.getTool.mockReturnValue({ displayName: 'Shell', diff --git a/packages/cli/src/acp-integration/session/Session.ts b/packages/cli/src/acp-integration/session/Session.ts index ad503a5296d..5dd850e72f5 100644 --- a/packages/cli/src/acp-integration/session/Session.ts +++ b/packages/cli/src/acp-integration/session/Session.ts @@ -172,6 +172,7 @@ import { NOT_CURRENTLY_GENERATING_CANCEL_MESSAGE } from '@qwen-code/acp-bridge/b // Single source of truth shared with the daemon-side answerer (BridgeClient), // so a rename can't desync caller and answerer into a silent -32601 latch. import { + type ActiveWorkHoldV1, DAEMON_CHANNEL_DELIVERY_META_KEY, MID_TURN_QUEUE_DRAIN_METHOD, isValidTrustedModelPrompt, @@ -1342,11 +1343,13 @@ export class Session implements SessionContext { private notificationProcessing = false; private notificationAbortController: AbortController | null = null; private notificationCompletion: Promise | null = null; + private currentAgentNotificationTaskId: string | null = null; private readonly persistedBackgroundNotificationTaskIds = new Set(); private readonly backgroundNotificationAcceptances = new Map< string, Promise >(); + private readonly activeAgentNotificationAcceptances = new Set(); // Set true in dispose(). Guards #drainCronQueue and #drainNotificationQueue // against the race where #drainNotificationQueue's finally block kicks off @@ -1358,6 +1361,9 @@ export class Session implements SessionContext { private closeGateCompletion: Promise | null = null; private resolveCloseGate: (() => void) | null = null; private unsubscribeChatRecordingFailure?: () => void; + /** The exact status-change callback this Session installed, so dispose can + * retract its own and nobody else's. */ + #statusChangeCallback: (() => void) | undefined; private readonly workflowApprovalAbortController = new AbortController(); private activeTodoPlanRevision?: { planId: string; @@ -1404,6 +1410,12 @@ export class Session implements SessionContext { readonly config: Config, private readonly client: AgentSideConnection, private readonly settings: LoadedSettings, + /** + * Invoked whenever work this Session owns may have started or finished. + * The owner (one reporter per ACP channel) coalesces these and republishes + * a full snapshot; the Session itself keeps no reporting state. + */ + private readonly onActiveWorkChanged?: () => void, ) { this.sessionId = id; this.runtimeBaseDir = config.storage.getRuntimeBaseDir(); @@ -2196,7 +2208,59 @@ export class Session implements SessionContext { } isIdle(): boolean { - return !this.closing && !this.#hasActiveTurn(); + return ( + !this.closing && + !this.#hasActiveTurn() && + this.collectActiveWorkHolds().length === 0 + ); + } + + /** + * The Session's current active-work holds, derived on every call. + * + * Nothing here is bookkeeping kept in parallel with the real work: agent + * holds come straight out of the registry's unfinalized set, notification + * holds out of the queue and the in-flight acceptance/continuation state. + * A hold therefore cannot leak past the work it names, and the daemon's + * cached copy converges on whatever these owners actually say. + * + * `hasUnfinalizedTasks()`'s predicate — not `hasRunningTasks()`' — backs the + * agent category on purpose: an agent that has been cancelled still owes its + * terminal task-notification, and treating it as finished would let the + * daemon reap the Session inside the cancel → finalizeCancelled() window and + * strand that notification. + * + * Prompts are absent by design. The daemon accepts, queues, dispatches, and + * settles them itself, so its own count is both authoritative and strictly + * wider than anything reported from here (it covers prompts still waiting in + * the FIFO, which the child cannot see). + */ + collectActiveWorkHolds(): ActiveWorkHoldV1[] { + if (this.disposed) return []; + const holds: ActiveWorkHoldV1[] = []; + for (const agentId of this.config + .getBackgroundTaskRegistry() + .listUnfinalizedBackgroundAgentIds()) { + holds.push({ category: 'agent', id: agentId }); + } + const notificationIds = new Set(); + for (const item of this.notificationQueue) { + if (item.kind === 'agent') notificationIds.add(item.taskId); + } + for (const taskId of this.activeAgentNotificationAcceptances) { + notificationIds.add(taskId); + } + if (this.currentAgentNotificationTaskId !== null) { + notificationIds.add(this.currentAgentNotificationTaskId); + } + for (const taskId of notificationIds) { + holds.push({ category: 'notification', id: taskId }); + } + return holds; + } + + #activeWorkChanged(): void { + this.onActiveWorkChanged?.(); } #hasActiveTurn(): boolean { @@ -2330,6 +2394,12 @@ export class Session implements SessionContext { this.config.getBackgroundTaskRegistry().abortAll({ notify: false }); this.config.getBackgroundTaskRegistry().setNotificationCallback(undefined); + if (this.#statusChangeCallback) { + this.config + .getBackgroundTaskRegistry() + .clearStatusChangeCallback(this.#statusChangeCallback); + this.#statusChangeCallback = undefined; + } this.config.getMonitorRegistry().setNotificationCallback(undefined); this.config.getBackgroundShellRegistry().setNotificationCallback(undefined); this.config.getChatRecordingService()?.setTitleRecordedCallback(undefined); @@ -2637,6 +2707,7 @@ export class Session implements SessionContext { } this.notificationQueue = []; this.notificationProcessing = false; + this.#activeWorkChanged(); // Stop scheduler and emit exit summary const scheduler = this.config.isCronEnabled() @@ -6083,6 +6154,14 @@ export class Session implements SessionContext { #registerBackgroundNotificationCallbacks(): void { const backgroundRegistry = this.config.getBackgroundTaskRegistry(); + // Single-slot setter, so remember exactly what we installed and only ever + // retract that. Under ACP nothing else claims the slot today, but a Session + // must not clear a callback it did not install — the TUI uses the same + // registry, and "clear on dispose" would silently unhook it. + this.#statusChangeCallback = () => { + this.#activeWorkChanged(); + }; + backgroundRegistry.setStatusChangeCallback(this.#statusChangeCallback); backgroundRegistry.setNotificationCallback( (displayText, modelText, meta) => { this.#enqueueBackgroundNotification({ @@ -6207,6 +6286,7 @@ export class Session implements SessionContext { ); } this.notificationQueue.push(item); + this.#activeWorkChanged(); void this.#drainNotificationQueue(); } @@ -6221,6 +6301,10 @@ export class Session implements SessionContext { const acceptance = this.#persistDaemonBackgroundNotification(item); this.backgroundNotificationAcceptances.set(item.taskId, acceptance); + if (item.kind === 'agent') { + this.activeAgentNotificationAcceptances.add(item.taskId); + this.#activeWorkChanged(); + } try { return { accepted: await acceptance }; } finally { @@ -6228,6 +6312,10 @@ export class Session implements SessionContext { this.backgroundNotificationAcceptances.get(item.taskId) === acceptance ) { this.backgroundNotificationAcceptances.delete(item.taskId); + if (item.kind === 'agent') { + this.activeAgentNotificationAcceptances.delete(item.taskId); + this.#activeWorkChanged(); + } } } } @@ -6323,16 +6411,25 @@ export class Session implements SessionContext { if (nextIndex < 0) break; const [item] = this.notificationQueue.splice(nextIndex, 1); if (!item) break; - await runWithInvocationContext(undefined, () => - sessionIdContext.run(this.config.getSessionId(), () => - this.#executeBackgroundNotificationPromptInner(item), - ), - ); + this.currentAgentNotificationTaskId = + item.kind === 'agent' ? item.taskId : null; + this.#activeWorkChanged(); + try { + await runWithInvocationContext(undefined, () => + sessionIdContext.run(this.config.getSessionId(), () => + this.#executeBackgroundNotificationPromptInner(item), + ), + ); + } finally { + this.currentAgentNotificationTaskId = null; + this.#activeWorkChanged(); + } } } finally { this.notificationProcessing = false; resolveCompletion(); this.notificationCompletion = null; + this.#activeWorkChanged(); void this.#drainCronQueue(); 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 273665726b0..fe27cd56706 100644 --- a/packages/cli/src/acp-integration/session/Session.worktree.test.ts +++ b/packages/cli/src/acp-integration/session/Session.worktree.test.ts @@ -159,6 +159,9 @@ describe('Session.pendingWorktreeNotice', () => { // these registries; provide no-op stubs so construction succeeds. getBackgroundTaskRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), + setStatusChangeCallback: vi.fn(), + clearStatusChangeCallback: vi.fn(), + listUnfinalizedBackgroundAgentIds: vi.fn().mockReturnValue([]), }), getMonitorRegistry: vi.fn().mockReturnValue({ setNotificationCallback: vi.fn(), diff --git a/packages/cli/src/serve/multi-workspace-sessions.test.ts b/packages/cli/src/serve/multi-workspace-sessions.test.ts index e2a25abc5b1..097cc5dc096 100644 --- a/packages/cli/src/serve/multi-workspace-sessions.test.ts +++ b/packages/cli/src/serve/multi-workspace-sessions.test.ts @@ -442,6 +442,17 @@ function makeBridge( get activePromptCount() { return 0; }, + get activeWork() { + return false; + }, + get activeWorkCoverage() { + return { + total: 0, + covered: 0, + onNegotiatedChannel: 0, + oldestCoveredReportAt: null, + }; + }, get pendingPromptTotal() { return 0; }, diff --git a/packages/cli/src/serve/routes/health-demo.ts b/packages/cli/src/serve/routes/health-demo.ts index 227b652557f..1ad2d802979 100644 --- a/packages/cli/src/serve/routes/health-demo.ts +++ b/packages/cli/src/serve/routes/health-demo.ts @@ -4,6 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ +import { gradeActiveWorkCoverage } from '@qwen-code/acp-bridge/bridgeTypes'; import type { Application, Request, Response } from 'express'; import { writeStderrLine } from '../../utils/stdioHelpers.js'; import { getDemoHtml } from '../demo.js'; @@ -107,8 +108,16 @@ export function createHealthDemoRoutes( let sessions = 0; let pendingPermissions = 0; let activePrompts = 0; + let activeWork = false; let channelAlive = false; let lastActivity: number | null = null; + // Coverage is summed as counts and graded once, at the end. Grading per + // runtime and combining the grades does not work: a runtime with zero + // Sessions is vacuously `full`, and treating that as evidence let an + // empty workspace vouch for another workspace's unreported Sessions. + let coveredSessions = 0; + let sessionsOnNegotiatedChannel = 0; + let oldestReportAt: number | null = null; for (const runtime of runtimes) { failedWorkspaceId = runtime.workspaceId; @@ -116,12 +125,24 @@ export function createHealthDemoRoutes( const runtimeSessions = bridge.sessionCount; const runtimePendingPermissions = bridge.pendingPermissionCount; const runtimeActivePrompts = bridge.activePromptCount; + const runtimeActiveWork = bridge.activeWork; + const runtimeCoverage = bridge.activeWorkCoverage; + const runtimeOldestReportAt = runtimeCoverage.oldestCoveredReportAt; const runtimeChannelAlive = bridge.isChannelLive(); const runtimeLastActivity = bridge.lastActivityAt; sessions += runtimeSessions; pendingPermissions += runtimePendingPermissions; activePrompts += runtimeActivePrompts; + activeWork = activeWork || runtimeActiveWork; + coveredSessions += runtimeCoverage.covered; + sessionsOnNegotiatedChannel += runtimeCoverage.onNegotiatedChannel; + if ( + runtimeOldestReportAt !== null && + (oldestReportAt === null || runtimeOldestReportAt < oldestReportAt) + ) { + oldestReportAt = runtimeOldestReportAt; + } channelAlive = channelAlive || runtimeChannelAlive; if ( runtimeLastActivity !== null && @@ -140,6 +161,17 @@ export function createHealthDemoRoutes( sessions, pendingPermissions, activePrompts, + activeWork, + activeWorkReporting: gradeActiveWorkCoverage({ + total: sessions, + covered: coveredSessions, + onNegotiatedChannel: sessionsOnNegotiatedChannel, + }), + // 0 rather than null when nothing is covered: an idle daemon with no + // sessions must not read as infinitely stale to a controller applying + // its own freshness floor. `oldestReportAt` is the oldest *covered* + // report, so this never disagrees with the grade above. + activeWorkStaleMs: oldestReportAt === null ? 0 : now - oldestReportAt, connectedClients: getActiveSseCount(), channelAlive, lastActivityAt: diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index 41ac3c204a1..7f3dd4a8eb1 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -458,6 +458,13 @@ function makeRuntimeBridge(): HttpAcpBridge { sessionCount: 0, pendingPermissionCount: 0, activePromptCount: 0, + activeWork: false, + activeWorkCoverage: { + total: 0, + covered: 0, + onNegotiatedChannel: 0, + oldestCoveredReportAt: null, + }, lastActivityAt: null, getDaemonStatusSnapshot: vi.fn().mockReturnValue(BASE_BRIDGE_SNAPSHOT), isChannelLive: vi.fn().mockReturnValue(true), diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 4e48e2e5aa3..be47e25321d 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -1805,6 +1805,17 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { get activePromptCount() { return 0; }, + get activeWork() { + return false; + }, + get activeWorkCoverage() { + return { + total: 0, + covered: 0, + onNegotiatedChannel: 0, + oldestCoveredReportAt: null, + }; + }, get lastActivityAt() { return null; }, @@ -20651,6 +20662,12 @@ describe('createServeApp', () => { expect(res.body).toMatchObject({ status: 'ok', activePrompts: 0, + activeWork: false, + activeWorkReporting: 'full', + // No covered session, so the boolean rests on nothing stale. Null here + // would read as infinitely stale to a controller with a freshness floor + // and make an idle daemon permanently un-restartable. + activeWorkStaleMs: 0, connectedClients: 0, channelAlive: false, lastActivityAt: null, @@ -20696,6 +20713,15 @@ describe('createServeApp', () => { sessionCount: { get: () => 2 }, pendingPermissionCount: { get: () => 1 }, activePromptCount: { get: () => 1 }, + activeWork: { get: () => false }, + activeWorkCoverage: { + get: () => ({ + total: 2, + covered: 2, + onNegotiatedChannel: 2, + oldestCoveredReportAt: now - 20_000, + }), + }, lastActivityAt: { get: () => now - 120_000 }, isChannelLive: { value: () => true }, }); @@ -20703,6 +20729,17 @@ describe('createServeApp', () => { sessionCount: { get: () => 3 }, pendingPermissionCount: { get: () => 2 }, activePromptCount: { get: () => 2 }, + activeWork: { get: () => true }, + // Sessions this runtime cannot vouch for drag the daemon-wide grade + // down, because `activeWork` is an OR across all of them. + activeWorkCoverage: { + get: () => ({ + total: 3, + covered: 1, + onNegotiatedChannel: 3, + oldestCoveredReportAt: now - 45_000, + }), + }, lastActivityAt: { get: () => now - 30_000 }, isChannelLive: { value: () => false }, }); @@ -20735,6 +20772,9 @@ describe('createServeApp', () => { sessions: 5, pendingPermissions: 3, activePrompts: 3, + activeWork: true, + activeWorkReporting: 'partial', + activeWorkStaleMs: 45_000, channelAlive: true, lastActivityAt: new Date(now - 30_000).toISOString(), idleSinceMs: 30_000, @@ -20744,6 +20784,63 @@ describe('createServeApp', () => { } }); + it('does not let an empty workspace vouch for another one at deep=1', async () => { + const now = 1_700_000_120_000; + // Nothing to report, so this runtime vouches for everything it has — + // vacuously. Grading per runtime and combining the grades would let that + // count as evidence about the *other* runtime's sessions. + const emptyBridge = fakeBridge(); + const unsupportedBridge = fakeBridge(); + Object.defineProperties(unsupportedBridge, { + sessionCount: { get: () => 2 }, + activeWork: { get: () => false }, + // Two live sessions, neither on a channel that negotiated reporting: + // `activeWork: false` here means "nobody told me", not "nothing runs". + activeWorkCoverage: { + get: () => ({ + total: 2, + covered: 0, + onNegotiatedChannel: 0, + oldestCoveredReportAt: null, + }), + }, + }); + const registry = createWorkspaceRegistry([ + makeWorkspaceRuntimeForTest({ + workspaceId: 'health-empty', + workspaceCwd: WS_BOUND, + primary: true, + bridge: emptyBridge, + }), + makeWorkspaceRuntimeForTest({ + workspaceId: 'health-unsupported', + workspaceCwd: WS_DIFFERENT, + primary: false, + bridge: unsupportedBridge, + }), + ]); + const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(now); + try { + const app = createServeApp(baseOpts, undefined, { + workspaceRegistry: registry, + }); + const res = await request(app) + .get('/health?deep=1') + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(res.status).toBe(200); + expect(res.body).toMatchObject({ + sessions: 2, + activeWork: false, + // `none`, not `partial`: no session anywhere is covered, which is the + // one case where a controller must not act on `activeWork` at all. + activeWorkReporting: 'none', + activeWorkStaleMs: 0, + }); + } finally { + nowSpy.mockRestore(); + } + }); + it('includes draining workspaces until registry removal completes', async () => { const primaryBridge = fakeBridge(); const secondaryBridge = fakeBridge(); @@ -20790,11 +20887,11 @@ describe('createServeApp', () => { it('does not short-circuit later workspace health getters', async () => { const primaryBridge = fakeBridge(); const secondaryBridge = fakeBridge(); - Object.defineProperty(primaryBridge, 'isChannelLive', { - value: () => true, + Object.defineProperty(primaryBridge, 'activeWork', { + get: () => true, }); - Object.defineProperty(secondaryBridge, 'isChannelLive', { - value: () => { + Object.defineProperty(secondaryBridge, 'activeWork', { + get: () => { throw new Error('secondary bridge wedged'); }, }); diff --git a/packages/core/src/agents/background-tasks.ts b/packages/core/src/agents/background-tasks.ts index ccda093b660..333a9f35338 100644 --- a/packages/core/src/agents/background-tasks.ts +++ b/packages/core/src/agents/background-tasks.ts @@ -1304,6 +1304,33 @@ export class BackgroundTaskRegistry { return false; } + /** + * The agent ids behind `hasUnfinalizedTasks()`, in registration order. + * + * Callers that must *name* the outstanding work — rather than just know + * that some exists — use this. The daemon's active-work snapshot builds + * one hold per id so a restart controller and the session-retention path + * both see the same set the registry itself would report, with no second + * ledger to drift out of sync. Deliberately shares + * `hasUnfinalizedTasks()`'s predicate (and not `hasRunningTasks()`'s): + * a cancelled entry still owes its terminal task-notification, and + * dropping it here would let the daemon reap the session inside the + * cancel → finalizeCancelled() window. + */ + listUnfinalizedBackgroundAgentIds(): string[] { + const ids: string[] = []; + for (const entry of this.agents.values()) { + if (!entry.isBackgrounded) continue; + if ( + entry.status === 'running' || + (entry.status === 'cancelled' && !entry.notified) + ) { + ids.push(entry.agentId); + } + } + return ids; + } + /** * True while any background entry is still actually executing. Unlike * `hasUnfinalizedTasks()`, a `cancelled`-but-not-yet-finalized entry @@ -1502,6 +1529,20 @@ export class BackgroundTaskRegistry { this.statusChangeCallback = cb; } + /** + * Retract `cb`, but only if it is still the installed one. + * + * The slot holds a single callback, so a subscriber that clears it + * unconditionally on teardown can unhook whoever claimed it afterwards. This + * makes the retraction safe to call from any owner's dispose path without + * having to know whether it is still the owner. + */ + clearStatusChangeCallback(cb: BackgroundStatusChangeCallback): void { + if (this.statusChangeCallback === cb) { + this.statusChangeCallback = undefined; + } + } + setActivityChangeCallback( cb: BackgroundActivityChangeCallback | undefined, ): void {