feat(daemon): Add ACP channel transport liveness

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.

packages/acp-bridge/src/bridge.test.ts

@@ -87,13 +87,20 @@ 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,
   DAEMON_MODEL_PROMPT_META_KEY,
   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<ReturnType<typeof deferred<unknown>>> = [];