diff --git a/docs/architecture.md b/docs/architecture.md index ee99ceec4f..04b7f111f6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -4,7 +4,7 @@ This document describes the phase-1 architecture of the DeepSeek Harness — the > **Microkernel approach. Everything is a plugin.** -The harness core is deliberately tiny: a handful of abstract services plus one concrete plugin (the agent loop). Every product feature — tools, hooks, compaction, sandboxing, UI, persistence, sub-agents, MCP, skills — is meant to be written as a plugin against the extension surface described here, without modifying the loop. +The harness core is deliberately tiny: a handful of abstract services plus one concrete loop plugin (`dsh-agent-loop`). Every product feature — tools, hooks, compaction, sandboxing, UI, persistence, sub-agents, MCP, skills — is meant to be written as a plugin against the extension surface described here, without modifying the loop. Requirement context: [Coding Harness MVP 需求分析][mvp-doc]. diff --git a/docs/development.md b/docs/development.md index 546be5dc86..2270aa0436 100644 --- a/docs/development.md +++ b/docs/development.md @@ -17,12 +17,12 @@ Install dependencies from the repo root: pnpm install ``` -The install also runs the root `postinstall` script, which installs lefthook from the repo dev dependency. +The install also runs the root `postinstall` script, which installs lefthook from the repo dev dependency through `scripts/install-lefthook.mjs`; the wrapper uses lefthook's reviewed `--force` mode so linked worktrees with an existing `core.hooksPath` do not fail normal `pnpm run …` commands. If hooks are missing because dependencies were restored from cache or `postinstall` was skipped, install them manually: ```sh -pnpm exec lefthook install +pnpm exec lefthook install --force ``` Run typecheck once after a fresh clone: diff --git a/docs/rfc/README.md b/docs/rfc/README.md index 9bb5448830..62d625cc84 100644 --- a/docs/rfc/README.md +++ b/docs/rfc/README.md @@ -31,6 +31,8 @@ Do NOT write one for a mechanical or local choice (a variable name, a one-file r | [Multiplex concurrent ACP sessions over one connection](proposed/2026-06-14-acp-multi-session.md) | 2026-06-14 | | [Optional Code Mode — model writes TypeScript against an SDK of all tools](proposed/2026-06-15-optional-code-mode.md) | 2026-06-15 | | [Runtime schemas for the event vocabulary (Zod vs the merge-extensible-map pattern)](proposed/2026-06-16-typed-event-schemas.md) | 2026-06-16 | +| [Agent lifecycle and ownership seams](proposed/2026-06-18-agent-lifecycle-and-ownership-seams.md) | 2026-06-18 | +| [Shared persistence write coordinator](proposed/2026-06-18-shared-persistence-write-coordinator.md) | 2026-06-18 | ## Implemented diff --git a/docs/rfc/implemented/2026-06-11-content-block-vocabulary.md b/docs/rfc/implemented/2026-06-11-content-block-vocabulary.md index 81ae260d53..49d55badac 100644 --- a/docs/rfc/implemented/2026-06-11-content-block-vocabulary.md +++ b/docs/rfc/implemented/2026-06-11-content-block-vocabulary.md @@ -12,10 +12,10 @@ The harness needs one internal language for messages that the loop, session log, Own it: messages are arrays of typed content blocks (`text`, `reasoning`, `tool-call`, `tool-result`, `image`), with the union derived from the merge-extensible `ContentBlockMap` so plugins add block types via declaration merging. The same merge-extensible-map pattern types every "stringly" field (`MessageSource`, `FinishReason`, `TurnTrigger`, `TurnEndReason`). Streaming is a raw chunk protocol; `BlockAssembler` is the single shared assembly implementation. Adapters translate to provider wire formats — mapping cost lives in adapters, where it belongs. -In-session context injection (`context/message`, `steering/message`) renders as tagged user-role envelopes (the system-reminder pattern) rather than a new role, so adapters carry zero burden. TODO(review): revisit once the DeepSeek V4 adapter exists. +In-session context injection (`context/message`, `steering/message`) renders as tagged user-role envelopes (the system-reminder pattern) rather than a new role, so adapters carry zero burden. Live-adapter review has since validated the tagged-envelope rendering against current DeepSeek behavior; a future provider-specific mismatch should be handled in that adapter rather than by adding a new role to the canonical content vocabulary. ## Consequences - Reasoning, prefill, cache hints, and multimodal content all have a home without provider contortions. -- Every adapter pays a translation cost; the streaming protocol carries a TODO(review) marker until the first real adapter validates it. +- Every adapter pays a translation cost; the first real adapters have since validated the streaming protocol, and new adapters should continue proving their provider-specific mapping in adapter-local tests. - IDs that cross package boundaries are branded (`CallId`, `SessionId`, `AgentId`) — nominal typing at zero runtime cost. diff --git a/docs/rfc/implemented/2026-06-11-microkernel-event-taxonomy.md b/docs/rfc/implemented/2026-06-11-microkernel-event-taxonomy.md index d5aa81fc25..942445a429 100644 --- a/docs/rfc/implemented/2026-06-11-microkernel-event-taxonomy.md +++ b/docs/rfc/implemented/2026-06-11-microkernel-event-taxonomy.md @@ -16,7 +16,7 @@ Pure Cordis event taxonomy. The loop's extension seams are typed events with del - **emit** (sync fire-and-forget) for notifications: turn/step boundaries, stream chunks, lifecycle, errors. - **parallel** (awaited) for the one durability checkpoint: `session/flush`. -The event vocabulary lives in interface packages (dsh-agent declares the agent/* events); `@deepseek-ai/dsh-agent-loop` is the only concrete plugin and is itself swappable — nothing outside it may depend on it. +The event vocabulary lives in interface packages (dsh-agent declares the agent/* events); `@deepseek-ai/dsh-agent-loop` is the only concrete loop plugin and is itself swappable — nothing outside it may depend on it. ## Consequences diff --git a/docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md b/docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md index f548d042f8..9b9dc8e7b6 100644 --- a/docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md +++ b/docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md @@ -3,7 +3,7 @@ Status: proposed -> **Implementation status (MVP landed):** steps 1, 2, 3, 4, 6, 7, 8 are implemented in `packages/acp` + `examples/acp-agent`. **Step 5 (the `session/request_permission` permission gate) is deferred** — the bridge ships a pass-through (tools run with the executor's full authority) marked `TODO(rfc010-permission-gate)`, and lays down only the `WeakMap` ownership seam the gate will build on. Status stays `proposed` until the gate lands. One further best-effort limitation is tracked as `TODO(rfc010-cancel-prestep)`: `session/cancel` aborts a running step and settles the RPC as `cancelled`, but a turn still queued (not yet started) when the cancel arrives may execute before the abort takes effect, pending a loop-level pre-step cancel. **Per-session `cwd` is now honored** (lifting the original "launch the server in the workspace root" restriction — see § Deferred): any absolute `cwd` is accepted and routed to the bash workdir via `session.header.cwd`, so an editor can open any project folder and N sessions can each target a different directory. +> **Implementation status (MVP landed):** steps 1, 2, 3, 4, 6, 7, 8 are implemented in `packages/acp` + `examples/acp-agent`. **Step 5 (the `session/request_permission` permission gate) is deferred** — the bridge ships a pass-through (tools run with the executor's full authority) marked `TODO(rfc010-permission-gate)`, and lays down only the `WeakMap` ownership seam the gate will build on. Status stays `proposed` until the gate lands. One further best-effort limitation is tracked as `TODO(rfc010-cancel-prestep)`: `session/cancel` aborts a running step and settles the RPC as `cancelled`, but a turn still queued (not yet started) when the cancel arrives may execute before the abort takes effect, pending a loop-level pre-step cancel. **Per-session `cwd` is now honored** (lifting the original "launch the server in the workspace root" restriction — see § Deferred): `session/new` accepts any absolute `cwd`, and `session/load` requires the request `cwd` to match the persisted session `cwd` so the editor and bash executor agree on the workspace. ## Problem @@ -24,8 +24,8 @@ The mapping between ACP and existing harness seams — each row names the seam a | ACP (client ⇄ agent) | Harness seam | Notes | |---|---|---| | `initialize` | static handler | negotiate `protocolVersion` (echo the supported version, else error); advertise text-only `promptCapabilities` and `loadSession: true`; report agent name/version | -| `session/new {cwd, mcpServers, additionalDirectories}` → `{sessionId}` | the `dsh-agent` create factory (see Dependency note + Plan) | the seam must accept `{ sessionId, meta }` so the ACP-generated `sessionId` becomes the live/persisted session id and the validated `cwd` is attached as the `SessionHeader` (today `AgentLoop.create(id)` hardcodes `${id}-session` and takes no metadata); reject a 2nd session (single-session MVP, see [ACP multi-session](2026-06-14-acp-multi-session.md)); `cwd` validated (require absolute) — any absolute cwd is honored: it becomes the session's `SessionHeader.cwd` and the default bash workdir (per-session cwd, see § Deferred → RESOLVED), so the server need not launch in the workspace; `mcpServers` ignored (no `mcpCapabilities` advertised); non-empty `additionalDirectories` rejected for the MVP (the bridge cannot yet widen bash/tool filesystem scope, so silently ignoring them would desync the client's filesystem-scope UI) | -| `session/load {sessionId, cwd, mcpServers, additionalDirectories}` | the `dsh-agent` resume factory ([session persistence](../implemented/2026-06-14-session-persistence.md) + Dependency note) | load `{ meta, events }`, seed the session, re-derive history via `deriveMessages()`, replay prior turns to the client as `session/update` per the ACP load contract; `additionalDirectories` rejected as in `session/new` | +| `session/new {cwd, mcpServers, additionalDirectories}` → `{sessionId}` | the `dsh-agent` create factory (see Dependency note + Plan) | the seam must accept `{ sessionId, meta }` so the ACP-generated `sessionId` becomes the live/persisted session id and the validated `cwd` is attached as the `SessionHeader` (today `AgentLoop.create(id)` hardcodes `${id}-session` and takes no metadata); reject a 2nd session (single-session MVP, see [ACP multi-session](2026-06-14-acp-multi-session.md)); `cwd` validated (require absolute) — any absolute cwd is honored: it becomes the session's `SessionHeader.cwd` and the default bash workdir (per-session cwd, see § Deferred → RESOLVED), so the server need not launch in the workspace; non-empty `mcpServers` and `additionalDirectories` are rejected for the MVP because silently ignoring requested servers/roots would desync the client's tool and filesystem-scope UI | +| `session/load {sessionId, cwd, mcpServers, additionalDirectories}` | the `dsh-agent` resume factory ([session persistence](../implemented/2026-06-14-session-persistence.md) + Dependency note) | load `{ meta, events }`, seed the session, re-derive history via `deriveMessages()`, replay prior turns to the client as `session/update` per the ACP load contract; `mcpServers` and `additionalDirectories` rejected as in `session/new` | | `session/prompt {prompt}` | `agent.send()` (idle) | text blocks → `TextBlock`; reject image/audio per advertised capabilities; one in-flight prompt per session | | resolve `session/prompt` → `{stopReason}` | `agent/turn-end` (extended, see Plan) | map the harness kebab `TurnEndReason` to the ACP snake_case `StopReason` wire enum: `completed`→`end_turn`, `max-tokens`→`max_tokens`, `aborted`(cancel)→`cancelled`, plus `refusal`/`max_turn_requests` when applicable; honor the batch-into-one-turn and send-not-synchronously-running settle semantics | | `session/update: agent_message_chunk` | `agent/stream-chunk` `text-delta` only | do NOT also emit on `block-end(TextBlock)` — it carries the fully-assembled block and would duplicate the streamed text | diff --git a/docs/rfc/proposed/2026-06-14-acp-multi-session.md b/docs/rfc/proposed/2026-06-14-acp-multi-session.md index 22887e38d1..388ec04880 100644 --- a/docs/rfc/proposed/2026-06-14-acp-multi-session.md +++ b/docs/rfc/proposed/2026-06-14-acp-multi-session.md @@ -3,12 +3,14 @@ Status: proposed -> **Implementation status:** the multi-session bridge (steps 1, 3, 4) and the bash task-ownership isolation are implemented in `packages/acp` + `packages/tool-bash`. **Per-session *permission* ownership is deferred** — it depends on the RFC 010 permission gate (`TODO(rfc010-permission-gate)`), which is itself deferred; the `agent→sessionId` reverse map the gate will route through is in place. Step 2's "real per-session disposer scope" is also deferred (`TODO(rfc010-agent-disposal)`): the bridge demuxes via id-keyed maps and global `ctx.on` listeners (correct and leak-free — disposal drains every session in parallel to quiescence), and a per-agent disposer seam is the follow-up. Status stays `proposed` until per-session permission ownership lands. +> **Implementation status:** the multi-session bridge (steps 1, 3, 4) and the bash task-ownership isolation are implemented in `packages/acp` + `packages/tool-bash`. **Per-session *permission* ownership is deferred** — it depends on [the ACP support permission gate](2026-06-14-acp-agent-client-protocol.md) (`TODO(rfc010-permission-gate)`), which is itself deferred; the `agent→sessionId` reverse map the gate will route through is in place. Step 2's "real per-session disposer scope" is also deferred (`TODO(rfc010-agent-disposal)`): the bridge demuxes via id-keyed maps and global `ctx.on` listeners (correct and leak-free — disposal drains every session in parallel to quiescence), and a per-agent disposer seam is the follow-up. Status stays `proposed` until per-session permission ownership lands. ## Problem [ACP support](2026-06-14-acp-agent-client-protocol.md) ships with a single active session per connection: a second `session/new` is rejected. Editors expect to run several conversations over one agent subprocess — a user opens multiple threads, or a client pre-warms sessions. The single-session guard is a deliberate MVP scope cut, not an architectural limit; this RFC lifts it. +This paragraph is historical: the multi-session bridge has landed. The remaining proposed work is per-session permission ownership plus the lifecycle seams now tracked in [agent lifecycle and ownership seams](2026-06-18-agent-lifecycle-and-ownership-seams.md). + ## Proposal The harness core already supports many agents (`AgentRegistry.list()` and `AgentLoop.create` impose no count limit), so multiplexing is a bridge-layer change in `@deepseek-ai/dsh-acp`, not a loop or core change. diff --git a/docs/rfc/proposed/2026-06-18-agent-lifecycle-and-ownership-seams.md b/docs/rfc/proposed/2026-06-18-agent-lifecycle-and-ownership-seams.md new file mode 100644 index 0000000000..ea98277d0e --- /dev/null +++ b/docs/rfc/proposed/2026-06-18-agent-lifecycle-and-ownership-seams.md @@ -0,0 +1,26 @@ +# RFC: Agent lifecycle and ownership seams + +Status: proposed + +## Problem + +Several ACP and tool-bash limitations are symptoms of the same missing seam: plugins can create or resume agents through `ctx.agents`, but they cannot own and dispose one agent independently, and long-running bash tasks carry no stable owner in the executor itself. ACP currently aborts and awaits agents on disconnect, but cannot unregister just that session's agent; `session/cancel` cannot cancel queued-but-not-yet-started work; and `tool-bash` keeps task ownership in a plugin-local `Map`, so an HMR reload can make an old task look unowned. + +## Proposal + +Add explicit lifecycle ownership to the agent factory and explicit ownership metadata to background tasks. + +1. `ctx.agents.create/resume` should return an `AgentHandle` (or add an adjacent method) that exposes the `Agent` plus an async disposer. The disposer unregisters the agent, aborts queued/running work, and resolves only when the driver loop reaches quiescence. +2. Add a queue-aware cancel primitive to the `Agent` interface. It must clear queued work that has not started, abort the current step if one exists, and make `whenIdle()` wait for the post-cancel quiescent state. ACP `session/cancel` and bridge teardown then become honest cancellation, not best-effort pre-step cancellation. +3. Move background task ownership into the bash seam. `BashExecSpec` or `BashTask` should carry a stable owner token, preferably the session id rather than the `Agent` object identity. `bash_output`/`bash_kill` then ask the executor for ownership rather than relying on a `tool-bash` instance-local map. + +## Acceptance Criteria + +- ACP disconnect/session close leaves no registered agent for that session, even when `session/load` races teardown. +- `session/cancel` before a queued prompt starts prevents that prompt from running and cannot batch the next prompt into the cancelled turn. +- A `tool-bash` HMR reload does not make an existing background task readable or killable by a different session. +- Existing non-ACP demos still work without managing handles explicitly; config-created agents remain owned by the `AgentLoop` plugin fiber. + +## Risks + +This touches public interfaces (`Agent`, `AgentFactory`, and the bash seam), so it should not be smuggled into a local ACP patch. The compatibility trap is preserving the simple synchronous `Agent.send()` ergonomics while adding a robust async lifecycle path for owners that need it. diff --git a/docs/rfc/proposed/2026-06-18-shared-persistence-write-coordinator.md b/docs/rfc/proposed/2026-06-18-shared-persistence-write-coordinator.md new file mode 100644 index 0000000000..06bbe1a93d --- /dev/null +++ b/docs/rfc/proposed/2026-06-18-shared-persistence-write-coordinator.md @@ -0,0 +1,24 @@ +# RFC: Shared persistence write coordinator + +Status: proposed + +## Problem + +`dsh-session-persistence-jsonl` and `dsh-session-persistence-sqlite` intentionally prove the same `SessionPersistence` contract over different storage media, but their write-path orchestration is now duplicated: per-session state, `session/created` adoption, backend-specific prefix reads, write-behind buffers, serialized flush chains, HMR seeding, and dispose drains. The pure seed-prefix collision and serializability guards have already moved into the seam package; the remaining orchestration is still correctness-heavy and already receives the same fixes twice. + +## Proposal + +Extract a backend-agnostic coordinator into `dsh-session-persistence`. The coordinator owns live-session adoption, buffering, cursor filtering, per-id serialization, and disposal quiescence. Concrete backends provide small hooks for durable operations: create lazy state, find/load stored prefix, append a contiguous batch, update summary, delete, and list. + +The public `SessionPersistence` service shape can stay the same. The coordinator can be an internal exported helper or protected base class used by first-party backends; third-party backends may still implement the abstract service directly if their write path is different. + +## Acceptance Criteria + +- JSONL and SQLite keep passing the existing shared `runPersistenceContract`. +- HMR/adoption/collision tests move to a shared coordinator test suite and run once for each backend through hook-driven fixtures. +- Backend-specific tests focus on storage mechanics only: JSONL path safety/fsync/sidecar behavior and SQLite schema/WAL/transaction behavior. +- A future backend does not need to copy the current `session/event` → buffer → flush orchestration. + +## Risks + +The current duplication is verbose but explicit. A coordinator must not hide storage-specific durability semantics or make unusual backends fight an inheritance hierarchy. Prefer narrow hooks and contract tests over a large framework. diff --git a/examples/acp-agent/README.md b/examples/acp-agent/README.md index 51d3f22414..6c7ae3115f 100644 --- a/examples/acp-agent/README.md +++ b/examples/acp-agent/README.md @@ -21,14 +21,14 @@ Add to your Zed `settings.json` under `agent_servers`: "agent_servers": { "DeepSeek Harness": { "command": "pnpm", - "args": ["run", "demo:acp"], + "args": ["--dir", "/path/to/deepseek-harness", "run", "demo:acp"], "env": { "DEEPSEEK_API_KEY": "sk-…" } } } } ``` -The editor sets each session's `cwd` to the project it opens; the agent's bash tools run there (see the per-session `cwd` note in `packages/acp`), so the server does not need to be launched in the workspace. +The editor sets each session's `cwd` to the project it opens; the agent's bash tools run there (see the per-session `cwd` note in `packages/acp`), so launch the server from the harness repo with `pnpm --dir …` and let ACP carry the workspace path per session. ## Snapshot tests (record-once / replay-deterministic) @@ -36,4 +36,4 @@ This example is the home of the harness's **snapshot tests** — they boot this ## MVP limitations -The bridge supports N concurrent sessions per connection, each in its own workspace `cwd` (RFC 011). Remaining limits: text-only prompts, `additionalDirectories` rejected (a session operates in its single `cwd`), and the tool-permission gate is deferred (`TODO(rfc010-permission-gate)` — tools run with the executor's full authority). See `packages/acp/README.md` for the full contract. +The bridge supports N concurrent sessions per connection, each in its own workspace `cwd` (RFC 011). Remaining limits: prompts support ACP's baseline `text` and `resource_link` blocks only, `additionalDirectories` and `mcpServers` are rejected, and the tool-permission gate is deferred (`TODO(rfc010-permission-gate)` — tools run with the executor's full authority). See `packages/acp/README.md` for the full contract. diff --git a/examples/acp-agent/start.ts b/examples/acp-agent/start.ts index f2ae6f60ba..11c2769603 100644 --- a/examples/acp-agent/start.ts +++ b/examples/acp-agent/start.ts @@ -1,4 +1,4 @@ -import { pathToFileURL } from 'node:url' +import { fileURLToPath, pathToFileURL } from 'node:url' import { Context } from 'cordis' import Loader from '@cordisjs/plugin-loader' @@ -33,6 +33,10 @@ if (snapshotMode !== 'replay') { } } +// Resolve relative cordis.yml paths from the repo root no matter where the +// editor launches this demo command. +process.chdir(fileURLToPath(new URL('../..', import.meta.url))) + const ctx = new Context() ctx.baseUrl = pathToFileURL(import.meta.dirname).href + '/' diff --git a/package.json b/package.json index b892288a0e..50319023cf 100644 --- a/package.json +++ b/package.json @@ -35,7 +35,7 @@ "demo:echo": "node --expose-internals --import tsx examples/echo-agent/start.ts", "demo:coding": "node --expose-internals --import tsx examples/coding-agent/start.ts", "demo:acp": "node --expose-internals --import tsx examples/acp-agent/start.ts", - "postinstall": "lefthook install" + "postinstall": "node scripts/install-lefthook.mjs" }, "devDependencies": { "@agentclientprotocol/sdk": "0.25.1", diff --git a/packages/AGENTS.md b/packages/AGENTS.md index e3123899c7..31f42c26c9 100644 --- a/packages/AGENTS.md +++ b/packages/AGENTS.md @@ -13,6 +13,6 @@ Naming notes: - A *service* `src/index.ts` exports the service class as `export default` + all public types; a *function/namespace plugin* `src/index.ts` exports `name`/`inject`/`Config`/`apply` as named exports and NO default (see the plugin-export-shape rule above) - `src/types.ts` contain only types — no runtime code - Tests live at package level under `tests/`, not `src/__tests__/` -- A package's README and module/JSDoc comments are part of the change: when you alter behavior (config keys, defaults, error codes, wire fields), update them in the same commit. CI runs `pnpm run doc-sync`, which typechecks fenced `ts` blocks in `packages/*/README.md` and verifies the event-taxonomy table — but it does NOT cover this file or prose drift (config keys, defaults, error codes), so those stay on the author. +- A package's README and module/JSDoc comments are part of the change: when you alter behavior (config keys, defaults, error codes, wire fields), update them in the same commit. CI runs `pnpm run doc-sync`, which typechecks fenced `ts` blocks in `packages/*/README.md`, verifies the event-taxonomy table, and checks markdown wrapping across this file too — but it does NOT catch prose drift (config keys, defaults, error codes), so those stay on the author. Read the per-package README.md for package-specific details: service API, events, extension points, TODOs. diff --git a/packages/README.md b/packages/README.md index 911cd7aefb..8d4c81286c 100644 --- a/packages/README.md +++ b/packages/README.md @@ -1,6 +1,6 @@ # Packages -Harness packages, all under the `@deepseek-ai/dsh-*` scope. Each package is a Cordis service (microkernel plugin-style): it exports a default `Service` class that gets registered via `ctx.plugin()`, declares its ctx key and events through declaration merging, and exposes extension points through `ctx.effect()`, `ctx.on()`, and `ctx.waterfall()`. +Harness packages, all under the `@deepseek-ai/dsh-*` scope. Each package is a Cordis plugin (microkernel-style): it exports either a default `Service` subclass or a functional plugin that gets registered via `ctx.plugin()`, declares its ctx key/events where applicable through declaration merging, and exposes extension points through `ctx.effect()`, `ctx.on()`, and `ctx.waterfall()`. ## Dependency graph @@ -31,7 +31,7 @@ The rule: plugins depend on interfaces, never on the concrete loop. `dsh-agent-l | `system-prompt/` | Prompt-section + tool-schema assembly registry | `ctx.systemPrompt` | | `tools/` | Tool registry + `tools/execute` waterfall | `ctx.tools` | | `agent/` | Agent interface, registry, `agent/*` event vocabulary | `ctx.agents` | -| `agent-loop/` | THE concrete plugin: `LoopAgent` + the loop driver | `ctx.agentLoop` | +| `agent-loop/` | THE concrete loop plugin: `LoopAgent` + the loop driver | `ctx.agentLoop` | | `bash/` | Abstract bash executor seam (interface + vocabulary) | `ctx.bash` | | `bash-local/` | Local-subprocess `BashExecutor` implementation | (registers `ctx.bash`) | | `tool-bash/` | Model-facing `bash`/`bash_output`/`bash_kill` tool schemas | (registers on `ctx.tools`) | diff --git a/packages/acp/README.md b/packages/acp/README.md index a47a7327c2..c514361b15 100644 --- a/packages/acp/README.md +++ b/packages/acp/README.md @@ -1,6 +1,6 @@ # @deepseek-ai/dsh-acp -The **Agent Client Protocol (ACP)** bridge: exposes the DeepSeek Harness coding agent as an ACP server over JSON-RPC stdio, so editors (Zed and other ACP clients) can drive it — streaming render, tool-call display, and resumable sessions. **N concurrent sessions per connection** (RFC 011): each maps to its own `LoopAgent`, and every event is demuxed strictly by session id so two sessions streaming at once never interleave. +The **Agent Client Protocol (ACP)** bridge: exposes the DeepSeek Harness coding agent as an ACP server over JSON-RPC stdio, so editors (Zed and other ACP clients) can drive it — streaming render, tool-call display, and resumable sessions. **N concurrent sessions per connection** (see [ACP multi-session](../../docs/rfc/proposed/2026-06-14-acp-multi-session.md)): each maps to its own `LoopAgent`, and every event is demuxed strictly by session id so two sessions streaming at once never interleave. It is a **client-driver / UI plugin**, the structured analogue of the readline `stdio-chat` plugin — NOT a loop change and NOT a [capability seam](../../docs/rfc/implemented/2026-06-13-capability-seams.md). It consumes the existing `agent/*` event taxonomy, the `dsh-agent` create/resume factory, and `dsh-session-persistence`. @@ -23,14 +23,14 @@ It is a **client-driver / UI plugin**, the structured analogue of the readline ` | ACP method | Harness seam | Notes | |---|---|---| -| `initialize` | static | negotiate `protocolVersion`; advertise text-only `promptCapabilities` and `loadSession: true` | -| `session/new` | `ctx.agents.create({ sessionId, meta:{cwd} })` | creates a new session/agent; N concurrent sessions are allowed (RFC 011), keyed by id; `cwd` must be absolute (it becomes the session's workspace — see Per-session cwd); `additionalDirectories` rejected; `mcpServers` ignored | -| `session/load` | `ctx.agents.resume(...)` | replays the persisted event log to the client as `session/update` — the USER side (`user/message` → `user_message_chunk`), assistant text/reasoning (`assistant/chunk`), and tool calls/results (`tool/call` + `tool/result`). Re-loading an already-live id is rejected; the id's load slot is reserved (`loadingIds`) BEFORE the async resume so a pipelined load of the SAME id can't leak a second agent (distinct ids load concurrently). The resumed session keeps its PERSISTED header `cwd`, so its bash tools run in the original workspace; the requested `cwd` only needs to be absolute. After the async resume a `closed` re-check refuses to install a record if the bridge tore down mid-load | -| `session/prompt` | `agent.send()` | text-only; rejects image/audio and empty prompts; one in-flight prompt PER session (independent); settles on the OWNING turn's end (a turn that ends in `error` rejects the RPC) | +| `initialize` | static | negotiate `protocolVersion`; advertise baseline prompt capabilities (`text`, plus `resource_link` rendered as text) and `loadSession: true` | +| `session/new` | `ctx.agents.create({ sessionId, meta:{cwd} })` | creates a new session/agent; N concurrent sessions are allowed, keyed by id; `cwd` must be absolute (it becomes the session's workspace — see Per-session cwd); non-empty `additionalDirectories` and `mcpServers` rejected | +| `session/load` | `ctx.agents.resume(...)` | replays the persisted event log to the client as `session/update` — the USER side (`user/message` → `user_message_chunk`), assistant text/reasoning (`assistant/chunk`), and tool calls/results (`tool/call` + `tool/result`). Re-loading an already-live id is rejected; the id's load slot is reserved (`loadingIds`) BEFORE the async resume so a pipelined load of the SAME id can't leak a second agent (distinct ids load concurrently). The resumed session keeps its PERSISTED header `cwd`, so its bash tools run in the original workspace; the requested `cwd` must be absolute and match the persisted `cwd`. After the async resume a `closed` re-check refuses to install a record if the bridge tore down mid-load | +| `session/prompt` | `agent.send()` | supports ACP `text` and `resource_link` blocks; rejects image/audio/embedded resource and empty prompts; one in-flight prompt PER session (independent); settles on the OWNING turn's end (a turn that ends in `error` rejects the RPC) | | `session/cancel` | `agent.abort()` | aborts a running step + settles the prompt `cancelled` for ONLY that session — a cancel never touches another session's stream or prompt (see limitation below) | | `session/update` | `session/event` | `agent_message_chunk` (text-delta), `agent_thought_chunk` (reasoning-delta), `user_message_chunk` (load replay), `tool_call`/`tool_call_update` (title/kind/rawInput/content owned by the TOOL via `presentCall`/`presentResult` — see Tool-call presentation) | -## Multi-session (RFC 011) +## Multi-session The bridge multiplexes N sessions over one connection. Live sessions are held in a `Map` (forward) with a `WeakMap` reverse map so `agent/*` events — which carry only the `Agent` — demux in O(1). Every `session/event` and `agent/status` is routed strictly to its owning record, so concurrent sessions never cross-settle or interleave their `session/update` notifications. State is per session: one in-flight prompt each, `session/cancel` aborts and settles only its own agent/prompt, and disposal drains every live session in parallel to quiescence. (Per-session *permission* ownership is reserved for the deferred permission gate — `TODO(rfc010-permission-gate)`.) @@ -38,7 +38,7 @@ Background-task isolation rides on `dsh-tool-bash`: bash task ids are global and ## Per-session cwd -Each session runs in its own workspace, recorded as the session's `SessionHeader.cwd`. On `session/new` the (absolute) request `cwd` becomes that header cwd; on `session/load` the resumed session keeps its PERSISTED header cwd (the request `cwd` is only shape-checked — it does not override the stored one), and a load whose persisted session has no absolute cwd is REJECTED up front via a metadata-only `list()` check, BEFORE resume constructs an agent (else bash would silently fall back to the server's launch dir, and a post-resume reject would leak the registered agent). `dsh-tool-bash` then defaults the bash workdir to the calling agent's `session.header.cwd` (an explicit model `workdir` still wins; a relative one resolves against the session cwd; with no session cwd the executor falls back to its own config / `process.cwd()`). So the server no longer has to be launched in the workspace — an editor can open any project folder, and N sessions over one connection can each target a different directory. (`additionalDirectories` is still rejected: widening the tool/filesystem scope beyond the single cwd is a separate sandbox concern.) +Each session runs in its own workspace, recorded as the session's `SessionHeader.cwd`. On `session/new` the (absolute) request `cwd` becomes that header cwd; on `session/load` the resumed session keeps its PERSISTED header cwd and the request `cwd` must be absolute and equal to it, so the editor and bash executor agree on the workspace before an agent is constructed. A load whose persisted session has no absolute cwd is REJECTED up front via a metadata-only `list()` check, BEFORE resume constructs an agent (else bash would silently fall back to the server's launch dir, and a post-resume reject would leak the registered agent). `dsh-tool-bash` then defaults the bash workdir to the calling agent's `session.header.cwd` (an explicit model `workdir` still wins; a relative one resolves against the session cwd; with no session cwd the executor falls back to its own config / `process.cwd()`). So the server no longer has to be launched in the workspace — an editor can open any project folder, and N sessions over one connection can each target a different directory. (`additionalDirectories` is still rejected: widening the tool/filesystem scope beyond the single cwd is a separate sandbox concern.) ## Tool-call presentation @@ -65,25 +65,25 @@ Teardown reaches quiescence: for EVERY live session settle any pending prompt as ## Known limitations (tracked TODOs) -- **`TODO(rfc010-permission-gate)`** — the `tools/execute` permission gate (`session/request_permission`) is NOT implemented; tools run with the executor's full authority. The `agent→sessionId` reverse map is in place so the gate can route a permission request (which receives only `exec.agent`) back to its originating session. RFC 010/011 stay `proposed` until the gate (and per-session permission ownership) land. +- **`TODO(rfc010-permission-gate)`** — the `tools/execute` permission gate (`session/request_permission`) is NOT implemented; tools run with the executor's full authority. The `agent→sessionId` reverse map is in place so the gate can route a permission request (which receives only `exec.agent`) back to its originating session. [ACP support](../../docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md) and [ACP multi-session](../../docs/rfc/proposed/2026-06-14-acp-multi-session.md) stay `proposed` until the gate (and per-session permission ownership) land. - **`TODO(rfc010-cancel-prestep)`** — `session/cancel` (and teardown/disconnect) is honest RPC/UI cancellation plus best-effort abort: a *running* step is aborted, but a turn that is queued-but-not-yet-started (the gap before `agent.abort()` has an `AbortController` to signal) may still run to completion. This same window means disposal/disconnect can return while one short queued turn per session still runs, and a prompt accepted right after a pre-step cancel can be batched into the cancelled turn (the loop merges queued messages into one turn). A loop-level queue-aware cancel will close this; the single-in-flight-per-session rule bounds the worst case to one extra prompt per session. - **`TODO(rfc010-agent-disposal)`** — the factory (`ctx.agents.create`/`resume`) returns no per-agent disposer, so teardown aborts+drains each agent but cannot individually unregister it; on a bare client disconnect (no host dispose) the idled agents linger in `ctx.agents` until the host context disposes. A reconnect spins up a fresh context, so this strands no work; a per-agent disposal seam is the follow-up. - **`additionalDirectories`** — rejected. A session operates in its single `cwd` (see Per-session cwd); widening the tool/filesystem scope to extra roots is a separate sandbox concern, not yet implemented. ## stdout is the protocol -The JSON-RPC frames go on stdout, so this plugin MUST run in an example that loads **no stdout logger** (the console logger writes to stdout and would corrupt the frames). The guarantee is config-only — see `examples/acp-agent` (no console logger) and RFC 010 § Risks. A stderr exporter is fine for logging. +The JSON-RPC frames go on stdout, so this plugin MUST run in an example that loads **no stdout logger** (the console logger writes to stdout and would corrupt the frames). The guarantee is config-only — see `examples/acp-agent` (no console logger) and [ACP support risks](../../docs/rfc/proposed/2026-06-14-acp-agent-client-protocol.md#risks). A stderr exporter is fine for logging. ## Running -`pnpm run demo:acp` boots `examples/acp-agent` (needs `DEEPSEEK_API_KEY`). Point an ACP client at it; for Zed, add to `agent_servers`: +`pnpm --dir /path/to/deepseek-harness run demo:acp` boots `examples/acp-agent` (needs `DEEPSEEK_API_KEY`). Point an ACP client at it; for Zed, add to `agent_servers`: ```json { "agent_servers": { "DeepSeek Harness": { "command": "pnpm", - "args": ["run", "demo:acp"] + "args": ["--dir", "/path/to/deepseek-harness", "run", "demo:acp"] } } } diff --git a/packages/acp/src/codec.ts b/packages/acp/src/codec.ts index 5ef31d00dd..dd28bc5ae6 100644 --- a/packages/acp/src/codec.ts +++ b/packages/acp/src/codec.ts @@ -59,8 +59,9 @@ export function turnEndToStopReason(reason: TurnEndReason): StopReason { /** * Translate a harness {@link ContentBlock} from a prompt into ACP content for * replay, or `undefined` for block kinds the bridge does not surface to the - * client as message content. Today only `text` maps (text-only - * `promptCapabilities`); `reasoning` is surfaced via `agent_thought_chunk` + * client as message content. Today only `text` maps; `resource_link` is an + * ACP prompt-only input rendered into text by {@link acpPromptToText}; + * `reasoning` is surfaced via `agent_thought_chunk` * streaming rather than as a message block, and `tool-call`/`tool-result`/ * `image` are handled by the tool-call update path or not advertised. */ @@ -70,34 +71,38 @@ export function harnessBlockToAcpContent(block: ContentBlock): AcpContentBlock | return { type: 'text', text: block.text } // reasoning → streamed as agent_thought_chunk, not a message block // tool-call / tool-result → the tool_call / tool_call_update path - // image → not advertised (text-only promptCapabilities) + // image → not advertised default: return undefined } } /** - * Extract plain text from an ACP prompt's content blocks, concatenating every - * `text` block. Non-text blocks are ignored here; the caller rejects a prompt - * carrying image/audio per the advertised text-only capabilities BEFORE - * calling this, so dropping them here only affects `resource`/`resource_link` - * (which carry no inline text to forward in the MVP). + * Extract plain text from an ACP prompt's content blocks. Text blocks are + * concatenated verbatim; resource links become explicit textual references so + * baseline ACP clients can point at files without the bridge silently dropping + * that context. */ export function acpPromptToText(prompt: readonly AcpContentBlock[]): string { return prompt - .filter((block): block is AcpContentBlock & { type: 'text'; text: string } => block.type === 'text') - .map(block => block.text) + .flatMap((block): string[] => { + switch (block.type) { + case 'text': + return [block.text] + case 'resource_link': + return [`\n[resource_link name=${JSON.stringify(block.name)} uri=${JSON.stringify(block.uri)}]\n`] + default: + return [] + } + }) .join('') } /** - * Whether an ACP prompt contains any content the text-only bridge cannot - * accept — i.e. ANY non-`text` block (image, audio, `resource`, `resource_link`, - * …). The caller rejects such a prompt up front rather than silently dropping - * the unsupported parts: a prompt like `[text, resource_link]` carries context - * the model would otherwise never see, so running it text-only would be silent - * data loss. When richer block kinds are supported, narrow this. + * Whether an ACP prompt contains content the bridge cannot accept. Baseline ACP + * requires `text` and `resource_link`; richer inline payloads (`resource`, + * image, audio, …) are rejected rather than silently dropped. */ export function promptHasUnsupportedContent(prompt: readonly AcpContentBlock[]): boolean { - return prompt.some(block => block.type !== 'text') + return prompt.some(block => block.type !== 'text' && block.type !== 'resource_link') } diff --git a/packages/acp/src/index.ts b/packages/acp/src/index.ts index 84c1b585ed..66be7458ec 100644 --- a/packages/acp/src/index.ts +++ b/packages/acp/src/index.ts @@ -59,7 +59,7 @@ import { } from '@agentclientprotocol/sdk' import type { ContentBlock } from '@deepseek-ai/dsh-llm' import type { Agent, AgentStatus } from '@deepseek-ai/dsh-agent' -import type { SessionEvent } from '@deepseek-ai/dsh-session' +import type { SessionEvent, TurnEndReason } from '@deepseek-ai/dsh-session' import type { ToolCallKind, ToolCallPresentation, ToolRegistry, ToolResultPresentation, ToolTerminal } from '@deepseek-ai/dsh-tools' // Side-effect type import: declaration-merges `ctx.sessionPersistence` onto // Context (the bridge injects it and reads `list()` for load cwd validation). @@ -100,6 +100,10 @@ function internalError(detail: string): RequestError { return RequestError.internalError(undefined, detail) } +function sameWorkspaceCwd(left: string, right: string): boolean { + return resolvePath(left) === resolvePath(right) +} + /** Plugin config: the agent template ACP sessions are created from. */ export interface AcpConfig { /** Model name for created agents (must have a registered adapter). */ @@ -278,6 +282,18 @@ export function apply(ctx: Context, config: AcpConfig): void { inflight.resolve(reason) } + /** Apply the single ACP prompt-settlement mapping for a completed turn. */ + const settleFromTurnEnd = ( + inflight: NonNullable, + reason: TurnEndReason, + ): void => { + if (reason.kind === 'error') { + inflight.reject(internalError(`turn failed: ${reason.message}`)) + } else { + inflight.resolve(turnEndToStopReason(reason)) + } + } + // --- Stream the harness event taxonomy to ACP session/update -------------- // All content streaming AND the prompt settle flow through `session/event`, @@ -303,7 +319,7 @@ export function apply(ctx: Context, config: AcpConfig): void { streamSessionEventUpdate(rec.sessionId, event, notify, rec.presenter, { enabled: rec.terminalEnabled, cwd: session.header.cwd, - }) + }, { includeUserMessages: false }) const inflight = rec.inflight if (inflight === undefined) return if (event.type === 'turn/start') { @@ -323,12 +339,7 @@ export function apply(ctx: Context, config: AcpConfig): void { // Settle only on the OWNING turn's end. if (event.type !== 'turn/end' || inflight.turn !== event.data.turn) return rec.inflight = undefined - const reason = event.data.reason - if (reason.kind === 'error') { - inflight.reject(internalError(`turn failed: ${reason.message}`)) - } else { - inflight.resolve(turnEndToStopReason(reason)) - } + settleFromTurnEnd(inflight, event.data.reason) }) // Settle fallback: a `session/event` listener registered BEFORE ACP that @@ -369,12 +380,7 @@ export function apply(ctx: Context, config: AcpConfig): void { inflight.resolve('cancelled') return } - const reason = end.data.reason - if (reason.kind === 'error') { - inflight.reject(internalError(`turn failed: ${reason.message}`)) - } else { - inflight.resolve(turnEndToStopReason(reason)) - } + settleFromTurnEnd(inflight, end.data.reason) } // On a settle to idle/disposed, reconcile any still-pending prompt from the @@ -409,7 +415,8 @@ export function apply(ctx: Context, config: AcpConfig): void { agentInfo: { name: agentName, version: agentVersion }, agentCapabilities: { loadSession: true, - // text-only: no image/audio/embeddedContext, no mcpCapabilities + // Baseline prompt blocks only: text plus resource_link rendered as + // text. No image/audio/embeddedContext, no mcpCapabilities. promptCapabilities: { image: false, audio: false, embeddedContext: false }, }, authMethods: [], @@ -425,6 +432,7 @@ export function apply(ctx: Context, config: AcpConfig): void { newSession(params: NewSessionRequest): Promise { assertOpen() validateWorkspaceParams(params) + validateMcpServers(params) const sessionId = randomUUID() const agent = agents.create({ agentId: sessionId, @@ -443,6 +451,7 @@ export function apply(ctx: Context, config: AcpConfig): void { throw invalidParams(`session ${params.sessionId} is already loaded`) } validateWorkspaceParams(params) + validateMcpServers(params) // Reserve THIS id's load slot BEFORE the await. Without it, two pipelined // loads for the same id could both pass the guard above while the first // resume() is pending, then both install a record and leak a second @@ -463,10 +472,16 @@ export function apply(ctx: Context, config: AcpConfig): void { // (An id unknown to `list()` falls through to resume, which rejects with // the backend's not-found error.) const meta = (await sessionPersistence.list()).find(m => m.id === params.sessionId) - if (meta !== undefined && (meta.cwd === undefined || !isAbsolute(meta.cwd))) { - throw invalidParams( - `session ${params.sessionId} has no absolute persisted cwd; cannot determine its workspace (it predates per-session cwd, or was created without one)`, - ) + if (meta !== undefined) { + const persistedCwd = meta.cwd + if (persistedCwd === undefined || !isAbsolute(persistedCwd)) { + throw invalidParams( + `session ${params.sessionId} has no absolute persisted cwd; cannot determine its workspace (it predates per-session cwd, or was created without one)`, + ) + } + if (!sameWorkspaceCwd(persistedCwd, params.cwd)) { + throw invalidParams(`session ${params.sessionId} cwd mismatch: persisted ${persistedCwd}, requested ${params.cwd}`) + } } const agent = await agents.resume({ agentId: params.sessionId, @@ -528,7 +543,7 @@ export function apply(ctx: Context, config: AcpConfig): void { throw invalidParams('a prompt is already in flight for this session') } if (promptHasUnsupportedContent(params.prompt)) { - throw invalidParams('only text prompt content is supported (text-only promptCapabilities); image/audio/resource blocks are rejected rather than silently dropped') + throw invalidParams('only text and resource_link prompt content is supported; image/audio/embedded resource blocks are rejected rather than silently dropped') } const text = acpPromptToText(params.prompt) if (text.trim().length === 0) { @@ -681,13 +696,13 @@ export function agentOptions(config: AcpConfig): { model?: string; systemPrompt? /** * Validate the `cwd`/`additionalDirectories` contract shared by `session/new` * and `session/load`: `cwd` must be absolute (a relative path would be ambiguous - * as a workspace root). What the cwd is USED for differs by method, and this - * validator only enforces shape: + * as a workspace root). The persisted-cwd equality check for `session/load` + * happens after the metadata lookup; this validator only enforces request shape: * - `session/new`: the validated `cwd` becomes the session's `SessionHeader.cwd` * (via `agents.create({meta:{cwd}})`) and thus the default bash workdir. - * - `session/load`: the request `cwd` is shape-checked only; the RESUMED - * session keeps its PERSISTED `header.cwd`, which stays authoritative for the - * bash workdir — the request cwd does not override it. + * - `session/load`: the request `cwd` must be absolute AND must match the + * PERSISTED `header.cwd`, which stays authoritative for the bash workdir — + * the request cwd does not override it. * Any absolute path is accepted (the per-session cwd flows to the bash executor * — see `dsh-tool-bash`), so the server no longer has to launch in the * workspace. `additionalDirectories` must still be empty: widening the @@ -705,6 +720,12 @@ function validateWorkspaceParams(params: { cwd: string; additionalDirectories?: } } +function validateMcpServers(params: { mcpServers?: unknown[] }): void { + if (params.mcpServers !== undefined && params.mcpServers.length > 0) { + throw invalidParams('mcpServers is not supported in this MVP') + } +} + /** * Translate a single harness {@link SessionEvent} into the `session/update` * notification(s) it produces, pushing each via `notify`. Shared by live @@ -712,8 +733,9 @@ function validateWorkspaceParams(params: { cwd: string; additionalDirectories?: * identical update stream from the same event log. * * - `assistant/chunk` text-delta/reasoning-delta → message/thought chunks - * - `user/message` → `user_message_chunk` (text blocks) — so a `session/load` - * replay reconstructs the USER side of each turn, not just the agent's + * - `user/message` → `user_message_chunk` during load replay only — so a + * loaded transcript reconstructs the USER side of each turn without echoing + * a live `session/prompt` back to the client * - `tool/call` → `tool_call` (pending) * - `tool/result` → `tool_call_update` (completed/failed) * @@ -734,7 +756,9 @@ export function streamSessionEventUpdate( notify: (notification: SessionNotification) => void, presenter: Pick = nullToolPresenter, terminal: TerminalRendering = noTerminalRendering, + options: { includeUserMessages?: boolean } = {}, ): void { + const includeUserMessages = options.includeUserMessages ?? true switch (event.type) { case 'assistant/chunk': { const chunk = event.data.chunk @@ -746,9 +770,10 @@ export function streamSessionEventUpdate( return } case 'user/message': { + if (!includeUserMessages) return // Replay the user's prompt so a loaded session shows both sides of each - // turn. Only text blocks carry inline content the bridge surfaces (the - // prompt path is text-only); other block kinds produce no chunk. + // turn. Live prompt turns suppress this path to avoid duplicating what + // the client just sent. for (const block of event.data.content) { const content = harnessBlockToAcpContent(block) if (content !== undefined) { diff --git a/packages/acp/tests/bridge.spec.ts b/packages/acp/tests/bridge.spec.ts index d537f4f684..dd10a88bcb 100644 --- a/packages/acp/tests/bridge.spec.ts +++ b/packages/acp/tests/bridge.spec.ts @@ -105,19 +105,20 @@ describe('acp bridge', () => { })).rejects.toThrow(/text/) }) - it('rejects a prompt carrying a non-text block alongside text (no silent context loss)', async () => { - // A text + resource_link prompt must be rejected, not run text-only with the - // resource silently dropped — that would feed the model an incomplete prompt. - harness = await makeBridgeHarness({ storageDir, script: [] }) + it('accepts a resource_link prompt by rendering the link into the text sent to the agent', async () => { + harness = await makeBridgeHarness({ storageDir, script: [textResponse('ok')] }) await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) const { sessionId } = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [] }) - await expect(harness.client.prompt({ + const result = await harness.client.prompt({ sessionId, prompt: [ { type: 'text', text: 'fix the bug in' }, { type: 'resource_link', uri: 'file:///x.ts', name: 'x.ts' }, ], - })).rejects.toThrow(/text/) + }) + expect(result.stopReason).toBe('end_turn') + const user = harness.ctx.agents.get(sessionId)!.session.events.find(event => event.type === 'user/message') + expect(JSON.stringify(user)).toContain('resource_link') }) it('rejects a prompt for an unknown session', async () => { diff --git a/packages/acp/tests/codec.spec.ts b/packages/acp/tests/codec.spec.ts index 58c5a4bcdf..38a7a6cb41 100644 --- a/packages/acp/tests/codec.spec.ts +++ b/packages/acp/tests/codec.spec.ts @@ -39,27 +39,29 @@ describe('harnessBlockToAcpContent', () => { }) describe('acpPromptToText', () => { - it('concatenates text blocks and ignores non-text', () => { + it('concatenates text blocks and renders resource links explicitly', () => { const prompt: AcpContentBlock[] = [ { type: 'text', text: 'hello ' }, { type: 'resource_link', uri: 'file:///x', name: 'x' }, { type: 'text', text: 'world' }, ] - expect(acpPromptToText(prompt)).toBe('hello world') + expect(acpPromptToText(prompt)).toBe('hello \n[resource_link name="x" uri="file:///x"]\nworld') }) it('returns empty string for a prompt with no text blocks', () => { - expect(acpPromptToText([{ type: 'resource_link', uri: 'file:///x', name: 'x' }])).toBe('') + expect(acpPromptToText([{ type: 'image', mimeType: 'image/png', data: 'AA==' }])).toBe('') }) }) describe('promptHasUnsupportedContent', () => { - it('detects image and audio blocks', () => { + it('detects image, audio, and embedded resource blocks', () => { expect(promptHasUnsupportedContent([{ type: 'image', mimeType: 'image/png', data: 'AA==' }])).toBe(true) expect(promptHasUnsupportedContent([{ type: 'audio', mimeType: 'audio/wav', data: 'AA==' }])).toBe(true) + expect(promptHasUnsupportedContent([{ type: 'resource', resource: { uri: 'file:///x', text: 'x' } }])).toBe(true) }) - it('passes a text-only prompt', () => { + it('passes baseline text and resource_link prompt blocks', () => { expect(promptHasUnsupportedContent([{ type: 'text', text: 'hi' }])).toBe(false) + expect(promptHasUnsupportedContent([{ type: 'resource_link', uri: 'file:///x', name: 'x' }])).toBe(false) }) }) diff --git a/packages/acp/tests/edges.spec.ts b/packages/acp/tests/edges.spec.ts index 2a48ae6d8d..9484368322 100644 --- a/packages/acp/tests/edges.spec.ts +++ b/packages/acp/tests/edges.spec.ts @@ -52,4 +52,13 @@ describe('acp bridge — demux & config edges', () => { const a = await harness.client.newSession({ cwd: process.cwd(), mcpServers: [], additionalDirectories: [] }) expect(a.sessionId).toBeTruthy() }) + + it('rejects non-empty mcpServers until MCP wiring is implemented', async () => { + harness = await makeBridgeHarness({ storageDir }) + await harness.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + await expect(harness.client.newSession({ + cwd: process.cwd(), + mcpServers: [{ name: 'fs', command: 'npx', args: ['server'], env: [] }], + })).rejects.toThrow(/mcpServers/) + }) }) diff --git a/packages/acp/tests/load.spec.ts b/packages/acp/tests/load.spec.ts index a10aca4bd4..f06c707d85 100644 --- a/packages/acp/tests/load.spec.ts +++ b/packages/acp/tests/load.spec.ts @@ -158,7 +158,7 @@ describe('acp bridge — session/load replay', () => { expect(loader.ctx.agents.get(sessionId)).toBeUndefined() }) - it('loads a session whose persisted cwd differs from the launch dir (honors per-session cwd)', async () => { + it('rejects load when the requested cwd does not match the persisted session cwd', async () => { // Seed a session on disk whose header.cwd is a DIFFERENT absolute path than // the server's launch dir. The bridge must LOAD it (per-session cwd is // honored — the resumed session keeps header.cwd, and bash routes there), no @@ -174,10 +174,12 @@ describe('acp bridge — session/load replay', () => { ]) await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) - // Load succeeds even though the requested cwd is the launch dir, not otherCwd. - const res = await loader.client.loadSession({ sessionId: 'elsewhere', cwd: process.cwd(), mcpServers: [] }) + await expect(loader.client.loadSession({ sessionId: 'elsewhere', cwd: process.cwd(), mcpServers: [] })) + .rejects.toThrow(/cwd mismatch/) + expect(loader.ctx.agents.get('elsewhere')).toBeUndefined() + + const res = await loader.client.loadSession({ sessionId: 'elsewhere', cwd: `${otherCwd}/.`, mcpServers: [] }) expect(res).toBeDefined() - // The resumed session retains its ORIGINAL workspace cwd (so bash runs there). expect(loader.ctx.agents.get('elsewhere')!.session.header.cwd).toBe(otherCwd) }) @@ -188,6 +190,13 @@ describe('acp bridge — session/load replay', () => { .rejects.toThrow(/absolute/) }) + it('lets persistence reject a load for an unknown id after metadata lookup misses', async () => { + loader = await makeBridgeHarness({ storageDir, script: [] }) + await loader.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + await expect(loader.client.loadSession({ sessionId: 'missing', cwd: process.cwd(), mcpServers: [] })) + .rejects.toThrow(/Internal error/) + }) + it('rejects loading a persisted session that has NO cwd (would silently run in the launch dir)', async () => { // A legacy / externally-created session log with no header.cwd. The bridge // must reject the load rather than accept it and let bash silently fall back diff --git a/packages/acp/tests/stream-update.spec.ts b/packages/acp/tests/stream-update.spec.ts index bf73fe0629..cdba5a3bf3 100644 --- a/packages/acp/tests/stream-update.spec.ts +++ b/packages/acp/tests/stream-update.spec.ts @@ -12,6 +12,13 @@ function updatesFor(event: SessionEvent): SessionNotification['update'][] { return out } +/** Collect the updates emitted by the live prompt stream (user echo suppressed). */ +function liveUpdatesFor(event: SessionEvent): SessionNotification['update'][] { + const out: SessionNotification['update'][] = [] + streamSessionEventUpdate('s1', event, n => out.push(n.update), undefined, undefined, { includeUserMessages: false }) + return out +} + /** A tiny tool registry stub exposing just `get` for {@link ToolPresenter}. */ function registryOf(...tools: ToolDefinition[]): Pick { const map = new Map(tools.map(t => [t.name, t])) @@ -99,6 +106,13 @@ describe('streamSessionEventUpdate', () => { expect(updatesFor(evt('user/message', { content: [], source: { kind: 'user' } }))).toEqual([]) }) + it('can suppress user/message chunks for live prompt turns', () => { + expect(liveUpdatesFor(evt('user/message', { + content: [{ type: 'text', text: 'hi' }], + source: { kind: 'user' }, + }))).toEqual([]) + }) + it('produces no update for boundary/other event types', () => { expect(updatesFor(evt('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }))).toEqual([]) expect(updatesFor(evt('turn/end', { turn: 1, reason: { kind: 'completed' } }))).toEqual([]) diff --git a/packages/agent-loop/src/agent.ts b/packages/agent-loop/src/agent.ts index 8040e1b107..96a1158c66 100644 --- a/packages/agent-loop/src/agent.ts +++ b/packages/agent-loop/src/agent.ts @@ -63,7 +63,11 @@ export class LoopAgent implements Agent { // waiter (AGENTS.md "contain callback exceptions" — a lifecycle await must // not hang on one bad listener). if (status !== 'running') this.settleIdleWaiters() - this.ctx.emit('agent/status', this, status) + try { + this.ctx.emit('agent/status', this, status) + } catch (error: unknown) { + this.ctx.logger.warn(`agent "${this.id}": agent/status listener threw on ${status}: ${String(error)}`) + } } /** @@ -177,16 +181,16 @@ export class LoopAgent implements Agent { * `running`. If it is already disposed, awaits {@link done} (the loop-exit * promise) — `agent/status('disposed')` fires in the disposer BEFORE the * driver loop has unwound, so it is NOT itself a quiescence signal. If it is - * idle, resolves immediately. Otherwise queues an internal waiter (see - * {@link idleWaiters}) released on the next running→idle/disposed transition, - * resolving on `idle` directly (the turn fully ended) or chaining {@link done} - * on `disposed` (wait for the loop to actually exit). Implements the - * {@link Agent.whenIdle} contract used by teardown (`abort()` then - * `await whenIdle()`). + * idle AND has no queued work, resolves immediately. Otherwise queues an + * internal waiter (see {@link idleWaiters}) released on the next + * running→idle/disposed transition, resolving on `idle` directly (the turn + * fully ended) or chaining {@link done} on `disposed` (wait for the loop to + * actually exit). Implements the {@link Agent.whenIdle} contract used by + * teardown (`abort()` then `await whenIdle()`). */ whenIdle(): Promise { if (this._status === 'disposed') return this.done - if (this._status !== 'running') return Promise.resolve() + if (this._status !== 'running' && !this.inbox.hasQueued) return Promise.resolve() // Register an internal waiter (resolved by settleIdleWaiters on the next // running→idle/disposed transition), NOT an effect-scoped `ctx.on` listener: // a concurrent fiber disposal runs this agent's listener disposers, which diff --git a/packages/agent-loop/src/loop.ts b/packages/agent-loop/src/loop.ts index 0d1708e617..359329e062 100644 --- a/packages/agent-loop/src/loop.ts +++ b/packages/agent-loop/src/loop.ts @@ -203,8 +203,8 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: // agent/step-end emit is contained: a throwing step-end listener must not // abort finalization and strand the turn open (turn/end balance > notifying // one bad listener). Appended before the emit (the event-sourcing RFC append-before-emit). - const closeStep = (): void => { - if (!stepOpen) return + const closeStep = (): boolean => { + if (!stepOpen) return false stepOpen = false // Session.append pushes step/end BEFORE notifying session/event listeners, // so a throwing listener leaves step/end in the log (balance holds) but @@ -226,7 +226,11 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: // a throwing listener from producing a silent "completed" turn when the step // itself succeeded, AND keeps finalization going when closeStep runs from // the outer catch. - if (failure !== undefined) failTurn(toError(failure)) + if (failure !== undefined) { + failTurn(toError(failure)) + return true + } + return false } // Record a step/turn failure exactly once: append the single `error` event @@ -358,7 +362,7 @@ async function runTurn(ctx: Context, agent: LoopAgent, handle: LoopHandle, turn: // Steering that arrived during streaming/tool execution. const steered = drainSteering(ctx, agent, turn) - closeStep() + if (closeStep()) break const defaultDecision = stepOutcome.hadToolCalls || steered let shouldContinue: boolean @@ -497,6 +501,18 @@ async function runStep( const stepError = finishError(assembler.finish) if (stepError) throw stepError + if (assembler.finish.kind === 'max-tokens') { + let message: Message = withoutToolCalls(assembler.message()) + message = withoutToolCalls(await ctx.waterfall('agent/step-result', agent, turn, step, message, () => Promise.resolve(message))) + if (message.content.length > 0) { + session.append('assistant/message', { turn, step, content: message.content }) + } + if (assembler.usage) { + session.append('usage', { turn, step, usage: assembler.usage }) + } + return { hadToolCalls: false, finish: assembler.finish } + } + // The step-result waterfall runs BEFORE the session append so the log (the // source of truth for derived history and replay) records the message that // tool dispatch actually uses. @@ -544,8 +560,6 @@ async function runStep( }) // signal CAN flip during the await above (abort() inside a tool); // the analyzer can't see through the await boundary. - // signal can flip during the await above (abort() inside a tool); - // the analyzer can't see through the await boundary. /* v8 ignore start -- signal.reason default unreachable via agent.abort() */ // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition if (signal.aborted) throw new Error(String(signal.reason ?? 'aborted')) @@ -555,6 +569,10 @@ async function runStep( return { hadToolCalls: toolCalls.length > 0, finish: assembler.finish } } +function withoutToolCalls(message: Message): Message { + return { ...message, content: message.content.filter(block => block.type !== 'tool-call') } +} + /** The last turn number in a (possibly seeded) session log, or 0. */ export function lastTurnNumber(session: Session): number { const lastStart = session.events.findLast(event => event.type === 'turn/start') diff --git a/packages/agent-loop/tests/agent.spec.ts b/packages/agent-loop/tests/agent.spec.ts index 4e2cd6c046..4fd7237061 100644 --- a/packages/agent-loop/tests/agent.spec.ts +++ b/packages/agent-loop/tests/agent.spec.ts @@ -32,6 +32,17 @@ function waitForIdle(ctx: Context, agent: LoopAgent): Promise { }) } +function waitForStatus(ctx: Context, agent: LoopAgent, expected: LoopAgent['status']): Promise { + return new Promise((resolve) => { + const dispose = ctx.on('agent/status', (subject, status) => { + if (subject === agent && status === expected) { + dispose() + resolve() + } + }) + }) +} + function send(agent: LoopAgent, text: string) { agent.send([{ type: 'text', text }]) } @@ -265,6 +276,24 @@ describe('LoopAgent', () => { expect(agent.status).not.toBe('running') }) + it('whenIdle() waits for queued work that has not flipped status yet', async () => { + const adapter = new MockAdapter(['hang']) + const ctx = await harness(adapter) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + + send(agent, 'queued') + let settled = false + const idle = agent.whenIdle().then(() => { settled = true }) + await Promise.resolve() + expect(settled).toBe(false) + + await waitForStatus(ctx, agent, 'running') + agent.abort('done') + await idle + expect(settled).toBe(true) + expect(agent.status).toBe('idle') + }) + it('whenIdle() awaits the running→idle transition, ignoring other subjects/running events', async () => { const adapter = new MockAdapter([textResponse('ok'), textResponse('ok')]) const ctx = await harness(adapter) @@ -366,6 +395,42 @@ describe('LoopAgent', () => { expect(doneResolved).toBe(true) }) + it('contains a throwing agent/status listener on the running transition', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + ctx.on('agent/status', (_subject, status) => { + if (status === 'running') throw new Error('bad running listener') + }) + + send(agent, 'go') + await agent.whenIdle() + + expect(adapter.requests).toHaveLength(1) + expect(agent.status).toBe('idle') + expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/status listener threw on running')) + warn.mockRestore() + }) + + it('contains a throwing agent/status listener on the idle transition', async () => { + const adapter = new MockAdapter([textResponse('ok')]) + const ctx = await harness(adapter) + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + ctx.on('agent/status', (_subject, status) => { + if (status === 'idle') throw new Error('bad idle listener') + }) + + send(agent, 'go') + await agent.whenIdle() + + expect(adapter.requests).toHaveLength(1) + expect(agent.status).toBe('idle') + expect(warn).toHaveBeenCalledWith(expect.stringContaining('agent/status listener threw on idle')) + warn.mockRestore() + }) + it('abort() resolves reason to "aborted" when no reason provided', async () => { const adapter = new MockAdapter(['hang']) const ctx = await harness(adapter) diff --git a/packages/agent-loop/tests/loop.spec.ts b/packages/agent-loop/tests/loop.spec.ts index 4d8cdefe6c..c4f09a24bf 100644 --- a/packages/agent-loop/tests/loop.spec.ts +++ b/packages/agent-loop/tests/loop.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' import { Context } from 'cordis' -import LlmService, { StreamChunk } from '@deepseek-ai/dsh-llm' +import LlmService, { CallId, StreamChunk } from '@deepseek-ai/dsh-llm' import SessionStore, { TurnEndReason } from '@deepseek-ai/dsh-session' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineTool } from '@deepseek-ai/dsh-tools' @@ -385,6 +385,10 @@ describe('agent loop', () => { expect(steps).toBe(2) expect(adapter.requests).toHaveLength(2) + expect(adapter.requests[1]!.messages).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'go' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'first half' }] }, + ]) expect(reasons).toEqual([{ kind: 'max-tokens' }]) }) @@ -406,6 +410,98 @@ describe('agent loop', () => { expect(reasons).toEqual([{ kind: 'max-tokens' }, { kind: 'completed' }]) }) + it('does not dispatch tool calls from a max-tokens-truncated step', async () => { + const callId = CallId('c1') + const adapter = new MockAdapter([[ + { type: 'block-start', index: 0, blockType: 'tool-call' }, + { type: 'tool-call-delta', index: 0, id: callId, name: 'echo', argumentsDelta: '{"text":"x"}' }, + { type: 'block-end', index: 0, block: { type: 'tool-call', id: callId, name: 'echo', arguments: '{"text":"x"}' } }, + { type: 'usage', usage: { inputTokens: 10, outputTokens: 5 } }, + { type: 'finish', reason: { kind: 'max-tokens' } }, + ]]) + const ctx = await harness(adapter) + let executions = 0 + ctx.tools.register(defineTool({ + name: 'echo', + description: '', + parameters: { text: { type: 'string' } }, + async execute() { + executions += 1 + return [{ type: 'text', text: 'should not run' }] + }, + })) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + + const reasons: TurnEndReason[] = [] + ctx.on('agent/turn-end', (_agent, _turn, reason) => void reasons.push(reason)) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + expect(executions).toBe(0) + expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false) + expect(agent.session.deriveMessages()).toEqual([{ role: 'user', content: [{ type: 'text', text: 'go' }] }]) + expect(reasons).toEqual([{ kind: 'max-tokens' }]) + }) + + it('keeps safe max-tokens assistant content while dropping truncated tool calls', async () => { + const callId = CallId('c1') + const adapter = new MockAdapter([[ + { type: 'block-start', index: 0, blockType: 'text' }, + { type: 'text-delta', index: 0, text: 'partial text' }, + { type: 'block-end', index: 0, block: { type: 'text', text: 'partial text' } }, + { type: 'block-start', index: 1, blockType: 'tool-call' }, + { type: 'tool-call-delta', index: 1, id: callId, name: 'echo', argumentsDelta: '{"text"' }, + { type: 'finish', reason: { kind: 'max-tokens' } }, + ]]) + const ctx = await harness(adapter) + let stepResults = 0 + ctx.on('agent/step-result', async (_agent, _turn, _step, message, next) => { + stepResults += 1 + expect(message.content).toEqual([{ type: 'text', text: 'partial text' }]) + return next() + }) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + expect(stepResults).toBe(1) + expect(agent.session.events.some(e => e.type === 'tool/call')).toBe(false) + expect(agent.session.deriveMessages()).toEqual([ + { role: 'user', content: [{ type: 'text', text: 'go' }] }, + { role: 'assistant', content: [{ type: 'text', text: 'partial text' }] }, + ]) + }) + + it('stops the turn when agent/step-end listener failure has recorded an error', async () => { + const adapter = new MockAdapter([ + toolCallResponse('c1', 'echo', { text: 'x' }), + textResponse('should not run'), + ]) + const ctx = await harness(adapter) + ctx.tools.register(defineTool({ + name: 'echo', + description: '', + parameters: { text: { type: 'string' } }, + async execute(args) { + return [{ type: 'text', text: String(args.text) }] + }, + })) + const agent = ctx.agentLoop.create('a1', { model: 'mock' }) + let threw = false + ctx.on('agent/step-end', () => { + if (!threw) { threw = true; throw new Error('bad step-end listener') } + }) + + send(agent, 'go') + await waitForIdle(ctx, agent) + + expect(adapter.requests).toHaveLength(1) + const turnEnd = agent.session.events.findLast(e => e.type === 'turn/end') + expect(turnEnd?.type === 'turn/end' && turnEnd.data.reason.kind).toBe('error') + }) + it('chains queued messages into consecutive turns', async () => { const adapter = new MockAdapter([textResponse('first'), textResponse('second')]) const ctx = await harness(adapter) diff --git a/packages/agent/src/index.ts b/packages/agent/src/index.ts index 5d43f87c25..c9181081a3 100644 --- a/packages/agent/src/index.ts +++ b/packages/agent/src/index.ts @@ -71,6 +71,9 @@ export interface AgentFactory { resume(options: ResumeAgentOptions): Promise } +/** Thrown when create/resume is called before an agent factory is registered. */ +const NO_FACTORY_MESSAGE = 'no agent factory registered (load an agent-loop plugin)' + /** * Agent registry (`ctx.agents`): tracks live agents so UI, hook, and * orchestrator plugins can find them without depending on the concrete loop @@ -107,7 +110,7 @@ export class AgentRegistry extends Service { * registered. */ create(options: CreateAgentOptions): Agent { - if (this.factory === undefined) throw new Error('no agent factory registered (load an agent-loop plugin)') + if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE) return this.factory.createAgent(options) } @@ -117,7 +120,7 @@ export class AgentRegistry extends Service { * session persistence is not configured. */ async resume(options: ResumeAgentOptions): Promise { - if (this.factory === undefined) throw new Error('no agent factory registered (load an agent-loop plugin)') + if (this.factory === undefined) throw new Error(NO_FACTORY_MESSAGE) return this.factory.resume(options) } diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index 86e35bcf54..56805a4a23 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -72,8 +72,9 @@ export interface Agent { * (inject is synchronous): a failing flush is reported via `agent/error` * (step `0`) and the logger, never thrown into the caller. * - * TODO(review): exact envelope/rendering rules live in dsh-session and need - * review once a real adapter exists. + * Live-adapter review has validated the tagged-envelope rendering against + * current DeepSeek behavior; provider-specific mismatches belong in that + * adapter, not in the canonical session vocabulary. */ inject(content: ContentBlock[], options?: SendOptions): void @@ -82,11 +83,12 @@ export interface Agent { /** * Resolve once the agent has reached quiescence after settling out of - * `running`, or immediately if it is already idle. The quiescence signal a - * teardown awaits: `agent.abort()` then `await agent.whenIdle()` guarantees - * the in-flight turn has fully stopped before the caller proceeds (a closing - * ACP connection, a disposing UI plugin), rather than returning while the - * driver is still streaming. + * `running`, or immediately if it is already idle with no queued work. The + * quiescence signal a teardown awaits: `agent.abort()` then + * `await agent.whenIdle()` guarantees queued/running work has fully stopped + * before the caller proceeds (a closing ACP connection, a disposing UI + * plugin), rather than returning while the driver is still streaming or about + * to start a queued turn. * * "Quiescence", not merely "status changed": a disposed agent emits * `agent/status('disposed')` from inside its disposer, BEFORE the driver loop diff --git a/packages/bash-local/README.md b/packages/bash-local/README.md index fe585492d8..d8062bc6ff 100644 --- a/packages/bash-local/README.md +++ b/packages/bash-local/README.md @@ -20,7 +20,7 @@ Design surveyed against the bash tools of Claude Code, OpenCode, Codex, and pi; - **Spawn per call, no shell state** — every call is a fresh non-login `bash -c` (deterministic; no rc files). All four surveyed tools spawn per call. `XXX(stateful-shell)` in `src/run.ts` records the two proven stateful designs (Claude Code's cwd-only persistence; Codex's PTY exec sessions) for when real workflows demand them. - **Process-group kills with escalation** — children are spawned `detached` (own process group); kills send SIGTERM to the group, then SIGKILL after a 3s grace (OpenCode's escalation; pipelines and subshells die with the parent). ESRCH is tolerated; daemons that re-parent away from the group can still survive — same caveat as the surveyed tools. -- **Tail-keep truncation + spill files** — output beyond `maxOutputBytes` keeps the in-memory TAIL (errors/results cluster at the end — pi/OpenCode rationale) while the FULL stream is appended to a temp file whose path is reported. The model can `grep`/`tail` the spill file with bash itself. +- **Tail-keep truncation + spill files** — output beyond `maxOutputBytes` keeps the in-memory TAIL (errors/results cluster at the end — pi/OpenCode rationale) while the FULL stream is appended to a temp file whose path is reported when available. If the final spill close reports a delayed writeback failure, the executor still returns the tail but withholds the path rather than advertising a possibly incomplete file. - **Model-friendly env** — `NO_COLOR=1 TERM=dumb PAGER=cat GIT_PAGER=cat` (Codex's hardcoded set) so pagers and ANSI color don't garble results. - **Background tasks** — `start()` returns immediately, no timeout applies (Claude Code detaches timeouts when backgrounding), `readOutput()` is incremental with whole-stream byte offsets, and disposal kills everything. diff --git a/packages/bash-local/src/index.ts b/packages/bash-local/src/index.ts index 1169160614..7320276a1a 100644 --- a/packages/bash-local/src/index.ts +++ b/packages/bash-local/src/index.ts @@ -38,6 +38,12 @@ export interface Config { /** The shape after schemastery applied the defaults (cwd has none). */ type ResolvedConfig = Required> & Pick +function assertPositiveFinite(name: string, value: number): void { + if (!Number.isFinite(value) || value <= 0) { + throw new Error(`bash-local: ${name} must be a positive finite number`) + } +} + interface TrackedTask extends BashTask { running: RunningBash /** Whole-stream byte offsets already delivered via {@link LocalBashExecutor.readOutput}. */ @@ -72,6 +78,9 @@ export class LocalBashExecutor extends BashExecutor { // schemastery (static Config) has already filled the defaulted fields; // the cast records that runtime fact for exactOptionalPropertyTypes. this.config = config as ResolvedConfig + assertPositiveFinite('timeoutMs', this.config.timeoutMs) + assertPositiveFinite('maxTimeoutMs', this.config.maxTimeoutMs) + assertPositiveFinite('maxOutputBytes', this.config.maxOutputBytes) ctx.effect(() => async () => { // Kill every live process group and WAIT for the processes to close so // nothing outlives the fiber (HMR safety) — a TERM-trapping child is @@ -98,6 +107,7 @@ export class LocalBashExecutor extends BashExecutor { * values and never re-default. */ resolve(request: BashExecRequest): BashExecSpec { + if (request.timeoutMs !== undefined) assertPositiveFinite('request.timeoutMs', request.timeoutMs) const timeoutMs = Math.min(request.timeoutMs ?? this.config.timeoutMs, this.config.maxTimeoutMs) return { command: request.command, diff --git a/packages/bash-local/src/run.ts b/packages/bash-local/src/run.ts index d78ef815c6..46d48cf5fd 100644 --- a/packages/bash-local/src/run.ts +++ b/packages/bash-local/src/run.ts @@ -195,7 +195,15 @@ export class OutputCollector { /** Close the spill file (if any) and return the final output. */ finalize(): CollectedOutput { if (this.spillFd !== undefined) { - closeSync(this.spillFd) + try { + closeSync(this.spillFd) + } catch { + // close can surface delayed writeback failures (for example EIO/ENOSPC) + // after writeSync appeared to succeed. Keep finalize total so runBash's + // close handler still resolves, but stop advertising a spill file that + // may be missing its tail. + this.spillFile = undefined + } this.spillFd = undefined } return this.snapshot() diff --git a/packages/bash-local/tests/executor.spec.ts b/packages/bash-local/tests/executor.spec.ts index 72383d0427..3dd7f7983a 100644 --- a/packages/bash-local/tests/executor.spec.ts +++ b/packages/bash-local/tests/executor.spec.ts @@ -59,6 +59,16 @@ describe('LocalBashExecutor.run', () => { expect(result.timeoutMs).toBe(2_000) }) + it('rejects invalid numeric config and timeout overrides', async () => { + await expect(setup({ timeoutMs: Number.NaN })).rejects.toThrow(/timeoutMs/) + await expect(setup({ maxTimeoutMs: 0 })).rejects.toThrow(/maxTimeoutMs/) + await expect(setup({ maxOutputBytes: -1 })).rejects.toThrow(/maxOutputBytes/) + + const { bash } = await setup() + expect(() => bash.resolve({ command: 'true', timeoutMs: Number.NaN })).toThrow(/request\.timeoutMs/) + expect(() => bash.resolve({ command: 'true', timeoutMs: -1 })).toThrow(/request\.timeoutMs/) + }) + it('per-call timeout takes precedence under the cap and kills on expiry', async () => { const { bash } = await setup({ timeoutMs: 60_000 }) const result = await bash.run(bash.resolve({ command: 'sleep 60', timeoutMs: 100 })) diff --git a/packages/bash-local/tests/run.spec.ts b/packages/bash-local/tests/run.spec.ts index 7f17ba1a94..ff1d92b519 100644 --- a/packages/bash-local/tests/run.spec.ts +++ b/packages/bash-local/tests/run.spec.ts @@ -4,6 +4,21 @@ import { dirname, join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { killGroup, OutputCollector, runBash } from '@deepseek-ai/dsh-bash-local' +const { failNextClose } = vi.hoisted(() => ({ failNextClose: { value: false } })) +vi.mock('node:fs', async (importOriginal) => { + const actual = await importOriginal() + return { + ...actual, + closeSync(fd: number): void { + if (failNextClose.value) { + failNextClose.value = false + throw Object.assign(new Error('simulated EIO on close'), { code: 'EIO' }) + } + actual.closeSync(fd) + }, + } +}) + const spillDir = mkdtempSync(join(tmpdir(), 'dsh-bash-spec-')) function spec(command: string, overrides: Partial[0]> = {}) { @@ -160,6 +175,19 @@ describe('output truncation and spill', () => { expect(result.stdout.text.length).toBe(500) expect(result.stdout.spillPath).toBeUndefined() }) + + it('settles with the tail and no spill path when final spill close fails', async () => { + failNextClose.value = true + const result = await runBash( + spec('for i in $(seq 1 200); do printf "line-%04d\\n" $i; done', { maxOutputBytes: 500 }), + { spillDir }, + ).done + expect(failNextClose.value).toBe(false) + expect(result.exitCode).toBe(0) + expect(result.stdout.truncated).toBe(true) + expect(result.stdout.text).toContain('line-0200') + expect(result.stdout.spillPath).toBeUndefined() + }) }) describe('OutputCollector', () => { @@ -200,6 +228,22 @@ describe('OutputCollector', () => { expect(collector.totalBytes).toBe(8) expect(collector.finalize().text).toBe('bbbb') }) + + it('contains close failures and drops the spill path', () => { + const collector = new OutputCollector(4, 'closefail', spillDir) + collector.push(Buffer.from('aaaa')) + collector.push(Buffer.from('bbbb')) + expect(collector.snapshot().spillPath).toBeDefined() + + failNextClose.value = true + let out: ReturnType + expect(() => { out = collector.finalize() }).not.toThrow() + + expect(failNextClose.value).toBe(false) + expect(out!.text).toBe('bbbb') + expect(out!.truncated).toBe(true) + expect(out!.spillPath).toBeUndefined() + }) }) describe('killGroup', () => { diff --git a/packages/bash/README.md b/packages/bash/README.md index 12976d551b..8de529fee1 100644 --- a/packages/bash/README.md +++ b/packages/bash/README.md @@ -27,4 +27,4 @@ Implementations subclass `BashExecutor`, implement the abstract methods, and cal ## Vocabulary -`BashExecSpec` (command, workdir?, timeoutMs?, signal?) → `BashRunResult` (exitCode, signal, timedOut, aborted, timeoutMs, stdout/stderr as `CollectedOutput`) and `BashTask`/`BashTaskRead` for the background side. See `src/types.ts` for the full contracts. +`BashExecRequest` (command, workdir?, timeoutMs?, signal?) resolves to `BashExecSpec` (command, workdir, timeoutMs, signal?) before execution; `run()` returns `BashRunResult` (exitCode, signal, timedOut, aborted, timeoutMs, stdout/stderr as `CollectedOutput`) and `start()`/`readOutput()` use `BashTask`/`BashTaskRead` for the background side. See `src/types.ts` for the full contracts. diff --git a/packages/bash/src/types.ts b/packages/bash/src/types.ts index 1860401ee2..cfa60ab331 100644 --- a/packages/bash/src/types.ts +++ b/packages/bash/src/types.ts @@ -44,7 +44,7 @@ export interface CollectedOutput { text: string /** True when bytes were dropped from `text`. */ truncated: boolean - /** Path to a file holding the COMPLETE stream, when truncated. */ + /** Path to a file holding the COMPLETE stream, when truncated and available. */ spillPath?: string } @@ -87,9 +87,9 @@ export interface BashTaskRead { delta: string /** True when truncation dropped unread bytes the delta cannot include. */ lossy: boolean - /** Full stdout spill file, when stdout truncation occurred. */ + /** Full stdout spill file, when stdout truncation occurred and a safe path is available. */ stdoutSpillPath?: string - /** Full stderr spill file, when stderr truncation occurred. */ + /** Full stderr spill file, when stderr truncation occurred and a safe path is available. */ stderrSpillPath?: string } diff --git a/packages/invariants/src/index.ts b/packages/invariants/src/index.ts index 1e5ac64315..b201e1ae43 100644 --- a/packages/invariants/src/index.ts +++ b/packages/invariants/src/index.ts @@ -57,6 +57,10 @@ interface SessionTrace { openTurn: number | null /** Open step within the current turn, or null between steps. */ openStep: number | null + /** The next turn number expected in this session log. */ + nextTurn: number + /** The next step number expected within the open turn. */ + nextStep: number /** * Tool-call ids issued in the OPEN step awaiting a result. Cleared at * `step/end` — a result must arrive in the same step as its call. @@ -114,7 +118,14 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { if (trace.openTurn !== null) { throw new InvariantError(`turn/start ${event.data.turn} while turn ${trace.openTurn} is still open`) } + // Current sessions replay full logs, so numbering starts at 1 and remains + // contiguous. If a future compaction/fork stores a partial log, it must + // seed `nextTurn` from retained metadata before this check runs. + if (event.data.turn !== trace.nextTurn) { + throw new InvariantError(`turn/start expected turn ${trace.nextTurn}, got ${event.data.turn}`) + } trace.openTurn = event.data.turn + trace.nextStep = 1 break } case 'turn/end': { @@ -125,6 +136,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { throw new InvariantError(`turn/end ${event.data.turn} while step ${trace.openStep} is still open`) } trace.openTurn = null + trace.nextTurn += 1 break } case 'step/start': { @@ -134,6 +146,10 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { if (trace.openStep !== null) { throw new InvariantError(`step/start ${event.data.step} while step ${trace.openStep} is still open`) } + // Steps are checked under the same full-log assumption as turns above. + if (event.data.step !== trace.nextStep) { + throw new InvariantError(`step/start expected step ${trace.nextStep} in turn ${event.data.turn}, got ${event.data.step}`) + } trace.openStep = event.data.step break } @@ -143,6 +159,7 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // (a step that errored before its result) do not carry to the next step. trace.pendingCalls.clear() trace.openStep = null + trace.nextStep += 1 break } case 'assistant/chunk': { @@ -163,7 +180,8 @@ function checkEvent(trace: SessionTrace, event: SessionEvent): void { // A result needs a prior matching call in the same step. (The converse // does NOT hold: a call may have no result — a throwing tools/execute // waterfall ends the step with no tool/result, which is legal.) - if (!trace.pendingCalls.delete(event.data.callId)) { + const syntheticInterrupted = event.data.isError && event.data.error?.code === 'interrupted' + if (!trace.pendingCalls.delete(event.data.callId) && !syntheticInterrupted) { throw new InvariantError(`tool/result for ${event.data.callId} with no prior tool/call in this step`) } break @@ -216,7 +234,14 @@ export function apply(ctx: Context, config: Config = {}): void { // (re-)apply seeds the baseline, so a reload never produces a false positive. const lastStatus = new WeakMap() - const freshTrace = (): SessionTrace => ({ lastSeq: -1, openTurn: null, openStep: null, pendingCalls: new Set() }) + const freshTrace = (): SessionTrace => ({ + lastSeq: -1, + openTurn: null, + openStep: null, + nextTurn: 1, + nextStep: 1, + pendingCalls: new Set(), + }) /** Build (or rebuild) a session's trace by replaying its whole log; freeze it. */ const seedSession = (session: Session): SessionTrace => { diff --git a/packages/invariants/tests/invariants.spec.ts b/packages/invariants/tests/invariants.spec.ts index 8df1629710..1086a21bfb 100644 --- a/packages/invariants/tests/invariants.spec.ts +++ b/packages/invariants/tests/invariants.spec.ts @@ -126,6 +126,28 @@ describe('session-log invariants', () => { .toThrow(/no prior tool\/call/) }) + it('allows a synthetic interrupted tool/result from crash repair without a prior tool/call event', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + expect(() => { + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('assistant/message', { turn: 1, step: 1, content: [ + { type: 'tool-call', id: CallId('crashed'), name: 'bash', arguments: '{}' }, + ] }) + session.append('tool/result', { + turn: 1, + step: 1, + callId: CallId('crashed'), + content: [{ type: 'text', text: 'interrupted' }], + isError: true, + error: { name: 'InterruptedError', code: 'interrupted' }, + }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'interrupted' } }) + }).not.toThrow() + }) + it('allows a tool/call with no matching tool/result (thrown waterfall ends the step)', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() @@ -177,6 +199,25 @@ describe('session-log invariants', () => { }).not.toThrow() }) + it('rejects a skipped turn number', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + expect(() => session.append('turn/start', { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } })) + .toThrow(/expected turn 2, got 3/) + }) + + it('rejects a skipped step number within a turn', async () => { + const { ctx } = await setup({ freeze: false }) + const session = ctx.sessions.create() + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + session.append('step/end', { turn: 1, step: 1 }) + expect(() => session.append('step/start', { turn: 1, step: 3 })) + .toThrow(/expected step 2 in turn 1, got 3/) + }) + it('rejects a turn/end while a step is still open', async () => { const { ctx } = await setup({ freeze: false }) const session = ctx.sessions.create() diff --git a/packages/llm-deepseek/tests/adapter.e2e.ts b/packages/llm-deepseek/tests/adapter.e2e.ts index f37484fd3c..ebec6e62ce 100644 --- a/packages/llm-deepseek/tests/adapter.e2e.ts +++ b/packages/llm-deepseek/tests/adapter.e2e.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { afterEach, describe, expect, it } from 'vitest' import { Context } from 'cordis' import LlmService, { CallId } from '@deepseek-ai/dsh-llm' import type { GenerateResult, Message, ToolSchema } from '@deepseek-ai/dsh-llm' @@ -13,14 +13,20 @@ import type { Config } from '@deepseek-ai/dsh-llm-deepseek' const FLASH = 'deepseek-v4-flash' const PRO = 'deepseek-v4-pro' +const contexts: Context[] = [] async function harness(model: string, config: Partial = {}) { const ctx = new Context() + contexts.push(ctx) await ctx.plugin(LlmService) await ctx.plugin(LlmDeepSeek, { models: [model], ...config }) return ctx } +afterEach(async () => { + await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose())) +}) + function ask(text: string): Message[] { return [{ role: 'user', content: [{ type: 'text', text }] }] } diff --git a/packages/llm-pi-ai/README.md b/packages/llm-pi-ai/README.md index ae9dc93301..70f61fdb62 100644 --- a/packages/llm-pi-ai/README.md +++ b/packages/llm-pi-ai/README.md @@ -6,10 +6,10 @@ DeepSeek adapter for the harness LLM seam backed by [`@earendil-works/pi-ai`](ht `@deepseek-ai/dsh-llm-deepseek` already talks to the same endpoint. This package is its **design-verification twin**: same models, same wire protocol, completely different internals — a unified LLM library with its own event vocabulary versus hand-rolled fetch/SSE. Anything the harness `StreamChunk` protocol cannot express for BOTH implementations is a core-vocabulary bug. The differences it exercised on purpose: -- pi-ai hands back tool-call `arguments` as **parsed objects**; the harness keeps raw JSON strings (re-stringified at `block-end`). +- pi-ai hands tool-call `arguments` around as **parsed objects**; the harness keeps raw JSON strings. The adapter patches replay payloads back to the original raw strings before sending them, and re-stringifies parsed output tool calls at `block-end`. - pi-ai reports failures as **in-stream error events** (it never throws mid-stream); these map to `finish {kind:'error'|'aborted'}` chunks — the protocol's other sanctioned error path besides throwing (which llm-deepseek uses). - pi-ai folds reasoning tokens into `usage.output`; there is no separate reasoning count to map. -- pi-ai's options omit stop sequences; `GenerateOptions.stop` is injected via its `onPayload` hook. +- pi-ai's options omit some DeepSeek/OpenAI-compatible details; the adapter uses its `onPayload` hook to preserve the harness contract (`stop`, per-tool `strict`, omitted reasoning effort, raw replayed tool arguments). ## Config diff --git a/packages/llm-pi-ai/src/adapter.ts b/packages/llm-pi-ai/src/adapter.ts index e9eac80295..38b05dc007 100644 --- a/packages/llm-pi-ai/src/adapter.ts +++ b/packages/llm-pi-ai/src/adapter.ts @@ -14,7 +14,7 @@ import { stream as piStream } from '@earendil-works/pi-ai' import type { Model } from '@earendil-works/pi-ai' import { LlmAdapter, LlmError } from '@deepseek-ai/dsh-llm' -import type { GenerateOptions, StreamChunk } from '@deepseek-ai/dsh-llm' +import type { GenerateOptions, StreamChunk, ToolSchema } from '@deepseek-ai/dsh-llm' import { toPiContext, toStreamChunks } from './convert.ts' /** Reasoning levels surfaced by this adapter (DeepSeek wire: high|max). */ @@ -59,12 +59,79 @@ export function buildModel(modelId: string, options: PiAiAdapterOptions): Model< } } +type Payload = { + tools?: { function?: { name?: unknown; strict?: unknown } }[] + messages?: { + role?: unknown + tool_calls?: { id?: unknown; function?: { arguments?: unknown } }[] + }[] + reasoning_effort?: unknown + stop?: unknown +} + +function rawToolArguments(options: GenerateOptions): Map { + const raw = new Map() + for (const message of options.messages) { + if (message.role !== 'assistant') continue + for (const block of message.content) { + if (block.type === 'tool-call') raw.set(block.id, block.arguments) + } + } + return raw +} + +function strictByToolName(tools: ToolSchema[] | undefined): Map { + return new Map((tools ?? []).map(tool => [tool.name, tool.strict])) +} + +function patchPayload(payload: unknown, options: GenerateOptions, reasoning: PiAiReasoning | undefined): unknown { + /* v8 ignore next -- pi-ai onPayload always receives an object; tolerate unusual future hooks defensively */ + if (typeof payload !== 'object' || payload === null) return payload + const body = payload as Payload + + if (reasoning === undefined) { + delete body.reasoning_effort + } + if (options.stop !== undefined) { + body.stop = options.stop + } + + const strictByName = strictByToolName(options.tools) + for (const tool of body.tools ?? []) { + /* v8 ignore next -- malformed pi-ai payload guard: real tool entries always carry function */ + if (tool.function === undefined) continue + const name = tool.function.name + /* v8 ignore next -- malformed pi-ai payload guard: real function entries always carry a string name */ + if (typeof name !== 'string') continue + const strict = strictByName.get(name) + if (strict === undefined) delete tool.function.strict + else tool.function.strict = strict + } + + const rawById = rawToolArguments(options) + /* v8 ignore next -- defensive for non-chat payloads; OpenAI chat payloads always carry messages */ + for (const message of body.messages ?? []) { + if (message.role !== 'assistant') continue + /* v8 ignore next -- assistant messages without tool_calls need no raw-argument patch */ + for (const call of message.tool_calls ?? []) { + /* v8 ignore next -- malformed pi-ai payload guard: real tool calls always carry a string id */ + if (typeof call.id !== 'string') continue + const raw = rawById.get(call.id) + /* v8 ignore next -- pi-ai always emits a function object for assistant tool_calls; guard malformed payloads defensively */ + if (raw !== undefined && call.function !== undefined) call.function.arguments = raw + } + } + + return body +} + /** * pi-ai-backed adapter. One instance serves every registered model name. * * Implementation notes: - * - `GenerateOptions.stop` is injected via pi-ai's `onPayload` hook (its - * public options omit stop sequences). + * - `onPayload` patches provider payload details pi-ai cannot express directly: + * stop sequences, per-tool strict, omitted reasoning effort, and raw replayed + * tool-call arguments. * - `prefill` throws UNSUPPORTED (same contract as dsh-llm-deepseek). * - pi-ai reports request failures as in-stream error events; convert.ts * maps them to `finish {kind:'error'|'aborted'}` chunks rather than @@ -86,8 +153,9 @@ export class PiAiAdapter extends LlmAdapter { const model = buildModel(options.model, this.options) // Undefined config means "provider default" (DeepSeek: thinking ENABLED), // matching llm-deepseek's omission semantics. pi-ai derives the wire - // thinking toggle from whether reasoningEffort is passed, so undefined - // maps to 'high' here; only an explicit 'off' disables thinking. + // thinking toggle from whether reasoningEffort is passed, so undefined maps + // internally to 'high' to get `thinking: enabled`; patchPayload then removes + // `reasoning_effort` so the provider chooses its default effort. const reasoning = this.options.reasoning ?? 'high' // pi-ai's event stream has no iterator-return cancellation hook: if our @@ -106,13 +174,7 @@ export class PiAiAdapter extends LlmAdapter { ...options.maxTokens !== undefined ? { maxTokens: options.maxTokens } : {}, signal: controller.signal, ...reasoning !== 'off' ? { reasoningEffort: reasoning } : {}, - ...options.stop !== undefined ? { - // pi-ai's options omit stop sequences; inject them into the raw body. - onPayload: (payload: unknown) => { - (payload as Record).stop = options.stop - return payload - }, - } : {}, + onPayload: payload => patchPayload(payload, options, this.options.reasoning), maxRetries: 0, }) diff --git a/packages/llm-pi-ai/src/convert.ts b/packages/llm-pi-ai/src/convert.ts index 065a31b89e..4610ff01c9 100644 --- a/packages/llm-pi-ai/src/convert.ts +++ b/packages/llm-pi-ai/src/convert.ts @@ -7,7 +7,8 @@ * exists — an independent implementation stress-tests the StreamChunk * protocol): * - pi-ai tool-call `arguments` are PARSED OBJECTS; the harness keeps the - * raw JSON string. We parse on the way in and re-stringify on the way out. + * raw JSON string. We parse on the way into pi-ai, patch provider payloads + * back to the original raw string in the adapter, and re-stringify on output. * - pi-ai reports errors as in-stream `error` events (it never throws * mid-stream); the harness expresses those as `finish {kind:'error'}` / * `{kind:'aborted'}` chunks. @@ -17,7 +18,7 @@ * @module dsh-llm-pi-ai/convert */ -import { CallId } from '@deepseek-ai/dsh-llm' +import { CallId, LlmError } from '@deepseek-ai/dsh-llm' import type { FinishReason, GenerateOptions, Message, StreamChunk, TokenUsage } from '@deepseek-ai/dsh-llm' import type { AssistantMessage, @@ -168,6 +169,14 @@ export function mapUsage(usage: PiUsage): TokenUsage { } } +function classifyPiAiError(message: string): string { + if (/\b(?:401|403)\b/.test(message)) return 'AUTH' + if (/\b429\b|rate.?limit/i.test(message)) return 'RATE_LIMIT' + if (/\b400\b|invalid.?request/i.test(message)) return 'INVALID_REQUEST' + if (/\b5\d\d\b/.test(message)) return 'SERVER' + return 'PI_AI_ERROR' +} + /** Map a terminal pi-ai event to the harness finish reason. */ export function mapStopReason(message: AssistantMessage): FinishReason { switch (message.stopReason) { @@ -175,10 +184,9 @@ export function mapStopReason(message: AssistantMessage): FinishReason { case 'length': return { kind: 'max-tokens' } case 'toolUse': return { kind: 'tool-calls' } case 'aborted': return { kind: 'aborted' } - case 'error': return { - kind: 'error', - message: message.errorMessage ?? 'pi-ai stream error', - code: 'PI_AI_ERROR', + case 'error': { + const text = message.errorMessage ?? 'pi-ai stream error' + return { kind: 'error', message: text, code: classifyPiAiError(text) } } } } @@ -264,4 +272,5 @@ export async function* toStreamChunks(events: AsyncIterable = {}) { const ctx = new Context() + contexts.push(ctx) await ctx.plugin(LlmService) await ctx.plugin(LlmPiAi, { models: [model], ...config }) return ctx } +afterEach(async () => { + await Promise.all(contexts.splice(0).map(ctx => ctx.fiber.dispose())) +}) + function ask(text: string): Message[] { return [{ role: 'user', content: [{ type: 'text', text }] }] } @@ -114,6 +120,7 @@ describe.skipIf(!process.env.DEEPSEEK_API_KEY)('llm-pi-ai e2e (real API)', () => // same block KINDS in the same order for a deterministic prompt — the // cross-implementation check that the StreamChunk design holds. const deepseekCtx = new Context() + contexts.push(deepseekCtx) await deepseekCtx.plugin(LlmService) await deepseekCtx.plugin(LlmDeepSeek, { models: [FLASH], thinking: 'disabled' }) diff --git a/packages/llm-pi-ai/tests/adapter.spec.ts b/packages/llm-pi-ai/tests/adapter.spec.ts index 6466084554..97b20f4617 100644 --- a/packages/llm-pi-ai/tests/adapter.spec.ts +++ b/packages/llm-pi-ai/tests/adapter.spec.ts @@ -148,6 +148,44 @@ describe('PiAiAdapter against a mock server', () => { expect(server.requests[0]).toMatchObject({ stop: ['END'] }) }) + it('preserves per-tool strict exactly through onPayload', async () => { + const server = await mockServer([{ events: textEvents }]) + const ctx = await harness(server.url) + await ctx.llm.generate({ + model: 'deepseek-v4-flash', + messages: [], + tools: [ + { name: 'strict_true', description: 'true', parameters: {}, strict: true }, + { name: 'strict_false', description: 'false', parameters: {}, strict: false }, + { name: 'strict_omitted', description: 'omitted', parameters: {} }, + ], + }) + + const request = server.requests[0] as { tools: { function: { name: string; strict?: boolean } }[] } + expect(request.tools.map(tool => [tool.function.name, tool.function.strict])).toEqual([ + ['strict_true', true], + ['strict_false', false], + ['strict_omitted', undefined], + ]) + expect('strict' in request.tools[2]!.function).toBe(false) + }) + + it('preserves raw replayed tool-call arguments in the provider payload', async () => { + const server = await mockServer([{ events: textEvents }]) + const ctx = await harness(server.url) + await ctx.llm.generate({ + model: 'deepseek-v4-flash', + messages: [{ + role: 'assistant', + content: [{ type: 'tool-call', id: CallId('broken'), name: 'f', arguments: '{broken' }], + }], + }) + + const request = server.requests[0] as { messages: { role: string; tool_calls?: { id: string; function: { arguments: string } }[] }[] } + const assistant = request.messages.find(message => message.role === 'assistant') + expect(assistant?.tool_calls?.[0]?.function.arguments).toBe('{broken') + }) + it('maps HTTP errors to error finish chunks (pi-ai in-stream style)', async () => { const server = await mockServer([{ status: 401, @@ -155,10 +193,21 @@ describe('PiAiAdapter against a mock server', () => { }]) const ctx = await harness(server.url) const result = await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }) - expect(result.finish.kind).toBe('error') + expect(result.finish).toMatchObject({ kind: 'error', code: 'AUTH' }) expect((result.finish as { message: string }).message).toMatch(/bad key|401/) }) + it.each([ + [400, 'INVALID_REQUEST'], + [429, 'RATE_LIMIT'], + [500, 'SERVER'], + ] as const)('maps HTTP %s to stable error code %s', async (status, code) => { + const server = await mockServer([{ status, body: JSON.stringify({ error: { message: `provider ${status}` } }) }]) + const ctx = await harness(server.url) + const result = await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }) + expect(result.finish).toMatchObject({ kind: 'error', code }) + }) + it('rejects prefill with UNSUPPORTED', async () => { const ctx = await harness('http://127.0.0.1:1') await expect(ctx.llm.generate({ @@ -263,10 +312,9 @@ describe('review fixes', () => { const server = await mockServer([{ events: textEvents }]) const ctx = await harness(server.url) // no reasoning key at all await ctx.llm.generate({ model: 'deepseek-v4-flash', messages: [] }) - expect(server.requests[0]).toMatchObject({ - thinking: { type: 'enabled' }, - reasoning_effort: 'high', - }) + const request = server.requests[0] as Record + expect(request.thinking).toEqual({ type: 'enabled' }) + expect('reasoning_effort' in request).toBe(false) }) it('replays reasoning_content on assistant tool-call turns (passback rule)', async () => { diff --git a/packages/llm-pi-ai/tests/convert.spec.ts b/packages/llm-pi-ai/tests/convert.spec.ts index a4394b1e1b..be42c9e9b4 100644 --- a/packages/llm-pi-ai/tests/convert.spec.ts +++ b/packages/llm-pi-ai/tests/convert.spec.ts @@ -271,6 +271,11 @@ describe('toStreamChunks', () => { const chunks = await collect(toStreamChunks(feed({ type: 'error', reason: 'aborted', error }))) expect(chunks.at(-1)).toEqual({ type: 'finish', reason: { kind: 'aborted' } }) }) + + it('rejects a stream that ends without done or error', async () => { + await expect(collect(toStreamChunks(feed({ type: 'start', partial: assistant() })))) + .rejects.toThrow(/without done\/error/) + }) }) describe('mapStopReason / mapUsage', () => { @@ -288,6 +293,15 @@ describe('mapStopReason / mapUsage', () => { .toEqual({ kind: 'error', message: 'pi-ai stream error', code: 'PI_AI_ERROR' }) }) + it('maps routable HTTP-ish error messages to stable codes', () => { + expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 401: bad key' }))) + .toMatchObject({ kind: 'error', code: 'AUTH' }) + expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 429: rate limit' }))) + .toMatchObject({ kind: 'error', code: 'RATE_LIMIT' }) + expect(mapStopReason(assistant({ stopReason: 'error', errorMessage: 'HTTP 500: backend down' }))) + .toMatchObject({ kind: 'error', code: 'SERVER' }) + }) + it('maps cache fields only when nonzero', () => { expect(mapUsage(usage(10, 5, 8, 2))).toEqual({ inputTokens: 10, diff --git a/packages/session-persistence-jsonl/src/index.ts b/packages/session-persistence-jsonl/src/index.ts index a73a75cade..7a0c97637c 100644 --- a/packages/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence-jsonl/src/index.ts @@ -8,8 +8,7 @@ * line, verbatim including `assistant/chunk` so `seq` stays contiguous) plus * a small atomic `.summary.json` sidecar for the mutable `SessionSummary`. * Lazy materialization (no file until the first `append`), atomic first - * write, and truncation-repair of a never-committed crash tail on the first - * `append` after a `load`. + * write, and load-time repair of a never-committed crash tail. * * 2. **The write path** — the `session/event` → buffer → `session/flush` drain * that generalizes the example `session-jsonl.ts`: snapshot each event when @@ -24,10 +23,12 @@ import { Context } from 'cordis' import z from 'schemastery' import { open, mkdir, readFile, readdir, rename, link, rm, truncate } from 'node:fs/promises' -import { resolve } from 'node:path' +import { dirname, resolve } from 'node:path' import { randomBytes } from 'node:crypto' -import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' -import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session' +import { + SessionPersistence, assertSerializable, seedCoversPrefix, +} from '@deepseek-ai/dsh-session-persistence' +import { interruptedTurnClosers } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' import { encodeSegment, eventLine, logPath, parseHeaderMeta, scanLog, sessionDir, sidecarPath, toHeaderLine, @@ -60,44 +61,6 @@ interface SessionState { owner?: Session } -/** - * Whether a live session's `seed` reproduces a persisted `prefix` exactly — the - * prefix is no longer than the seed, and each prefix event DEEP-equals the seed - * event at the same index. Used to tell a session legitimately continuing a - * persisted log (HMR re-seeing its own session, or a resume) from a different - * session that merely reuses the id: the latter would have its already-counted - * seq 0..prefix-1 events filtered out on flush and its conversation silently - * grafted onto the old log. - * - * The comparison is a full structural equality (via canonical JSON) of each - * event INCLUDING its `data` payload, not just `seq`/`type`/`time` — a session - * built from loaded events but with mutated message/tool payloads (same seq/ - * type/time) must NOT be accepted, or the live history and durable log diverge. - * Both sides are JSON-serializable by contract (Session.append enforces it), so - * JSON.stringify is a sound canonical form here. - */ -function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean { - return prefix.length <= seed.length - && prefix.every((e, i) => { - const s = seed[i] - return s !== undefined && JSON.stringify(s) === JSON.stringify(e) - }) -} - -/** - * Reject non-JSON-serializable `event.data`, naming the offending type. Used on - * the backend's `append(events)` entry point (replay/fork paths that bypass a - * live `Session`); events that flow through `Session.append` are already - * validated at the source, so the live write path never needs this. - */ -function assertSerializable(events: readonly SessionEvent[]): void { - for (const event of events) { - if (!isJsonValue(event.data)) { - throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`) - } - } -} - /** * Whether `error` is a "no such file/directory" (`ENOENT`) failure — the ONLY * filesystem error that legitimately means "this session/root is absent" for a @@ -111,6 +74,15 @@ function isENOENT(error: unknown): boolean { return (error as NodeJS.ErrnoException | null)?.code === 'ENOENT' } +async function settledErrors(promises: Iterable>): Promise { + const settled = await Promise.allSettled([...promises]) + const errors: unknown[] = [] + for (const result of settled) { + if (result.status === 'rejected') errors.push(result.reason) + } + return errors +} + /** * The JSONL persistence backend. Load as a plugin; it registers as * `ctx.sessionPersistence` and installs the write-path listeners. @@ -306,6 +278,34 @@ export class SessionPersistenceJsonl extends SessionPersistence { return { meta: fullMeta, events: balanced } } + private async adoptLiveDiskPrefix( + session: Session, + seed: readonly SessionEvent[], + file: { path: string; cwd: string | undefined }, + ): Promise { + const buffer = await readFile(file.path) + const { meta, events, committedBytes } = scanLog(buffer) + this.assertVersion(meta) + if (!seedCoversPrefix(seed, events)) { + throw new Error(`session "${session.header.id}" already has a persisted log on disk that does not match this live session (id collision)`) + } + + const summary = await this.readSidecar(session.header.id, meta.cwd) + const state: SessionState = { + meta: { ...meta, ...summary }, + cursor: events.length, + materialized: true, + owner: session, + } + this.states.set(session.header.id, state) + + if (committedBytes < buffer.byteLength) { + await this.repair(state, committedBytes) + } + const suffix = seed.slice(events.length) + if (suffix.length > 0) await this.appendCore(session.header.id, suffix) + } + async list(): Promise { const metas: SessionMeta[] = [] for (const dir of await this.listCwdDirs()) { @@ -318,7 +318,7 @@ export class SessionPersistenceJsonl extends SessionPersistence { if (first === undefined) continue // empty/half-written file const meta = parseHeaderMeta(first) if (meta === undefined) continue // not a session header - const summary = await this.readSidecar(meta.id, meta.cwd) + const summary = await this.readSidecarForList(meta.id, meta.cwd) metas.push({ ...meta, ...summary }) } } @@ -397,8 +397,8 @@ export class SessionPersistenceJsonl extends SessionPersistence { // sidecar is best-effort) — but if we mutated state.meta first, a later // touchSummary() on a successful append would persist the rejected // title/firstPrompt, making a failed update durable after the fact. - const nextMeta: SessionMeta = { ...state.meta, ...summary } - await this.writeSidecar(nextMeta) + const nextMeta: SessionMeta = { ...state.meta, ...summary, updatedAt: summary.updatedAt ?? Date.now() } + if (state.materialized) await this.writeSidecar(nextMeta) state.meta = nextMeta } @@ -407,7 +407,10 @@ export class SessionPersistenceJsonl extends SessionPersistence { /** Atomically write the header line + first batch (temp-write, fsync, rename). */ private async materialize(state: SessionState, events: readonly SessionEvent[]): Promise { const dir = sessionDir(this.root, state.meta.cwd) + await mkdir(this.root, { recursive: true, mode: 0o700 }) + await this.syncDir(dirname(this.root)) await mkdir(dir, { recursive: true, mode: 0o700 }) + await this.syncDir(this.root) const finalPath = logPath(this.root, state.meta.cwd, state.meta.id) // Never rename over an existing committed log: materialize is the FIRST // write of a session the backend believes is new. A file here means a @@ -561,17 +564,29 @@ export class SessionPersistenceJsonl extends SessionPersistence { } /** - * Read the mutable-summary sidecar, or `undefined` if it is absent/unreadable - * (a session that has never been `update()`d, or a failed sidecar write). The - * caller keeps the header-derived `updatedAt` (the session's createdAt) in - * that case rather than overlaying `0` — reporting an active session as - * updated at the Unix epoch would be wrong. + * Read the mutable-summary sidecar, or `undefined` if it is absent (a session + * that has never been `update()`d). Non-ENOENT failures surface on strict + * load/adopt paths so corrupt metadata does not masquerade as a clean default. */ private async readSidecar(id: SessionId, cwd: string | undefined): Promise { try { const raw = await readFile(sidecarPath(this.root, cwd, id), 'utf8') return JSON.parse(raw) as SessionSummary - } catch { + } catch (error) { + if (isENOENT(error)) return undefined + throw error + } + } + + /** + * Best-effort summary read for list(): a corrupt sidecar should degrade one + * row to header metadata, not hide every session from a picker. + */ + private async readSidecarForList(id: SessionId, cwd: string | undefined): Promise { + try { + return await this.readSidecar(id, cwd) + } catch (error: unknown) { + this.ctx.logger.warn(`session-persistence-jsonl: ignoring unreadable summary for session "${id}" while listing: ${String(error)}`) return undefined } } @@ -675,9 +690,14 @@ export class SessionPersistenceJsonl extends SessionPersistence { // Dispose must reach quiescence: await every session's init + final drain // BEFORE returning, so no write lands after teardown (orphan rename/ENOENT). ctx.effect(() => async () => { - await Promise.allSettled([...this.inits.values()]) - await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s))) - await Promise.allSettled([...this.chains.values()]) + const errors = [ + ...await settledErrors(this.inits.values()), + ...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))), + ...await settledErrors(this.chains.values()), + ] + if (errors.length > 0) { + throw new AggregateError(errors, 'session-persistence-jsonl dispose failed') + } }, 'session-persistence-jsonl write path') // HMR: a hot reload does not replay session/created, so seed existing live @@ -797,28 +817,11 @@ export class SessionPersistenceJsonl extends SessionPersistence { const onDisk = await this.findLog(id, session.header.cwd) if (onDisk !== undefined) { - // Read the committed on-disk events and check they are a seq-aligned - // prefix of the live session (HMR re-seeing its own session) vs. an - // unrelated session colliding on the id. - const { events: diskEvents } = scanLog(await readFile(onDisk.path)) - if (!seedCoversPrefix(seed, diskEvents)) { - // case 3: genuine collision — fail loudly rather than clobber. - throw new Error(`session "${id}" already has a persisted log on disk that does not match this live session (id collision)`) - } - // case 2: adopt. loadCore sets the state (cursor = committed length, - // repair offset if a crash tail exists). - await this.serialize(id, () => this.loadCore(id)) - const adopted = this.states.get(id) - /* v8 ignore next -- loadCore always sets the state for the id */ - if (adopted !== undefined) adopted.owner = session - // Persist the live SUFFIX beyond the on-disk prefix. These events live - // ONLY in `seed` (the live session was ahead of disk — mid-turn at - // reload, or events appended while the previous backend was disposed); - // this backend never buffered them via session/event, so without this - // they would be lost and the next flush (starting at a later seq) would - // mismatch or skip them. - const suffix = seed.slice(diskEvents.length) - if (suffix.length > 0) await this.append(id, suffix) + // case 2: adopt a LIVE prefix. Do NOT route through loadCore(): loadCore + // crash-repairs open turns as interrupted, which is right for a true load + // after a crash but wrong for HMR while the live Session is still the + // authority and may append the real step/turn end later. + await this.serialize(id, () => this.adoptLiveDiskPrefix(session, seed, onDisk)) return } diff --git a/packages/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence-jsonl/tests/jsonl.spec.ts index 563dd8c627..8257a9853c 100644 --- a/packages/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence-jsonl/tests/jsonl.spec.ts @@ -1,6 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' -import { mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises' +import { appendFile, mkdtemp, mkdir, rm, readFile, writeFile, readdir, stat } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import SessionStore, { SessionId } from '@deepseek-ai/dsh-session' @@ -644,25 +644,40 @@ describe('SessionPersistenceJsonl: edge cases', () => { expect(loaded.meta.title).toBeUndefined() }) - it('delete removes the sidecar of a lazy session that has no log', async () => { - // update() before the first append() writes a .summary.json sidecar but no - // .jsonl log (lazy create). delete() must still remove that sidecar. - const m = meta('lazy-del', '/a') + it('update before the first append keeps summary in memory and writes no orphan sidecar', async () => { + const m = meta('lazy-update', '/a') await ctx.sessionPersistence.create(m) await ctx.sessionPersistence.update(m.id, { title: 'secret', firstPrompt: 'sensitive' }) const sidecar = sidecarPath(root, '/a', m.id) - expect((await stat(sidecar)).isFile()).toBe(true) // sidecar exists, no log - await expect(stat(logPath(root, '/a', m.id))).rejects.toThrow() // no log - await ctx.sessionPersistence.delete(m.id) - await expect(stat(sidecar)).rejects.toThrow() // sidecar gone + await expect(stat(sidecar)).rejects.toThrow() + await expect(stat(logPath(root, '/a', m.id))).rejects.toThrow() + + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + const loaded = await ctx.sessionPersistence.load(m.id) + expect(loaded.meta.title).toBe('secret') + expect(loaded.meta.firstPrompt).toBe('sensitive') + expect((await stat(sidecar)).isFile()).toBe(true) }) - it('delete removes a cwd-bucket sidecar even after a restart loses the in-memory cwd', async () => { - // A lazy session writes a sidecar under cwd /a (no log). Restart the backend - // (fresh instance, empty state) and delete: the in-memory cwd is gone and - // there is no log to recover it from, so delete must scan every bucket for - // the sidecar rather than only the _no-cwd bucket. + it('a lazy update leaves no sidecar that can leak into a future same-id session after restart', async () => { + await ctx.sessionPersistence.create(meta('restart-lazy', '/a')) + await ctx.sessionPersistence.update(SessionId('restart-lazy'), { title: 'secret' }) + await expect(stat(sidecarPath(root, '/a', SessionId('restart-lazy')))).rejects.toThrow() + + const ctx2 = new Context() + await ctx2.plugin(SessionStore) + await ctx2.plugin(SessionPersistenceJsonl, { root }) + const m2 = meta('restart-lazy', '/a') + await ctx2.sessionPersistence.create(m2) + await ctx2.sessionPersistence.append(m2.id, oneTurnLog()) + const loaded = await ctx2.sessionPersistence.load(m2.id) + expect(loaded.meta.title).toBeUndefined() + await ctx2.fiber.dispose() + }) + + it('delete removes a materialized cwd-bucket sidecar after a restart', async () => { await ctx.sessionPersistence.create(meta('restart-del', '/a')) + await ctx.sessionPersistence.append(SessionId('restart-del'), oneTurnLog()) await ctx.sessionPersistence.update(SessionId('restart-del'), { title: 'secret' }) const sidecar = sidecarPath(root, '/a', SessionId('restart-del')) expect((await stat(sidecar)).isFile()).toBe(true) @@ -671,7 +686,8 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.plugin(SessionStore) await ctx2.plugin(SessionPersistenceJsonl, { root }) await ctx2.sessionPersistence.delete(SessionId('restart-del')) - await expect(stat(sidecar)).rejects.toThrow() // sidecar gone despite no in-memory cwd + await expect(stat(sidecar)).rejects.toThrow() + await expect(stat(logPath(root, '/a', SessionId('restart-del')))).rejects.toThrow() await ctx2.fiber.dispose() }) @@ -749,6 +765,27 @@ describe('SessionPersistenceJsonl: edge cases', () => { expect(ids).toEqual(['p1', 'p2', 'p3']) }) + it('list tolerates one corrupt sidecar and still returns other sessions', async () => { + const warn = vi.spyOn(ctx.logger, 'warn').mockImplementation(() => undefined) + const bad = meta('bad-list-summary', '/proj') + await ctx.sessionPersistence.create(bad) + await ctx.sessionPersistence.append(bad.id, oneTurnLog()) + await ctx.sessionPersistence.update(bad.id, { title: 'hidden by corrupt sidecar' }) + await writeFile(sidecarPath(root, '/proj', bad.id), '{not json') + const good = meta('good-list-summary', '/proj') + await ctx.sessionPersistence.create(good) + await ctx.sessionPersistence.append(good.id, oneTurnLog()) + await ctx.sessionPersistence.update(good.id, { title: 'visible' }) + + const listed = await ctx.sessionPersistence.list() + + const badListed = listed.find(m => m.id === bad.id) + expect(badListed).toMatchObject({ id: bad.id }) + expect(badListed).not.toHaveProperty('title') + expect(listed.find(m => m.id === good.id)).toMatchObject({ id: good.id, title: 'visible' }) + expect(warn).toHaveBeenCalledWith(expect.stringContaining('bad-list-summary')) + }) + it('list on an empty root returns nothing', async () => { expect(await ctx.sessionPersistence.list()).toEqual([]) }) @@ -821,6 +858,30 @@ describe('SessionPersistenceJsonl: edge cases', () => { await ctx2.fiber.dispose() }) + it('HMR adoption does not crash-repair an active open turn as interrupted', async () => { + const dir = await freshRoot() + const hmr = new Context() + await hmr.plugin(SessionStore) + const first = await hmr.plugin(SessionPersistenceJsonl, { root: dir }) + const session = hmr.sessions.create('hmr-open', { meta: { cwd: '/hmr' } }) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + await hmr.parallel('session/flush', session) + + await first.dispose() + await appendFile(logPath(dir, '/hmr', SessionId('hmr-open')), '{"torn":') + const second = await hmr.plugin(SessionPersistenceJsonl, { root: dir }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await hmr.parallel('session/flush', session) + + const loaded = await hmr.sessionPersistence.load(SessionId('hmr-open')) + expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end']) + expect(loaded.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'completed' } } }) + await second.dispose() + await hmr.fiber.dispose() + }) + it('a NEW live session whose id collides with an on-disk log is rejected, not silently adopted', async () => { // Persist a session on disk. const s1 = ctx.sessions.create('collide', { meta: { cwd: '/a' } }) @@ -1023,6 +1084,14 @@ describe('SessionPersistenceJsonl: edge cases', () => { expect(loaded.meta.updatedAt).toBe(5) }) + it('load rejects a corrupt sidecar instead of treating it as absent', async () => { + const m = meta('bad-sidecar') + await ctx.sessionPersistence.create(m) + await ctx.sessionPersistence.append(m.id, oneTurnLog()) + await writeFile(sidecarPath(root, undefined, m.id), '{not json') + await expect(ctx.sessionPersistence.load(m.id)).rejects.toThrow() + }) + it('list returns nothing when the root directory does not exist', async () => { const ctx2 = new Context() await ctx2.plugin(SessionStore) diff --git a/packages/session-persistence-sqlite/README.md b/packages/session-persistence-sqlite/README.md index 051d054766..2025bd8a80 100644 --- a/packages/session-persistence-sqlite/README.md +++ b/packages/session-persistence-sqlite/README.md @@ -14,7 +14,7 @@ The repo targets Node ≥ 24 (the root `engines` field), which includes the stab - **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.) - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `has()`/`list()` (which report exactly the sessions that have a row). -- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`. +- **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (an error `tool/result` for every assistant tool call left unanswered, a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `has()`/`list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`. ## Configuration (schemastery) diff --git a/packages/session-persistence-sqlite/src/index.ts b/packages/session-persistence-sqlite/src/index.ts index 6976d3d193..d28fc8d6f7 100644 --- a/packages/session-persistence-sqlite/src/index.ts +++ b/packages/session-persistence-sqlite/src/index.ts @@ -24,8 +24,10 @@ import z from 'schemastery' import { DatabaseSync } from 'node:sqlite' import { mkdir } from 'node:fs/promises' import { dirname, resolve } from 'node:path' -import { SessionPersistence } from '@deepseek-ai/dsh-session-persistence' -import { isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session' +import { + SessionPersistence, assertSerializable, seedCoversPrefix, +} from '@deepseek-ai/dsh-session-persistence' +import { interruptedTurnClosers } from '@deepseek-ai/dsh-session' import type { Session, SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' import { openDatabase, rowToMeta, scanRows, type EventRow, type SessionRow, @@ -54,30 +56,13 @@ interface SessionState { owner?: Session } -/** - * Whether a live session's `seed` reproduces a persisted `prefix` exactly (the - * prefix is no longer than the seed and each event DEEP-equals the seed event - * at the same index). Distinguishes a session legitimately continuing a - * persisted log (HMR re-seeing its own session, or a resume) from a different - * session that merely reuses the id. Mirrors the JSONL backend's check; both - * sides are JSON-serializable by contract, so `JSON.stringify` is a sound - * canonical form. - */ -function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean { - return prefix.length <= seed.length - && prefix.every((e, i) => { - const s = seed[i] - return s !== undefined && JSON.stringify(s) === JSON.stringify(e) - }) -} - -/** Reject non-JSON-serializable `event.data`, naming the offending type. */ -function assertSerializable(events: readonly SessionEvent[]): void { - for (const event of events) { - if (!isJsonValue(event.data)) { - throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`) - } +async function settledErrors(promises: Iterable>): Promise { + const settled = await Promise.allSettled([...promises]) + const errors: unknown[] = [] + for (const result of settled) { + if (result.status === 'rejected') errors.push(result.reason) } + return errors } /** @@ -276,6 +261,35 @@ export class SessionPersistenceSqlite extends SessionPersistence { return { meta, events: balanced } } + private async adoptLiveStoredPrefix(session: Session, seed: readonly SessionEvent[]): Promise { + await this.ready + const row = this.rowFor(session.header.id) + /* v8 ignore next -- caller checked row existence */ + if (row === undefined) throw new Error(`session "${session.header.id}" not found`) + const meta = rowToMeta(row) + this.assertVersion(meta) + + const rows = this.db + .prepare('SELECT seq, type, time, data FROM events WHERE session_id = ? ORDER BY seq') + .all(session.header.id) as unknown as EventRow[] + const { preserved, tornFrom } = scanRows(rows) + if (!seedCoversPrefix(seed, preserved)) { + throw new Error(`session "${session.header.id}" already has a persisted log that does not match this live session (id collision)`) + } + + if (tornFrom !== undefined) { + this.db.prepare('DELETE FROM events WHERE session_id = ? AND seq >= ?').run(session.header.id, tornFrom) + } + this.states.set(session.header.id, { + meta: { ...meta }, + cursor: preserved.length, + materialized: true, + owner: session, + }) + const suffix = seed.slice(preserved.length) + if (suffix.length > 0) await this.appendCore(session.header.id, suffix) + } + async list(): Promise { await this.ready // Every metadata row is a materialized session: the row is written only by @@ -314,7 +328,7 @@ export class SessionPersistenceSqlite extends SessionPersistence { await this.ready let state = this.states.get(id) if (state === undefined) state = await this.adopt(id) - const nextMeta: SessionMeta = { ...state.meta, ...summary } + const nextMeta: SessionMeta = { ...state.meta, ...summary, updatedAt: summary.updatedAt ?? Date.now() } // update's only durable effect is the summary fields; the event log is // untouched. If the row is not materialized yet (a lazy session updated // before its first append) there is nothing to write — keep the pending @@ -410,11 +424,31 @@ export class SessionPersistenceSqlite extends SessionPersistence { // Dispose must reach quiescence: await every init + final drain, then close // the database, BEFORE returning, so no write lands after teardown. ctx.effect(() => async () => { - await Promise.allSettled([...this.inits.values()]) - await Promise.allSettled([...this.buffers.keys()].map(s => this.flush(s))) - await Promise.allSettled([...this.chains.values()]) - await this.ready - this.db.close() + let disposeError: unknown + try { + const errors = [ + ...await settledErrors(this.inits.values()), + ...await settledErrors([...this.buffers.keys()].map(s => this.flush(s))), + ...await settledErrors(this.chains.values()), + ] + if (errors.length > 0) { + throw new AggregateError(errors, 'session-persistence-sqlite dispose failed') + } + } catch (error: unknown) { + disposeError = error + throw error + } finally { + try { + await this.ready + this.db.close() + } catch (error: unknown) { + /* v8 ignore next -- open/close failure racing disposal is a defensive teardown edge */ + if (disposeError === undefined) throw error + // Opening/closing the database can only add teardown context here; keep + // the already-captured init/flush/chain AggregateError as the primary + // disposal failure instead of masking it from callers. + } + } }, 'session-persistence-sqlite write path') // HMR: a hot reload does not replay session/created, so seed existing live @@ -472,16 +506,9 @@ export class SessionPersistenceSqlite extends SessionPersistence { const row = this.rowFor(id) if (row !== undefined) { - const stored = this.eventsFor(id) - if (!seedCoversPrefix(seed, stored)) { - throw new Error(`session "${id}" already has a persisted log that does not match this live session (id collision)`) - } - await this.serialize(id, () => this.loadCore(id)) - const adopted = this.states.get(id) - /* v8 ignore next -- loadCore always sets the state for the id */ - if (adopted !== undefined) adopted.owner = session - const suffix = seed.slice(stored.length) - if (suffix.length > 0) await this.append(id, suffix) + // Adopt a LIVE prefix without crash-repairing an open turn as interrupted; + // HMR may still append the real completion from the live Session. + await this.serialize(id, () => this.adoptLiveStoredPrefix(session, seed)) return } diff --git a/packages/session-persistence-sqlite/tests/sqlite.spec.ts b/packages/session-persistence-sqlite/tests/sqlite.spec.ts index 44d788ab72..d6216b7b50 100644 --- a/packages/session-persistence-sqlite/tests/sqlite.spec.ts +++ b/packages/session-persistence-sqlite/tests/sqlite.spec.ts @@ -108,6 +108,35 @@ describe('scanRows', () => { }) }) +describe('SessionPersistenceSqlite: HMR adoption', () => { + it('does not crash-repair an active open turn as interrupted', async () => { + const path = await freshDbPath() + const ctx = new Context() + await ctx.plugin(SessionStore) + const first = await ctx.plugin(SessionPersistenceSqlite, { path }) + const session = ctx.sessions.create('hmr-open', { meta: { cwd: '/hmr' } }) + session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }) + session.append('step/start', { turn: 1, step: 1 }) + await ctx.parallel('session/flush', session) + + await first.dispose() + const db = openDatabase(path) + db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)') + .run('hmr-open', 2, 'step/end', 2, '{"torn":') + db.close() + const second = await ctx.plugin(SessionPersistenceSqlite, { path }) + session.append('step/end', { turn: 1, step: 1 }) + session.append('turn/end', { turn: 1, reason: { kind: 'completed' } }) + await ctx.parallel('session/flush', session) + + const loaded = await ctx.sessionPersistence.load(SessionId('hmr-open')) + expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'step/start', 'step/end', 'turn/end']) + expect(loaded.events.at(-1)).toMatchObject({ type: 'turn/end', data: { reason: { kind: 'completed' } } }) + await second.dispose() + await ctx.fiber.dispose() + }) +}) + describe('SessionPersistenceSqlite: durability and crash semantics', () => { it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => { const path = await freshDbPath() diff --git a/packages/session-persistence/src/index.ts b/packages/session-persistence/src/index.ts index 02e62940d7..0ce1e25146 100644 --- a/packages/session-persistence/src/index.ts +++ b/packages/session-persistence/src/index.ts @@ -22,6 +22,7 @@ */ import { Context, Service } from 'cordis' +import { isJsonValue } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionId, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session' // Re-export the metadata vocabulary so consumers import it from the seam. @@ -33,6 +34,35 @@ declare module 'cordis' { } } +/** + * Whether a live session's seed reproduces a persisted prefix exactly. Backends + * use this collision check to distinguish a legitimate resume/HMR rebind from a + * different live session reusing an existing session id. + * + * The comparison includes the full event payload, not just seq/type/time, so a + * mutated seed cannot be grafted onto a durable log with the same envelope. + */ +export function seedCoversPrefix(seed: readonly SessionEvent[], prefix: readonly SessionEvent[]): boolean { + return prefix.length <= seed.length + && prefix.every((event, index) => { + const seedEvent = seed[index] + return seedEvent !== undefined && JSON.stringify(seedEvent) === JSON.stringify(event) + }) +} + +/** + * Reject non-JSON-serializable event data before a backend serializes a batch. + * Live session appends already enforce this; persistence append paths also + * accept replay/fork batches that may bypass a live session instance. + */ +export function assertSerializable(events: readonly SessionEvent[]): void { + for (const event of events) { + if (!isJsonValue(event.data)) { + throw new Error(`event "${event.type}" carries non-JSON-serializable data (seq ${event.seq})`) + } + } +} + /** * Abstract durable session-persistence service. Subclass, implement the * abstract methods, and load the subclass as a plugin — it registers as diff --git a/packages/session-persistence/tests/contract.ts b/packages/session-persistence/tests/contract.ts index f8535a70f9..73f6f653bd 100644 --- a/packages/session-persistence/tests/contract.ts +++ b/packages/session-persistence/tests/contract.ts @@ -8,7 +8,7 @@ * @module @deepseek-ai/dsh-session-persistence/tests/contract */ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import { SessionId } from '@deepseek-ai/dsh-session' import type { SessionEvent, SessionMeta } from '@deepseek-ai/dsh-session' import { CallId } from '@deepseek-ai/dsh-llm' @@ -250,11 +250,19 @@ export function runPersistenceContract(name: string, make: () => Promise): Promise { const entry = this.store.get(id) - if (entry) Object.assign(entry.meta, summary) + if (entry) Object.assign(entry.meta, summary, { updatedAt: summary.updatedAt ?? Date.now() }) } } @@ -104,3 +104,38 @@ describe('SessionPersistence service registration', () => { await fiber.dispose() }) }) + +describe('shared persistence helpers', () => { + it('accepts a seed that reproduces the persisted prefix exactly', () => { + const log = oneTurnLog() + expect(seedCoversPrefix(log, log.slice(0, 3))).toBe(true) + expect(seedCoversPrefix(log, [])).toBe(true) + }) + + it('rejects a prefix longer than the seed', () => { + const log = oneTurnLog() + expect(seedCoversPrefix(log.slice(0, 2), log)).toBe(false) + }) + + it('rejects a same-envelope event with mutated data', () => { + const log = oneTurnLog() + const tampered = structuredClone(log) + const event = tampered[1]! + tampered[1] = { + ...event, + data: { ...event.data, content: [{ type: 'text', text: 'tampered' }] }, + } as SessionEvent + expect(seedCoversPrefix(tampered, log.slice(0, 2))).toBe(false) + }) + + it('accepts JSON-serializable event data', () => { + expect(() => { assertSerializable(oneTurnLog()) }).not.toThrow() + }) + + it('rejects non-JSON-serializable event data with type and seq context', () => { + const bad = [ + { type: 'user/message', seq: 0, time: 1, data: { content: 1n } }, + ] as unknown as SessionEvent[] + expect(() => { assertSerializable(bad) }).toThrow(/"user\/message".*seq 0/) + }) +}) diff --git a/packages/session/src/index.ts b/packages/session/src/index.ts index 34eefb24a3..4796c05f51 100644 --- a/packages/session/src/index.ts +++ b/packages/session/src/index.ts @@ -30,7 +30,7 @@ declare module 'cordis' { /** * Awaited durability checkpoint. The agent loop awaits * `ctx.parallel('session/flush', session)` at every turn end; persistence - * plugins (JSONL, sqlite — TODO, future phase) drain their write-behind + * plugins (JSONL, SQLite) drain their write-behind * buffers here and on fiber dispose. */ 'session/flush'(session: Session): Promise | void @@ -42,7 +42,9 @@ declare module 'cordis' { * synthetic user-role message (the system-reminder pattern: zero adapter * burden, models distinguish it from real user prompts by the envelope). * - * TODO(review): revisit the envelope once a real adapter exists. + * Live-adapter review has validated the tagged-envelope rendering against + * current DeepSeek behavior; provider-specific mismatches belong in that + * adapter, not in the canonical session vocabulary. */ function renderTagged(tag: string, content: ContentBlock[], source: MessageSource): ContentBlock[] { const open = `<${tag} source=${JSON.stringify(source.kind)}>` diff --git a/packages/session/src/repair.ts b/packages/session/src/repair.ts index b7afeb1562..5cc62b37c7 100644 --- a/packages/session/src/repair.ts +++ b/packages/session/src/repair.ts @@ -27,9 +27,10 @@ * — which every provider rejects as an invalid transcript on the next request. * Synthesizing an error result per orphaned call keeps resume safe. * - * This module computes those synthetic closers from an event list; the backend - * returns them inline from `load` (so the reconstructed session is balanced and - * immediately usable) and persists them on the first post-load `append`. + * This module computes those synthetic closers from an event list; backends + * return them inline from `load` (so the reconstructed session is balanced and + * immediately usable) and persist them during that mutating load before any + * later append continues the log. * * @module @deepseek-ai/dsh-session/repair */ @@ -78,6 +79,7 @@ export function interruptedTurnClosers(events: readonly SessionEvent[]): Session openStep = event.data.step break case 'step/end': + pendingCalls.clear() openStep = null break case 'assistant/message': diff --git a/packages/session/tests/repair.spec.ts b/packages/session/tests/repair.spec.ts index 893015b218..57422e7719 100644 --- a/packages/session/tests/repair.spec.ts +++ b/packages/session/tests/repair.spec.ts @@ -82,6 +82,21 @@ describe('interruptedTurnClosers', () => { expect(closers.map(e => e.type)).toEqual(['step/end', 'turn/end']) }) + it('does NOT synthesize a result after the owning step already closed', () => { + const events: SessionEvent[] = [ + userTurnStart(2, 0), + { type: 'step/start', seq: 1, time: 1, data: { turn: 2, step: 1 } }, + { type: 'assistant/message', seq: 2, time: 2, data: { turn: 2, step: 1, content: [ + { type: 'tool-call', id: CallId('call-1'), name: 'bash', arguments: '{}' }, + ] } }, + { type: 'step/end', seq: 3, time: 3, data: { turn: 2, step: 1 } }, + ] + + const closers = interruptedTurnClosers(events) + expect(closers.map(e => e.type)).toEqual(['turn/end']) + expect(closers[0]?.seq).toBe(4) + }) + it('synthesizes results only for the still-open turn, not a committed earlier turn', () => { // Turn 1 completed with its own tool call+result (balanced). Turn 2 crashed // with an unanswered call. Only turn 2's call must get a synthetic result. diff --git a/packages/system-prompt/src/index.ts b/packages/system-prompt/src/index.ts index ec89a3460c..25d87d2194 100644 --- a/packages/system-prompt/src/index.ts +++ b/packages/system-prompt/src/index.ts @@ -116,15 +116,21 @@ export class SystemPrompt extends Service { /** * Assemble the current prompt (sections sorted by order, tools collected - * from all providers). Runs through the `system-prompt/assemble` waterfall, - * giving listeners the opportunity to mutate or replace the assembly before - * it reaches the model. Await the result before reading the assembly values — + * from all providers). Section records are top-level clones (the `text` + * provider may be a function and is intentionally shared); tool schemas are + * deep-cloned because adapters and request waterfalls may mutate schema + * objects. Runs through the `system-prompt/assemble` waterfall, giving + * listeners the opportunity to mutate or replace the assembly before it + * reaches the model. Await the result before reading the assembly values — * waterfall listeners may be async. */ assemble(): Promise { const assembly: PromptAssembly = { - sections: [...this.sections].sort((a, b) => a.order - b.order), - tools: this.toolProviders.flatMap(provider => provider()), + sections: this.sections + .map(section => ({ ...section })) + .sort((a, b) => a.order - b.order), + tools: this.toolProviders.flatMap(provider => + provider().map(tool => ({ ...tool, parameters: structuredClone(tool.parameters) }))), } return this.ctx.waterfall(this, 'system-prompt/assemble', assembly, () => Promise.resolve(assembly)) } diff --git a/packages/system-prompt/tests/system-prompt.spec.ts b/packages/system-prompt/tests/system-prompt.spec.ts index 43e9240412..cfbf36cbec 100644 --- a/packages/system-prompt/tests/system-prompt.spec.ts +++ b/packages/system-prompt/tests/system-prompt.spec.ts @@ -107,6 +107,23 @@ describe('SystemPrompt', () => { expect(assembly.sections).toHaveLength(0) }) + it('assembles snapshots so one-step mutations do not leak into future assemblies', async () => { + const ctx = new Context() + await ctx.plugin(SystemPrompt) + ctx.systemPrompt.section({ name: 'base', order: 0, text: 'base' }) + ctx.systemPrompt.tools(() => [{ name: 't', description: 'tool', parameters: { type: 'object', properties: {} } }]) + + const first = await ctx.systemPrompt.assemble() + first.sections[0]!.name = 'mutated' + first.tools[0]!.description = 'mutated' + const firstParameters = first.tools[0]!.parameters as { properties: Record } + firstParameters.properties['leak'] = { type: 'string' } + + const second = await ctx.systemPrompt.assemble() + expect(second.sections.map(section => section.name)).toEqual(['base']) + expect(second.tools).toEqual([{ name: 't', description: 'tool', parameters: { type: 'object', properties: {} } }]) + }) + it('filters out empty section text from renderPrompt', () => { // Direct test of renderPrompt: function returning empty string, and empty static text const result = renderPrompt({ diff --git a/packages/tool-bash/README.md b/packages/tool-bash/README.md index 5a53bce17f..33ac71bc05 100644 --- a/packages/tool-bash/README.md +++ b/packages/tool-bash/README.md @@ -12,21 +12,21 @@ Requires a loaded executor implementation (e.g. `@deepseek-ai/dsh-bash-local`); |---|---|---| | `command` | string (required) | Run via `bash -c`. No state persists between calls — use `workdir`, not `cd`. | | `description` | string (required) | One-line, active-voice summary of the command (5-10 words), for UI/log display only — no effect on execution. | -| `timeoutMs` | number | Default/max from executor config (120s/600s for bash-local). | +| `timeoutMs` | number | Timeout override in milliseconds. The executor applies its configured default and cap. | | `workdir` | string | Working directory for this call. Defaults to the calling agent's session cwd (`session.header.cwd`) so each session runs in its own workspace; a relative `workdir` is resolved against that session cwd. | | `run_in_background` | boolean | Return a task id immediately; no timeout applies. | `command`, `workdir`, and `timeoutMs` are resolved against the executor's config defaults via `ctx.bash.resolve()` before execution, so the executor seam (`BashExecSpec`) receives explicit `workdir`/`timeoutMs` values. The workdir default is applied in the tool layer (from the calling agent's `session.header.cwd`) BEFORE `resolve()` — the per-session cwd must come from `exec.agent`, since N sessions share one executor; only when no session cwd is available does the executor fall back to its own config / `process.cwd()`. -Result text: stdout, then a `[stderr]` section, then status markers — `[timed out after Nms]` whenever the executor's timer fired (reported independently of how the process ended, so a command that traps SIGTERM and exits 0 still shows it), `[killed by signal: …]` for a signal death, `[exit code: N]` for a non-zero exit (reported, **not** `isError`: the model decides how to react), and `[output truncated; full output: ]` when the tail was kept. Only infrastructure failures (spawn errors, aborts) surface as `isError` results. +Result text: stdout, then a `[stderr]` section, then status markers — `[timed out after Nms]` whenever the executor's timer fired (reported independently of how the process ended, so a command that traps SIGTERM and exits 0 still shows it), `[killed by signal: …]` for a signal death, `[exit code: N]` for a non-zero exit (reported, **not** `isError`: the model decides how to react), and `[output truncated; full output: ]` when the tail was kept and a safe spill file is available. If the executor knows output was dropped but cannot safely advertise a complete spill file, the path is reported as `(unavailable)`. Only infrastructure failures (spawn errors, aborts) surface as `isError` results. ### `bash_output` -`task_id` → output produced **since the previous `bash_output` call** plus a status line (`running` / `completed, exit code: N` / `killed`). Reads that lost data to buffer bounds say so and point at the full-output spill file. +`task_id` → output produced **since the previous `bash_output` call** plus a status line (`running` / `completed, exit code: N` / `killed`). Reads that lost data to buffer bounds say so and point at the full-output spill file when one is safely available, otherwise `(unavailable)`. ### `bash_kill` -`task_id` → SIGTERM→SIGKILL on the task's process group. Killing an already-finished task is a reported no-op; unknown ids are errors. +`task_id` → ask the executor to kill the background task. The concrete executor decides how to signal or stop the process; killing an already-finished task is a reported no-op, and unknown ids are errors. ### Task ownership (cross-session isolation) diff --git a/packages/tool-bash/src/index.ts b/packages/tool-bash/src/index.ts index 1b01880e5a..2cdc28b1f0 100644 --- a/packages/tool-bash/src/index.ts +++ b/packages/tool-bash/src/index.ts @@ -316,7 +316,7 @@ export function apply(ctx: Context): void { description: 'Execute a bash command (`bash -c`) and return its stdout/stderr. ' + 'Each call runs in a fresh shell: no state (cwd, variables, functions) persists between calls — ' + 'pass `workdir` instead of using `cd`. Non-zero exits are reported as `[exit code: N]`. ' - + 'Long output is truncated to its tail; the full output is saved to a file whose path is reported. ' + + 'Long output is truncated to its tail; the full output is saved to a file whose path is reported when available. ' + 'Set `run_in_background: true` for long-running commands: the call returns a task id immediately; ' + 'poll it with `bash_output` and stop it with `bash_kill`.', parameters: { @@ -328,7 +328,7 @@ export function apply(ctx: Context): void { + '5-10 words (shown in the UI). Examples: "ls" → "List files in current directory"; ' + '"git status" → "Show working tree status"; "npm install" → "Install package dependencies".', }, - timeoutMs: { type: 'number', description: 'Timeout in milliseconds (default 120000, max 600000). The command is killed on expiry.' }, + timeoutMs: { type: 'number', description: 'Timeout in milliseconds. The executor applies its configured default and cap, and kills the command on expiry.' }, workdir: { type: 'string', description: 'Working directory for this command. Defaults to the session workspace; a relative path is resolved against it.' }, run_in_background: { type: 'boolean', description: 'Run in the background and return a task id immediately. No timeout applies.' }, }, @@ -377,7 +377,8 @@ export function apply(ctx: Context): void { let text = read.delta.length > 0 ? read.delta : '(no new output)' if (read.lossy) { const paths = [read.stdoutSpillPath, read.stderrSpillPath].filter((p): p is string => p !== undefined) - text += `\n[some output was dropped from memory; full output: ${paths.join(', ')}]` + const fullOutput = paths.length > 0 ? paths.join(', ') : '(unavailable)' + text += `\n[some output was dropped from memory; full output: ${fullOutput}]` } text += `\n${statusLine(read.task)}` return Promise.resolve([{ type: 'text', text }]) @@ -387,7 +388,7 @@ export function apply(ctx: Context): void { ctx.tools.register(defineTool({ name: 'bash_kill', - description: 'Kill a running background bash task (SIGTERM, then SIGKILL) by task id.', + description: 'Ask the executor to kill a running background bash task by task id.', parameters: { task_id: { type: 'string', required: true, description: 'Task id returned by the bash tool.' }, }, diff --git a/packages/tool-bash/tests/tools.spec.ts b/packages/tool-bash/tests/tools.spec.ts index 46194f33d2..1e75d7fb24 100644 --- a/packages/tool-bash/tests/tools.spec.ts +++ b/packages/tool-bash/tests/tools.spec.ts @@ -4,6 +4,8 @@ import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { Context } from 'cordis' import { CallId } from '@deepseek-ai/dsh-llm' +import { BashExecutor } from '@deepseek-ai/dsh-bash' +import type { BashExecRequest, BashExecSpec, BashRunResult, BashTask, BashTaskRead } from '@deepseek-ai/dsh-bash' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry from '@deepseek-ai/dsh-tools' import { LocalBashExecutor } from '@deepseek-ai/dsh-bash-local' @@ -31,6 +33,51 @@ function text(result: { content: { type: string; text?: string }[] }): string { return result.content.filter(block => block.type === 'text').map(block => block.text).join('') } +class LossyReadBashExecutor extends BashExecutor { + private readonly task: BashTask = { + id: 'bash-lossy', + command: 'fake', + status: 'running', + exitCode: null, + signal: null, + done: Promise.resolve(), + } + + resolve(request: BashExecRequest): BashExecSpec { + return { + command: request.command, + workdir: request.workdir ?? process.cwd(), + timeoutMs: request.timeoutMs ?? 0, + ...request.signal ? { signal: request.signal } : {}, + } + } + + run(): Promise { + return Promise.reject(new Error('not used')) + } + + start(): BashTask { + return this.task + } + + get(id: string): BashTask | undefined { + return id === this.task.id ? this.task : undefined + } + + list(): BashTask[] { + return [this.task] + } + + readOutput(id: string): BashTaskRead { + if (id !== this.task.id) throw new Error(`unknown bash task "${id}"`) + return { task: this.task, delta: 'tail', lossy: true } + } + + kill(): boolean { + return false + } +} + describe('bash tool', () => { it('returns stdout for a successful command', async () => { const ctx = await setup() @@ -226,6 +273,17 @@ describe('background tools', () => { expect(text(read)).toContain('[some output was dropped from memory; full output: ') }) + it('bash_output reports unavailable when a lossy read has no safe spill path', async () => { + const ctx = new Context() + await ctx.plugin(SystemPrompt) + await ctx.plugin(ToolRegistry) + await ctx.plugin(LossyReadBashExecutor) + await ctx.plugin(ToolBash) + + const read = await call(ctx, 'bash_output', { task_id: 'bash-lossy' }) + expect(text(read)).toBe('tail\n[some output was dropped from memory; full output: (unavailable)]\n[status: running]') + }) + it('bash_kill stops a running task; repeat reports already-finished', async () => { const ctx = await setup() const started = await call(ctx, 'bash', { command: 'sleep 60', description: 'test command', run_in_background: true }) diff --git a/packages/tools/src/index.ts b/packages/tools/src/index.ts index 5730b778c7..eee77445eb 100644 --- a/packages/tools/src/index.ts +++ b/packages/tools/src/index.ts @@ -326,37 +326,46 @@ export class ToolRegistry extends Service { return [...this.store.values()].map(({ name, description, parameters, strict }): ToolSchema => ({ name, description, - parameters, + parameters: structuredClone(parameters), ...strict !== undefined ? { strict } : {}, })) } /** - * Execute one tool call through the `tools/execute` waterfall. If the tool - * is not registered, the result is an `isError` carrying a `UNKNOWN_TOOL` - * structured error. If the tool throws, the error is caught and returned as - * an `isError` result so the loop never sees an uncaught exception; a thrown - * {@link HarnessError} surfaces its `{ name, code }` on the result. + * Execute one tool call through the `tools/execute` waterfall. If the tool is + * not registered, the result is an `isError` carrying a `UNKNOWN_TOOL` + * structured error. If the tool or a waterfall listener throws, the error is + * caught and returned as an `isError` result so the loop records a failed tool + * call instead of failing the whole turn; a thrown {@link HarnessError} + * surfaces its `{ name, code }` on the result. */ - execute(exec: ToolExecution): Promise { - return this.ctx.waterfall(this, 'tools/execute', exec, async (): Promise => { - try { - const tool = this.store.get(exec.name) - // Unknown tool routes through the same catch as a tool-thrown error, so - // both failure classes get structured `{ name, code }` from one path. - if (!tool) throw new ToolNotFoundError(exec.name) - const content = await tool.execute(exec.arguments, exec) - return { callId: exec.callId, content, isError: false } - } catch (error: unknown) { - const info = errorInfo(error) - return { - callId: exec.callId, - content: [{ type: 'text', text: `Error: ${errorMessage(error)}` }], - isError: true, - ...info ? { error: info } : {}, + async execute(exec: ToolExecution): Promise { + try { + return await this.ctx.waterfall(this, 'tools/execute', exec, async (): Promise => { + try { + const tool = this.store.get(exec.name) + // Unknown tool routes through the same catch as a tool-thrown error, so + // both failure classes get structured `{ name, code }` from one path. + if (!tool) throw new ToolNotFoundError(exec.name) + const content = await tool.execute(exec.arguments, exec) + return { callId: exec.callId, content, isError: false } + } catch (error: unknown) { + return toolErrorResult(exec.callId, error) } - } - }) + }) + } catch (error: unknown) { + return toolErrorResult(exec.callId, error) + } + } +} + +function toolErrorResult(callId: ToolExecution['callId'], error: unknown): ToolExecutionResult { + const info = errorInfo(error) + return { + callId, + content: [{ type: 'text', text: `Error: ${errorMessage(error)}` }], + isError: true, + ...info ? { error: info } : {}, } } diff --git a/packages/tools/tests/tools.spec.ts b/packages/tools/tests/tools.spec.ts index 9706790d55..c88963ecc1 100644 --- a/packages/tools/tests/tools.spec.ts +++ b/packages/tools/tests/tools.spec.ts @@ -1,6 +1,6 @@ import { describe, expect, expectTypeOf, it } from 'vitest' import { Context } from 'cordis' -import { CallId } from '@deepseek-ai/dsh-llm' +import { CallId, HarnessError } from '@deepseek-ai/dsh-llm' import SystemPrompt from '@deepseek-ai/dsh-system-prompt' import ToolRegistry, { defineTool, schemaSpecToJsonSchema, validateArgs, ToolArgsError, ToolNotFoundError, @@ -155,6 +155,54 @@ describe('ToolRegistry', () => { expect(order).toEqual(['first:before', 'second:before', 'second:after', 'first:after']) }) + it('returns an isError result when a tools/execute listener throws', async () => { + const ctx = await setup() + ctx.tools.register(echoTool) + ctx.on('tools/execute', async () => { + throw new Error('permission hook broke') + }) + + const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'echo', arguments: { text: 'hi' } }) + + expect(result).toEqual({ + callId: CallId('c1'), + content: [{ type: 'text', text: 'Error: permission hook broke' }], + isError: true, + }) + }) + + it('preserves structured error info when a tools/execute listener throws HarnessError', async () => { + const ctx = await setup() + ctx.tools.register(echoTool) + ctx.on('tools/execute', async () => { + throw new HarnessError('denied', 'DENIED') + }) + + const result = await ctx.tools.execute({ callId: CallId('c1'), name: 'echo', arguments: { text: 'hi' } }) + + expect(result).toMatchObject({ + callId: CallId('c1'), + isError: true, + error: { name: 'HarnessError', code: 'DENIED' }, + }) + }) + + it('schemas() snapshots tool schemas instead of exposing registry objects', async () => { + const ctx = await setup() + ctx.tools.register(echoTool) + + const first = ctx.tools.schemas() + const firstParameters = first[0]!.parameters as { properties: Record } + firstParameters.properties['mutated'] = { type: 'string' } + first[0]!.description = 'mutated' + + expect(ctx.tools.schemas()).toEqual([{ + name: 'echo', + description: 'echo arguments back', + parameters: { type: 'object', properties: { text: { type: 'string' } } }, + }]) + }) + it('rejects duplicate names and unregisters on fiber dispose (HMR safety)', async () => { const ctx = await setup() ctx.tools.register(echoTool) diff --git a/scripts/install-lefthook.mjs b/scripts/install-lefthook.mjs new file mode 100644 index 0000000000..be81daa153 --- /dev/null +++ b/scripts/install-lefthook.mjs @@ -0,0 +1,13 @@ +#!/usr/bin/env node +import { existsSync } from 'node:fs' +import { spawnSync } from 'node:child_process' +import { join } from 'node:path' + +const git = spawnSync('git', ['rev-parse', '--git-dir'], { stdio: 'ignore' }) +if (git.status !== 0) process.exit(0) + +const lefthook = join(process.cwd(), 'node_modules', '.bin', process.platform === 'win32' ? 'lefthook.cmd' : 'lefthook') +if (!existsSync(lefthook)) process.exit(0) + +const result = spawnSync(lefthook, ['install', '--force'], { stdio: 'inherit' }) +process.exit(result.status ?? 1)