diff --git a/docs/design/2026-08-24-daemon-channel-transport-liveness.md b/docs/design/2026-08-24-daemon-channel-transport-liveness.md new file mode 100644 index 00000000000..6c47d239e76 --- /dev/null +++ b/docs/design/2026-08-24-daemon-channel-transport-liveness.md @@ -0,0 +1,89 @@ +# Daemon ACP channel transport liveness + +## Problem + +The daemon knows when an ACP child exits or its bounded NDJSON transport rejects a frame, but it cannot detect a child that remains alive while its event loop or transport stops answering. Session active-work snapshots cannot fill this gap: silence from one Session is not proof that the shared channel is dead, and using that signal to recycle the child would destroy every multiplexed Session. + +Layer 2 of #8586 adds a separate channel-level liveness contract. It does not change `activeWork`, Session cleanup, or logical Agent progress detection. + +## Scope and ownership + +One `AcpSessionBridge` owns one workspace runtime and at most one attach-available ACP child channel. The liveness monitor therefore belongs to `ChannelInfo`, not to a Session. A failure condemns the whole child because every Session on that transport is already unreachable. + +The monitor covers only daemon-owned ACP children that acknowledge the v1 capability during `initialize`. Older or alternate children that do not acknowledge it keep the current behavior. Direct ACP consumers and channel adapters are unchanged. + +## Wire contract + +The daemon advertises this initialize metadata: + +```json +{ + "qwen.daemon.channelLiveness": { "v": 1 } +} +``` + +A supporting child echoes the same metadata in its initialize response. Once negotiated, the daemon periodically calls the read-only extension method: + +```text +qwen/status/channel/ping { v: 1, nonce: N } + -> { v: 1, nonce: N } +``` + +`nonce` is a non-negative safe integer allocated monotonically for the life of the channel and wraps to zero after `Number.MAX_SAFE_INTEGER`. The daemon accepts only an exact version and nonce echo. A malformed response or a request rejection after a v1 acknowledgement is a definite protocol failure; it does not spend the timeout retry budget. + +No Session id, work state, timeout, or process detail crosses this method. + +## Timing and escalation + +The fixed internal policy is: + +- wait 15 seconds between healthy probes; +- allow 10 seconds for one response; +- after the first on-time timeout, retry immediately; +- after a second consecutive on-time timeout, fail the transport and recycle the channel. + +This bounds detection to 35 seconds when a child wedges just after a healthy probe, while requiring two independent unanswered requests before taking down multiplexed Sessions. A successful response resets the consecutive-timeout count. These values are implementation policy, not public configuration, matching #8586's non-goal. + +At most two requests can remain unresolved: the first timed-out request and its retry. ACP request ids keep a late first response separate from the retry. Once the channel is condemned, the existing transport teardown closes both. + +## Host suspend and parent stalls + +Every scheduled callback records its expected firing time using `performance.now()`. If either the interval callback or a probe-timeout callback runs more than one second after its expected monotonic deadline, the daemon treats the observation as a local scheduling gap and clears the miss count. A delayed interval starts a fresh interval; a delayed probe timeout keeps the same bounded request outstanding and gives it a fresh response window. It does not infer anything about the child from a timer the parent could not run on time. + +This covers platforms whose monotonic clock advances across host suspend. On platforms where that clock pauses during suspend, the timer deadline pauses with it and no overdue callback is produced. Wall-clock changes never participate. + +The rule also prevents a long daemon event-loop pause from charging a child timeout. It does not excuse an on-time timeout merely because the daemon was busy earlier; the immediate retry supplies the second observation. + +## Lifecycle and failure path + +The monitor starts only after initialize succeeds and the channel is published. Its timers are unreferenced so an otherwise idle daemon can exit. It stops on transport failure, channel exit, synchronous kill-all, and graceful shutdown; every callback also rechecks that the same channel remains live and is not dying. + +A terminal liveness failure enters the existing transport-failure path before starting teardown: + +1. mark the `ChannelInfo` dying synchronously so no new Session work is admitted; +2. retain a bounded failure code for telemetry; +3. clear in-flight extension refresh bookkeeping; +4. ask the daemon transport guard to fail and terminate the child, or call the channel's ordinary kill fallback for injected channels without a guard; +5. let the existing `channel.exited` handler remove every multiplexed Session, publish `session_died`, and release channel state. + +This preserves the existing overlap invariant: a replacement can be spawned while the dying channel remains reachable to `killAllSync` until OS reap. + +## Observability + +The daemon emits one `channel.liveness_failed` telemetry event and one bounded stderr line when escalation begins. The existing `channel.exited` event then reports `transport_failed=true`, `transport_failure_initiated_teardown=true`, and either `acp_channel_liveness_timeout` or `acp_channel_liveness_protocol_error`. + +No health or status response field is added. Public health remains an observation surface, not a synchronous probe. + +## Compatibility and non-goals + +- An unacknowledged capability disables the monitor for that channel. +- Existing Session active-work reporting and conditional close behavior do not change. +- A responsive process with a logically stalled Agent still answers ping; Layer 3 owns progress watchdogs. +- Runtime generation draining and recovery remain Layers 4 and 5. +- There is no public timeout setting, persistence change, transport retry, Session-specific kill, or inference from active-work snapshots. + +## Verification + +Unit coverage pins negotiation, exact ping echo validation, healthy cadence, timeout reset, two-timeout escalation, local timer-delay suppression, cleanup, legacy compatibility, and teardown of every Session on the failed channel. A child-side test pins the initialize acknowledgement and stateless ping response. + +The E2E plan freezes only the ACP child with `SIGSTOP` to reproduce the current gap and verify channel recycle. Host-suspend safety is deterministic unit coverage because a real machine suspend cannot be made reliable or portable in CI. diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 52eac20257e..94314c21e9b 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -87,6 +87,8 @@ import { ACTIVE_WORK_MAX_SNAPSHOT_SESSIONS, ACTIVE_WORK_NOTIFICATION_METHOD, ACTIVE_WORK_STALE_INTERVALS, + CHANNEL_LIVENESS_META_KEY, + CHANNEL_LIVENESS_VERSION, gradeActiveWorkCoverage, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, @@ -94,6 +96,11 @@ import { WORKTREE_MCP_DEFER_META_KEY, LOAD_REPLAY_HIDE_INHERITED_META_KEY, } from './bridgeTypes.js'; +import { + CHANNEL_LIVENESS_INTERVAL_MS, + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, + CHANNEL_LIVENESS_TIMEOUT_CODE, +} from './channel-liveness.js'; import { ApprovalMode, SESSION_ARTIFACT_PERSISTENCE_VERSION, @@ -276,6 +283,18 @@ function activeWorkInitializeResponse( }; } +function channelLivenessInitializeResponse(): InitializeResponse { + return { + protocolVersion: PROTOCOL_VERSION, + agentInfo: { name: 'channel-liveness-agent', version: '0' }, + authMethods: [], + agentCapabilities: {}, + _meta: { + [CHANNEL_LIVENESS_META_KEY]: { v: CHANNEL_LIVENESS_VERSION }, + }, + }; +} + function agentHold(id: string) { return { category: 'agent' as const, id }; } @@ -11857,6 +11876,155 @@ describe('createAcpSessionBridge', () => { void killPromise; }); + describe('channel liveness', () => { + it('advertises v1 but stays disabled when the child does not acknowledge it', async () => { + vi.useFakeTimers(); + try { + const handle = makeChannel(); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + expect(handle.agent.initializeCalls[0]?._meta).toMatchObject({ + [CHANNEL_LIVENESS_META_KEY]: { v: CHANNEL_LIVENESS_VERSION }, + }); + await vi.advanceTimersByTimeAsync( + CHANNEL_LIVENESS_INTERVAL_MS + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS * 2, + ); + expect(handle.agent.extMethodCalls).not.toContainEqual( + expect.objectContaining({ + method: SERVE_STATUS_EXT_METHODS.channelPing, + }), + ); + expect(handle.killed).toBe(false); + await bridge.shutdown(); + } finally { + vi.useRealTimers(); + } + }); + + it('fails the transport after two unanswered channel probes', async () => { + vi.useFakeTimers(); + try { + const event = vi.fn(); + const channelLifecycle = vi.fn(); + const handle = makeChannel({ + initializeImpl: () => channelLivenessInitializeResponse(), + extMethodImpl: (method) => + method === SERVE_STATUS_EXT_METHODS.channelPing + ? new Promise(() => {}) + : {}, + }); + const baseKill = handle.channel.kill; + const transportFail = vi.fn((error: unknown) => { + void baseKill(); + return error; + }); + handle.channel = { + ...handle.channel, + transportGuard: { + maxActiveHandlers: 10, + maxActiveHandlerBytes: 1024 * 1024, + reserveOutboundOperation: () => () => {}, + reservePreparedResponse: () => {}, + fail: transportFail, + }, + }; + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + sessionScope: 'thread', + telemetry: { + captureContext: () => undefined, + runWithContext: async (_captured, fn) => await fn(), + withSpan: async (_operation, _attributes, fn) => await fn(), + event, + injectPromptContext: (request) => request, + metrics: { + sessionLifecycle: vi.fn(), + channelLifecycle, + promptQueueWait: vi.fn(), + promptDuration: vi.fn(), + cancelled: vi.fn(), + }, + }, + }); + const first = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const second = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + expect(bridge.sessionCount).toBe(2); + const firstEvents: BridgeEvent[] = []; + const secondEvents: BridgeEvent[] = []; + const firstDrain = (async () => { + for await (const event of bridge.subscribeEvents(first.sessionId)) { + firstEvents.push(event); + } + })(); + const secondDrain = (async () => { + for await (const event of bridge.subscribeEvents(second.sessionId)) { + secondEvents.push(event); + } + })(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + const firstProbeCalls = handle.agent.extMethodCalls.filter( + ({ method }) => method === SERVE_STATUS_EXT_METHODS.channelPing, + ); + expect(firstProbeCalls).toEqual([ + { + method: SERVE_STATUS_EXT_METHODS.channelPing, + params: { v: CHANNEL_LIVENESS_VERSION, nonce: 0 }, + }, + ]); + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(transportFail).not.toHaveBeenCalled(); + expect( + handle.agent.extMethodCalls.filter( + ({ method }) => method === SERVE_STATUS_EXT_METHODS.channelPing, + ), + ).toHaveLength(2); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + await handle.channel.exited; + await Promise.all([firstDrain, secondDrain]); + + expect(transportFail).toHaveBeenCalledOnce(); + expect(transportFail.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_TIMEOUT_CODE, + }); + expect(bridge.sessionCount).toBe(0); + expect(firstEvents.at(-1)).toMatchObject({ + type: 'session_died', + data: { sessionId: first.sessionId, reason: 'channel_closed' }, + }); + expect(secondEvents.at(-1)).toMatchObject({ + type: 'session_died', + data: { sessionId: second.sessionId, reason: 'channel_closed' }, + }); + expect(channelLifecycle).toHaveBeenCalledWith('exit', false); + expect(event).toHaveBeenCalledWith( + 'channel.liveness_failed', + expect.objectContaining({ + 'qwen-code.daemon.channel.session_count': 2, + 'qwen-code.daemon.channel.transport_error_code': + CHANNEL_LIVENESS_TIMEOUT_CODE, + }), + ); + expect(event).toHaveBeenCalledWith( + 'channel.exited', + expect.objectContaining({ + 'qwen-code.daemon.channel.transport_failed': true, + 'qwen-code.daemon.channel.transport_failure_initiated_teardown': true, + 'qwen-code.daemon.channel.transport_error_code': + CHANNEL_LIVENESS_TIMEOUT_CODE, + }), + ); + await bridge.shutdown(); + } finally { + vi.useRealTimers(); + } + }); + }); + it('transport failure marks the channel dying before process exit', async () => { const handles: ChannelHandle[] = []; const failures: Array>> = []; diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index b01835a5aa1..496977e4b8d 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -135,6 +135,8 @@ import { ACTIVE_WORK_HOLD_CATEGORIES, ACTIVE_WORK_MAX_SESSION_HOLDS, ACTIVE_WORK_STALE_INTERVALS, + CHANNEL_LIVENESS_META_KEY, + CHANNEL_LIVENESS_VERSION, clampActiveWorkIntervalMs, type ActiveWorkHeartbeatCapabilityV1, type ActiveWorkHoldCategory, @@ -163,6 +165,11 @@ import { isValidTrustedModelPrompt, sessionCloseDrainBudgetMs, } from './bridgeTypes.js'; +import { + startChannelLivenessMonitor, + type ChannelLivenessMonitor, + type ChannelLivenessFailure, +} from './channel-liveness.js'; import { getChannelStartupProfileAttributes } from './channel-startup-profile.js'; import type { BridgeSession, @@ -966,6 +973,7 @@ interface ChannelInfo { /** Highest snapshot sequence applied; guards against reordering only. */ seq: number; }; + channelLiveness?: ChannelLivenessMonitor; handshakeComplete: boolean; } @@ -3281,6 +3289,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { context?: string, ): Promise { ci.isDying = true; + ci.channelLiveness?.stop(); await ci.channel.kill().catch((err) => { writeStderrLine( `qwen serve: channel kill failed${context ? ` (${context})` : ''}: ${String(err)}`, @@ -3432,6 +3441,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (!channelShouldReapWhenIdle(ci) || !hasNoChannelWork(ci)) return; ci.emptyReapPending = false; ci.isDying = true; + ci.channelLiveness?.stop(); await ci.channel.kill().catch(() => { /* best-effort — channel.exited handler still runs */ }); @@ -4131,6 +4141,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { info.transportFailed = true; info.transportFailureCode = safeTransportFailureCode(error); info.isDying = true; + info.channelLiveness?.stop(); clearInFlightExtensionRefreshes(info.connection); }; void channel.transportFailed?.then( @@ -4176,6 +4187,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // the SIGTERM grace window — even if a concurrent `spawnOrAttach` // has already reassigned `channelInfo` to a fresh channel. void channel.exited.then((exitInfo) => { + info.channelLiveness?.stop(); clearInFlightExtensionRefreshes(info.connection); if (channelInfo === info) cancelIdleTimer(); if (info.workspaceMcpDiscoveryTimer) { @@ -4295,6 +4307,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // registered, so failure paths (init throw, timeout, late // shutdown) only need to mark dying + kill — the handler does // the alive-set cleanup when the OS reaps the child. + let channelLivenessNegotiated = false; try { await telemetry.withSpan( 'channel.initialize', @@ -4316,6 +4329,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { [CHANNEL_STARTUP_PROFILE_META_KEY]: { v: CHANNEL_STARTUP_PROFILE_VERSION, }, + [CHANNEL_LIVENESS_META_KEY]: { + v: CHANNEL_LIVENESS_VERSION, + }, [PRIVATE_PARENT_CAPABILITY_META_KEY]: privateParentCapability, }, @@ -4364,6 +4380,12 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { seq: 0, }; } + const channelLivenessCapability = isRecord(response._meta) + ? response._meta[CHANNEL_LIVENESS_META_KEY] + : undefined; + channelLivenessNegotiated = + isRecord(channelLivenessCapability) && + channelLivenessCapability['v'] === CHANNEL_LIVENESS_VERSION; try { const attributes = getChannelStartupProfileAttributes( response, @@ -4418,6 +4440,38 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // caller. channelInfo = info; info.handshakeComplete = true; + if (channelLivenessNegotiated) { + const failChannelLiveness = (error: ChannelLivenessFailure) => { + if (info.isDying || !aliveChannels.has(info)) return; + markTransportFailed(error); + telemetry.event('channel.liveness_failed', { + 'qwen-code.daemon.acp_channel.id': info.id, + 'qwen-code.daemon.channel.session_count': info.sessionIds.size, + 'qwen-code.daemon.channel.transport_error_code': error.code, + }); + writeStderrLine( + `qwen serve: channel liveness failed (${error.code}); killing channel`, + ); + if (info.channel.transportGuard) { + info.channel.transportGuard.fail(error); + } else { + void killChannelWithLog(info, 'channel liveness failure'); + } + }; + info.channelLiveness = startChannelLivenessMonitor({ + probe: (nonce) => + info.connection.extMethod(SERVE_STATUS_EXT_METHODS.channelPing, { + v: CHANNEL_LIVENESS_VERSION, + nonce, + }), + onFailure: failChannelLiveness, + isActive: () => + channelInfo === info && + aliveChannels.has(info) && + !info.isDying && + !shuttingDown, + }); + } telemetry.metrics?.channelLifecycle('spawn'); return info; })(); @@ -4559,6 +4613,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // stays set until OS reap so `killAllSync` mid-SIGTERM still // finds a target (BkUyD invariant). ci.isDying = true; + ci.channelLiveness?.stop(); await ci.channel.kill().catch(() => { /* best-effort — channel.exited handler still runs */ }); @@ -12565,6 +12620,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }); } for (const info of channels) { + info.channelLiveness?.stop(); try { info.channel.killSync(); } catch { @@ -12605,7 +12661,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // `ensureChannel` past the gate would still see the dying // state and not attach). const channels = Array.from(aliveChannels); - for (const ci of channels) ci.isDying = true; + for (const ci of channels) { + ci.isDying = true; + ci.channelLiveness?.stop(); + } // Drain mediator pending state before clearing byId so awaiting // `requestPermission` callers unwind. Each `forgetSession` // settles all matching pending as session_closed; the bridge's diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index f7af4b1da43..635a92ef79c 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -252,6 +252,8 @@ 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 CHANNEL_LIVENESS_META_KEY = 'qwen.daemon.channelLiveness'; +export const CHANNEL_LIVENESS_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 diff --git a/packages/acp-bridge/src/channel-liveness.test.ts b/packages/acp-bridge/src/channel-liveness.test.ts new file mode 100644 index 00000000000..3d1a35f66fd --- /dev/null +++ b/packages/acp-bridge/src/channel-liveness.test.ts @@ -0,0 +1,267 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { CHANNEL_LIVENESS_VERSION } from './bridgeTypes.js'; +import { + CHANNEL_LIVENESS_INTERVAL_MS, + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, + CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE, + CHANNEL_LIVENESS_TIMER_LATE_TOLERANCE_MS, + CHANNEL_LIVENESS_TIMEOUT_CODE, + startChannelLivenessMonitor, +} from './channel-liveness.js'; + +function deferred(): { + promise: Promise; + resolve(value: T): void; +} { + let resolve!: (value: T) => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +afterEach(() => { + vi.useRealTimers(); +}); + +describe('startChannelLivenessMonitor', () => { + it('keeps a healthy channel on the negotiated cadence', async () => { + vi.useFakeTimers(); + const probe = vi.fn(async (nonce: number) => ({ + v: CHANNEL_LIVENESS_VERSION, + nonce, + })); + const onFailure = vi.fn(); + const monitor = startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + + expect(probe.mock.calls).toEqual([[0], [1]]); + expect(onFailure).not.toHaveBeenCalled(); + monitor.stop(); + }); + + it('fails only after two consecutive on-time timeouts', async () => { + vi.useFakeTimers(); + const probe = vi.fn(() => new Promise(() => {})); + const onFailure = vi.fn(); + startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(probe).toHaveBeenCalledTimes(2); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(onFailure).toHaveBeenCalledOnce(); + expect(onFailure.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_TIMEOUT_CODE, + }); + }); + + it('resets the timeout count when the immediate retry succeeds', async () => { + vi.useFakeTimers(); + const probe = vi + .fn<(nonce: number) => Promise>() + .mockImplementationOnce(() => new Promise(() => {})) + .mockImplementation(async (nonce) => ({ + v: CHANNEL_LIVENESS_VERSION, + nonce, + })); + const onFailure = vi.fn(); + const monitor = startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync( + CHANNEL_LIVENESS_INTERVAL_MS + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, + ); + expect(probe).toHaveBeenCalledTimes(2); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + expect(probe).toHaveBeenCalledTimes(3); + expect(onFailure).not.toHaveBeenCalled(); + monitor.stop(); + }); + + it('does not charge a callback delayed by the parent event loop', async () => { + vi.useFakeTimers(); + let monotonicNow = 0; + const probe = vi.fn(async (nonce: number) => ({ + v: CHANNEL_LIVENESS_VERSION, + nonce, + })); + const onFailure = vi.fn(); + const monitor = startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => monotonicNow, + }); + + monotonicNow = + CHANNEL_LIVENESS_INTERVAL_MS + + CHANNEL_LIVENESS_TIMER_LATE_TOLERANCE_MS + + 1; + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + expect(probe).not.toHaveBeenCalled(); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + expect(probe).toHaveBeenCalledOnce(); + expect(onFailure).not.toHaveBeenCalled(); + monitor.stop(); + }); + + it('clears an existing timeout streak after a delayed callback', async () => { + vi.useFakeTimers(); + let monotonicNow = 0; + const probe = vi.fn(() => new Promise(() => {})); + const onFailure = vi.fn(); + startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => monotonicNow, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(probe).toHaveBeenCalledTimes(2); + expect(onFailure).not.toHaveBeenCalled(); + + monotonicNow = + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS + + CHANNEL_LIVENESS_TIMER_LATE_TOLERANCE_MS + + 1; + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(probe).toHaveBeenCalledTimes(2); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(probe).toHaveBeenCalledTimes(3); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(onFailure).toHaveBeenCalledOnce(); + expect(onFailure.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_TIMEOUT_CODE, + }); + }); + + it('does not let a late response reset the current probe streak', async () => { + vi.useFakeTimers(); + const first = deferred(); + const probe = vi + .fn<(nonce: number) => Promise>() + .mockImplementationOnce(() => first.promise) + .mockImplementation(() => new Promise(() => {})); + const onFailure = vi.fn(); + startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync( + CHANNEL_LIVENESS_INTERVAL_MS + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS, + ); + expect(probe).toHaveBeenCalledTimes(2); + + first.resolve({ v: CHANNEL_LIVENESS_VERSION, nonce: 0 }); + await Promise.resolve(); + expect(onFailure).not.toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + expect(onFailure).toHaveBeenCalledOnce(); + expect(onFailure.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_TIMEOUT_CODE, + }); + }); + + it('fails a malformed response without spending the retry budget', async () => { + vi.useFakeTimers(); + const probe = vi.fn(async () => ({ + v: CHANNEL_LIVENESS_VERSION, + nonce: 99, + })); + const onFailure = vi.fn(); + startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + + expect(probe).toHaveBeenCalledOnce(); + expect(onFailure).toHaveBeenCalledOnce(); + expect(onFailure.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE, + }); + }); + + it('fails a rejected probe without spending the retry budget', async () => { + vi.useFakeTimers(); + const probe = vi.fn(() => { + throw new Error('ping rejected'); + }); + const onFailure = vi.fn(); + startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + + expect(probe).toHaveBeenCalledOnce(); + expect(onFailure).toHaveBeenCalledOnce(); + expect(onFailure.mock.calls[0]?.[0]).toMatchObject({ + code: CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE, + }); + }); + + it('cancels an active timeout when stopped', async () => { + vi.useFakeTimers(); + const probe = vi.fn(() => new Promise(() => {})); + const onFailure = vi.fn(); + const monitor = startChannelLivenessMonitor({ + probe, + onFailure, + isActive: () => true, + now: () => 0, + }); + + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_INTERVAL_MS); + monitor.stop(); + await vi.advanceTimersByTimeAsync(CHANNEL_LIVENESS_PROBE_TIMEOUT_MS * 3); + + expect(probe).toHaveBeenCalledOnce(); + expect(onFailure).not.toHaveBeenCalled(); + }); +}); diff --git a/packages/acp-bridge/src/channel-liveness.ts b/packages/acp-bridge/src/channel-liveness.ts new file mode 100644 index 00000000000..080bad3a9df --- /dev/null +++ b/packages/acp-bridge/src/channel-liveness.ts @@ -0,0 +1,168 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { performance } from 'node:perf_hooks'; +import { CHANNEL_LIVENESS_VERSION } from './bridgeTypes.js'; + +export const CHANNEL_LIVENESS_INTERVAL_MS = 15_000; +export const CHANNEL_LIVENESS_PROBE_TIMEOUT_MS = 10_000; +export const CHANNEL_LIVENESS_FAILURE_THRESHOLD = 2; +export const CHANNEL_LIVENESS_TIMER_LATE_TOLERANCE_MS = 1_000; + +export const CHANNEL_LIVENESS_TIMEOUT_CODE = 'acp_channel_liveness_timeout'; +export const CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE = + 'acp_channel_liveness_protocol_error'; + +export class ChannelLivenessFailure extends Error { + constructor( + readonly code: + | typeof CHANNEL_LIVENESS_TIMEOUT_CODE + | typeof CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE, + ) { + super( + code === CHANNEL_LIVENESS_TIMEOUT_CODE + ? 'ACP channel failed consecutive liveness probes' + : 'ACP channel returned an invalid liveness response', + ); + this.name = 'ChannelLivenessFailure'; + } +} + +interface ChannelLivenessMonitorOptions { + probe(nonce: number): Promise; + onFailure(error: ChannelLivenessFailure): void; + isActive(): boolean; + now?: () => number; +} + +export interface ChannelLivenessMonitor { + stop(): void; +} + +type ProbeOutcome = + | { kind: 'response'; response: unknown } + | { kind: 'rejected' } + | { kind: 'timeout' } + | { kind: 'local_delay' } + | { kind: 'stopped' }; + +function isValidResponse(response: unknown, nonce: number): boolean { + return ( + typeof response === 'object' && + response !== null && + !Array.isArray(response) && + (response as Record)['v'] === CHANNEL_LIVENESS_VERSION && + (response as Record)['nonce'] === nonce + ); +} + +export function startChannelLivenessMonitor( + options: ChannelLivenessMonitorOptions, +): ChannelLivenessMonitor { + const now = options.now ?? (() => performance.now()); + let stopped = false; + let timer: NodeJS.Timeout | undefined; + let consecutiveTimeouts = 0; + let nextNonce = 0; + let resolveStopped!: (outcome: ProbeOutcome) => void; + const stoppedOutcome = new Promise((resolve) => { + resolveStopped = resolve; + }); + + const isActive = () => !stopped && options.isActive(); + const stop = () => { + stopped = true; + if (timer !== undefined) { + clearTimeout(timer); + timer = undefined; + } + resolveStopped({ kind: 'stopped' }); + }; + const fail = (error: ChannelLivenessFailure) => { + if (!isActive()) return; + stop(); + options.onFailure(error); + }; + const timerWasLate = (expectedAt: number) => + now() > expectedAt + CHANNEL_LIVENESS_TIMER_LATE_TOLERANCE_MS; + + const runProbe = async () => { + if (!isActive()) return; + const nonce = nextNonce; + nextNonce = nonce === Number.MAX_SAFE_INTEGER ? 0 : Math.max(0, nonce + 1); + const response = Promise.resolve() + .then(() => options.probe(nonce)) + .then( + (value) => ({ kind: 'response', response: value }), + () => ({ kind: 'rejected' }), + ); + + let outcome: ProbeOutcome; + while (true) { + const expectedAt = now() + CHANNEL_LIVENESS_PROBE_TIMEOUT_MS; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => { + timer = undefined; + resolve( + timerWasLate(expectedAt) + ? { kind: 'local_delay' } + : { kind: 'timeout' }, + ); + }, CHANNEL_LIVENESS_PROBE_TIMEOUT_MS); + timer.unref(); + }); + outcome = await Promise.race([response, timeout, stoppedOutcome]); + if (timer !== undefined) { + clearTimeout(timer); + timer = undefined; + } + if (outcome.kind === 'stopped' || !isActive()) return; + if (outcome.kind !== 'local_delay') break; + consecutiveTimeouts = 0; + } + + if (outcome.kind === 'response') { + if (!isValidResponse(outcome.response, nonce)) { + fail(new ChannelLivenessFailure(CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE)); + return; + } + consecutiveTimeouts = 0; + schedule(CHANNEL_LIVENESS_INTERVAL_MS); + return; + } + if (outcome.kind === 'rejected') { + await Promise.resolve(); + fail(new ChannelLivenessFailure(CHANNEL_LIVENESS_PROTOCOL_ERROR_CODE)); + return; + } + + consecutiveTimeouts++; + if (consecutiveTimeouts >= CHANNEL_LIVENESS_FAILURE_THRESHOLD) { + fail(new ChannelLivenessFailure(CHANNEL_LIVENESS_TIMEOUT_CODE)); + return; + } + void runProbe(); + }; + + function schedule(delayMs: number): void { + if (!isActive()) return; + const expectedAt = now() + delayMs; + timer = setTimeout(() => { + timer = undefined; + if (!isActive()) return; + if (timerWasLate(expectedAt)) { + consecutiveTimeouts = 0; + schedule(CHANNEL_LIVENESS_INTERVAL_MS); + return; + } + void runProbe(); + }, delayMs); + timer.unref(); + } + + schedule(CHANNEL_LIVENESS_INTERVAL_MS); + return { stop }; +} diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index 26a05494ffa..2a631a8098e 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -108,6 +108,7 @@ export class MissingCliEntryError extends Error { } export const SERVE_STATUS_EXT_METHODS = { + channelPing: 'qwen/status/channel/ping', workspaceMcp: 'qwen/status/workspace/mcp', workspaceMcpTools: 'qwen/status/workspace/mcp/tools', workspaceMcpResources: 'qwen/status/workspace/mcp/resources', diff --git a/packages/cli/src/acp-integration/acpAgent.test.ts b/packages/cli/src/acp-integration/acpAgent.test.ts index 191812d9843..3022087944a 100644 --- a/packages/cli/src/acp-integration/acpAgent.test.ts +++ b/packages/cli/src/acp-integration/acpAgent.test.ts @@ -1036,6 +1036,8 @@ import { ACTIVE_WORK_HOLD_CATEGORIES, ACTIVE_WORK_LEGACY_HOLD_CATEGORIES, ACTIVE_WORK_NOTIFICATION_METHOD, + CHANNEL_LIVENESS_META_KEY, + CHANNEL_LIVENESS_VERSION, CHANNEL_STARTUP_PROFILE_META_KEY, CHANNEL_STARTUP_PROFILE_VERSION, PROMPT_CANCEL_METHOD, @@ -2845,6 +2847,69 @@ describe('QwenAgent MCP SSE/HTTP support', () => { await agentPromise; }); + it('negotiates and answers channel liveness without a Session', async () => { + 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_LIVENESS_META_KEY]: { v: CHANNEL_LIVENESS_VERSION }, + }, + })) as { _meta?: Record }; + + expect(response._meta).toMatchObject({ + [CHANNEL_LIVENESS_META_KEY]: { v: CHANNEL_LIVENESS_VERSION }, + }); + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.channelPing, { + v: CHANNEL_LIVENESS_VERSION, + nonce: 42, + }), + ).resolves.toEqual({ v: CHANNEL_LIVENESS_VERSION, nonce: 42 }); + await expect( + agent.extMethod(SERVE_STATUS_EXT_METHODS.channelPing, { + v: CHANNEL_LIVENESS_VERSION, + nonce: -1, + }), + ).rejects.toThrow('Invalid channel liveness ping'); + + mockConnectionState.resolve(); + await agentPromise; + }); + + it('does not acknowledge an unrequested channel liveness capability', async () => { + 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: {}, + })) as { _meta?: Record }; + + expect(response._meta?.[CHANNEL_LIVENESS_META_KEY]).toBeUndefined(); + + mockConnectionState.resolve(); + await agentPromise; + }); + /** Installs a beginClose mock that tracks gate hold state; the returned * getter is asserted false after refusal/timeout paths to pin that the * close gate is always released. */ diff --git a/packages/cli/src/acp-integration/acpAgent.ts b/packages/cli/src/acp-integration/acpAgent.ts index c8aad29652c..ac778cd57f9 100644 --- a/packages/cli/src/acp-integration/acpAgent.ts +++ b/packages/cli/src/acp-integration/acpAgent.ts @@ -373,6 +373,8 @@ import { ACTIVE_WORK_HEARTBEAT_VERSION, ACTIVE_WORK_HOLD_CATEGORIES, ACTIVE_WORK_LEGACY_HOLD_CATEGORIES, + CHANNEL_LIVENESS_META_KEY, + CHANNEL_LIVENESS_VERSION, clampActiveWorkIntervalMs, type ActiveWorkHoldV1, CHANNEL_STARTUP_PROFILE_META_KEY, @@ -4401,6 +4403,13 @@ class QwenAgent implements Agent { ) : ACTIVE_WORK_LEGACY_HOLD_CATEGORIES : undefined; + const requestedChannelLiveness = args._meta?.[CHANNEL_LIVENESS_META_KEY]; + const channelLivenessRequested = + requestedChannelLiveness !== null && + typeof requestedChannelLiveness === 'object' && + !Array.isArray(requestedChannelLiveness) && + (requestedChannelLiveness as Record)['v'] === + CHANNEL_LIVENESS_VERSION; if (activeWorkIntervalMs !== undefined) { this.activeWorkReporter?.dispose(); this.activeWorkReporter = new ActiveWorkReporter( @@ -4430,6 +4439,13 @@ class QwenAgent implements Agent { }, } : {}), + ...(channelLivenessRequested + ? { + [CHANNEL_LIVENESS_META_KEY]: { + v: CHANNEL_LIVENESS_VERSION, + }, + } + : {}), }; return Object.keys(responseMeta).length > 0 ? { ...response, _meta: responseMeta } @@ -7649,6 +7665,21 @@ class QwenAgent implements Agent { const SESSION_ID_RE = /^[0-9a-fA-F-]{32,36}$/; switch (method) { + case SERVE_STATUS_EXT_METHODS.channelPing: { + const nonce = params['nonce']; + if ( + params['v'] !== CHANNEL_LIVENESS_VERSION || + typeof nonce !== 'number' || + !Number.isSafeInteger(nonce) || + nonce < 0 + ) { + throw RequestError.invalidParams( + undefined, + 'Invalid channel liveness ping', + ); + } + return { v: CHANNEL_LIVENESS_VERSION, nonce }; + } case PROMPT_CANCEL_METHOD: { const sessionId = params['sessionId']; if (typeof sessionId !== 'string' || sessionId.length === 0) {