Merge remote-tracking branch 'origin/master' into feat/acp-snapshot-tests

This commit is contained in:
Tianyi Cui
2026-06-19 10:03:04 +08:00
63 changed files with 1297 additions and 330 deletions
+1 -1
View File
@@ -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].
+2 -2
View File
@@ -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:
+2
View File
@@ -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
@@ -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.
@@ -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
@@ -3,7 +3,7 @@
Status: proposed
<!-- XXX: legacy ADR/RFC body format, not yet normalized to a unified RFC template. -->
> **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<Agent, sessionId>` 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<Agent, sessionId>` 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 |
@@ -3,12 +3,14 @@
Status: proposed
<!-- XXX: legacy ADR/RFC body format, not yet normalized to a unified RFC template. -->
> **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.
@@ -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.
@@ -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.
+3 -3
View File
@@ -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.
+5 -1
View File
@@ -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 + '/'
+1 -1
View File
@@ -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",
+1 -1
View File
@@ -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.
+2 -2
View File
@@ -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`) |
+11 -11
View File
@@ -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<sessionId, SessionRecord>` (forward) with a `WeakMap<Agent, sessionId>` 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"]
}
}
}
+22 -17
View File
@@ -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')
}
+54 -29
View File
@@ -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<SessionRecord['inflight']>,
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<NewSessionResponse> {
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<ToolPresenter, 'call' | 'result'> = 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) {
+7 -6
View File
@@ -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 () => {
+7 -5
View File
@@ -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)
})
})
+9
View File
@@ -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/)
})
})
+13 -4
View File
@@ -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
+14
View File
@@ -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<ToolRegistry, 'get'> {
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([])
+12 -8
View File
@@ -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<void> {
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
+24 -6
View File
@@ -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')
+65
View File
@@ -32,6 +32,17 @@ function waitForIdle(ctx: Context, agent: LoopAgent): Promise<void> {
})
}
function waitForStatus(ctx: Context, agent: LoopAgent, expected: LoopAgent['status']): Promise<void> {
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)
+97 -1
View File
@@ -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)
+5 -2
View File
@@ -71,6 +71,9 @@ export interface AgentFactory {
resume(options: ResumeAgentOptions): Promise<Agent>
}
/** 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<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.resume(options)
}
+9 -7
View File
@@ -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
+1 -1
View File
@@ -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.
+10
View File
@@ -38,6 +38,12 @@ export interface Config {
/** The shape after schemastery applied the defaults (cwd has none). */
type ResolvedConfig = Required<Omit<Config, 'cwd'>> & Pick<Config, 'cwd'>
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,
+9 -1
View File
@@ -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()
@@ -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 }))
+44
View File
@@ -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<typeof import('node:fs')>()
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<Parameters<typeof runBash>[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<typeof collector.finalize>
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', () => {
+1 -1
View File
@@ -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.
+3 -3
View File
@@ -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
}
+27 -2
View File
@@ -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<Agent, AgentStatus>()
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 => {
@@ -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()
+7 -1
View File
@@ -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<Config> = {}) {
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 }] }]
}
+2 -2
View File
@@ -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
+74 -12
View File
@@ -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<string, string> {
const raw = new Map<string, string>()
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<string, boolean | undefined> {
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<string, unknown>).stop = options.stop
return payload
},
} : {},
onPayload: payload => patchPayload(payload, options, this.options.reasoning),
maxRetries: 0,
})
+15 -6
View File
@@ -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<AssistantMessageEven
// when one is added (switch covers all current variants).
}
}
throw new LlmError('pi-ai event stream ended without done/error', 'STREAM_CLOSED')
}
+8 -1
View File
@@ -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'
@@ -15,14 +15,20 @@ import * as LlmDeepSeek 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<Config> = {}) {
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' })
+53 -5
View File
@@ -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<string, unknown>
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 () => {
+14
View File
@@ -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,
+80 -77
View File
@@ -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<unknown>>): Promise<unknown[]> {
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<void> {
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<SessionMeta[]> {
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<void> {
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<SessionSummary | undefined> {
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<SessionSummary | undefined> {
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
}
@@ -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)
@@ -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)
@@ -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<unknown>>): Promise<unknown[]> {
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<void> {
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<SessionMeta[]> {
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
}
@@ -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()
+30
View File
@@ -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
+10 -2
View File
@@ -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<Contrac
const log = oneTurnLog()
await persistence.create(m)
await persistence.append(m.id, log)
await persistence.update(m.id, { title: 'My session', firstPrompt: 'hi' })
const beforeUpdate = (await persistence.load(m.id)).meta.updatedAt
vi.useFakeTimers()
vi.setSystemTime(beforeUpdate + 1_000)
try {
await persistence.update(m.id, { title: 'My session', firstPrompt: 'hi' })
} finally {
vi.useRealTimers()
}
const loaded = await persistence.load(m.id)
expect(loaded.meta.title).toBe('My session')
expect(loaded.meta.firstPrompt).toBe('hi')
expect(loaded.meta.updatedAt).toBe(beforeUpdate + 1_000)
expect(loaded.events).toEqual(log) // log untouched
} finally {
await dispose()
@@ -2,7 +2,7 @@ import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { SessionId, isJsonValue, interruptedTurnClosers } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionMeta, SessionSummary } from '@deepseek-ai/dsh-session'
import { SessionPersistence } from '../src/index.ts'
import { SessionPersistence, assertSerializable, seedCoversPrefix } from '../src/index.ts'
import { runPersistenceContract, meta, oneTurnLog } from './contract.ts'
/**
@@ -69,7 +69,7 @@ class MemoryPersistence extends SessionPersistence {
async update(id: SessionId, summary: Partial<SessionSummary>): Promise<void> {
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/)
})
})
+4 -2
View File
@@ -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> | 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)}>`
+5 -3
View File
@@ -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':
+15
View File
@@ -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.
+11 -5
View File
@@ -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<PromptAssembly> {
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))
}
@@ -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<string, unknown> }
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({
+4 -4
View File
@@ -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: <path>]` 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: <path>]` 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)
+5 -4
View File
@@ -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.' },
},
+58
View File
@@ -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<BashRunResult> {
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 })
+33 -24
View File
@@ -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<ToolExecutionResult> {
return this.ctx.waterfall(this, 'tools/execute', exec, async (): Promise<ToolExecutionResult> => {
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<ToolExecutionResult> {
try {
return await this.ctx.waterfall(this, 'tools/execute', exec, async (): Promise<ToolExecutionResult> => {
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 } : {},
}
}
+49 -1
View File
@@ -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<string, unknown> }
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)
+13
View File
@@ -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)