From f393043b036dfa411f7c13655d178a690b5b1a64 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 22 Jun 2026 10:47:02 +0800 Subject: [PATCH 1/5] Add the ACP subagent backend: out-of-process delegation (PR3) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first OUT-OF-PROCESS subagent backend, proving the seam generalizes past the in-process backends. @deepseek-ai/dsh-subagent-acp runs each child agent in a spawned subprocess, driven over the Agent Client Protocol as the CLIENT — the direction-inverted twin of the dsh-acp server bridge. Point the configured command at the acp-agent example and the harness talks to its own process. - Fresh process per run: start spawns, runs one ACP session (initialize → newSession → prompt), dispose kills the subprocess and awaits its exit. - Minimal client stub: advertises no fs/terminal; accumulates agent_message_chunk text as the result output; auto-answers session/request_permission by a configured policy (reject default / allow). No start-time capabilities (an out-of-process child can't enforce the parent's depth/tool-filter); ignores request.parent; injects only `subagents`. - StopReason mapping (end_turn→completed, cancelled→aborted, …); result resolves error/aborted on a child failure, never rejects (seam contract). - Security: credential-shaped ambient env vars are scrubbed; the child's own key is forwarded only via explicit config.env. A spawn-level error (ENOENT) is captured and raced against the ACP drive so a bad command settles error rather than crashing the parent. Testing designed at every tier: keyless integration drives a scripted mock ACP server subprocess (cancellation incl. the pre-newSession race and a torn-pipe-after-cancel, permission auto-answer, non-message updates, spawn failure, HMR, export shape) at 100% coverage; a with-key e2e drives the REAL acp-agent example process (PONG + real file write, verified on disk) — the harness driving itself. Snapshot coverage of an ACP child is deferred as TODO(acp-subagent-replay) (each child is its own process with its own replay). Stayed on @agentclientprotocol/sdk 0.25.1: the proposed 0.28.x bump only deprecates the stable ClientSideConnection/AgentSideConnection API this layer uses (33 sites incl. the server bridge), turning no-deprecated red across code this PR shouldn't rewrite — that fluent-API migration is its own follow-up. The backend needs nothing 0.28.x adds. This completes the subagent seam stack (PR1 interface → PR2 in-process → PR2.5 snapshot infra → PR3 ACP); the seam RFC moves to implemented/, amended. --- docs/architecture.md | 2 +- docs/core-data-structures/subagent.md | 2 +- docs/module-graph.md | 4 + docs/rfc/README.md | 3 +- .../2026-06-21-subagent-capability-seam.md | 4 +- .../2026-06-22-acp-subagent-backend.md | 47 +++ .../2026-06-22-subagent-snapshot-replay.md | 2 +- .../2026-06-20-unify-agent-and-session-id.md | 2 +- knip.json | 4 + packages/README.md | 2 + packages/subagent/README.md | 5 +- packages/subagent/subagent-acp/README.md | 69 ++++ packages/subagent/subagent-acp/package.json | 39 +++ packages/subagent/subagent-acp/src/index.ts | 90 +++++ packages/subagent/subagent-acp/src/run.ts | 292 ++++++++++++++++ .../subagent-acp/tests/mock-acp-server.ts | 150 +++++++++ .../subagent-acp/tests/subagent-acp.e2e.ts | 110 ++++++ .../subagent-acp/tests/subagent-acp.spec.ts | 315 ++++++++++++++++++ packages/subagent/subagent-acp/tsconfig.json | 30 ++ packages/subagent/subagent/README.md | 2 +- packages/subagent/tool-subagent/README.md | 2 +- pnpm-lock.yaml | 25 ++ tsconfig.build.json | 3 +- 23 files changed, 1192 insertions(+), 12 deletions(-) rename docs/rfc/{proposed => implemented}/feature/2026-06-21-subagent-capability-seam.md (94%) create mode 100644 docs/rfc/implemented/feature/2026-06-22-acp-subagent-backend.md create mode 100644 packages/subagent/subagent-acp/README.md create mode 100644 packages/subagent/subagent-acp/package.json create mode 100644 packages/subagent/subagent-acp/src/index.ts create mode 100644 packages/subagent/subagent-acp/src/run.ts create mode 100644 packages/subagent/subagent-acp/tests/mock-acp-server.ts create mode 100644 packages/subagent/subagent-acp/tests/subagent-acp.e2e.ts create mode 100644 packages/subagent/subagent-acp/tests/subagent-acp.spec.ts create mode 100644 packages/subagent/subagent-acp/tsconfig.json diff --git a/docs/architecture.md b/docs/architecture.md index 612e73ddce..216ed949b6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -116,7 +116,7 @@ Tool schemas are deliberately **part of the assembly**: "what the model is told - `whenIdle()` — resolves once the agent reaches quiescence after settling out of `running` (resolves immediately when already idle; awaits the loop exit when disposed). A non-owner's quiescence-observation hook: it lets a consumer await the current work settling **without** disposing the agent. It is NOT teardown — it does not stop queued work, unregister the agent, or detach the session; a lifecycle owner tears an agent down with `await AgentHandle.dispose()` (which stops the loop, awaits its exit, and unregisters). - `session`, `status`, `options` -**Subagents**: `spawn`/`fork` are realized by the [`@deepseek-ai/dsh-subagent`](../packages/subagent/subagent) seam (a named-provider registry on `ctx.subagents`), not a method on `Agent`. The in-process backends create the child via `ctx.agents.create` — fork seeds the child Session with a balanced completed-turn prefix of the parent's log (`CreateAgentOptions.seed`), spawn starts fresh; children are ordinary `Agent` handles so `steer()` and event subscription work uniformly. Out-of-process transports (ACP, and later A2A / Codex app-server / Claude Code SDK) register as sibling providers. See [docs/core-data-structures/subagent.md](core-data-structures/subagent.md) and [the subagent RFC](rfc/proposed/feature/2026-06-21-subagent-capability-seam.md). Inter-agent channels beyond delegation remain deferred. +**Subagents**: `spawn`/`fork` are realized by the [`@deepseek-ai/dsh-subagent`](../packages/subagent/subagent) seam (a named-provider registry on `ctx.subagents`), not a method on `Agent`. The in-process backends create the child via `ctx.agents.create` — fork seeds the child Session with a balanced completed-turn prefix of the parent's log (`CreateAgentOptions.seed`), spawn starts fresh; children are ordinary `Agent` handles so `steer()` and event subscription work uniformly. Out-of-process transports (ACP, and later A2A / Codex app-server / Claude Code SDK) register as sibling providers. See [docs/core-data-structures/subagent.md](core-data-structures/subagent.md) and [the subagent RFC](rfc/implemented/feature/2026-06-21-subagent-capability-seam.md). Inter-agent channels beyond delegation remain deferred. ### Loop lifecycle (session / turn / step) diff --git a/docs/core-data-structures/subagent.md b/docs/core-data-structures/subagent.md index ec58de91bf..c64d370ff2 100644 --- a/docs/core-data-structures/subagent.md +++ b/docs/core-data-structures/subagent.md @@ -2,7 +2,7 @@ The subagent seam — an agent delegating work to a child agent. Like [bash](bash.md) it is **one optional capability**, not part of the agent-loop spine, so its vocabulary lives here rather than in [core.md](core.md). But it differs from every other seam on one axis: **multiple provider implementations coexist** in one context, registered by name (`ctx.subagents`), where bash allows only one executor. The registry shape mirrors the [LLM adapter registry](llm-streaming.md), not the single-service bash executor. -Interface: [dsh-subagent](../../packages/subagent/subagent) (`ctx.subagents` + the vocabulary below). Implementations are sibling packages (`dsh-subagent-spawn`, `-fork`, `-acp`); the model-facing consumer is [dsh-tool-subagent](../../packages/subagent/tool-subagent). The proposal and rationale: [the subagent RFC](../rfc/proposed/feature/2026-06-21-subagent-capability-seam.md). +Interface: [dsh-subagent](../../packages/subagent/subagent) (`ctx.subagents` + the vocabulary below). Implementations are sibling packages (`dsh-subagent-spawn`, `-fork`, `-acp`); the model-facing consumer is [dsh-tool-subagent](../../packages/subagent/tool-subagent). The proposal and rationale: [the subagent RFC](../rfc/implemented/feature/2026-06-21-subagent-capability-seam.md). Source: [`packages/subagent/subagent/src/types.ts`](../../packages/subagent/subagent/src/types.ts) diff --git a/docs/module-graph.md b/docs/module-graph.md index 0724130ced..95308a5207 100644 --- a/docs/module-graph.md +++ b/docs/module-graph.md @@ -60,6 +60,9 @@ graph TD agent-core --> system-prompt agent-core --> tool-bash agent-core --> tools + subagent-acp --> agent + subagent-acp --> llm + subagent-acp --> subagent subagent-mock --> agent subagent-mock --> llm subagent-mock --> subagent @@ -108,6 +111,7 @@ graph TD | `subagent` | `agent`, `llm`, `tools` | | `tool-bash` | `agent`, `bash`, `llm`, `tools` | | `agent-core` | `agent`, `agent-loop`, `invariants`, `llm`, `session`, `system-prompt`, `tool-bash`, `tools` | +| `subagent-acp` | `agent`, `llm`, `subagent` | | `subagent-mock` | `agent`, `llm`, `subagent` | | `subagent-spawn` | `agent`, `llm`, `session`, `subagent` | | `tool-subagent` | `agent`, `llm`, `subagent`, `tools` | diff --git a/docs/rfc/README.md b/docs/rfc/README.md index 87b6456ab8..3ec7b7e5dc 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -44,7 +44,6 @@ Do NOT write one for a mechanical or local choice (a variable name, a one-file r | [Agent Client Protocol (ACP) support for external editors](proposed/feature/2026-06-14-acp-agent-client-protocol.md) | 2026-06-14 | | [Multiplex concurrent ACP sessions over one connection](proposed/feature/2026-06-14-acp-multi-session.md) | 2026-06-14 | | [Optional Code Mode — model writes TypeScript against an SDK of all tools](proposed/feature/2026-06-15-optional-code-mode.md) | 2026-06-15 | -| [Subagent capability seam](proposed/feature/2026-06-21-subagent-capability-seam.md) | 2026-06-21 | ### Simplification @@ -83,6 +82,8 @@ Do NOT write one for a mechanical or local choice (a variable name, a one-file r | Title | First proposed | |---|---| | [Rich ACP bash rendering — the terminal card (`_meta`) and command classification](implemented/feature/2026-06-18-acp-terminal-and-tool-rendering.md) | 2026-06-18 | +| [Subagent capability seam](implemented/feature/2026-06-21-subagent-capability-seam.md) | 2026-06-21 | +| [ACP subagent backend (out-of-process delegation)](implemented/feature/2026-06-22-acp-subagent-backend.md) | 2026-06-22 | ### Simplification diff --git a/docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md b/docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md similarity index 94% rename from docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md rename to docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md index 71a277ff5b..f733daca7b 100644 --- a/docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md +++ b/docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md @@ -1,8 +1,8 @@ # RFC: Subagent capability seam -Status: proposed +Status: implemented -> **Implementation status:** PR1 (this proposal + the `dsh-subagent` interface, the `dsh-subagent-mock` test backend, and the `dsh-tool-subagent` consumer) is the first of three PRs. The two in-process backends (`dsh-subagent-spawn`, `dsh-subagent-fork`) and the out-of-process `dsh-subagent-acp` backend land in PR2 and PR3. Status stays `proposed` until all three ship; the file moves to `implemented/feature/` then, amended to describe what actually landed. +> **Implementation status:** shipped across four PRs. PR1 landed this proposal + the `dsh-subagent` interface, the `dsh-subagent-mock` test backend, and the `dsh-tool-subagent` consumer; PR2 the two in-process backends (`dsh-subagent-spawn`, `dsh-subagent-fork`); PR2.5 the nested-agent snapshot infrastructure (see [Per-session snapshot replay for nested agents](../testing/2026-06-22-subagent-snapshot-replay.md)); PR3 the out-of-process `dsh-subagent-acp` backend (see [ACP subagent backend](2026-06-22-acp-subagent-backend.md)). The design below is amended to describe what actually landed. ## Problem diff --git a/docs/rfc/implemented/feature/2026-06-22-acp-subagent-backend.md b/docs/rfc/implemented/feature/2026-06-22-acp-subagent-backend.md new file mode 100644 index 0000000000..6be0f6746b --- /dev/null +++ b/docs/rfc/implemented/feature/2026-06-22-acp-subagent-backend.md @@ -0,0 +1,47 @@ +# RFC: ACP subagent backend (out-of-process delegation) + +Status: implemented + +## Problem + +The subagent seam ([the seam RFC](2026-06-21-subagent-capability-seam.md)) was built so multiple backends coexist by name on `ctx.subagents`. The in-process backends (`-spawn`/`-fork`) run a child as a second `Agent` on the SAME cordis context — cheap, but the child shares the parent's process, model client, and tools. The seam's whole point was to also support an OUT-OF-PROCESS child reached over a protocol, proving the abstraction generalizes across a process boundary. This RFC adds the first such backend: an Agent Client Protocol (ACP) client. + +## Decision + +`@deepseek-ai/dsh-subagent-acp` registers a `SubagentProvider` that runs each child agent in a SPAWNED SUBPROCESS, driven over ACP as the *client*. It is the direction-inverted twin of the existing server-side bridge `@deepseek-ai/dsh-acp` (the ACP *agent*): the bridge ANSWERS `initialize`/`newSession`/`prompt`; this backend CALLS them and IMPLEMENTS the `Client` callbacks (`sessionUpdate`, `requestPermission`). Pointing the configured spawn command at the `acp-agent` example makes the harness talk to its own process. + +### Fresh process per run + +Each `start` spawns a new child, runs exactly one ACP session (`initialize` → `newSession` → `prompt`), and `dispose` kills the subprocess and awaits its exit. This is the simplest lifecycle and mirrors the in-process one-child-per-run shape. Persistent-process pooling (reuse a warm child across runs) is a performance optimization deferred to future work — it adds session-lifecycle and crash-recovery complexity the first cut does not need. + +### Minimal client stub + +The client advertises NO optional capabilities (no `fs`, no `terminal`): the child self-serves file/terminal access in its own process. `session/update` notifications are consumed — the backend accumulates `agent_message_chunk` text as the result output and ignores the rest (thoughts, tool-call cards) in this cut, which surfaces only the child's final answer. `session/request_permission` is auto-answered by a configured policy (`reject` declines every prompt, `allow` approves via the first allow-shaped option) — the first cut surfaces no prompt to a human. Proxying `fs`/`terminal` back to the parent (a shared-workspace mode) remains future work, as the seam RFC noted. + +### No start-time capabilities + +The provider's `capabilities` are all `false`. An out-of-process child cannot honor the parent's `maxDepth` (it has no access to `parent.options.subagentDepth`) or `toolFilter` (it owns its own tool registry), and the first cut does not implement `outputSchema`. The service rejects a request needing any of them before `start` runs. The backend injects only `subagents` (not `ctx.agents`) and ignores `request.parent`. + +### StopReason mapping + +ACP `StopReason` → harness `SubagentStopReason`: `end_turn`→`completed`, `max_tokens`→`max-tokens`, `refusal`→`refusal`, `cancelled`→`aborted`, `max_turn_requests`→`error` (no clean equivalent — the task did not finish), unknown→`error`. A spawn/transport/RPC failure resolves `error` (or `aborted` if a cancel was requested); `result` never rejects on a child-level failure, per the seam contract. + +### SDK version: stayed on 0.25.1 + +The plan proposed bumping `@agentclientprotocol/sdk` 0.25.1 → 0.28.x for the new fluent `acp.client()` / `ActiveSession.nextUpdate()` API. Validating that against the code (the AGENTS.md "RFC is a proposal, not golden truth" discipline) reversed the decision: the backend only needs `ClientSideConnection` + `ndJsonStream` + `PROTOCOL_VERSION` + the `Client`/`Agent`/`StopReason` types, **all present and non-deprecated in 0.25.1**. The fluent API and `unstable_forkSession` that motivated the bump are never used here, so the "cleaner client code" benefit did not materialize. Worse, 0.28.x **deprecates both** `ClientSideConnection` AND `AgentSideConnection` (it wants all callers on the fluent builders), which turns the `no-deprecated` lint red across the entire existing ACP layer — 33 usages including the server bridge this PR has no business rewriting. That cross-cutting connection-API migration is its own PR, not baggage for "add an ACP subagent backend". So the bump was reverted and the backend is written against 0.25.1 (the plan's own fallback clause: "if the bump proves disruptive, fall back to `ClientSideConnection` (0.25.1), which is sufficient"). Migrating the whole ACP layer to the fluent API on a later 0.28.x bump is a worthwhile standalone follow-up. + +### Security: scrubbed child environment + +The child is a separate process, so it inherits an environment. Credential-shaped ambient vars (`/KEY|SECRET|TOKEN/i`) are NOT forwarded by default — the parent harness's own secrets must not leak into a spawned process implicitly (the same policy the bash executor applies). The child's OWN credentials (it needs a model key) are supplied EXPLICITLY via `config.env`, layered AFTER the scrub, so an intended `DEEPSEEK_API_KEY` survives while an incidental `AWS_SECRET_ACCESS_KEY` does not. Child stderr is inherited to the parent's stderr (diagnostics surface naturally); a spawn-level `error` event (e.g. ENOENT for a bad command) is captured and raced against the ACP drive, so a bad command settles `error` instead of crashing the parent with an unhandled error. + +## Testing + +Designed at every tier the backend touches, per the AGENTS.md "design test infrastructure up front" rule: + +- **Keyless unit/integration** (`subagent-acp.spec.ts`): spawns a scripted mock ACP server subprocess (`tests/mock-acp-server.ts`) and drives it through the real backend over real ACP stdio. Covers: the prompt round-trip + output accumulation; every StopReason mapping; cancellation via `run.cancel()` and via the request signal; the already-aborted-before-start case; the cancel-races-ahead-of-newSession case; a torn-pipe-after-cancel (child crashes on cancel) settling `aborted`; permission auto-answer under both policies (including the allow-policy-no-allow-option fallback); a non-message update consumed but not accumulated; a nonexistent-command spawn failure settling `error`; HMR provider cleanup; and the namespace export shape. 100% per-file coverage. +- **With-key e2e** (`subagent-acp.e2e.ts`): the harness drives ITSELF — the backend spawns the real `acp-agent` example process and a real model in that child answers a prompt (PONG) and does real file work (writes `proof.txt`, verified on disk). Self-skips without `DEEPSEEK_API_KEY`. This is the "talk to our own process" smoke and the out-of-process analogue of the in-process spawn e2e. +- **Snapshot**: deferred as `TODO(acp-subagent-replay)`. An ACP child is a distinct replay shape — each child is its own PROCESS with its own single-agent replay (booted under `DSH_SNAPSHOT=replay` with its own sessions-root + fixture), unlike the in-process per-session keying that [PR2.5](../testing/2026-06-22-subagent-snapshot-replay.md) added. The keyless mock-server tests give deterministic coverage of the backend in the meantime; the snapshot follow-up would record the parent driving a real-but-replayed ACP child. + +## Future providers + +The same out-of-process spawn/prompt/stream/cancel shape generalizes to other transports named in the seam RFC — A2A, the Codex app-server, and the Claude Code Agent SDK — each a sibling provider registered by name. The ACP backend is the proof that the seam supports the boundary; those are mechanically similar. diff --git a/docs/rfc/implemented/testing/2026-06-22-subagent-snapshot-replay.md b/docs/rfc/implemented/testing/2026-06-22-subagent-snapshot-replay.md index 422c8d0301..acb29e401d 100644 --- a/docs/rfc/implemented/testing/2026-06-22-subagent-snapshot-replay.md +++ b/docs/rfc/implemented/testing/2026-06-22-subagent-snapshot-replay.md @@ -11,7 +11,7 @@ It was built for ONE session per process, and that assumption is wired into two - **`dsh-llm-replay` keyed nothing.** It served the Nth `llm/stream` call the Nth recorded entry from a single global cursor. With a parent agent AND an in-process subagent both streaming on one context, the calls interleave and the single cursor hands the child the parent's script (and vice versa). - **The harness harvested one log.** `findSessionLog` walked the sessions root and returned the FIRST `.jsonl` it found. A subagent runs as a second `Session` with its own log in the same cwd bucket, so the child's transcript was silently dropped. -This was the `TODO(subagent-snapshots)` deferral recorded in the [subagent seam RFC](../../proposed/feature/2026-06-21-subagent-capability-seam.md): the in-process backends (PR2) shipped with unit + e2e coverage, but the full-transcript snapshot tier could not express a nested-agent shape until this infrastructure landed. This RFC is that stacked follow-up. +This was the `TODO(subagent-snapshots)` deferral recorded in the [subagent seam RFC](../../implemented/feature/2026-06-21-subagent-capability-seam.md): the in-process backends (PR2) shipped with unit + e2e coverage, but the full-transcript snapshot tier could not express a nested-agent shape until this infrastructure landed. This RFC is that stacked follow-up. ## Decision diff --git a/docs/rfc/proposed/simplification/2026-06-20-unify-agent-and-session-id.md b/docs/rfc/proposed/simplification/2026-06-20-unify-agent-and-session-id.md index 2455981ba8..cc0db091dc 100644 --- a/docs/rfc/proposed/simplification/2026-06-20-unify-agent-and-session-id.md +++ b/docs/rfc/proposed/simplification/2026-06-20-unify-agent-and-session-id.md @@ -46,7 +46,7 @@ The genuine risks of collapsing the two ids into one (the case AGAINST this prop - **It forecloses a one-agent-resumes-many-sessions / one-session-driven-by-many-agents future.** Today the separate ids leave room for an agent (a stable actor) to detach from one session and attach to another, or for a handoff where a new agent process adopts an existing session under a new actor handle. Unifying makes "agent" and "session" the same lifetime, so any such future needs a NEW seam (e.g. an explicit `actorId` distinct from the session) — re-introducing the very separation we removed. We judge this generality currently unused, but it is a door this change closes. -- **Subagents / fork / spawn may WANT a stable actor id across forked sessions.** The [subagent seam](../feature/2026-06-21-subagent-capability-seam.md) runs a child agent seeded from a parent's event log (fork). If a future design wants "the same agent identity across a fork" (parent and child share an actor but have distinct session logs), a unified id blocks it. The implementing PR must check the intended fork/spawn model BEFORE unifying, or accept that fork always mints a fresh combined id. (As shipped, each subagent child mints its own distinct agent id — `parentSession` records lineage — so the seam does not currently rely on a shared actor id, but unifying would foreclose adding one.) +- **Subagents / fork / spawn may WANT a stable actor id across forked sessions.** The [subagent seam](../../implemented/feature/2026-06-21-subagent-capability-seam.md) runs a child agent seeded from a parent's event log (fork). If a future design wants "the same agent identity across a fork" (parent and child share an actor but have distinct session logs), a unified id blocks it. The implementing PR must check the intended fork/spawn model BEFORE unifying, or accept that fork always mints a fresh combined id. (As shipped, each subagent child mints its own distinct agent id — `parentSession` records lineage — so the seam does not currently rely on a shared actor id, but unifying would foreclose adding one.) - **The config-driven resume-or-create policy becomes load-bearing, not cosmetic.** Today the per-run-uuid session id quietly sidesteps the "a fixed id collides with its own on-disk log on the second run" problem. Once the id is unified and stable, a config agent restarting MUST decide resume-vs-fresh deliberately — there is no longer a throwaway session id to hide behind. Getting this wrong reintroduces the create-collision the uuid was avoiding (a durable backend refuses to re-create an id whose log exists). This is the one real design decision the implementing PR owns, and it is easy to get subtly wrong. diff --git a/knip.json b/knip.json index aaf0e105d3..67d99a861d 100644 --- a/knip.json +++ b/knip.json @@ -40,6 +40,10 @@ "packages/subagent/subagent-spawn": { "entry": ["tests/**/*.spec.ts", "tests/**/*.e2e.ts"], "project": ["src/**/*.ts", "tests/**/*.ts"] + }, + "packages/subagent/subagent-acp": { + "entry": ["tests/**/*.spec.ts", "tests/**/*.e2e.ts", "tests/mock-acp-server.ts"], + "project": ["src/**/*.ts", "tests/**/*.ts"] } } } diff --git a/packages/README.md b/packages/README.md index 7cf32781bf..e5e66c63a7 100644 --- a/packages/README.md +++ b/packages/README.md @@ -42,6 +42,7 @@ dsh-subagent ← dsh-agent, dsh-llm, dsh-tools (abstract subagent provid dsh-subagent-mock ← dsh-subagent (scripted provider for tests) dsh-subagent-spawn ← dsh-subagent, dsh-agent, dsh-session, dsh-llm (in-process fresh child + shared run driver) dsh-subagent-fork ← dsh-subagent-spawn, dsh-agent, dsh-session (in-process child seeded from parent log) +dsh-subagent-acp ← dsh-subagent, dsh-agent, dsh-llm, @agentclientprotocol/sdk (out-of-process child over ACP) dsh-tool-subagent ← dsh-subagent, dsh-tools, dsh-agent (model-facing delegation tool) dsh-agent-core ← timer, dsh-llm, dsh-session, dsh-system-prompt, dsh-tools, dsh-agent, dsh-invariants, dsh-tool-bash, dsh-agent-loop (the providerless spine, as one bundle plugin) dsh-stdio-agent ← dsh-agent-core, dsh-ui-stdio, dsh-session-persistence-jsonl, dsh-agent, dsh-session (stdio chat APP + bin) @@ -78,6 +79,7 @@ The rule: **extension** plugins depend on interfaces, never on the concrete loop | `subagent/` | `subagent` | Abstract subagent seam: named-provider registry for delegating to child agents | `ctx.subagents` | | `subagent-spawn/` | `subagent` | In-process backend: a fresh child agent (+ the shared in-process run driver) | (registers on `ctx.subagents`) | | `subagent-fork/` | `subagent` | In-process backend: a child agent seeded with the parent's completed-turn prefix | (registers on `ctx.subagents`) | +| `subagent-acp/` | `subagent` | Out-of-process backend: a child agent in a spawned subprocess, driven over the Agent Client Protocol | (registers on `ctx.subagents`) | | `subagent-mock/` | `support` | Scripted `SubagentProvider` for testing the seam through the real load path | (registers on `ctx.subagents`) | | `tool-subagent/` | `subagent` | Model-facing `subagent` delegation tool over `ctx.subagents` | (registers on `ctx.tools`) | | `brand/` | `util` | Type-only `Branded` nominal-typing primitive (no runtime code, no harness deps) | (none — type-only) | diff --git a/packages/subagent/README.md b/packages/subagent/README.md index 582172dfd1..1cffff1cb1 100644 --- a/packages/subagent/README.md +++ b/packages/subagent/README.md @@ -7,8 +7,9 @@ The subagent seam: an agent delegating work to a child agent. Like the [bash](.. | `subagent/` | Abstract subagent seam: named-provider registry + vocabulary | `ctx.subagents` | | `subagent-spawn/` | In-process backend: a fresh child agent (+ the shared run driver) | (registers on `ctx.subagents`) | | `subagent-fork/` | In-process backend: a child seeded with the parent's completed-turn prefix | (registers on `ctx.subagents`) | +| `subagent-acp/` | Out-of-process backend: a child agent in a spawned subprocess, driven over ACP | (registers on `ctx.subagents`) | | `tool-subagent/` | Model-facing `subagent` delegation tool over `ctx.subagents` | (registers on `ctx.tools`) | -The interface lives at `subagent/subagent/`. The in-process `subagent-spawn` / `subagent-fork` backends ship here; the out-of-process `dsh-subagent-acp` and the test-only `dsh-subagent-mock` (in [support](../support/README.md)) are separate. All **product** packages except the mock. +The interface lives at `subagent/subagent/`. The in-process `subagent-spawn` / `subagent-fork` and the out-of-process `subagent-acp` backends ship here; the test-only `dsh-subagent-mock` (in [support](../support/README.md)) is separate. All **product** packages except the mock. -The proposal and design rationale: [docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md](../../docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md). +The proposal and design rationale: [docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md](../../docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md). diff --git a/packages/subagent/subagent-acp/README.md b/packages/subagent/subagent-acp/README.md new file mode 100644 index 0000000000..03990a7369 --- /dev/null +++ b/packages/subagent/subagent-acp/README.md @@ -0,0 +1,69 @@ +# @deepseek-ai/dsh-subagent-acp + +The out-of-process **ACP subagent backend**: runs each child agent in a spawned subprocess, driven over the [Agent Client Protocol](https://github.com/zed-industries/agent-client-protocol) (ACP) as the *client*. Registers a `SubagentProvider` on `ctx.subagents` (the [subagent seam](../subagent)), alongside the in-process [`-spawn`](../subagent-spawn)/[`-fork`](../subagent-fork) backends — multiple backends coexist by name. + +It is the direction-inverted twin of the server-side bridge in [`@deepseek-ai/dsh-acp`](../../ui/acp): that package is the ACP *agent* (it answers `initialize`/`newSession`/`prompt`); this one is the ACP *client* (it *calls* them and implements the `sessionUpdate`/`requestPermission` callbacks). Point the configured command at the `acp-agent` example to "talk to our own process". + +## What it does + +`start(request)` spawns the configured command, wraps its stdio in an ACP `ClientSideConnection`, and drives one session: `initialize` → `newSession` → `prompt`. The child's streamed `agent_message_chunk` text becomes the `SubagentResult.output`; the prompt's terminal `StopReason` maps to the stop reason. `dispose()` kills the subprocess and awaits its exit. + +**Fresh process per run.** Each `start` spawns a new child, runs exactly one ACP session, and disposes it. Persistent-process pooling is a future optimization (see the RFC). + +Unlike the in-process backends, the child does NOT share this cordis context — it is a separate process with its own session, model client, and tools. So this backend: +- injects only `subagents` (no `ctx.agents`); +- advertises NO start-time capabilities (an out-of-process child can't enforce the parent's depth/tool-filter); +- ignores `request.parent`. + +## Config + +| Key | Type | Default | Notes | +|---|---|---|---| +| `providerName` | string | `acp` | Registry name on `ctx.subagents`. | +| `command` | string | — (required) | The executable to spawn for each run (the child ACP agent). | +| `args` | string[] | `[]` | Arguments passed to `command`. | +| `cwd` | string | parent cwd | Working directory for the child process and its ACP session. | +| `permission` | `'allow' \| 'reject'` | `reject` | How to auto-answer the child's `session/request_permission` prompts. `reject` declines every prompt (answer `cancelled`); `allow` approves via the first allow-shaped option. The first cut surfaces no prompt to a human. | +| `env` | Record | `{}` | Extra env vars for the child (e.g. its own `DEEPSEEK_API_KEY`). Forwarded on top of a credential-scrubbed copy of the parent env, so an explicit key reaches the child while ambient secrets do not leak implicitly. | + +```yaml +- id: subagent-acp + name: '@deepseek-ai/dsh-subagent-acp' + config: + providerName: acp + command: node + args: ['--import', 'tsx', './packages/ui/acp-agent/src/bin.ts', './examples/acp-agent/cordis.yml'] + permission: reject + env: + DEEPSEEK_API_KEY: !!js process.env.DEEPSEEK_API_KEY +``` + +## StopReason mapping + +ACP `StopReason` → harness `SubagentStopReason`: + +| ACP | harness | +|---|---| +| `end_turn` | `completed` | +| `max_tokens` | `max-tokens` | +| `refusal` | `refusal` | +| `cancelled` | `aborted` | +| `max_turn_requests` | `error` (no clean equivalent; the task did not finish) | +| _(unknown)_ | `error` | + +A spawn/transport/RPC failure resolves `error` (or `aborted` if a cancel was requested) — `result` never rejects on a child-level failure, per the seam contract. + +## Environment scrub + +Credential-shaped ambient vars (`/KEY|SECRET|TOKEN/i`) are NOT forwarded to the child by default — the parent harness's own secrets must not leak into a spawned process implicitly. The child's OWN credentials are supplied explicitly via `config.env`, layered AFTER the scrub, so an intended `DEEPSEEK_API_KEY` survives while an incidental `AWS_SECRET_ACCESS_KEY` does not. + +## Testing + +- **Keyless** (`subagent-acp.spec.ts`): spawns a scripted mock ACP server subprocess (`tests/mock-acp-server.ts`) and drives it through the real backend over real ACP stdio — connection setup, client callbacks, the prompt round-trip, stop-reason mapping, cancellation (including the early-cancel race and a torn-pipe-after-cancel), permission auto-answer, and quiescent disposal. No model, no key. +- **With-key e2e** (`subagent-acp.e2e.ts`): the harness drives ITSELF — the backend spawns the real `acp-agent` example process and a real model in that child answers a prompt and does real file work (verified on disk). Self-skips without `DEEPSEEK_API_KEY`. + +`TODO(acp-subagent-replay)`: snapshot-tier coverage of an ACP child is a separate replay shape (each child is its own PROCESS with its own single-agent replay, distinct from the in-process per-session keying), deferred — see the RFC. + +## Plugin export shape + +Named `name` / `inject` / `Config` / `apply`, with **no default export**: the cordis Loader's `unwrapExports` does `exports.default ?? exports`, so a stray default would collapse the module to the bare function and drop the `inject` namespace (see [docs/postmortem/0001](../../../docs/postmortem/0001-acp-default-export-drops-inject.md)). diff --git a/packages/subagent/subagent-acp/package.json b/packages/subagent/subagent-acp/package.json new file mode 100644 index 0000000000..ee9c55a169 --- /dev/null +++ b/packages/subagent/subagent-acp/package.json @@ -0,0 +1,39 @@ +{ + "name": "@deepseek-ai/dsh-subagent-acp", + "description": "Out-of-process ACP subagent backend: drives a child agent in a spawned subprocess over the Agent Client Protocol", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/index.d.ts", + "exports": { + ".": { + "types": "./lib/index.d.ts", + "default": "./lib/index.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib", + "src" + ], + "license": "BSD-3-Clause", + "peerDependencies": { + "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-llm": "^0.0.1", + "@deepseek-ai/dsh-subagent": "^0.0.1", + "cordis": "^4.0.0-rc.6" + }, + "dependencies": { + "@agentclientprotocol/sdk": "0.25.1", + "schemastery": "^3.18.0" + }, + "devDependencies": { + "@deepseek-ai/dsh-agent": "workspace:^", + "@deepseek-ai/dsh-llm": "workspace:^", + "@deepseek-ai/dsh-subagent": "workspace:^", + "@cordisjs/plugin-loader": "^1.0.0-rc.4", + "cordis": "^4.0.0-rc.6" + } +} diff --git a/packages/subagent/subagent-acp/src/index.ts b/packages/subagent/subagent-acp/src/index.ts new file mode 100644 index 0000000000..66f254b831 --- /dev/null +++ b/packages/subagent/subagent-acp/src/index.ts @@ -0,0 +1,90 @@ +/** + * The out-of-process ACP subagent backend: registers a {@link SubagentProvider} + * on `ctx.subagents` that runs each child agent in a SPAWNED SUBPROCESS, driven + * over the Agent Client Protocol (ACP) as the client. The parent process is the + * ACP client; the child is any ACP agent (point the configured command at the + * `acp-agent` example to "talk to our own process"). + * + * Unlike the in-process backends (`-spawn`/`-fork`), the child does NOT share + * this cordis context — it is a separate process with its own session, model + * client, and tools. So this backend injects only `subagents` (no `agents`), + * advertises NO start-time capabilities (an out-of-process child cannot enforce + * the parent's depth/tool-filter), and ignores `request.parent`. + * + * Plugin export shape: named `name`/`inject`/`Config`/`apply`, NO default + * export (the cordis Loader's `unwrapExports` does `exports.default ?? exports`, + * so a stray default would drop the namespace — see docs/postmortem/0001). + * + * @module @deepseek-ai/dsh-subagent-acp + */ + +import type { Context } from 'cordis' +import z from 'schemastery' +import type { SubagentCapabilities, SubagentProvider, SubagentStartRequest } from '@deepseek-ai/dsh-subagent' +import { type AcpRunSpec, type PermissionPolicy, startAcpRun } from './run.ts' + +export const name = 'subagent-acp' +export const inject = ['subagents'] + +/** Config: how to spawn and drive the child ACP agent process. */ +export interface Config { + /** Provider name on `ctx.subagents` (default `acp`). */ + providerName: string + /** The executable to spawn for each run (the child ACP agent). */ + command: string + /** Arguments passed to {@link command}. */ + args: string[] + /** + * Working directory for the child process and its ACP session. Defaults to + * the parent process's cwd when omitted. + */ + cwd?: string + /** + * How to auto-answer the child's `session/request_permission` prompts: + * `reject` (default — decline every prompt) or `allow` (approve via the first + * allow-shaped option). The first cut surfaces no prompt to a human. + */ + permission: PermissionPolicy + /** + * Extra environment variables for the child process — e.g. the child + * harness's own `DEEPSEEK_API_KEY`. Forwarded on top of a credential-scrubbed + * copy of the parent env, so an explicit key here reaches the child while + * ambient secrets do not leak implicitly. + */ + env: Record +} + +export const Config: z = z.object({ + providerName: z.string().default('acp'), + command: z.string().required(), + args: z.array(z.string()).default([]), + cwd: z.string(), + permission: z.union(['allow', 'reject'] as const).default('reject'), + env: z.dict(z.string()).default({}), +}) + +/** + * The ACP provider. Advertises NO start-time capabilities: an out-of-process + * child cannot honor `outputSchema`/`maxDepth`/`toolFilter` (the service rejects + * a request needing any of them before `start` runs). + */ +class AcpProvider implements SubagentProvider { + readonly capabilities: SubagentCapabilities = { outputSchema: false, depthLimit: false, toolFilter: false } + + constructor(readonly name: string, private readonly config: Config) {} + + start(request: SubagentStartRequest) { + const spec: AcpRunSpec = { + command: this.config.command, + args: this.config.args, + cwd: this.config.cwd ?? process.cwd(), + permission: this.config.permission, + env: this.config.env, + } + return startAcpRun(request, spec) + } +} + +export function apply(ctx: Context, config: Config): void { + ctx.subagents.registerProvider(new AcpProvider(config.providerName, config)) +} diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts new file mode 100644 index 0000000000..18f20c53c1 --- /dev/null +++ b/packages/subagent/subagent-acp/src/run.ts @@ -0,0 +1,292 @@ +/** + * The out-of-process ACP subagent run driver. Spawns a child agent as a + * subprocess, speaks the Agent Client Protocol (ACP) to it over stdio as the + * CLIENT, drives one session to completion, and shapes the result into a + * {@link SubagentResult}. The mirror image of the server-side bridge in + * `@deepseek-ai/dsh-acp` (which is the ACP *agent* side): here we are the ACP + * *client*, so we CALL `initialize`/`newSession`/`prompt`/`cancel` and we + * IMPLEMENT the `Client` callbacks (`sessionUpdate`, `requestPermission`). + * + * One subprocess per run (fresh-process-per-run): `start` spawns, runs exactly + * one ACP session, and `dispose` kills the subprocess and awaits its exit. + * Persistent-process pooling is a future optimization (see the RFC). + * + * TODO(acp-subagent-replay): snapshot-tier coverage of an ACP child is a + * distinct replay shape — each child is its own PROCESS with its own + * single-agent replay (the child boots under `DSH_SNAPSHOT=replay` with its own + * sessions-root + fixture), unlike the in-process per-session keying in + * `dsh-llm-replay`. Deferred to a follow-up; keyless coverage here is via a + * scripted mock ACP server subprocess, and the with-key e2e drives the real + * `acp-agent` example. See the ACP-subagent-backend RFC. + * + * @module @deepseek-ai/dsh-subagent-acp/run + */ + +import { spawn, type ChildProcess } from 'node:child_process' +import { randomUUID } from 'node:crypto' +import { Readable, Writable } from 'node:stream' +import { + ClientSideConnection, + ndJsonStream, + PROTOCOL_VERSION, + type Agent as AcpAgent, + type Client, + type ContentBlock as AcpContentBlock, + type RequestPermissionRequest, + type RequestPermissionResponse, + type SessionNotification, + type StopReason, +} from '@agentclientprotocol/sdk' +import { AgentId } from '@deepseek-ai/dsh-agent' +import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import type { SubagentResult, SubagentRun, SubagentStartRequest, SubagentStopReason } from '@deepseek-ai/dsh-subagent' + +/** + * How the client answers a child's `session/request_permission`. The first cut + * does not surface permission prompts to a human, so every request is + * auto-answered by this fixed policy: + * + * - `reject` — decline every prompt (answer `cancelled`). Safe default: a child + * that asks before a side effect does not get to take it. + * - `allow` — approve every prompt by selecting its first `allow_*` option (or, + * if none is offered, `cancelled`). Use when the child is trusted to act. + */ +export type PermissionPolicy = 'allow' | 'reject' + +/** Resolved spawn spec for an ACP child process (no defaults — see Config). */ +export interface AcpRunSpec { + /** The executable to spawn (the child ACP agent). */ + command: string + /** Arguments passed to {@link command}. */ + args: string[] + /** Working directory for the child process AND its ACP session `cwd`. */ + cwd: string + /** How to auto-answer the child's permission prompts. */ + permission: PermissionPolicy + /** + * Extra environment variables to ADD for the child (e.g. the child harness's + * `DEEPSEEK_API_KEY`). Merged on top of the scrubbed ambient env — see + * {@link buildChildEnv}. A value here is forwarded even if its name matches + * the credential-scrub pattern (an explicit opt-in for the child's own creds). + */ + env: Record +} + +/** + * Credential-shaped ambient env vars are NOT forwarded to the child by default + * (the parent harness's own `DEEPSEEK_API_KEY`/secrets must not leak into a + * spawned process implicitly). Same pattern as the bash executor. The child + * agent needs its OWN credentials to reach a model — those are supplied + * explicitly via {@link AcpRunSpec.env}, which is layered on top AFTER the + * scrub, so an intended `DEEPSEEK_API_KEY` survives while an incidental + * `AWS_SECRET_ACCESS_KEY` does not. + */ +export const SENSITIVE_ENV_PATTERN = /KEY|SECRET|TOKEN/i + +/** The ambient env minus credential-shaped vars, plus the spec's explicit env. */ +export function buildChildEnv(extra: Record): NodeJS.ProcessEnv { + const env: NodeJS.ProcessEnv = {} + for (const [key, value] of Object.entries(process.env)) { + if (!SENSITIVE_ENV_PATTERN.test(key)) env[key] = value + } + return { ...env, ...extra } +} + +/** Map an ACP {@link StopReason} to a harness {@link SubagentStopReason}. */ +export function acpStopReason(reason: StopReason): SubagentStopReason { + switch (reason) { + case 'end_turn': + return 'completed' + case 'max_tokens': + return 'max-tokens' + case 'refusal': + return 'refusal' + case 'cancelled': + return 'aborted' + // `max_turn_requests` (the child hit its turn-request budget) has no direct + // harness equivalent and means the task did NOT finish cleanly — surface it + // as a generic failure so the consumer maps it to an isError result rather + // than reporting a partial answer as success. + case 'max_turn_requests': + return 'error' + // ACP StopReason is a closed wire union, but a future SDK could add a + // variant; treat an unknown terminal reason as a failure (never silently + // 'completed'). + default: + return 'error' + } +} + +/** Collect the text of an ACP content block (non-text blocks contribute nothing). */ +export function acpContentText(content: AcpContentBlock): string { + return content.type === 'text' ? content.text : '' +} + +/** Translate the harness prompt blocks into ACP prompt blocks (text only). */ +export function toAcpPrompt(prompt: ContentBlock[]): AcpContentBlock[] { + const blocks: AcpContentBlock[] = [] + for (const block of prompt) { + if (block.type === 'text') blocks.push({ type: 'text', text: block.text }) + } + return blocks +} + +/** Resolve once the child process exits (any code/signal); immediate if gone. */ +function waitForExit(child: ChildProcess): Promise { + if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve() + return new Promise(resolve => child.once('exit', () => { resolve() })) +} + +/** + * Start an out-of-process ACP child for `request` and return a {@link SubagentRun}. + * + * Spawns the configured command, wraps its stdio in an ACP `ClientSideConnection`, + * and drives one session: `initialize` → `newSession` → `prompt`. The accumulated + * `agent_message_chunk` text is the result output; the prompt's terminal + * `StopReason` maps to the stop reason. `result` never REJECTS on a child-level + * failure (a spawn/transport/RPC error resolves with `stopReason: 'error'`), per + * the seam contract. `cancel()` sends `session/cancel`; `dispose()` kills the + * subprocess and awaits its exit (quiescent teardown). + */ +export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): SubagentRun { + const id = AgentId(randomUUID()) + + // Spawn the child ACP agent. stdin = ACP request channel, stdout = ACP + // response channel, stderr = INHERIT so the child's diagnostics surface on the + // parent's stderr (no separate capture to drain — we don't fold child stderr + // into the result; the seam reports only output + stop reason). + const child = spawn(spec.command, spec.args, { + cwd: spec.cwd, + env: buildChildEnv(spec.env), + stdio: ['pipe', 'pipe', 'inherit'], + }) + // A spawn-level failure (e.g. ENOENT for a bad command) is emitted as an + // `error` event, NOT a thrown exception — without a listener Node treats it as + // an unhandled error and crashes the parent. Capture it into a promise the + // result path races, so a bad command settles `error` like any child failure. + const spawnFailed = new Promise((resolve) => { + child.once('error', (err) => { resolve(err) }) + }) + + // Accumulate the child's streamed assistant text — the SubagentResult output. + const output: string[] = [] + // `cancelled` records that a cancel was requested (signal or cancel()), so a + // run torn down before the prompt resolves settles `aborted` rather than the + // generic error mapping. + let cancelled = false + + const makeClient = (_agent: AcpAgent): Client => ({ + sessionUpdate(params: SessionNotification): Promise { + const update = params.update + if (update.sessionUpdate === 'agent_message_chunk') { + output.push(acpContentText(update.content)) + } + // Other updates (thoughts, tool calls, plans) are consumed but not + // surfaced in this cut — the subagent returns only its final answer. + return Promise.resolve() + }, + requestPermission(params: RequestPermissionRequest): Promise { + // Auto-answer by the configured policy. `allow` selects the first + // allow-shaped option the child offered; if it offered none (or we + // reject), answer `cancelled` so the child does not proceed. + if (spec.permission === 'allow') { + const allow = params.options.find(o => o.kind === 'allow_once' || o.kind === 'allow_always') + if (allow !== undefined) { + return Promise.resolve({ outcome: { outcome: 'selected', optionId: allow.optionId } }) + } + } + return Promise.resolve({ outcome: { outcome: 'cancelled' } }) + }, + }) + + const conn = new ClientSideConnection( + makeClient, + ndJsonStream( + Writable.toWeb(child.stdin) as WritableStream, + Readable.toWeb(child.stdout) as ReadableStream, + ), + ) + + let sessionId: string | undefined + const requestCancel = (): void => { + cancelled = true + // Best-effort: tell the child to cancel the in-flight turn. Swallows a + // rejection — the session may not exist yet, or the pipe may be gone; the + // dispose path kills the process regardless. If the session has NOT been + // created yet (cancel raced ahead of `newSession`), the `cancelled` flag + // alone carries it: the result path re-checks the flag after each await and + // settles `aborted` without running the prompt. The `.catch` swallow is + // defensive for a narrow transport race (child gone mid-send) — v8-ignored + // because dispose kills the process regardless, so it can't be hit in tests. + /* v8 ignore next */ + if (sessionId !== undefined) void conn.cancel({ sessionId }).catch(() => { /* child gone / no session */ }) + } + const onAbort = (): void => { requestCancel() } + request.signal?.addEventListener('abort', onAbort, { once: true }) + + const result: Promise = (async (): Promise => { + // The accumulated child text as harness ContentBlocks (empty array when the + // child streamed nothing). Read at every return so a partial answer survives + // a later cancel/error. + const collectOutput = (): ContentBlock[] => { + const text = output.join('') + return text.length > 0 ? [{ type: 'text', text }] : [] + } + try { + // An already-aborted request never runs the child. + if (request.signal?.aborted) { + cancelled = true + return { output: [], stopReason: 'aborted' } + } + // Race the ACP drive against a spawn failure: a bad command never speaks + // ACP, so `initialize` would hang forever — the spawn `error` event is the + // only signal, and a rejected race settles the run `error` via the catch. + const driveAcp = async (): Promise => { + await conn.initialize({ + protocolVersion: PROTOCOL_VERSION, + // Advertise NO optional client capabilities (no fs, no terminal): the + // child self-serves in its own process. + clientCapabilities: {}, + }) + const session = await conn.newSession({ cwd: spec.cwd, mcpServers: [] }) + sessionId = session.sessionId + // A cancel that raced ahead of `newSession` set `cancelled` but could not + // send `session/cancel` (no session id yet). Honor it here: settle + // `aborted` without ever issuing the prompt, rather than running the child + // to completion and ignoring the cancel. + if (cancelled) return { output: collectOutput(), stopReason: 'aborted' } + const promptResult = await conn.prompt({ sessionId, prompt: toAcpPrompt(request.prompt) }) + return { output: collectOutput(), stopReason: acpStopReason(promptResult.stopReason) } + } + return await Promise.race([ + driveAcp(), + spawnFailed.then((err): SubagentResult => { throw err }), + ]) + } catch { + // The seam contract: result resolves (never rejects) on a child-level + // failure. A spawn/transport/RPC error becomes an error/aborted result — + // `aborted` if a cancel was requested (the failure is the cancellation + // surfacing as a torn pipe / rejected RPC), else a genuine `error`. + return { output: collectOutput(), stopReason: cancelled ? 'aborted' : 'error' } + } + })() + + return { + id, + result, + cancel(_reason?: string): void { + requestCancel() + }, + async dispose(): Promise { + request.signal?.removeEventListener('abort', onAbort) + // Kill the subprocess and AWAIT its exit (quiescent teardown — dispose + // must reach quiescence, not merely request it). SIGTERM first; the child + // is our own short-lived ACP agent, so a graceful term is enough. Guard + // the kill: the process may already be gone. + if (child.exitCode === null && child.signalCode === null) { + child.kill('SIGTERM') + } + await waitForExit(child) + }, + } +} diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts new file mode 100644 index 0000000000..e383bbe8eb --- /dev/null +++ b/packages/subagent/subagent-acp/tests/mock-acp-server.ts @@ -0,0 +1,150 @@ +/** + * A minimal mock ACP AGENT, run as a subprocess, for the keyless + * `dsh-subagent-acp` tests. It speaks the agent side of ACP over stdio and is + * fully scripted by environment variables — no model, no network: + * + * - `MOCK_TEXT` — the assistant text it streams as one `agent_message_chunk`. + * - `MOCK_STOP` — the ACP `StopReason` it returns from `prompt` + * (`end_turn` default, or `max_tokens`/`refusal`/…). + * - `MOCK_HANG` — if `1`, `prompt` never resolves on its own (it waits for + * a `session/cancel`), to exercise the client's cancel path. + * - `MOCK_PERMISSION` — if `1`, the agent calls `session/request_permission` + * before answering, to exercise the client's auto-answer. + * - `MOCK_READY_FILE` — if set, the path the agent touches once its `prompt` + * handler is in flight (it has streamed its chunk). A test + * polls for this file to cancel on a CONDITION rather than + * an arbitrary timeout (subprocess cold-start is variable). + * + * It is NOT a test spec (no `describe`/`it`) — it is spawned BY the specs as the + * child process the ACP backend drives. Kept as a `.ts` run under tsx by the + * spec (which passes its own tsconfig), mirroring how the snapshot harness boots + * the real example. + * + * @module @deepseek-ai/dsh-subagent-acp/tests/mock-acp-server + */ + +import { randomUUID } from 'node:crypto' +import { existsSync, writeFileSync } from 'node:fs' +import { Readable, Writable } from 'node:stream' +import { + AgentSideConnection, + ndJsonStream, + PROTOCOL_VERSION, + type Agent, + type CancelNotification, + type AuthenticateRequest, + type InitializeRequest, + type InitializeResponse, + type NewSessionRequest, + type NewSessionResponse, + type PromptRequest, + type PromptResponse, + type StopReason, +} from '@agentclientprotocol/sdk' + +const TEXT = process.env.MOCK_TEXT ?? 'mock child answer' +const STOP = (process.env.MOCK_STOP ?? 'end_turn') as StopReason +const HANG = process.env.MOCK_HANG === '1' +const WANT_PERMISSION = process.env.MOCK_PERMISSION === '1' +const NO_ALLOW = process.env.MOCK_NO_ALLOW === '1' +const THOUGHT = process.env.MOCK_THOUGHT === '1' +const CRASH_ON_CANCEL = process.env.MOCK_CRASH_ON_CANCEL === '1' +const READY_FILE = process.env.MOCK_READY_FILE +// When MOCK_NEWSESSION_READY/GO are set, newSession touches READY then blocks +// until GO appears — letting a test cancel mid-newSession deterministically. +const NEWSESSION_GATE = process.env.MOCK_NEWSESSION_READY !== undefined && process.env.MOCK_NEWSESSION_GO !== undefined + ? { ready: process.env.MOCK_NEWSESSION_READY, go: process.env.MOCK_NEWSESSION_GO } + : undefined + +function makeAgent(conn: AgentSideConnection): Agent { + // Pending cancel resolver for the HANG path: a `session/cancel` resolves the + // prompt with `cancelled`. + let resolveCancel: ((reason: StopReason) => void) | undefined + + return { + initialize(_params: InitializeRequest): Promise { + return Promise.resolve({ + protocolVersion: PROTOCOL_VERSION, + agentCapabilities: { loadSession: false, promptCapabilities: { image: false, audio: false, embeddedContext: false } }, + authMethods: [], + }) + }, + async newSession(_params: NewSessionRequest): Promise { + // Optionally signal "newSession reached" and block until released, so a + // test can cancel DURING newSession (the early-cancel race window) on a + // condition rather than a timeout. + if (NEWSESSION_GATE !== undefined) { + writeFileSync(NEWSESSION_GATE.ready, 'at-newSession') + while (!existsSync(NEWSESSION_GATE.go)) await new Promise(r => setTimeout(r, 10)) + } + return { sessionId: randomUUID() } + }, + authenticate(_params: AuthenticateRequest): Promise { + // No auth methods advertised; nothing to do. + return Promise.resolve() + }, + async prompt(params: PromptRequest): Promise { + if (WANT_PERMISSION) { + // Ask the client to approve before answering; honor its decision. Under + // MOCK_NO_ALLOW the only options are reject-shaped, so an `allow`-policy + // client finds no allow option and must fall back to cancelled. + const options = NO_ALLOW + ? [{ optionId: 'no', name: 'Reject', kind: 'reject_once' as const }] + : [ + { optionId: 'yes', name: 'Allow', kind: 'allow_once' as const }, + { optionId: 'no', name: 'Reject', kind: 'reject_once' as const }, + ] + const decision = await conn.requestPermission({ + sessionId: params.sessionId, + toolCall: { toolCallId: 'mock-call', title: 'mock side effect' }, + options, + }) + if (decision.outcome.outcome === 'cancelled') { + return { stopReason: 'cancelled' } + } + } + // Optionally emit a NON-message update first (a thought), so the client's + // sessionUpdate sees an update it must consume-but-not-accumulate. + if (THOUGHT) { + await conn.sessionUpdate({ + sessionId: params.sessionId, + update: { sessionUpdate: 'agent_thought_chunk', content: { type: 'text', text: 'thinking…' } }, + }) + } + // Stream the canned assistant text as one chunk. + await conn.sessionUpdate({ + sessionId: params.sessionId, + update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: TEXT } }, + }) + // Signal "prompt is in flight" by touching the readiness file, so a test + // can wait on a CONDITION (file exists) rather than an arbitrary timeout + // before cancelling — deterministic regardless of subprocess cold-start. + if (READY_FILE !== undefined) writeFileSync(READY_FILE, 'ready') + if (HANG) { + // Never resolve on our own: wait for session/cancel to settle us. + return new Promise((resolve) => { + resolveCancel = (reason) => { resolve({ stopReason: reason }) } + }) + } + return { stopReason: STOP } + }, + cancel(_params: CancelNotification): Promise { + if (CRASH_ON_CANCEL) { + // Exit hard instead of answering — tears the ACP pipe, so the client's + // pending prompt REJECTS (exercises the backend's catch-while-cancelled + // path: a transport failure after a cancel settles `aborted`). + process.exit(1) + } + resolveCancel?.('cancelled') + return Promise.resolve() + }, + } +} + +new AgentSideConnection( + makeAgent, + ndJsonStream( + Writable.toWeb(process.stdout) as WritableStream, + Readable.toWeb(process.stdin) as ReadableStream, + ), +) diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.e2e.ts b/packages/subagent/subagent-acp/tests/subagent-acp.e2e.ts new file mode 100644 index 0000000000..826ef198dd --- /dev/null +++ b/packages/subagent/subagent-acp/tests/subagent-acp.e2e.ts @@ -0,0 +1,110 @@ +import { mkdtemp, readFile, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fileURLToPath } from 'node:url' +import { afterEach, describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import type { Agent } from '@deepseek-ai/dsh-agent' +import SubagentService from '@deepseek-ai/dsh-subagent' +import * as acp from '../src/index.ts' + +/** + * With-key e2e for the ACP subagent backend: the harness drives ITSELF as an ACP + * server. The backend spawns the real `acp-agent` example as a child PROCESS, + * speaks ACP to it over stdio, and the child runs the REAL model in its own + * process to answer a prompt. We verify the child's real answer comes back + * through the seam — the "talk to our own process" smoke the design called for. + * Key-gated (self-skips without DEEPSEEK_API_KEY). + * + * This is the out-of-process analogue of the in-process spawn e2e: there a + * parent agent on the same context drove a child; here the child is a separate + * process reached over ACP, proving the seam generalizes across the boundary. + */ + +// The real acp-agent example: its bin + cordis.yml (the live DeepSeek config). +const binScript = fileURLToPath(new URL('../../../ui/acp-agent/src/bin.ts', import.meta.url)) +const exampleConfig = fileURLToPath(new URL('../../../../examples/acp-agent/cordis.yml', import.meta.url)) +const tsxLoader = fileURLToPath(import.meta.resolve('tsx')) +const repoTsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url)) + +/** The ACP backend ignores the parent, but the seam requires one. */ +const fakeParent = { id: 'parent', session: { header: {} } } as unknown as Agent + +let ctx: Context | undefined +let workdir: string | undefined + +afterEach(async () => { + await ctx?.fiber.dispose() + ctx = undefined + if (workdir !== undefined) await rm(workdir, { recursive: true, force: true }) + workdir = undefined +}) + +describe.skipIf(!process.env.DEEPSEEK_API_KEY)('ACP backend with-key e2e (drive our own acp-agent)', () => { + it('drives the real acp-agent example process to answer a prompt', async () => { + workdir = await mkdtemp(join(tmpdir(), 'dsh-subagent-acp-e2e-')) + ctx = new Context() + await ctx.plugin(SubagentService) + await ctx.plugin(acp, { + providerName: 'acp', + command: process.execPath, + args: ['--import', tsxLoader, binScript, exampleConfig], + cwd: workdir, + permission: 'reject', + // The child harness needs the key to reach the model; forward it + // explicitly (buildChildEnv scrubs ambient creds but keeps these extras). + env: { + ...process.env.DEEPSEEK_API_KEY !== undefined ? { DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY } : {}, + ...process.env.DEEPSEEK_BASE_URL !== undefined ? { DEEPSEEK_BASE_URL: process.env.DEEPSEEK_BASE_URL } : {}, + TSX_TSCONFIG_PATH: repoTsconfig, + }, + }) + + const run = ctx.subagents.start('acp', { + prompt: [{ type: 'text', text: 'Reply with exactly the word PONG and nothing else. Do not use any tools.' }], + parent: fakeParent, + }) + const result = await run.result + await run.dispose() + + // The real child process completed its turn and streamed a real answer back + // across the ACP boundary. + expect(result.stopReason).toBe('completed') + const text = result.output.filter(b => b.type === 'text').map(b => (b as { text: string }).text).join('') + expect(text.length).toBeGreaterThan(0) + expect(text.toUpperCase()).toContain('PONG') + }, 180_000) + + it('drives the child to do real file work via its own bash tool', async () => { + workdir = await mkdtemp(join(tmpdir(), 'dsh-subagent-acp-e2e-')) + ctx = new Context() + await ctx.plugin(SubagentService) + await ctx.plugin(acp, { + providerName: 'acp', + command: process.execPath, + args: ['--import', tsxLoader, binScript, exampleConfig], + cwd: workdir, + // The child needs to act (run bash), so approve its permission prompts. + permission: 'allow', + env: { + ...process.env.DEEPSEEK_API_KEY !== undefined ? { DEEPSEEK_API_KEY: process.env.DEEPSEEK_API_KEY } : {}, + ...process.env.DEEPSEEK_BASE_URL !== undefined ? { DEEPSEEK_BASE_URL: process.env.DEEPSEEK_BASE_URL } : {}, + TSX_TSCONFIG_PATH: repoTsconfig, + }, + }) + + const run = ctx.subagents.start('acp', { + prompt: [{ type: 'text', text: + 'Use the bash tool to write the text ACP_CHILD_WAS_HERE into a file named proof.txt ' + + 'in the current directory. Then reply DONE.' }], + parent: fakeParent, + }) + const result = await run.result + await run.dispose() + + expect(result.stopReason).toBe('completed') + // Verify the WORLD: the child process actually wrote the file in its cwd. + const proof = await readFile(join(workdir, 'proof.txt'), 'utf8') + expect(proof).toContain('ACP_CHILD_WAS_HERE') + }, 180_000) +}) diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts new file mode 100644 index 0000000000..ac74ace7e8 --- /dev/null +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -0,0 +1,315 @@ +import { describe, expect, it } from 'vitest' +import { Context } from 'cordis' +import Loader from '@cordisjs/plugin-loader' +import { existsSync, mkdtempSync, rmSync, writeFileSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { fileURLToPath } from 'node:url' +import SubagentService from '@deepseek-ai/dsh-subagent' +import type { Agent } from '@deepseek-ai/dsh-agent' +import * as acp from '../src/index.ts' +import { acpStopReason, acpContentText, buildChildEnv, SENSITIVE_ENV_PATTERN, toAcpPrompt } from '../src/run.ts' + +/** + * Keyless integration tests for the ACP subagent backend. Each spawns a REAL + * subprocess — the scripted mock ACP server (tests/mock-acp-server.ts) — and + * drives it through the REAL backend over real ACP JSON-RPC stdio, so the + * connection setup, the client callbacks, the prompt round-trip, the stop-reason + * mapping, cancellation, and quiescent disposal are all exercised end to end. + * No model, no key. + */ + +const mockServer = fileURLToPath(new URL('./mock-acp-server.ts', import.meta.url)) +const tsxLoader = fileURLToPath(import.meta.resolve('tsx')) +const repoTsconfig = fileURLToPath(new URL('../../../../tsconfig.json', import.meta.url)) + +/** A throwaway parent Agent — the ACP backend ignores it, but the seam requires one. */ +const fakeParent = { id: 'parent', session: { header: {} } } as unknown as Agent + +interface SetupEnv { + /** Mock-server scripting env: MOCK_TEXT / MOCK_STOP / MOCK_HANG / MOCK_PERMISSION. */ + [key: string]: string +} + +/** + * Mount the ACP backend pointed at the mock server, scripted by `mockEnv`. + * `permission` selects the backend's auto-answer policy. + */ +async function setup(mockEnv: SetupEnv = {}, permission: 'allow' | 'reject' = 'reject') { + const ctx = new Context() + await ctx.plugin(SubagentService) + await ctx.plugin(acp, { + providerName: 'acp', + command: process.execPath, + args: ['--import', tsxLoader, mockServer], + permission, + // The mock-server scripting vars must reach the child; TSX_TSCONFIG_PATH lets + // tsx resolve @deepseek-ai/* from a child cwd outside the repo. + env: { ...mockEnv, TSX_TSCONFIG_PATH: repoTsconfig }, + }) + return ctx +} + +function text(blocks: { type: string; text?: string }[]): string { + return blocks.filter(b => b.type === 'text').map(b => b.text).join('') +} + +/** + * Poll until `file` exists (the mock touches it once its prompt is in flight), + * so a cancel test waits on a CONDITION rather than an arbitrary timeout — the + * subprocess cold-start under tsx is variable, and a fixed sleep both flakes and + * slows the suite. Fails loud if the child never signals readiness. + */ +async function waitForFile(file: string, timeoutMs = 5000): Promise { + const deadline = Date.now() + timeoutMs + while (!existsSync(file)) { + if (Date.now() > deadline) throw new Error(`mock child never became ready (${file})`) + await new Promise(r => setTimeout(r, 10)) + } +} + +describe('acpStopReason', () => { + it('maps each ACP stop reason to the harness vocabulary', () => { + expect(acpStopReason('end_turn')).toBe('completed') + expect(acpStopReason('max_tokens')).toBe('max-tokens') + expect(acpStopReason('refusal')).toBe('refusal') + expect(acpStopReason('cancelled')).toBe('aborted') + expect(acpStopReason('max_turn_requests')).toBe('error') + }) + + it('treats an unknown terminal reason as an error', () => { + expect(acpStopReason('something-new' as never)).toBe('error') + }) +}) + +describe('acpContentText / toAcpPrompt', () => { + it('extracts text from a text content block, empty for non-text', () => { + expect(acpContentText({ type: 'text', text: 'hi' })).toBe('hi') + // A non-text ACP content block (e.g. an image) contributes no text. + expect(acpContentText({ type: 'image', data: 'x', mimeType: 'image/png' })).toBe('') + }) + + it('keeps text prompt blocks and drops non-text ones', () => { + expect(toAcpPrompt([{ type: 'text', text: 'a' }])).toEqual([{ type: 'text', text: 'a' }]) + // A non-text harness block (e.g. reasoning) is dropped from the ACP prompt. + expect(toAcpPrompt([{ type: 'text', text: 'a' }, { type: 'reasoning', text: 'think' }])) + .toEqual([{ type: 'text', text: 'a' }]) + }) +}) + +describe('buildChildEnv', () => { + it('drops credential-shaped ambient vars but keeps the explicit extras', () => { + process.env.DSH_ACP_TEST_SECRET_TOKEN = 'leak-me' + try { + const env = buildChildEnv({ DEEPSEEK_API_KEY: 'explicit' }) + // The credential-shaped ambient var is scrubbed. + expect(env.DSH_ACP_TEST_SECRET_TOKEN).toBeUndefined() + // The explicitly-supplied key survives (an opt-in for the child's creds). + expect(env.DEEPSEEK_API_KEY).toBe('explicit') + // A normal ambient var is forwarded. + expect(SENSITIVE_ENV_PATTERN.test('PATH')).toBe(false) + expect(env.PATH).toBe(process.env.PATH) + } finally { + delete process.env.DSH_ACP_TEST_SECRET_TOKEN + } + }) +}) + +describe('dsh-subagent-acp', () => { + it('drives a child process to completion and returns its streamed output', async () => { + const ctx = await setup({ MOCK_TEXT: 'hello from acp child', MOCK_STOP: 'end_turn' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'do X' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('completed') + expect(text(result.output)).toBe('hello from acp child') + await run.dispose() + }) + + it('maps a max_tokens stop reason', async () => { + const ctx = await setup({ MOCK_TEXT: 'cut off', MOCK_STOP: 'max_tokens' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('max-tokens') + await run.dispose() + }) + + it('maps a refusal stop reason', async () => { + const ctx = await setup({ MOCK_TEXT: '', MOCK_STOP: 'refusal' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('refusal') + await run.dispose() + }) + + it('cancelling a running child settles aborted (session/cancel via run.cancel)', async () => { + const tmp = mkdtempSync(join(tmpdir(), 'acp-cancel-')) + const readyFile = join(tmp, 'ready') + try { + const ctx = await setup({ MOCK_TEXT: 'partial', MOCK_HANG: '1', MOCK_READY_FILE: readyFile }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + // Wait until the child's prompt is in flight (condition, not a sleep), + // then cancel — so we exercise the mid-run session/cancel path. + await waitForFile(readyFile) + run.cancel('test') + const result = await run.result + expect(result.stopReason).toBe('aborted') + await run.dispose() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('settles aborted without running the child when the signal is already aborted', async () => { + const controller = new AbortController() + controller.abort() + const ctx = await setup({ MOCK_TEXT: 'never seen' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent, signal: controller.signal }) + const result = await run.result + expect(result.stopReason).toBe('aborted') + expect(result.output).toEqual([]) + await run.dispose() + }) + + it('honors a cancel that races AHEAD of newSession (no session id yet) without running the prompt', async () => { + // Gate the child at newSession: it signals `ready` and blocks until `go`. + // We cancel WHILE newSession is pending (sessionId still undefined, so the + // backend cannot send session/cancel) — the `cancelled` flag alone must + // settle the run aborted after newSession resolves, never issuing the prompt. + const tmp = mkdtempSync(join(tmpdir(), 'acp-early-')) + const ready = join(tmp, 'ready') + const go = join(tmp, 'go') + try { + const ctx = await setup({ MOCK_NEWSESSION_READY: ready, MOCK_NEWSESSION_GO: go, MOCK_TEXT: 'should not run' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + await waitForFile(ready) // newSession is now in flight, sessionId undefined + run.cancel('early') // sets cancelled; cannot send session/cancel yet + writeFileSync(go, 'go') // let newSession resolve + const result = await run.result + expect(result.stopReason).toBe('aborted') + expect(result.output).toEqual([]) + await run.dispose() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('bridges the request signal to a session/cancel mid-run', async () => { + const tmp = mkdtempSync(join(tmpdir(), 'acp-signal-')) + const readyFile = join(tmp, 'ready') + try { + const controller = new AbortController() + const ctx = await setup({ MOCK_TEXT: 'partial', MOCK_HANG: '1', MOCK_READY_FILE: readyFile }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent, signal: controller.signal }) + await waitForFile(readyFile) + controller.abort() + const result = await run.result + expect(result.stopReason).toBe('aborted') + await run.dispose() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('auto-rejects a permission prompt by default (child settles cancelled→aborted)', async () => { + const ctx = await setup({ MOCK_TEXT: 'x', MOCK_PERMISSION: '1' }, 'reject') + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + // The child asked permission, the backend rejected, the child returned cancelled. + expect(result.stopReason).toBe('aborted') + await run.dispose() + }) + + it('auto-approves a permission prompt under the allow policy', async () => { + const ctx = await setup({ MOCK_TEXT: 'approved answer', MOCK_PERMISSION: '1', MOCK_STOP: 'end_turn' }, 'allow') + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('completed') + expect(text(result.output)).toBe('approved answer') + await run.dispose() + }) + + it('falls back to cancelled under the allow policy when the child offers no allow option', async () => { + // The child asks permission but offers ONLY reject-shaped options, so an + // allow-policy client finds nothing to select and must answer cancelled. + const ctx = await setup({ MOCK_PERMISSION: '1', MOCK_NO_ALLOW: '1' }, 'allow') + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('aborted') + await run.dispose() + }) + + it('consumes a non-message update (a thought) without adding it to the output', async () => { + // The child streams an agent_thought_chunk before its answer; the backend + // must consume it but NOT include it in the result output. + const ctx = await setup({ MOCK_THOUGHT: '1', MOCK_TEXT: 'final answer', MOCK_STOP: 'end_turn' }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + expect(result.stopReason).toBe('completed') + // Only the message text, NOT the thought. + expect(text(result.output)).toBe('final answer') + await run.dispose() + }) + + it('resolves error (not reject) when the spawn command does not exist', async () => { + const ctx = new Context() + await ctx.plugin(SubagentService) + await ctx.plugin(acp, { + providerName: 'acp', + command: '/nonexistent/acp-agent-binary', + args: [], + permission: 'reject', + env: {}, + }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + const result = await run.result + // The seam contract: a child-level failure resolves error, never rejects. + expect(result.stopReason).toBe('error') + await run.dispose() + }) + + it('settles aborted when the child crashes (tears the pipe) AFTER a cancel', async () => { + // The child hangs, we cancel, and instead of answering the child exits hard + // — the pending prompt RPC rejects. With a cancel already requested, the + // backend's catch path must settle `aborted` (the failure is the cancel + // surfacing as a torn pipe), not `error`. + const tmp = mkdtempSync(join(tmpdir(), 'acp-crash-')) + const ready = join(tmp, 'ready') + try { + const ctx = await setup({ MOCK_TEXT: 'partial', MOCK_HANG: '1', MOCK_CRASH_ON_CANCEL: '1', MOCK_READY_FILE: ready }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + await waitForFile(ready) + run.cancel('crash it') + const result = await run.result + expect(result.stopReason).toBe('aborted') + await run.dispose() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('advertises no start-time capabilities (out-of-process child)', async () => { + const ctx = await setup() + const provider = ctx.subagents.getProvider('acp')! + expect(provider.capabilities).toEqual({ outputSchema: false, depthLimit: false, toolFilter: false }) + }) + + it('unregisters the provider when its fiber is disposed (HMR safety)', async () => { + const ctx = new Context() + await ctx.plugin(SubagentService) + const fiber = await ctx.plugin(acp, { providerName: 'acp', command: 'x', args: [], permission: 'reject', env: {} }) + expect(ctx.subagents.list()).toEqual(['acp']) + await fiber.dispose() + expect(ctx.subagents.list()).toEqual([]) + }) + + it('has the namespace-plugin export shape (no stray default)', () => { + expect('default' in acp).toBe(false) + expect(acp.name).toBe('subagent-acp') + expect(acp.inject).toEqual(['subagents']) + const loader = Object.create(Loader.prototype) as Loader + const unwrapped = loader.unwrapExports(acp) as Record + expect(unwrapped).toBe(acp) + expect(unwrapped.name).toBe('subagent-acp') + expect(typeof unwrapped.apply).toBe('function') + }) +}) diff --git a/packages/subagent/subagent-acp/tsconfig.json b/packages/subagent/subagent-acp/tsconfig.json new file mode 100644 index 0000000000..77a7b76e5f --- /dev/null +++ b/packages/subagent/subagent-acp/tsconfig.json @@ -0,0 +1,30 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib" + }, + "include": [ + "src" + ], + "references": [ + { + "path": "../../../vendor/cosmokit" + }, + { + "path": "../../../vendor/cordis" + }, + { + "path": "../../../vendor/schemastery" + }, + { + "path": "../../llm/llm" + }, + { + "path": "../../core/agent" + }, + { + "path": "../subagent" + } + ] +} diff --git a/packages/subagent/subagent/README.md b/packages/subagent/subagent/README.md index de55315a79..57862ca8ab 100644 --- a/packages/subagent/subagent/README.md +++ b/packages/subagent/subagent/README.md @@ -34,6 +34,6 @@ Unlike the bash seam (one executor per context, second load throws), **multiple ## Scope (first cut) -The consumer collects **synchronously**: it starts a run and awaits `result`. Steering (`sendMessage`) is part of the contract but intentionally unused. Background / poll / spill semantics are deferred to a future redesign unifying long-running-tool handling across subagents and bash. See the RFC: [docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md](../../../docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md). +The consumer collects **synchronously**: it starts a run and awaits `result`. Steering (`sendMessage`) is part of the contract but intentionally unused. Background / poll / spill semantics are deferred to a future redesign unifying long-running-tool handling across subagents and bash. See the RFC: [docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md](../../../docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md). See `src/types.ts` for the full contracts. diff --git a/packages/subagent/tool-subagent/README.md b/packages/subagent/tool-subagent/README.md index 22b66fbf80..1bb48f29ff 100644 --- a/packages/subagent/tool-subagent/README.md +++ b/packages/subagent/tool-subagent/README.md @@ -16,4 +16,4 @@ This plugin binds to **exactly one** provider (`Config.provider`). The model see `execute` starts a run on the configured provider and **awaits `run.result` inside a `try/finally` that always `dispose()`s the run** — the owned child agent/session is torn down on every path (success, error, abort), never leaked. The tool's abort signal (`exec.signal`) is bridged to `run.cancel()`. A non-`completed` stop reason (aborted/error/max-tokens/refusal) maps to an `isError` tool result rather than returning partial output as success. -Background / poll collection is deferred (see the [RFC](../../../docs/rfc/proposed/feature/2026-06-21-subagent-capability-seam.md)); this cut blocks the parent turn until the child finishes. +Background / poll collection is deferred (see the [RFC](../../../docs/rfc/implemented/feature/2026-06-21-subagent-capability-seam.md)); this cut blocks the parent turn until the child finishes. diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index b262257b14..4003119c65 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -333,6 +333,31 @@ importers: specifier: ^4.0.0-rc.6 version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4) + packages/subagent/subagent-acp: + dependencies: + '@agentclientprotocol/sdk': + specifier: 0.25.1 + version: 0.25.1(zod@4.4.3) + schemastery: + specifier: ^3.18.0 + version: 3.18.0 + devDependencies: + '@cordisjs/plugin-loader': + specifier: ^1.0.0-rc.4 + version: 1.0.0-rc.4(cordis@4.0.0-rc.6) + '@deepseek-ai/dsh-agent': + specifier: workspace:^ + version: link:../../core/agent + '@deepseek-ai/dsh-llm': + specifier: workspace:^ + version: link:../../llm/llm + '@deepseek-ai/dsh-subagent': + specifier: workspace:^ + version: link:../subagent + cordis: + specifier: ^4.0.0-rc.6 + version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4) + packages/subagent/subagent-fork: dependencies: schemastery: diff --git a/tsconfig.build.json b/tsconfig.build.json index 4f0528961d..8b3010b3fe 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -36,6 +36,7 @@ { "path": "./packages/support/subagent-mock" }, { "path": "./packages/subagent/tool-subagent" }, { "path": "./packages/subagent/subagent-spawn" }, - { "path": "./packages/subagent/subagent-fork" } + { "path": "./packages/subagent/subagent-fork" }, + { "path": "./packages/subagent/subagent-acp" } ] } From 6801130f8cc79c5001c458e52150e622024ac840 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 22 Jun 2026 11:39:30 +0800 Subject: [PATCH 2/5] Bound ACP dispose with SIGKILL escalation; skip spawn when pre-aborted (Codex review round 1) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two lifecycle findings from the review: - A (blocker): dispose() could hang forever. It only sent SIGTERM and awaited exit, with no escalation — a child that traps SIGTERM (or our acp-agent if it doesn't quiesce on stdin EOF) would wedge dispose, stranding tool-subagent's finally cleanup and orphaning child-owned work (e.g. bash subprocesses). dispose now: ends stdin (graceful ACP close so the child can flush + exit), SIGTERM, then escalates to SIGKILL if it doesn't exit within a grace period (DEFAULT_DISPOSE_GRACE_MS, injectable via spec.disposeGraceMs), awaiting the certain exit. Mirrors the bash executor's bounded teardown. Regression test drives a SIGTERM-trapping mock subprocess and asserts dispose returns promptly — proven to hang (red) without the escalation. - B: an already-aborted request still spawned the configured binary. startAcpRun now returns an inert already-aborted run BEFORE spawning, so a pre-cancelled request launches nothing. Test points the command at `touch ` and asserts the sentinel never appears. The dispose regression test exposed (via systematic-debugging) that the child must signal trap-armed readiness before the test cancels — a bare timeout raced the trap install and the default SIGTERM handler killed the child, making the guard a no-op. The mock now touches its ready file once the trap is in place and the test waits on that condition. The `cancelled` flag moved onto a holder object so TS control-flow doesn't narrow the catch-time read to always-false. --- packages/subagent/subagent-acp/src/run.ts | 71 ++++++++++++++----- .../subagent-acp/tests/mock-acp-server.ts | 14 ++++ .../subagent-acp/tests/subagent-acp.spec.ts | 68 +++++++++++++++--- 3 files changed, 126 insertions(+), 27 deletions(-) diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 18f20c53c1..612fe87074 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -70,8 +70,17 @@ export interface AcpRunSpec { * the credential-scrub pattern (an explicit opt-in for the child's own creds). */ env: Record + /** + * Grace period (ms) between `SIGTERM` and the `SIGKILL` escalation in + * {@link SubagentRun.dispose}. Defaults to {@link DEFAULT_DISPOSE_GRACE_MS}; + * a test injects a small value to exercise the escalation without a long wait. + */ + disposeGraceMs?: number } +/** Default grace between SIGTERM and SIGKILL on dispose (mirrors the bash executor). */ +export const DEFAULT_DISPOSE_GRACE_MS = 3_000 + /** * Credential-shaped ambient env vars are NOT forwarded to the child by default * (the parent harness's own `DEEPSEEK_API_KEY`/secrets must not leak into a @@ -133,6 +142,9 @@ export function toAcpPrompt(prompt: ContentBlock[]): AcpContentBlock[] { /** Resolve once the child process exits (any code/signal); immediate if gone. */ function waitForExit(child: ChildProcess): Promise { + // Already-exited fast path: dispose guards on exitCode before calling, so in + // tests the child is always still alive here. + /* v8 ignore next */ if (child.exitCode !== null || child.signalCode !== null) return Promise.resolve() return new Promise(resolve => child.once('exit', () => { resolve() })) } @@ -151,6 +163,18 @@ function waitForExit(child: ChildProcess): Promise { export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): SubagentRun { const id = AgentId(randomUUID()) + // A request already aborted before it starts never spawns the child at all — + // return an inert run that settled `aborted`, rather than launching the + // configured binary just to tear it down. `dispose`/`cancel` are no-ops. + if (request.signal?.aborted) { + return { + id, + result: Promise.resolve({ output: [], stopReason: 'aborted' }), + cancel(_reason?: string): void { /* nothing was started */ }, + dispose(): Promise { return Promise.resolve() }, + } + } + // Spawn the child ACP agent. stdin = ACP request channel, stdout = ACP // response channel, stderr = INHERIT so the child's diagnostics surface on the // parent's stderr (no separate capture to drain — we don't fold child stderr @@ -172,8 +196,11 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su const output: string[] = [] // `cancelled` records that a cancel was requested (signal or cancel()), so a // run torn down before the prompt resolves settles `aborted` rather than the - // generic error mapping. - let cancelled = false + // generic error mapping. Held on a mutable object so the async closures that + // set it (the abort listener) and the IIFE that reads it don't fight TS's + // control-flow narrowing of a bare `let` (which would type the catch-time read + // as always-`false`). + const flags = { cancelled: false } const makeClient = (_agent: AcpAgent): Client => ({ sessionUpdate(params: SessionNotification): Promise { @@ -209,7 +236,7 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su let sessionId: string | undefined const requestCancel = (): void => { - cancelled = true + flags.cancelled = true // Best-effort: tell the child to cancel the in-flight turn. Swallows a // rejection — the session may not exist yet, or the pipe may be gone; the // dispose path kills the process regardless. If the session has NOT been @@ -233,11 +260,6 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su return text.length > 0 ? [{ type: 'text', text }] : [] } try { - // An already-aborted request never runs the child. - if (request.signal?.aborted) { - cancelled = true - return { output: [], stopReason: 'aborted' } - } // Race the ACP drive against a spawn failure: a bad command never speaks // ACP, so `initialize` would hang forever — the spawn `error` event is the // only signal, and a rejected race settles the run `error` via the catch. @@ -254,7 +276,7 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su // send `session/cancel` (no session id yet). Honor it here: settle // `aborted` without ever issuing the prompt, rather than running the child // to completion and ignoring the cancel. - if (cancelled) return { output: collectOutput(), stopReason: 'aborted' } + if (flags.cancelled) return { output: collectOutput(), stopReason: 'aborted' } const promptResult = await conn.prompt({ sessionId, prompt: toAcpPrompt(request.prompt) }) return { output: collectOutput(), stopReason: acpStopReason(promptResult.stopReason) } } @@ -267,7 +289,7 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su // failure. A spawn/transport/RPC error becomes an error/aborted result — // `aborted` if a cancel was requested (the failure is the cancellation // surfacing as a torn pipe / rejected RPC), else a genuine `error`. - return { output: collectOutput(), stopReason: cancelled ? 'aborted' : 'error' } + return { output: collectOutput(), stopReason: flags.cancelled ? 'aborted' : 'error' } } })() @@ -279,14 +301,29 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su }, async dispose(): Promise { request.signal?.removeEventListener('abort', onAbort) - // Kill the subprocess and AWAIT its exit (quiescent teardown — dispose - // must reach quiescence, not merely request it). SIGTERM first; the child - // is our own short-lived ACP agent, so a graceful term is enough. Guard - // the kill: the process may already be gone. - if (child.exitCode === null && child.signalCode === null) { - child.kill('SIGTERM') + // Reach quiescence, not merely request it (dispose must AWAIT the child + // actually stopping). If the child is already gone, nothing to do. + if (child.exitCode !== null || child.signalCode !== null) return + // 1. Graceful: end the ACP request stream (stdin EOF). Our own acp-agent + // disposes its fiber on stdin 'end' — flushing persistence and stopping + // child-owned work (e.g. bash subprocesses) — then exits, which the + // server bridge's connection-close quiesce path drives. A child that + // ignores EOF is handled by the signal escalation below. + child.stdin.end() + // 2. SIGTERM, then escalate to SIGKILL if it does not exit within the + // grace period — a child that traps SIGTERM must not wedge dispose + // forever (the seam requires bounded quiescence). Race the exit against + // a grace timer; on timeout, SIGKILL and await the (now-certain) exit. + child.kill('SIGTERM') + const graceMs = spec.disposeGraceMs ?? DEFAULT_DISPOSE_GRACE_MS + const exited = await Promise.race([ + waitForExit(child).then(() => true), + new Promise(resolve => setTimeout(() => { resolve(false) }, graceMs).unref()), + ]) + if (!exited) { + child.kill('SIGKILL') + await waitForExit(child) } - await waitForExit(child) }, } } diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts index e383bbe8eb..8260051f63 100644 --- a/packages/subagent/subagent-acp/tests/mock-acp-server.ts +++ b/packages/subagent/subagent-acp/tests/mock-acp-server.ts @@ -148,3 +148,17 @@ new AgentSideConnection( Readable.toWeb(process.stdin) as ReadableStream, ), ) + +// Under MOCK_TRAP_SIGTERM, ignore SIGTERM and keep stdin open so the process +// neither quiesces on EOF nor dies on the graceful signal — exercising the +// backend dispose path's SIGKILL escalation. Without this the process exits +// normally on SIGTERM / stdin end. Touch READY_FILE once the trap is armed, so +// a test waits for that CONDITION before disposing (the trap must be in place, +// not merely the process spawned — otherwise SIGTERM hits the default handler). +if (process.env.MOCK_TRAP_SIGTERM === '1') { + process.on('SIGTERM', () => { /* trapped: refuse to exit on the graceful signal */ }) + // Keep the event loop alive (a bare timer) so nothing else lets it exit. + setInterval(() => { /* stay alive until SIGKILL */ }, 1000) + if (READY_FILE !== undefined) writeFileSync(READY_FILE, 'trap-armed') +} + diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index ac74ace7e8..0e4604e638 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -8,7 +8,7 @@ import { fileURLToPath } from 'node:url' import SubagentService from '@deepseek-ai/dsh-subagent' import type { Agent } from '@deepseek-ai/dsh-agent' import * as acp from '../src/index.ts' -import { acpStopReason, acpContentText, buildChildEnv, SENSITIVE_ENV_PATTERN, toAcpPrompt } from '../src/run.ts' +import { acpStopReason, acpContentText, buildChildEnv, SENSITIVE_ENV_PATTERN, startAcpRun, toAcpPrompt, type AcpRunSpec } from '../src/run.ts' /** * Keyless integration tests for the ACP subagent backend. Each spawns a REAL @@ -159,15 +159,63 @@ describe('dsh-subagent-acp', () => { } }) - it('settles aborted without running the child when the signal is already aborted', async () => { - const controller = new AbortController() - controller.abort() - const ctx = await setup({ MOCK_TEXT: 'never seen' }) - const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent, signal: controller.signal }) - const result = await run.result - expect(result.stopReason).toBe('aborted') - expect(result.output).toEqual([]) - await run.dispose() + it('settles aborted WITHOUT spawning the child when the signal is already aborted', async () => { + // A pre-aborted request must not even launch the configured binary. Point + // the command at one that would create a sentinel file if it ever ran, and + // assert the sentinel never appears. + const tmp = mkdtempSync(join(tmpdir(), 'acp-preabort-')) + const sentinel = join(tmp, 'spawned') + try { + const controller = new AbortController() + controller.abort() + const run = startAcpRun( + { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent, signal: controller.signal }, + // `touch ` — runs only if the process is actually spawned. + { command: 'touch', args: [sentinel], cwd: tmp, permission: 'reject', env: {} }, + ) + const result = await run.result + expect(result.stopReason).toBe('aborted') + expect(result.output).toEqual([]) + // cancel/dispose on the inert run are safe no-ops. + run.cancel('noop') + await run.dispose() + // The binary was never launched — no sentinel. + expect(existsSync(sentinel)).toBe(false) + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('dispose escalates SIGTERM → SIGKILL for a child that traps SIGTERM (bounded quiescence)', async () => { + // The child traps SIGTERM and keeps its event loop alive, so a graceful + // term alone would hang dispose forever. With a short grace, dispose must + // escalate to SIGKILL and return once the process is actually gone. + const tmp = mkdtempSync(join(tmpdir(), 'acp-trap-')) + const ready = join(tmp, 'trap-armed') + try { + const spec: AcpRunSpec = { + command: process.execPath, + args: ['--import', tsxLoader, mockServer], + cwd: process.cwd(), + permission: 'reject', + env: { MOCK_TRAP_SIGTERM: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, TSX_TSCONFIG_PATH: repoTsconfig }, + disposeGraceMs: 150, + } + const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) + // Wait until the child has BOOTED AND ARMED THE TRAP (a condition, not a + // sleep) — otherwise SIGTERM races the trap install and the default handler + // terminates the child, never exercising the escalation. + await waitForFile(ready) + // Don't await result (the child hangs). Dispose must still return promptly + // via the SIGKILL escalation — bound it so a regression (no escalation) + // fails loud instead of hanging the suite. + await expect(Promise.race([ + run.dispose(), + new Promise((_r, reject) => { setTimeout(() => { reject(new Error('dispose did not return — no SIGKILL escalation')) }, 4000) }), + ])).resolves.toBeUndefined() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } }) it('honors a cancel that races AHEAD of newSession (no session id yet) without running the prompt', async () => { From 4565161c64b86a7db497106a76f0d5dd81e2e119 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 22 Jun 2026 12:11:17 +0800 Subject: [PATCH 3/5] Give the ACP child an EOF window to quiesce before SIGTERM (Codex review round 2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit dispose() ended stdin and sent SIGTERM in the same tick, so the child's EOF-driven quiesce had no window to run. The real acp-agent has no SIGTERM handler in a normal session — it flushes persistence and stops child-owned work via the server bridge's connection-close path (conn.closed → per-agent dispose → final session/flush), driven by stdin EOF, NOT by a signal. A prompt response can resolve from a turn/end before that post-turn flush lands, so the child still owes durable work when dispose runs; a same-tick default SIGTERM terminated it mid-flush, orphaning child-owned bash and dropping the flush. dispose now waits for the child's natural exit after stdin EOF first, then escalates SIGTERM (grace), then SIGKILL — a three-tier ladder. Add an `exitsWithin` helper for the bounded waits. Regression coverage: a new mock mode (MOCK_FLUSH_ON_EOF) flushes a marker asynchronously on EOF then self-exits; the tier-1 test asserts the marker lands (proven RED on the same-tick-SIGTERM ordering — child killed mid-flush). MOCK_IGNORE_EOF covers the middle tier (ignores EOF, dies on default SIGTERM); the existing MOCK_TRAP_SIGTERM test covers the SIGKILL tier. --- packages/subagent/subagent-acp/src/run.ts | 47 ++++++++------ .../subagent-acp/tests/mock-acp-server.ts | 38 ++++++++++++ .../subagent-acp/tests/subagent-acp.spec.ts | 61 +++++++++++++++++++ 3 files changed, 127 insertions(+), 19 deletions(-) diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 612fe87074..c78e8ae08d 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -149,6 +149,15 @@ function waitForExit(child: ChildProcess): Promise { return new Promise(resolve => child.once('exit', () => { resolve() })) } +/** Resolve `true` if the child exits within `ms`, `false` on timeout. */ +function exitsWithin(child: ChildProcess, ms: number): Promise { + return Promise.race([ + waitForExit(child).then(() => true), + // `.unref()` so a pending grace timer never keeps the parent's loop alive. + new Promise(resolve => setTimeout(() => { resolve(false) }, ms).unref()), + ]) +} + /** * Start an out-of-process ACP child for `request` and return a {@link SubagentRun}. * @@ -304,26 +313,26 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su // Reach quiescence, not merely request it (dispose must AWAIT the child // actually stopping). If the child is already gone, nothing to do. if (child.exitCode !== null || child.signalCode !== null) return - // 1. Graceful: end the ACP request stream (stdin EOF). Our own acp-agent - // disposes its fiber on stdin 'end' — flushing persistence and stopping - // child-owned work (e.g. bash subprocesses) — then exits, which the - // server bridge's connection-close quiesce path drives. A child that - // ignores EOF is handled by the signal escalation below. - child.stdin.end() - // 2. SIGTERM, then escalate to SIGKILL if it does not exit within the - // grace period — a child that traps SIGTERM must not wedge dispose - // forever (the seam requires bounded quiescence). Race the exit against - // a grace timer; on timeout, SIGKILL and await the (now-certain) exit. - child.kill('SIGTERM') const graceMs = spec.disposeGraceMs ?? DEFAULT_DISPOSE_GRACE_MS - const exited = await Promise.race([ - waitForExit(child).then(() => true), - new Promise(resolve => setTimeout(() => { resolve(false) }, graceMs).unref()), - ]) - if (!exited) { - child.kill('SIGKILL') - await waitForExit(child) - } + // 1. Graceful: end the ACP request stream (stdin EOF) and let the child + // quiesce ON ITS OWN. Our acp-agent has NO SIGTERM handler in a normal + // session — it tears down via the server bridge's connection-close path + // (conn.closed → per-agent dispose → final session/flush), driven by the + // stdin EOF, NOT by a signal. A prompt response can resolve from a + // turn/end BEFORE that post-turn flush lands, so the child still has + // durable work owed when dispose runs. Give the EOF-driven quiesce a real + // window to finish (flush persistence, stop child-owned bash) and EXIT; + // sending SIGTERM in the same tick would default-terminate it mid-flush. + child.stdin.end() + if (await exitsWithin(child, graceMs)) return + // 2. SIGTERM, then escalate to SIGKILL if it still does not exit within the + // grace period — a child that ignores EOF and traps SIGTERM must not + // wedge dispose forever (the seam requires bounded quiescence). + child.kill('SIGTERM') + if (await exitsWithin(child, graceMs)) return + // 3. Force-kill and await the (now-certain) exit. + child.kill('SIGKILL') + await waitForExit(child) }, } } diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts index 8260051f63..b30492258a 100644 --- a/packages/subagent/subagent-acp/tests/mock-acp-server.ts +++ b/packages/subagent/subagent-acp/tests/mock-acp-server.ts @@ -14,6 +14,18 @@ * handler is in flight (it has streamed its chunk). A test * polls for this file to cancel on a CONDITION rather than * an arbitrary timeout (subprocess cold-start is variable). + * - `MOCK_FLUSH_ON_EOF` — if set, on stdin EOF the agent takes an async beat + * (simulating the real acp-agent's EOF-driven + * quiesce+flush), then touches this path and exits ON ITS + * OWN — no signal. Stands in for a child whose durable + * flush completes only if dispose gives EOF a real window + * before escalating to SIGTERM. + * - `MOCK_IGNORE_EOF` — if `1`, keep the event loop alive past stdin EOF (a bare + * timer) but leave SIGTERM at its DEFAULT handler, so the + * child ignores the graceful EOF window yet still dies on + * SIGTERM — exercising dispose's middle tier (exit during + * the SIGTERM grace, before the SIGKILL escalation). It + * touches MOCK_READY_FILE once the keepalive is armed. * * It is NOT a test spec (no `describe`/`it`) — it is spawned BY the specs as the * child process the ACP backend drives. Kept as a `.ts` run under tsx by the @@ -50,6 +62,7 @@ const NO_ALLOW = process.env.MOCK_NO_ALLOW === '1' const THOUGHT = process.env.MOCK_THOUGHT === '1' const CRASH_ON_CANCEL = process.env.MOCK_CRASH_ON_CANCEL === '1' const READY_FILE = process.env.MOCK_READY_FILE +const FLUSH_ON_EOF = process.env.MOCK_FLUSH_ON_EOF // When MOCK_NEWSESSION_READY/GO are set, newSession touches READY then blocks // until GO appears — letting a test cancel mid-newSession deterministically. const NEWSESSION_GATE = process.env.MOCK_NEWSESSION_READY !== undefined && process.env.MOCK_NEWSESSION_GO !== undefined @@ -162,3 +175,28 @@ if (process.env.MOCK_TRAP_SIGTERM === '1') { if (READY_FILE !== undefined) writeFileSync(READY_FILE, 'trap-armed') } +// Under MOCK_FLUSH_ON_EOF, model the real acp-agent's EOF-driven quiesce: on +// stdin 'end' (the dispose path's `child.stdin.end()`), take an ASYNC beat to +// "flush", then touch the marker and exit ON OUR OWN — no signal involved. A +// dispose that sends SIGTERM in the same tick as the EOF (no graceful window) +// default-terminates this process before the beat completes, so the marker is +// missing; a dispose that waits for natural exit first lets the flush land. +if (FLUSH_ON_EOF !== undefined) { + process.stdin.on('end', () => { + setTimeout(() => { + writeFileSync(FLUSH_ON_EOF, 'flushed') + process.exit(0) + }, 150) + }) +} + +// Under MOCK_IGNORE_EOF, keep the loop alive past stdin EOF but leave SIGTERM at +// its DEFAULT handler — the child ignores the graceful EOF window yet still dies +// on SIGTERM, exercising dispose's middle tier (exit during the SIGTERM grace, +// before the SIGKILL escalation). Touch the ready file once the keepalive is +// armed, so a test disposes on that condition rather than a timeout. +if (process.env.MOCK_IGNORE_EOF === '1') { + setInterval(() => { /* stay alive past EOF; default SIGTERM still kills us */ }, 1000) + if (READY_FILE !== undefined) writeFileSync(READY_FILE, 'ignore-eof-armed') +} + diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index 0e4604e638..96b749218d 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -218,6 +218,67 @@ describe('dsh-subagent-acp', () => { } }) + it('dispose gives the child an EOF window to quiesce before escalating (graceful flush)', async () => { + // The real acp-agent flushes ASYNCHRONOUSLY on stdin EOF (its bridge tears + // down on connection close, NOT on a signal) — and it has no SIGTERM handler. + // The mock models that: on stdin 'end' it takes a beat to "flush", touches a + // marker, and exits on its own. dispose() must end stdin and WAIT for that + // natural exit before sending SIGTERM; a same-tick SIGTERM default-kills the + // child mid-flush and the marker never appears. + const tmp = mkdtempSync(join(tmpdir(), 'acp-eof-')) + const ready = join(tmp, 'ready') + const flushed = join(tmp, 'flushed') + try { + const spec: AcpRunSpec = { + command: process.execPath, + args: ['--import', tsxLoader, mockServer], + cwd: process.cwd(), + permission: 'reject', + // MOCK_HANG so the prompt never resolves on its own — we tear down a live + // child. MOCK_FLUSH_ON_EOF is the marker the child writes iff its EOF + // quiesce was allowed to finish. + env: { MOCK_HANG: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, MOCK_FLUSH_ON_EOF: flushed, TSX_TSCONFIG_PATH: repoTsconfig }, + } + const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) + // Wait until the child is fully booted with its prompt in flight (its ACP + // stdin reader is attached), so dispose's stdin EOF reaches a live child. + await waitForFile(ready) + await run.dispose() + // dispose returned via the natural-exit tier — the EOF-driven flush landed. + expect(existsSync(flushed)).toBe(true) + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + + it('escalates to SIGTERM for a child that ignores EOF but is not SIGTERM-trapping', async () => { + // A child that keeps its loop alive past stdin EOF (so the graceful window + // times out) but leaves SIGTERM at the default handler must die on the + // SIGTERM tier — dispose returns there, never reaching the SIGKILL tier. + const tmp = mkdtempSync(join(tmpdir(), 'acp-ignore-eof-')) + const ready = join(tmp, 'ready') + try { + const spec: AcpRunSpec = { + command: process.execPath, + args: ['--import', tsxLoader, mockServer], + cwd: process.cwd(), + permission: 'reject', + env: { MOCK_HANG: '1', MOCK_IGNORE_EOF: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, TSX_TSCONFIG_PATH: repoTsconfig }, + disposeGraceMs: 150, + } + const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) + await waitForFile(ready) + // Bound it: a regression (no SIGTERM tier, only EOF + SIGKILL) would still + // pass, but a hang would fail loud rather than stall the suite. + await expect(Promise.race([ + run.dispose(), + new Promise((_r, reject) => { setTimeout(() => { reject(new Error('dispose did not return')) }, 4000) }), + ])).resolves.toBeUndefined() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + it('honors a cancel that races AHEAD of newSession (no session id yet) without running the prompt', async () => { // Gate the child at newSession: it signals `ready` and blocks until `go`. // We cancel WHILE newSession is pending (sessionId still undefined, so the From 2ff112962b10c0f97389c065a655002e9a2029d5 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 22 Jun 2026 13:29:23 +0800 Subject: [PATCH 4/5] Widen the dispose EOF grace past nested-teardown headroom; prove the SIGTERM rung (Codex review round 3) Two round-3 findings: (A) The EOF-quiesce window reused the 3000ms SIGTERM grace, the SAME value as dsh-bash-local's own SIGTERM->SIGKILL grace. The child acp-agent's EOF teardown disposes its loop, which stops child-owned bash -- and a SIGTERM-trapping bash grandchild can hold that for up to ~3s before its own SIGKILL, then the child still owes a final flush. With both graces equal, the parent's SIGTERM fired exactly as the child reached its own SIGKILL+flush, cutting it off. Split the EOF grace into its own knob (disposeEofGraceMs, default 6000ms) that exceeds a single signal-grace of nested-teardown headroom. The child is an arbitrary ACP agent, so the value is a standalone generous default, NOT derived from any child's internals. Tier-1 test now uses a flush that outlasts the SIGTERM grace but fits the EOF grace, so it lands only because the EOF tier honors its own wider window (proven RED when tier 1 reuses the small SIGTERM grace). (B) The middle-tier (SIGTERM) test only asserted dispose returned in time, so an EOF->SIGKILL ladder with the rung removed would still pass. The mock's MOCK_IGNORE_EOF mode now installs a SIGTERM handler that touches an observable marker before exiting; SIGKILL is uncatchable, so removing the SIGTERM rung leaves the marker absent (proven RED). The test asserts the marker exists. --- packages/subagent/subagent-acp/src/run.ts | 30 +++++++++- .../subagent-acp/tests/mock-acp-server.ts | 54 +++++++++++------- .../subagent-acp/tests/subagent-acp.spec.ts | 57 +++++++++++++------ 3 files changed, 101 insertions(+), 40 deletions(-) diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index c78e8ae08d..8aeb136f2d 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -70,6 +70,13 @@ export interface AcpRunSpec { * the credential-scrub pattern (an explicit opt-in for the child's own creds). */ env: Record + /** + * Grace period (ms) for the child's EOF-driven quiesce in + * {@link SubagentRun.dispose} — the window to flush persistence and tear down + * its OWN nested subprocesses before the parent escalates to a signal. Defaults + * to {@link DEFAULT_DISPOSE_EOF_GRACE_MS}; a test injects a small value. + */ + disposeEofGraceMs?: number /** * Grace period (ms) between `SIGTERM` and the `SIGKILL` escalation in * {@link SubagentRun.dispose}. Defaults to {@link DEFAULT_DISPOSE_GRACE_MS}; @@ -78,6 +85,19 @@ export interface AcpRunSpec { disposeGraceMs?: number } +/** + * Default grace for the child's EOF-driven quiesce on dispose — the window for it + * to flush persistence and tear down its OWN nested subprocesses (which may run + * their own `SIGTERM`→`SIGKILL` escalation) before the parent escalates to a + * signal. Deliberately LARGER than {@link DEFAULT_DISPOSE_GRACE_MS}: a cooperative + * child whose teardown is itself waiting on a signal-trapping grandchild (e.g. a + * bash subprocess in its own ~3s SIGTERM→SIGKILL grace) plus a final flush needs + * MORE than a single signal-grace of headroom, or the parent's SIGTERM cuts it off + * exactly as it reaches its own SIGKILL+flush. The child is an arbitrary ACP agent, + * so this is a standalone generous default, NOT derived from any child's internals. + */ +export const DEFAULT_DISPOSE_EOF_GRACE_MS = 6_000 + /** Default grace between SIGTERM and SIGKILL on dispose (mirrors the bash executor). */ export const DEFAULT_DISPOSE_GRACE_MS = 3_000 @@ -313,6 +333,7 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su // Reach quiescence, not merely request it (dispose must AWAIT the child // actually stopping). If the child is already gone, nothing to do. if (child.exitCode !== null || child.signalCode !== null) return + const eofGraceMs = spec.disposeEofGraceMs ?? DEFAULT_DISPOSE_EOF_GRACE_MS const graceMs = spec.disposeGraceMs ?? DEFAULT_DISPOSE_GRACE_MS // 1. Graceful: end the ACP request stream (stdin EOF) and let the child // quiesce ON ITS OWN. Our acp-agent has NO SIGTERM handler in a normal @@ -321,10 +342,13 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su // stdin EOF, NOT by a signal. A prompt response can resolve from a // turn/end BEFORE that post-turn flush lands, so the child still has // durable work owed when dispose runs. Give the EOF-driven quiesce a real - // window to finish (flush persistence, stop child-owned bash) and EXIT; - // sending SIGTERM in the same tick would default-terminate it mid-flush. + // window — wider than a single signal-grace, since the child's own + // teardown may itself be awaiting a signal-trapping grandchild (a bash + // subprocess in its own SIGTERM→SIGKILL grace) plus a flush — and only + // escalate if it overruns. Sending SIGTERM in the same tick (or too soon) + // would default-terminate the child mid-flush, orphaning its nested work. child.stdin.end() - if (await exitsWithin(child, graceMs)) return + if (await exitsWithin(child, eofGraceMs)) return // 2. SIGTERM, then escalate to SIGKILL if it still does not exit within the // grace period — a child that ignores EOF and traps SIGTERM must not // wedge dispose forever (the seam requires bounded quiescence). diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts index b30492258a..74ae340bde 100644 --- a/packages/subagent/subagent-acp/tests/mock-acp-server.ts +++ b/packages/subagent/subagent-acp/tests/mock-acp-server.ts @@ -15,17 +15,19 @@ * polls for this file to cancel on a CONDITION rather than * an arbitrary timeout (subprocess cold-start is variable). * - `MOCK_FLUSH_ON_EOF` — if set, on stdin EOF the agent takes an async beat - * (simulating the real acp-agent's EOF-driven - * quiesce+flush), then touches this path and exits ON ITS - * OWN — no signal. Stands in for a child whose durable - * flush completes only if dispose gives EOF a real window - * before escalating to SIGTERM. + * (MOCK_FLUSH_DELAY_MS, default 150) simulating the real + * acp-agent's EOF-driven quiesce+flush, then touches this + * path and exits ON ITS OWN — no signal. Stands in for a + * child whose durable flush completes only if dispose + * gives EOF a real window before escalating to SIGTERM. * - `MOCK_IGNORE_EOF` — if `1`, keep the event loop alive past stdin EOF (a bare - * timer) but leave SIGTERM at its DEFAULT handler, so the - * child ignores the graceful EOF window yet still dies on - * SIGTERM — exercising dispose's middle tier (exit during - * the SIGTERM grace, before the SIGKILL escalation). It - * touches MOCK_READY_FILE once the keepalive is armed. + * timer) but install a SIGTERM handler that exits (and, if + * MOCK_SIGTERM_FILE is set, touches it as an observable + * proof the SIGTERM rung fired). The child ignores the + * graceful EOF window yet dies cooperatively on SIGTERM — + * exercising dispose's middle tier (exit during the SIGTERM + * grace, before the SIGKILL escalation). Touches + * MOCK_READY_FILE once armed. * * It is NOT a test spec (no `describe`/`it`) — it is spawned BY the specs as the * child process the ACP backend drives. Kept as a `.ts` run under tsx by the @@ -177,26 +179,36 @@ if (process.env.MOCK_TRAP_SIGTERM === '1') { // Under MOCK_FLUSH_ON_EOF, model the real acp-agent's EOF-driven quiesce: on // stdin 'end' (the dispose path's `child.stdin.end()`), take an ASYNC beat to -// "flush", then touch the marker and exit ON OUR OWN — no signal involved. A -// dispose that sends SIGTERM in the same tick as the EOF (no graceful window) -// default-terminates this process before the beat completes, so the marker is -// missing; a dispose that waits for natural exit first lets the flush land. +// "flush", then touch the marker and exit ON OUR OWN — no signal involved. The +// beat is MOCK_FLUSH_DELAY_MS (default 150). A dispose that sends SIGTERM before +// the beat completes (no graceful window, or an EOF grace shorter than the +// flush) default-terminates this process and the marker is missing; a dispose +// that gives the EOF quiesce enough window first lets the flush land. if (FLUSH_ON_EOF !== undefined) { + const flushDelayMs = Number(process.env.MOCK_FLUSH_DELAY_MS ?? '150') process.stdin.on('end', () => { setTimeout(() => { writeFileSync(FLUSH_ON_EOF, 'flushed') process.exit(0) - }, 150) + }, flushDelayMs) }) } -// Under MOCK_IGNORE_EOF, keep the loop alive past stdin EOF but leave SIGTERM at -// its DEFAULT handler — the child ignores the graceful EOF window yet still dies -// on SIGTERM, exercising dispose's middle tier (exit during the SIGTERM grace, -// before the SIGKILL escalation). Touch the ready file once the keepalive is -// armed, so a test disposes on that condition rather than a timeout. +// Under MOCK_IGNORE_EOF, keep the loop alive past stdin EOF (so the graceful EOF +// window times out) but INSTALL A SIGTERM HANDLER that records it and exits — the +// child ignores the graceful EOF window yet dies cooperatively on SIGTERM, +// exercising dispose's MIDDLE tier (exit during the SIGTERM grace, before the +// SIGKILL escalation). When MOCK_SIGTERM_FILE is set the handler touches it, an +// OBSERVABLE proof that the SIGTERM rung fired: if dispose skipped the middle +// rung and jumped EOF→SIGKILL, SIGKILL is uncatchable so the handler never runs +// and the marker is missing. Touch READY_FILE once armed (a test waits on it). if (process.env.MOCK_IGNORE_EOF === '1') { - setInterval(() => { /* stay alive past EOF; default SIGTERM still kills us */ }, 1000) + const sigtermFile = process.env.MOCK_SIGTERM_FILE + process.on('SIGTERM', () => { + if (sigtermFile !== undefined) writeFileSync(sigtermFile, 'sigterm') + process.exit(0) + }) + setInterval(() => { /* stay alive past EOF until SIGTERM */ }, 1000) if (READY_FILE !== undefined) writeFileSync(READY_FILE, 'ignore-eof-armed') } diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index 96b749218d..926319ad87 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -199,6 +199,10 @@ describe('dsh-subagent-acp', () => { cwd: process.cwd(), permission: 'reject', env: { MOCK_TRAP_SIGTERM: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, TSX_TSCONFIG_PATH: repoTsconfig }, + // Short on BOTH tiers: the trap ignores EOF and SIGTERM, so dispose must + // burn the EOF window, then the SIGTERM window, then SIGKILL — keep each + // small so the whole ladder finishes well within the 4000ms bound. + disposeEofGraceMs: 150, disposeGraceMs: 150, } const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) @@ -218,13 +222,16 @@ describe('dsh-subagent-acp', () => { } }) - it('dispose gives the child an EOF window to quiesce before escalating (graceful flush)', async () => { + it('dispose gives the child an EOF window that outlasts the SIGTERM grace (graceful flush)', async () => { // The real acp-agent flushes ASYNCHRONOUSLY on stdin EOF (its bridge tears // down on connection close, NOT on a signal) — and it has no SIGTERM handler. - // The mock models that: on stdin 'end' it takes a beat to "flush", touches a - // marker, and exits on its own. dispose() must end stdin and WAIT for that - // natural exit before sending SIGTERM; a same-tick SIGTERM default-kills the - // child mid-flush and the marker never appears. + // Its EOF teardown can itself await a signal-trapping grandchild (a bash + // subprocess in its own SIGTERM→SIGKILL grace) plus a flush, so the EOF window + // must be a SEPARATE, WIDER grace than the SIGTERM tier — not the same value. + // The mock models a flush that takes LONGER than the SIGTERM grace but well + // under the EOF grace: it lands only because tier 1 waits eofGraceMs, not + // graceMs. (If dispose reused the small SIGTERM grace for the EOF wait — the + // round-2 bug — SIGTERM would fire mid-flush and the marker would be missing.) const tmp = mkdtempSync(join(tmpdir(), 'acp-eof-')) const ready = join(tmp, 'ready') const flushed = join(tmp, 'flushed') @@ -235,16 +242,23 @@ describe('dsh-subagent-acp', () => { cwd: process.cwd(), permission: 'reject', // MOCK_HANG so the prompt never resolves on its own — we tear down a live - // child. MOCK_FLUSH_ON_EOF is the marker the child writes iff its EOF - // quiesce was allowed to finish. - env: { MOCK_HANG: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, MOCK_FLUSH_ON_EOF: flushed, TSX_TSCONFIG_PATH: repoTsconfig }, + // child. The flush beat (400ms) outlasts the 50ms SIGTERM grace but fits + // the 2000ms EOF grace; the marker lands iff the EOF tier honored its own + // wider grace. + env: { + MOCK_HANG: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, + MOCK_FLUSH_ON_EOF: flushed, MOCK_FLUSH_DELAY_MS: '400', TSX_TSCONFIG_PATH: repoTsconfig, + }, + disposeEofGraceMs: 2000, + disposeGraceMs: 50, } const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) // Wait until the child is fully booted with its prompt in flight (its ACP // stdin reader is attached), so dispose's stdin EOF reaches a live child. await waitForFile(ready) await run.dispose() - // dispose returned via the natural-exit tier — the EOF-driven flush landed. + // dispose returned via the natural-exit tier — the EOF-driven flush landed + // despite taking longer than the SIGTERM grace. expect(existsSync(flushed)).toBe(true) } finally { rmSync(tmp, { recursive: true, force: true }) @@ -253,27 +267,38 @@ describe('dsh-subagent-acp', () => { it('escalates to SIGTERM for a child that ignores EOF but is not SIGTERM-trapping', async () => { // A child that keeps its loop alive past stdin EOF (so the graceful window - // times out) but leaves SIGTERM at the default handler must die on the - // SIGTERM tier — dispose returns there, never reaching the SIGKILL tier. + // times out) but exits cooperatively on SIGTERM must die on the SIGTERM tier + // — dispose returns there, never reaching the SIGKILL tier. The child touches + // a SIGTERM marker from its signal handler: SIGKILL is uncatchable, so if + // dispose had skipped the middle rung (EOF→SIGKILL) the handler would never + // run and the marker would be absent — making this a GENUINE middle-tier guard. const tmp = mkdtempSync(join(tmpdir(), 'acp-ignore-eof-')) const ready = join(tmp, 'ready') + const sigterm = join(tmp, 'sigterm') try { const spec: AcpRunSpec = { command: process.execPath, args: ['--import', tsxLoader, mockServer], cwd: process.cwd(), permission: 'reject', - env: { MOCK_HANG: '1', MOCK_IGNORE_EOF: '1', MOCK_TEXT: 'x', MOCK_READY_FILE: ready, TSX_TSCONFIG_PATH: repoTsconfig }, - disposeGraceMs: 150, + env: { + MOCK_HANG: '1', MOCK_IGNORE_EOF: '1', MOCK_TEXT: 'x', + MOCK_READY_FILE: ready, MOCK_SIGTERM_FILE: sigterm, TSX_TSCONFIG_PATH: repoTsconfig, + }, + // Tiny EOF grace so the ignored-EOF window elapses fast, then SIGTERM. + disposeEofGraceMs: 150, + disposeGraceMs: 2000, } const run = startAcpRun({ prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, spec) await waitForFile(ready) - // Bound it: a regression (no SIGTERM tier, only EOF + SIGKILL) would still - // pass, but a hang would fail loud rather than stall the suite. + // Bound it so a hang fails loud rather than stalling the suite. await expect(Promise.race([ run.dispose(), - new Promise((_r, reject) => { setTimeout(() => { reject(new Error('dispose did not return')) }, 4000) }), + new Promise((_r, reject) => { setTimeout(() => { reject(new Error('dispose did not return')) }, 5000) }), ])).resolves.toBeUndefined() + // The child caught SIGTERM and exited — proof the middle rung fired (not a + // jump straight to the uncatchable SIGKILL). + expect(existsSync(sigterm)).toBe(true) } finally { rmSync(tmp, { recursive: true, force: true }) } From 083af62785fea35fd16b21965fcf453dcd3d36f1 Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 22 Jun 2026 16:57:38 +0800 Subject: [PATCH 5/5] Settle ACP cancel without the child's cooperation; preserve flattened errors (review feedback) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two findings on the ACP backend: Blocking: cancel() only sent session/cancel, so a child that ignores the notify or wedges the prompt left result hung forever — the model-facing tool awaits result before its finally disposes, so the parent cancellation hung and the child stayed alive, violating the SubagentRun.cancel() contract (result settles aborted). The result path now races the ACP drive against a cancelSettled promise that requestCancel resolves, so result settles aborted the instant a cancel is requested, regardless of the child. dispose() still kills+reaps the process. New MOCK_IGNORE_CANCEL mock mode (receives cancel, never resolves the prompt, never exits) drives a regression proven to hang without the race. Nit: the drive-path catch was an empty broad catch that discarded the error (AGENTS.md forbids). Because cancellation is now handled by the race arm, a rejection reaching the catch is always a genuine child-level error — bind it, flatten to error, and surface the original via a new AcpRunSpec.onError sink that the provider wires to ctx.logger.warn, so a real fault is preserved. --- packages/subagent/subagent-acp/src/index.ts | 9 ++- packages/subagent/subagent-acp/src/run.ts | 55 +++++++++++++--- .../subagent-acp/tests/mock-acp-server.ts | 14 ++++ .../subagent-acp/tests/subagent-acp.spec.ts | 65 ++++++++++++++++++- 4 files changed, 132 insertions(+), 11 deletions(-) diff --git a/packages/subagent/subagent-acp/src/index.ts b/packages/subagent/subagent-acp/src/index.ts index 66f254b831..037d32889e 100644 --- a/packages/subagent/subagent-acp/src/index.ts +++ b/packages/subagent/subagent-acp/src/index.ts @@ -71,7 +71,7 @@ export const Config: z = z.object({ class AcpProvider implements SubagentProvider { readonly capabilities: SubagentCapabilities = { outputSchema: false, depthLimit: false, toolFilter: false } - constructor(readonly name: string, private readonly config: Config) {} + constructor(readonly name: string, private readonly ctx: Context, private readonly config: Config) {} start(request: SubagentStartRequest) { const spec: AcpRunSpec = { @@ -80,11 +80,16 @@ class AcpProvider implements SubagentProvider { cwd: this.config.cwd ?? process.cwd(), permission: this.config.permission, env: this.config.env, + onError: (error, stopReason) => { + // The seam forbids `result` rejecting, so a child-level failure is + // flattened to a stop reason — preserve it here rather than losing it. + this.ctx.logger.warn(`subagent-acp "${this.name}": child run failed (${stopReason}): ${error.message}`) + }, } return startAcpRun(request, spec) } } export function apply(ctx: Context, config: Config): void { - ctx.subagents.registerProvider(new AcpProvider(config.providerName, config)) + ctx.subagents.registerProvider(new AcpProvider(config.providerName, ctx, config)) } diff --git a/packages/subagent/subagent-acp/src/run.ts b/packages/subagent/subagent-acp/src/run.ts index 8aeb136f2d..06f7a9ece8 100644 --- a/packages/subagent/subagent-acp/src/run.ts +++ b/packages/subagent/subagent-acp/src/run.ts @@ -83,6 +83,14 @@ export interface AcpRunSpec { * a test injects a small value to exercise the escalation without a long wait. */ disposeGraceMs?: number + /** + * Sink for a child-level failure that the run flattened into a stop reason + * (the seam contract forbids `result` rejecting). The driver calls this with + * the original error and the chosen stop reason so the fault is preserved + * rather than silently lost; the provider wires it to `ctx.logger.warn`. + * Optional — omitted in a unit test that asserts the stop reason directly. + */ + onError?: (error: Error, stopReason: SubagentStopReason) => void } /** @@ -160,6 +168,15 @@ export function toAcpPrompt(prompt: ContentBlock[]): AcpContentBlock[] { return blocks } +/** Normalize an unknown thrown value to an Error (the catch binding is `unknown`). */ +function toError(value: unknown): Error { + // The catch only sees rejections from the ACP SDK RPCs and the spawn `error` + // event, which are always `Error`s; the `String(value)` arm is a defensive + // fallback for a non-Error throw that the typed surfaces cannot produce. + /* v8 ignore next */ + return value instanceof Error ? value : new Error(String(value)) +} + /** Resolve once the child process exits (any code/signal); immediate if gone. */ function waitForExit(child: ChildProcess): Promise { // Already-exited fast path: dispose guards on exitCode before calling, so in @@ -264,8 +281,19 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su ) let sessionId: string | undefined + // Resolves when a cancel is requested, so `result` can settle `aborted` even + // if the child never cooperates with `session/cancel` (it ignores the notify, + // or the prompt wedges). The result path races this against the ACP drive: the + // FIRST to settle wins, so `cancel()` always honors the contract (`result` + // settles `aborted`) without waiting on a non-cooperative child. `dispose` + // still kills the process and reaps it; this only unblocks `result`. The + // executor runs synchronously, so `signalCancelSettled` is assigned before the + // Promise constructor returns (the `!` asserts the definite assignment). + let signalCancelSettled!: () => void + const cancelSettled = new Promise((resolve) => { signalCancelSettled = resolve }) const requestCancel = (): void => { flags.cancelled = true + signalCancelSettled() // Best-effort: tell the child to cancel the in-flight turn. Swallows a // rejection — the session may not exist yet, or the pipe may be gone; the // dispose path kills the process regardless. If the session has NOT been @@ -289,9 +317,14 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su return text.length > 0 ? [{ type: 'text', text }] : [] } try { - // Race the ACP drive against a spawn failure: a bad command never speaks - // ACP, so `initialize` would hang forever — the spawn `error` event is the - // only signal, and a rejected race settles the run `error` via the catch. + // Race three outcomes, first to settle wins: + // - driveAcp: the normal initialize → newSession → prompt path; + // - spawnFailed: a bad command never speaks ACP, so `initialize` would + // hang forever — the spawn `error` event is the only signal, and a + // rejected race settles the run `error` via the catch; + // - cancelSettled: a cancel was requested — settle `aborted` immediately + // rather than waiting on a child that may ignore `session/cancel` or + // wedge the prompt (the `cancel()` contract: `result` settles `aborted`). const driveAcp = async (): Promise => { await conn.initialize({ protocolVersion: PROTOCOL_VERSION, @@ -312,13 +345,19 @@ export function startAcpRun(request: SubagentStartRequest, spec: AcpRunSpec): Su return await Promise.race([ driveAcp(), spawnFailed.then((err): SubagentResult => { throw err }), + cancelSettled.then((): SubagentResult => ({ output: collectOutput(), stopReason: 'aborted' })), ]) - } catch { + } catch (error: unknown) { // The seam contract: result resolves (never rejects) on a child-level - // failure. A spawn/transport/RPC error becomes an error/aborted result — - // `aborted` if a cancel was requested (the failure is the cancellation - // surfacing as a torn pipe / rejected RPC), else a genuine `error`. - return { output: collectOutput(), stopReason: flags.cancelled ? 'aborted' : 'error' } + // failure. Cancellation is handled by the `cancelSettled` race arm above + // (it settles `aborted` the instant cancel is requested, beating any + // rejection), so a rejection that reaches HERE is always a genuine + // child-level error — the awaited ACP RPCs or the spawn-failure race + // (initialize/newSession/prompt transport/RPC errors, or ENOENT), not a + // local bug. Flatten to `error` and surface the original via onError so a + // real fault is preserved rather than silently lost. + spec.onError?.(toError(error), 'error') + return { output: collectOutput(), stopReason: 'error' } } })() diff --git a/packages/subagent/subagent-acp/tests/mock-acp-server.ts b/packages/subagent/subagent-acp/tests/mock-acp-server.ts index 74ae340bde..9cfeac1f44 100644 --- a/packages/subagent/subagent-acp/tests/mock-acp-server.ts +++ b/packages/subagent/subagent-acp/tests/mock-acp-server.ts @@ -8,6 +8,11 @@ * (`end_turn` default, or `max_tokens`/`refusal`/…). * - `MOCK_HANG` — if `1`, `prompt` never resolves on its own (it waits for * a `session/cancel`), to exercise the client's cancel path. + * - `MOCK_IGNORE_CANCEL` — if `1` (with MOCK_HANG), the agent receives + * `session/cancel` but NEVER resolves the pending prompt + * and never exits — a non-cooperative child. The backend's + * `result` must still settle `aborted` on its own and + * `dispose()` must still kill the process. * - `MOCK_PERMISSION` — if `1`, the agent calls `session/request_permission` * before answering, to exercise the client's auto-answer. * - `MOCK_READY_FILE` — if set, the path the agent touches once its `prompt` @@ -63,6 +68,7 @@ const WANT_PERMISSION = process.env.MOCK_PERMISSION === '1' const NO_ALLOW = process.env.MOCK_NO_ALLOW === '1' const THOUGHT = process.env.MOCK_THOUGHT === '1' const CRASH_ON_CANCEL = process.env.MOCK_CRASH_ON_CANCEL === '1' +const IGNORE_CANCEL = process.env.MOCK_IGNORE_CANCEL === '1' const READY_FILE = process.env.MOCK_READY_FILE const FLUSH_ON_EOF = process.env.MOCK_FLUSH_ON_EOF // When MOCK_NEWSESSION_READY/GO are set, newSession touches READY then blocks @@ -150,6 +156,14 @@ function makeAgent(conn: AgentSideConnection): Agent { // path: a transport failure after a cancel settles `aborted`). process.exit(1) } + if (IGNORE_CANCEL) { + // A NON-COOPERATIVE child: receive session/cancel but never resolve the + // pending prompt and never exit. The backend's `result` must still settle + // `aborted` on its own (the cancel-settle race), and `dispose()` must + // still kill the process — proving cancellation does not depend on the + // child cooperating. + return Promise.resolve() + } resolveCancel?.('cancelled') return Promise.resolve() }, diff --git a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts index 926319ad87..9819320ec5 100644 --- a/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts +++ b/packages/subagent/subagent-acp/tests/subagent-acp.spec.ts @@ -385,6 +385,20 @@ describe('dsh-subagent-acp', () => { }) it('resolves error (not reject) when the spawn command does not exist', async () => { + // Direct startAcpRun with NO onError sink — the catch must still flatten the + // spawn failure to `error` (the onError call is optional, covering the + // absent-sink branch). + const run = startAcpRun( + { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, + { command: '/nonexistent/acp-agent-binary', args: [], cwd: process.cwd(), permission: 'reject', env: {} }, + ) + const result = await run.result + // The seam contract: a child-level failure resolves error, never rejects. + expect(result.stopReason).toBe('error') + await run.dispose() + }) + + it('resolves error via the provider (real load path) when the command does not exist', async () => { const ctx = new Context() await ctx.plugin(SubagentService) await ctx.plugin(acp, { @@ -396,11 +410,35 @@ describe('dsh-subagent-acp', () => { }) const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) const result = await run.result - // The seam contract: a child-level failure resolves error, never rejects. expect(result.stopReason).toBe('error') await run.dispose() }) + it('reports a flattened child failure through onError (preserved, not silently lost)', async () => { + // The seam forbids `result` rejecting, so a child-level failure is flattened + // to a stop reason — onError must still surface the original error so a real + // fault is logged, not swallowed. A nonexistent command triggers the spawn + // failure path; the spy records the error + the chosen stop reason. + const errors: { message: string; stopReason: string }[] = [] + const run = startAcpRun( + { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }, + { + command: '/nonexistent/acp-agent-binary', + args: [], + cwd: process.cwd(), + permission: 'reject', + env: {}, + onError: (error, stopReason) => { errors.push({ message: error.message, stopReason }) }, + }, + ) + const result = await run.result + expect(result.stopReason).toBe('error') + expect(errors).toHaveLength(1) + expect(errors[0]!.stopReason).toBe('error') + expect(errors[0]!.message.length).toBeGreaterThan(0) + await run.dispose() + }) + it('settles aborted when the child crashes (tears the pipe) AFTER a cancel', async () => { // The child hangs, we cancel, and instead of answering the child exits hard // — the pending prompt RPC rejects. With a cancel already requested, the @@ -421,6 +459,31 @@ describe('dsh-subagent-acp', () => { } }) + it('settles aborted on cancel even when the child IGNORES session/cancel (non-cooperative)', async () => { + // The contract: run.cancel() → result settles `aborted`. A child that hangs + // its prompt AND ignores session/cancel must not wedge the parent — the + // backend's own cancel-settle path resolves `aborted` without the child's + // cooperation, and dispose() still reaps the process. + const tmp = mkdtempSync(join(tmpdir(), 'acp-ignorecancel-')) + const ready = join(tmp, 'ready') + try { + const ctx = await setup({ MOCK_TEXT: 'partial', MOCK_HANG: '1', MOCK_IGNORE_CANCEL: '1', MOCK_READY_FILE: ready }) + const run = ctx.subagents.start('acp', { prompt: [{ type: 'text', text: 'p' }], parent: fakeParent }) + await waitForFile(ready) + run.cancel('test') + // Bound it: a regression (cancel only notifies the child, which ignores it) + // would hang result forever — fail loud instead of stalling the suite. + const result = await Promise.race([ + run.result, + new Promise((_r, reject) => { setTimeout(() => { reject(new Error('result did not settle on cancel — backend waited on the child')) }, 4000) }), + ]) + expect(result.stopReason).toBe('aborted') + await run.dispose() + } finally { + rmSync(tmp, { recursive: true, force: true }) + } + }) + it('advertises no start-time capabilities (out-of-process child)', async () => { const ctx = await setup() const provider = ctx.subagents.getProvider('acp')!