diff --git a/docs/config-catalog.md b/docs/config-catalog.md index 1e8bd9e508..0183b44d37 100644 --- a/docs/config-catalog.md +++ b/docs/config-catalog.md @@ -328,6 +328,42 @@ export interface Config { Source: [`packages/hooks/hooks-codex/src/index.ts:43`](../packages/hooks/hooks-codex/src/index.ts) +## `@deepseek-ai/dsh-jsonrpc` + +Requires: `agents` + +```ts config-catalog +/** + * Plugin config. Every field is a runtime-only test seam — none is part of the + * schemastery {@link Config}, so nothing here is settable from a `cordis.yml` + * (production always serves the process stdio and exits via `process.exit`). + */ +export interface JsonRpcConfig { + /** + * Transport input override. Production omits this (the plugin reads + * `process.stdin`); tests inject an in-memory `Readable` to drive the server + * without a subprocess. + */ + input?: Readable + /** + * Transport output override. Production omits this (the plugin writes + * `process.stdout` — the protocol channel); tests inject an in-memory + * `Writable` to capture frames. + */ + output?: Writable + /** + * Process-exit override for the `shutdown` request path. Production omits + * this (`process.exit`); tests inject a recorder so a driven shutdown does + * not kill the test process. + */ + exit?: (code: number) => void +} +``` + +Depends on: `Readable` (`node:stream`) · `Writable` (`node:stream`) + +Source: [`packages/ui/jsonrpc/src/index.ts:55`](../packages/ui/jsonrpc/src/index.ts) + ## `@deepseek-ai/dsh-llm-deepseek` Requires: `llm` @@ -1168,6 +1204,7 @@ Imported as libraries by other packages; a `cordis.yml` cannot load them. - `@deepseek-ai/dsh-app-boot` ([`packages/ui/app-boot/src/index.ts`](../packages/ui/app-boot/src/index.ts)) - `@deepseek-ai/dsh-brand` ([`packages/util/brand/src/index.ts`](../packages/util/brand/src/index.ts)) - `@deepseek-ai/dsh-hook-protocol` ([`packages/hooks/hook-protocol/src/index.ts`](../packages/hooks/hook-protocol/src/index.ts)) +- `@deepseek-ai/dsh-jsonrpc-agent` ([`packages/ui/jsonrpc-agent/src/index.ts`](../packages/ui/jsonrpc-agent/src/index.ts)) - `@deepseek-ai/dsh-scope` ([`packages/core/scope/src/index.ts`](../packages/core/scope/src/index.ts)) - `@deepseek-ai/dsh-subagent-inprocess` ([`packages/subagent/subagent-inprocess/src/index.ts`](../packages/subagent/subagent-inprocess/src/index.ts)) - `@deepseek-ai/dsh-subagent-subprocess` ([`packages/subagent/subagent-subprocess/src/index.ts`](../packages/subagent/subagent-subprocess/src/index.ts)) diff --git a/docs/event-producer-consumer.md b/docs/event-producer-consumer.md index a8810a01fe..5b0a82673e 100644 --- a/docs/event-producer-consumer.md +++ b/docs/event-producer-consumer.md @@ -7,7 +7,7 @@ This matrix shows which packages dispatch each harness-owned event and which pac | Event | Mode | Declared in | Dispatchers | Listeners | | --- | --- | --- | --- | --- | -| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:316`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) | +| `agent/created` | `emit` | [`packages/core/agent/src/types.ts:316`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`jsonrpc`](../packages/ui/jsonrpc), [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/disposed` | `emit` | [`packages/core/agent/src/types.ts:331`](../packages/core/agent/src/types.ts) | [`agent`](../packages/core/agent) (`events.dispatch`) | [`stdio-agent`](../packages/ui/stdio-agent) | | `agent/error` | `emit` | [`packages/core/agent/src/types.ts:605`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`emit`) | - | | `agent/pre-step` | `serial` | [`packages/core/agent/src/types.ts:438`](../packages/core/agent/src/types.ts) | [`agent-loop`](../packages/core/agent-loop) (`serial`) | [`compact-basic`](../packages/compact/compact-basic), [`user-approval`](../packages/ui/user-approval) | @@ -25,13 +25,13 @@ This matrix shows which packages dispatch each harness-owned event and which pac | `fs/observed` | `emit` | [`packages/fs/fs/src/index.ts:138`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`emit`) | [`fs-policy`](../packages/fs/fs-policy) | | `fs/write-intent` | `waterfall` | [`packages/fs/fs/src/index.ts:109`](../packages/fs/fs/src/index.ts) | [`tool-fs`](../packages/fs/tool-fs) (`waterfall`) | [`fs-policy`](../packages/fs/fs-policy) | | `llm/stream` | `waterfall` | [`packages/llm/llm/src/index.ts:39`](../packages/llm/llm/src/index.ts) | [`llm`](../packages/llm/llm) (`waterfall`) | [`invariants`](../packages/support/invariants), [`llm-replay`](../packages/support/llm-replay) | -| `session/created` | `emit` | [`packages/core/session/src/index.ts:52`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence) | +| `session/created` | `emit` | [`packages/core/session/src/index.ts:52`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence) | | `session/disposed` | `emit` | [`packages/core/session/src/index.ts:64`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | - | -| `session/event` | `emit` | [`packages/core/session/src/index.ts:83`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio-agent`](../packages/ui/stdio-agent) | +| `session/event` | `emit` | [`packages/core/session/src/index.ts:83`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`acp`](../packages/ui/acp), [`invariants`](../packages/support/invariants), [`jsonrpc`](../packages/ui/jsonrpc), [`session-persistence`](../packages/session-persistence/session-persistence), [`stdio-agent`](../packages/ui/stdio-agent) | | `session/flush` | `parallel` | [`packages/core/session/src/index.ts:101`](../packages/core/session/src/index.ts) | [`session`](../packages/core/session) (`events.dispatch`) | [`session-persistence`](../packages/session-persistence/session-persistence) | | `skill/provider-added` | `emit` | [`packages/skill/skill/src/index.ts:131`](../packages/skill/skill/src/index.ts) | [`skill`](../packages/skill/skill) (`emit`) | - | | `skill/provider-removed` | `emit` | [`packages/skill/skill/src/index.ts:137`](../packages/skill/skill/src/index.ts) | [`skill`](../packages/skill/skill) (`emit`) | - | -| `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:90`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude) | +| `subagent/end` | `emit` | [`packages/subagent/subagent/src/index.ts:90`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude), [`jsonrpc`](../packages/ui/jsonrpc) | | `subagent/provider-added` | `emit` | [`packages/subagent/subagent/src/index.ts:66`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`emit`) | [`tool-subagent`](../packages/subagent/tool-subagent) | | `subagent/provider-removed` | `emit` | [`packages/subagent/subagent/src/index.ts:72`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`tool-subagent`](../packages/subagent/tool-subagent) | | `subagent/start` | `emit` | [`packages/subagent/subagent/src/index.ts:82`](../packages/subagent/subagent/src/index.ts) | [`subagent`](../packages/subagent/subagent) (`events.dispatch`) | [`hooks-claude`](../packages/hooks/hooks-claude) | diff --git a/docs/module-graph.md b/docs/module-graph.md index 2acee6ffc5..e4d0b3367d 100644 --- a/docs/module-graph.md +++ b/docs/module-graph.md @@ -92,6 +92,8 @@ flowchart TD pkg_acp["acp"] pkg_acp_agent["acp-agent"] pkg_app_boot["app-boot"] + pkg_jsonrpc["jsonrpc"] + pkg_jsonrpc_agent["jsonrpc-agent"] pkg_stdio_agent["stdio-agent"] pkg_tool_ask_user["tool-ask-user"] pkg_user_approval["user-approval"] @@ -284,6 +286,11 @@ flowchart TD pkg_subagent_mock --> pkg_agent pkg_subagent_mock --> pkg_llm pkg_subagent_mock --> pkg_subagent + pkg_jsonrpc --> pkg_agent + pkg_jsonrpc --> pkg_llm + pkg_jsonrpc --> pkg_llm_deepseek + pkg_jsonrpc --> pkg_session + pkg_jsonrpc --> pkg_subagent pkg_workflow_workerthread --> pkg_agent pkg_workflow_workerthread --> pkg_brand pkg_workflow_workerthread --> pkg_llm @@ -323,6 +330,7 @@ flowchart TD | [`subagent-subprocess`](../packages/subagent/subagent-subprocess) | `subagent` | — | | [`acp-snapshot`](../packages/support/acp-snapshot) | `support` | — | | [`app-boot`](../packages/ui/app-boot) | `ui` | — | +| [`jsonrpc-agent`](../packages/ui/jsonrpc-agent) | `ui` | — | | [`code-runtime`](../packages/code-runtime/code-runtime) | `code-runtime` | — | | [`llm`](../packages/llm/llm) | `llm` | [`brand`](../packages/util/brand) | | [`code-runtime-worker`](../packages/code-runtime/code-runtime-worker) | `code-runtime` | [`code-runtime`](../packages/code-runtime/code-runtime) | @@ -377,6 +385,7 @@ flowchart TD | [`tool-subagent`](../packages/subagent/tool-subagent) | `subagent` | [`agent`](../packages/core/agent), [`llm`](../packages/llm/llm), [`subagent`](../packages/subagent/subagent), [`tools`](../packages/core/tools) | | [`hooks-claude`](../packages/hooks/hooks-claude) | `hooks` | [`agent`](../packages/core/agent), [`hook-protocol`](../packages/hooks/hook-protocol), [`llm`](../packages/llm/llm), [`session`](../packages/core/session), [`subagent`](../packages/subagent/subagent), [`tools`](../packages/core/tools) | | [`subagent-mock`](../packages/support/subagent-mock) | `support` | [`agent`](../packages/core/agent), [`llm`](../packages/llm/llm), [`subagent`](../packages/subagent/subagent) | +| [`jsonrpc`](../packages/ui/jsonrpc) | `ui` | [`agent`](../packages/core/agent), [`llm`](../packages/llm/llm), [`llm-deepseek`](../packages/llm/llm-deepseek), [`session`](../packages/core/session), [`subagent`](../packages/subagent/subagent) | | [`workflow-workerthread`](../packages/workflow/workflow-workerthread) | `workflow` | [`agent`](../packages/core/agent), [`brand`](../packages/util/brand), [`llm`](../packages/llm/llm), [`session`](../packages/core/session), [`subagent`](../packages/subagent/subagent), [`tools`](../packages/core/tools), [`workflow`](../packages/workflow/workflow) | | [`subagent-fork`](../packages/subagent/subagent-fork) | `subagent` | [`agent`](../packages/core/agent), [`session`](../packages/core/session), [`subagent`](../packages/subagent/subagent), [`subagent-inprocess`](../packages/subagent/subagent-inprocess) | | [`subagent-spawn`](../packages/subagent/subagent-spawn) | `subagent` | [`subagent`](../packages/subagent/subagent), [`subagent-inprocess`](../packages/subagent/subagent-inprocess) | diff --git a/knip.json b/knip.json index cf43b90e34..f37f1d7636 100644 --- a/knip.json +++ b/knip.json @@ -81,6 +81,9 @@ "entry": ["tests/**/*.spec.ts", "tests/**/*.e2e.ts"], "project": ["src/**/*.ts", "tests/**/*.ts"] }, + "packages/ui/jsonrpc-agent": { + "project": ["src/**/*.ts"] + }, "packages/subagent/subagent-spawn": { "entry": ["tests/**/*.spec.ts", "tests/**/*.e2e.ts"], "project": ["src/**/*.ts", "tests/**/*.ts"] diff --git a/packages/ui/README.md b/packages/ui/README.md index 604eff6521..4f5a6dc920 100644 --- a/packages/ui/README.md +++ b/packages/ui/README.md @@ -10,10 +10,12 @@ Integrations that expose the agent to an external editor or client. These are ** | `tool-ask-user/` | Model-facing `ask_user_question` tool over `ctx.userInteraction` | (registers on `ctx.tools`) | | `stdio-agent/` | Terminal stdio chat APP: the agent-core spine + console logger + readline UI + a pre-created `main` agent, with a `bin` | (composition + `bin`) | | `acp-agent/` | ACP server APP: the agent-core spine + JSONL persistence + the `acp` bridge (no stdout logger), with a `bin` | (composition + `bin`) | -| `app-boot/` | Shared boot glue for the two app bins: `.env` loading, fail-loud Loader guards, snapshot-aware config resolution, the settle-the-tree boot sequence | (library for the bins) | +| `jsonrpc/` | Stdio JSON-RPC SDK server plugin: serves `HarnessSdkServer` to out-of-process SDK clients (the Python SDK) on the process stdio | (drives `ctx.agents`) | +| `jsonrpc-agent/` | JSON-RPC SDK server APP: a bin-only boot of an external `cordis.yml` whose `jsonrpc` entry is the serving face; the single-exe runtime entrypoint | (`bin` only) | +| `app-boot/` | Shared boot glue for the app bins: `.env` loading, fail-loud Loader guards, snapshot-aware config resolution, the settle-the-tree boot sequence | (library for the bins) | -A UI integration is a client-driver plugin, not a loop change and not a capability seam: it consumes the existing `agent/*` event taxonomy and the `dsh-agent` factory. The readline UI is the unstructured analogue of the `acp` bridge and lives INSIDE the stdio app (the `stdio-chat` module of [`stdio-agent/`](stdio-agent/README.md)): it is scaffolding for that one front door, not an independently swappable integration, so it carries no package boundary of its own. +A UI integration is a client-driver plugin, not a loop change and not a capability seam: it consumes the existing `agent/*` event taxonomy and the `dsh-agent` factory. The `jsonrpc` plugin is the SDK-client sibling of the `acp` bridge (a JSON-RPC server over `ctx.agents` for out-of-process SDK clients rather than editors). The readline UI is the unstructured analogue of the `acp` bridge and lives INSIDE the stdio app (the `stdio-chat` module of [`stdio-agent/`](stdio-agent/README.md)): it is scaffolding for that one front door, not an independently swappable integration, so it carries no package boundary of its own. `user-approval`, `user-interaction`, and `tool-ask-user` live here because asking a human is a UI-backed product affordance, not part of the providerless core spine. `user-approval` owns the one-shot `ctx.approval` decision mechanism and its policy tier; answerers remain with their UI channel owners. `user-interaction` remains provider-neutral (`ctx.userInteraction`), while `tool-ask-user` is its model-facing consumer and the app/bridge packages provide concrete providers. -`stdio-agent` and `acp-agent` are the two **app packages**: each composes the [`core/agent-core`](../core/agent-core/README.md) spine with its coupled front-door cluster (and owns the boot `bin`), so a leaf `cordis.yml` is the swappable backends plus one app entry plus any optional product tools. They live in `ui/` because each IS a user-facing front door; the stdout-purity coupling (logger vs. no logger) becomes a property of the artifact rather than a leaf convention. +`stdio-agent` and `acp-agent` are the two composing **app packages**: each composes the [`core/agent-core`](../core/agent-core/README.md) spine with its coupled front-door cluster (and owns the boot `bin`), so a leaf `cordis.yml` is the swappable backends plus one app entry plus any optional product tools. `jsonrpc-agent` is the third app but bin-only — no composition plugin, because the SDK runtime's hard semantic is that the external `cordis.yml` composes everything, the serving `jsonrpc` entry included. They live in `ui/` because each IS a user-facing front door; the stdout-purity coupling (logger vs. no logger) becomes a property of the artifact rather than a leaf convention. diff --git a/packages/ui/jsonrpc-agent/README.md b/packages/ui/jsonrpc-agent/README.md new file mode 100644 index 0000000000..d1836c737d --- /dev/null +++ b/packages/ui/jsonrpc-agent/README.md @@ -0,0 +1,17 @@ +# @deepseek-ai/dsh-jsonrpc-agent + +The **JSON-RPC SDK server app bin** (`dsh-jsonrpc-agent`): boot a harness from an externally supplied `cordis.yml` and let its [`@deepseek-ai/dsh-jsonrpc`](../jsonrpc/README.md) entry serve SDK clients over newline-delimited JSON-RPC on stdio. Structurally the SDK-runtime sibling of [`acp-agent`](../acp-agent/README.md)'s bin, but bin-only: there is no composition plugin here, because "the plugins that actually start come from the external config" is the SDK runtime's hard semantic — the leaf `cordis.yml` composes the spine, the backends, AND the serving face. This package is the entrypoint of the single-exe distribution (its `lib/bin.js` is what the packaged executable runs) — see [docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md](../../../docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md). + +## Config discovery + +Two channels, environment first: `$DSH_CORDIS_CONFIG` (the existing SDK-client convention, wins), then the `argv[2]` positional path (`dsh-jsonrpc-agent `, the human channel, isomorphic to `dsh-acp-agent`). An empty value counts as absent on either channel. Neither given, or the path missing on disk: the bin prints a one-line usage naming both channels to stderr and exits 1 — there is no default `./cordis.yml` and no built-in fallback config. A config that names a plugin which fails to load fails loud through the shared [`dsh-app-boot`](../app-boot/README.md) guards (`assertEntriesLoaded` + the unhandled-rejection handler), never a silent half-boot. There is no `DSH_SNAPSHOT` handling: this protocol is not part of the ACP snapshot tier. + +Note the deliberate flip side of config-decides-everything: a config that loads no `dsh-jsonrpc` entry boots fine and serves nothing — the bin cannot know which plugin is "the server". + +## Exit lifecycle + +The bin owns the PROCESS-level exits: stdin EOF (the SDK client is gone — an in-flight turn is deliberately cut off, see the risk note in docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md) and `SIGTERM` dispose the root context to quiescence and exit 0; `SIGINT` does the same but exits 130. The PROTOCOL-level exit — a `shutdown` JSON-RPC request answered first, then exit 0 — is owned by the `dsh-jsonrpc` plugin, which holds the server and transport; the two paths are individually idempotent and safe to race. + +## stdout is the protocol + +stdout carries only JSON-RPC frames; the bin and the app-boot guards write diagnostics to stderr only, and the booted config must load no stdout logger (see the `dsh-jsonrpc` README). diff --git a/packages/ui/jsonrpc-agent/package.json b/packages/ui/jsonrpc-agent/package.json new file mode 100644 index 0000000000..aebff4ab3a --- /dev/null +++ b/packages/ui/jsonrpc-agent/package.json @@ -0,0 +1,41 @@ +{ + "name": "@deepseek-ai/dsh-jsonrpc-agent", + "description": "JSON-RPC SDK server app bin: boots an externally supplied cordis.yml (DSH_CORDIS_CONFIG or argv, no built-in fallback) whose dsh-jsonrpc entry serves SDK clients over stdio; the single-exe runtime entrypoint", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/types/index.d.ts", + "bin": { + "dsh-jsonrpc-agent": "lib/bin.js" + }, + "exports": { + ".": { + "types": "./lib/types/index.d.ts", + "default": "./lib/index.js" + }, + "./bin": { + "types": "./lib/types/bin.d.ts", + "default": "./lib/bin.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib/index.js", + "lib/bin.js", + "lib/types/**/*.d.ts", + "lib/types/**/*.d.ts.map", + "src" + ], + "license": "BSD-3-Clause", + "dependencies": { + "@deepseek-ai/dsh-app-boot": "workspace:^" + }, + "peerDependencies": { + "cordis": "^4.0.0-rc.6" + }, + "devDependencies": { + "cordis": "^4.0.0-rc.6" + } +} diff --git a/packages/ui/jsonrpc-agent/src/bin.ts b/packages/ui/jsonrpc-agent/src/bin.ts new file mode 100644 index 0000000000..b2c9f0c3d2 --- /dev/null +++ b/packages/ui/jsonrpc-agent/src/bin.ts @@ -0,0 +1,75 @@ +#!/usr/bin/env node +/** + * The `dsh-jsonrpc-agent` bin: boot a harness from an externally supplied + * `cordis.yml` whose `@deepseek-ai/dsh-jsonrpc` entry serves SDK clients over + * newline-delimited JSON-RPC on stdio. The shared boot glue — `.env` loading, + * the fail-loud Loader guards, the settle-the-tree boot sequence — lives in + * {@link @deepseek-ai/dsh-app-boot}, shared with the stdio/ACP bins; this bin + * owns only config discovery and the process-level exit lifecycle: + * + * - Config discovery is `$DSH_CORDIS_CONFIG` (the existing SDK-client + * convention, wins) or the `argv[2]` positional path (the human channel, + * isomorphic to `dsh-acp-agent`); an empty value counts as absent. Neither + * given, or the path missing on disk, prints the one-line usage to stderr + * and exits 1. No built-in fallback — the external config IS the deployment + * (docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md). + * No `DSH_SNAPSHOT` handling: this + * protocol is not part of the ACP snapshot tier. + * - stdin EOF (the SDK client is gone) and SIGTERM dispose the root context + * to quiescence and exit 0; SIGINT does the same but exits 130. The + * `shutdown` JSON-RPC request's answer-then-exit-0 path is owned by the + * `dsh-jsonrpc` plugin, which holds the server (see its README). + * + * IMPORTANT: stdout is the JSON-RPC channel. Diagnostics go to STDERR only (a + * stray stdout write corrupts the protocol frames), which the app-boot guards + * already honor. + * + * @module @deepseek-ai/dsh-jsonrpc-agent/bin + */ + +import { existsSync } from 'node:fs' +import { boot, installFailLoud, loadEnv, resolveConfigPath } from '@deepseek-ai/dsh-app-boot' + +const NAME = 'dsh-jsonrpc-agent' + +/* v8 ignore start -- thin self-executing composition over the unit-tested + dsh-app-boot helpers; the serving lifecycle it boots is unit-tested in + @deepseek-ai/dsh-jsonrpc, and the composed artifact is exercised by the + single-exe acceptance drive (docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md) */ +installFailLoud(NAME) +loadEnv(NAME) + +// Env wins over the positional argument; an empty value on either channel +// counts as absent. There is deliberately NO default `./cordis.yml`: "the +// plugins that actually start come from an explicit external config" is a +// hard semantic of the SDK runtime. +const fromEnv = process.env['DSH_CORDIS_CONFIG'] +const fromArgv = process.argv[2] +const requested = fromEnv !== undefined && fromEnv !== '' + ? fromEnv + : fromArgv !== undefined && fromArgv !== '' ? fromArgv : undefined +const configPath = requested === undefined ? undefined : resolveConfigPath(requested, undefined) +if (configPath === undefined || !existsSync(configPath)) { + process.stderr.write( + `usage: ${NAME} (or set DSH_CORDIS_CONFIG=, which wins); the config is required — there is no built-in fallback\n`, + ) + process.exit(1) +} + +const ctx = await boot(NAME, configPath) +let exiting = false + +async function disposeAndExit(code: number): Promise { + if (exiting) return + exiting = true + try { + await ctx.fiber.dispose() + } finally { + process.exit(code) + } +} + +process.stdin.on('end', () => { void disposeAndExit(0) }) +process.on('SIGTERM', () => { void disposeAndExit(0) }) +process.on('SIGINT', () => { void disposeAndExit(130) }) +/* v8 ignore stop */ diff --git a/packages/ui/jsonrpc-agent/src/index.ts b/packages/ui/jsonrpc-agent/src/index.ts new file mode 100644 index 0000000000..39a2206dca --- /dev/null +++ b/packages/ui/jsonrpc-agent/src/index.ts @@ -0,0 +1,14 @@ +/** + * The `dsh-jsonrpc-agent` app package IS its bin (see `./bin.ts`): config + * discovery plus the process-level exit lifecycle around a booted + * `cordis.yml`. This module deliberately exports nothing — unlike the + * stdio/ACP app packages there is no composition plugin here, because the + * serving face is the {@link @deepseek-ai/dsh-jsonrpc} plugin the external + * config loads like any other entry (which plugins actually start is the + * config's decision, the hard semantic of the SDK runtime; see + * docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md). + * + * @module @deepseek-ai/dsh-jsonrpc-agent + */ + +export {} diff --git a/packages/ui/jsonrpc-agent/tsconfig.json b/packages/ui/jsonrpc-agent/tsconfig.json new file mode 100644 index 0000000000..e76defe8e2 --- /dev/null +++ b/packages/ui/jsonrpc-agent/tsconfig.json @@ -0,0 +1,21 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib/types" + }, + "include": [ + "src" + ], + "references": [ + { + "path": "../../../vendor/cordis" + }, + { + "path": "../../../vendor/loader" + }, + { + "path": "../app-boot" + } + ] +} diff --git a/packages/ui/jsonrpc-agent/tsdown.config.ts b/packages/ui/jsonrpc-agent/tsdown.config.ts new file mode 100644 index 0000000000..5b09ae2704 --- /dev/null +++ b/packages/ui/jsonrpc-agent/tsdown.config.ts @@ -0,0 +1,19 @@ +import { defineConfig } from 'tsdown' + +/** + * jsonrpc-agent ships TWO entries: the doc-only module (`index`) and the CLI + * `bin` (`bin`), the latter referenced by package.json `bin`/`exports["./bin"]`. + * The root tsdown builds only `lib/types/index.js`, so this override adds + * `lib/types/bin.js`. Declarations come from `tsc -b` (dts: false), + * matching every package. + */ +export default defineConfig({ + entry: ['lib/types/index.js', 'lib/types/bin.js'], + outDir: 'lib', + format: ['esm'], + platform: 'node', + target: 'es2024', + fixedExtension: false, + dts: false, + clean: false, +}) diff --git a/packages/ui/jsonrpc/README.md b/packages/ui/jsonrpc/README.md new file mode 100644 index 0000000000..dbc3fddcd0 --- /dev/null +++ b/packages/ui/jsonrpc/README.md @@ -0,0 +1,23 @@ +# @deepseek-ai/dsh-jsonrpc + +The **SDK server plugin** (`jsonrpc`): mounting it serves a stdio JSON-RPC server that lets an out-of-process SDK client (e.g. the Python `deepseek_harness` package) drive DeepSeek Harness agents without touching Cordis. The client speaks newline-delimited JSON-RPC on the process stdin/stdout ([`HarnessSdkServer`](src/server.ts): `initialize` → `session/prompt` → `shutdown`, with `session.event` / `session.finished` / `subagent.*` notifications over [`JsonRpcLineTransport`](src/transport.ts)). The SDK-client analogue of the [`acp`](../acp/README.md) bridge, split the same way: this package is the protocol plugin, [`jsonrpc-agent`](../jsonrpc-agent/README.md) is the app bin that boots a `cordis.yml` around it — which process serves this protocol is a config decision, not a hardcoded bin. This plugin is the serving face of the single-exe distribution plan — see [docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md](../../../docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md). + +## Wiring + +`inject: ['agents']` — the server creates one agent per SDK `sessionId` (get-or-create on `session/prompt`) and demuxes `subagent/end` through the registry. The LLM seam is read opportunistically via `ctx.get('llm')` (not injected): when `initialize.model` has no registered adapter, the plugin mounts `dsh-llm-deepseek` for it (credentials from `$DEEPSEEK_API_KEY` / `$DEEPSEEK_BASE_URL`); a config-registered adapter for the model wins. Everything else — persistence, the tool stacks, the adapter set — comes from the surrounding `cordis.yml`. + +## Config + +No `cordis.yml`-settable keys. The `JsonRpcConfig` fields (`input`, `output`, `exit`) are runtime-only test seams so a spec can drive the server over in-memory streams without a subprocess or a killed test process; production always serves the process stdio and exits via `process.exit`. + +## stdout is the protocol + +The process stdout this plugin runs in carries only JSON-RPC frames. The tree that loads it must load NO stdout logger (a console logger corrupts the frames) — the guarantee is config-only, same as the ACP bridge. Diagnostics go to stderr. + +## Shutdown and exit semantics + +The plugin owns the PROTOCOL-level exit: a `shutdown` request is answered first (the response frame flushes), then the plugin disposes its own fiber — running the effect disposer: an idempotent `server.shutdown()` (every SDK-created agent disposed to quiescence, event subscriptions detached) plus `transport.close()` — and exits the process with code 0. Own-fiber disposal is deliberate: the request's `server.shutdown()` already flushed all SDK-owned session state, and the process exit that follows is the teardown of the rest of the tree. Process-level exits (stdin EOF → 0, SIGTERM → 0, SIGINT → 130) belong to the app bin, which disposes the whole root context. Fiber disposal WITHOUT a `shutdown` request (HMR-style unload) just stops serving — it never exits the process. + +## Wire notes + +`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (SDK clients key on it, independent of this package's name). The `initialize` params `sessionRoot`, `systemPrompt`, and `clientInfo`, and the `session/prompt` param `profile`, are accepted for wire compatibility but currently unused — persistence roots and the deployment persona come from the `cordis.yml` (see the TODO in [`src/server.ts`](src/server.ts)). diff --git a/packages/ui/jsonrpc/package.json b/packages/ui/jsonrpc/package.json new file mode 100644 index 0000000000..be19ab7217 --- /dev/null +++ b/packages/ui/jsonrpc/package.json @@ -0,0 +1,46 @@ +{ + "name": "@deepseek-ai/dsh-jsonrpc", + "description": "Stdio JSON-RPC SDK server plugin: serves HarnessSdkServer over newline-delimited JSON-RPC on the process stdio, letting an out-of-process SDK client (e.g. the Python SDK) drive DeepSeek Harness agents", + "version": "0.0.1", + "private": true, + "type": "module", + "main": "lib/index.js", + "types": "lib/types/index.d.ts", + "exports": { + ".": { + "types": "./lib/types/index.d.ts", + "default": "./lib/index.js" + }, + "./src/*": "./src/*", + "./package.json": "./package.json" + }, + "files": [ + "lib/index.js", + "lib/types/**/*.d.ts", + "lib/types/**/*.d.ts.map", + "src" + ], + "license": "BSD-3-Clause", + "dependencies": { + "schemastery": "^3.17.0" + }, + "peerDependencies": { + "@deepseek-ai/dsh-agent": "^0.0.1", + "@deepseek-ai/dsh-llm": "^0.0.1", + "@deepseek-ai/dsh-llm-deepseek": "^0.0.1", + "@deepseek-ai/dsh-session": "^0.0.1", + "@deepseek-ai/dsh-subagent": "^0.0.1", + "cordis": "^4.0.0-rc.6" + }, + "devDependencies": { + "@cordisjs/plugin-loader": "workspace:^", + "@deepseek-ai/dsh-agent": "workspace:^", + "@deepseek-ai/dsh-agent-core": "workspace:^", + "@deepseek-ai/dsh-llm": "workspace:^", + "@deepseek-ai/dsh-llm-deepseek": "workspace:^", + "@deepseek-ai/dsh-session": "workspace:^", + "@deepseek-ai/dsh-session-persistence-jsonl": "workspace:^", + "@deepseek-ai/dsh-subagent": "workspace:^", + "cordis": "^4.0.0-rc.6" + } +} diff --git a/packages/ui/jsonrpc/src/index.ts b/packages/ui/jsonrpc/src/index.ts new file mode 100644 index 0000000000..9147f63e34 --- /dev/null +++ b/packages/ui/jsonrpc/src/index.ts @@ -0,0 +1,142 @@ +/** + * The SDK-facing stdio JSON-RPC server plugin: mounting it wires a + * {@link JsonRpcLineTransport} over the process stdio and serves + * {@link HarnessSdkServer} (`initialize` → `session/prompt`* → `shutdown`, + * plus the `session.*`/`subagent.*` notifications) to an out-of-process SDK + * client (e.g. the Python `deepseek_harness` package). The structured + * SDK-client analogue of the `acp` bridge: a client-driver plugin over + * `ctx.agents`, not a loop change and not a capability seam. Which process + * actually serves this protocol is a `cordis.yml` decision — the tree that + * loads this plugin IS the SDK server (the `dsh-jsonrpc-agent` bin boots such + * a tree for the single-exe distribution; see + * docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md). + * + * stdout is the protocol: this plugin must run in a tree that loads NO stdout + * logger (the console logger writes to stdout and would corrupt the JSON-RPC + * frames). The guarantee is config-only — see the package README. + * + * Exit-lifecycle split: this plugin owns the PROTOCOL-level exit (the + * `shutdown` request answers first, then the plugin disposes its own fiber and + * exits 0 — see {@link apply}); process-level exits (stdin EOF, SIGTERM, + * SIGINT) belong to the app bin (`dsh-jsonrpc-agent`), which disposes the + * whole root context. + * + * Plugin export shape: named `name`/`inject`/`Config`/`apply`, NO default + * export — the cordis Loader's `unwrapExports` does `exports.default ?? + * exports`, so a stray default would collapse the module to the bare `apply` + * and silently drop `inject`/`name`/`Config` (see docs/postmortem/0001). + * + * @module @deepseek-ai/dsh-jsonrpc + */ + +import type { Context } from 'cordis' +import type { Readable, Writable } from 'node:stream' +import Schema from 'schemastery' +import { HarnessSdkServer } from './server.ts' +import { JsonRpcLineTransport } from './transport.ts' + +export * from './server.ts' +export * from './transport.ts' + +export const name = 'jsonrpc' +// The server programs against the agent factory only: `agents` is read on +// every `session/prompt` (get-or-create) and on `subagent/end` demux. The LLM +// seam is deliberately NOT injected — `initialize` reads it opportunistically +// via `ctx.get('llm')` (the topology-independent lookup for a non-injected +// service, per packages/AGENTS.md) to decide whether to lazily mount the +// DeepSeek adapter for the requested model. +export const inject = ['agents'] + +/** + * Plugin config. Every field is a runtime-only test seam — none is part of the + * schemastery {@link Config}, so nothing here is settable from a `cordis.yml` + * (production always serves the process stdio and exits via `process.exit`). + */ +export interface JsonRpcConfig { + /** + * Transport input override. Production omits this (the plugin reads + * `process.stdin`); tests inject an in-memory `Readable` to drive the server + * without a subprocess. + */ + input?: Readable + /** + * Transport output override. Production omits this (the plugin writes + * `process.stdout` — the protocol channel); tests inject an in-memory + * `Writable` to capture frames. + */ + output?: Writable + /** + * Process-exit override for the `shutdown` request path. Production omits + * this (`process.exit`); tests inject a recorder so a driven shutdown does + * not kill the test process. + */ + exit?: (code: number) => void +} + +export const Config: Schema = Schema.object({}) + +/** + * Mount the SDK server on the process stdio: build the line transport and + * {@link HarnessSdkServer}, dispatch incoming requests, and start reading + * frames. Disposal is an effect: disposing this plugin's fiber runs + * `server.shutdown()` (disposes every SDK-created agent to quiescence and + * detaches the event subscriptions) and `transport.close()`. + * + * The `shutdown` request's process-exit semantics live HERE, because the + * plugin owns the server and transport: the request is answered first + * (`setImmediate` lets the response frame flush), then the plugin disposes its + * OWN fiber and calls `exit(0)`. Own-fiber disposal is sufficient — the + * request's `server.shutdown()` already brought every SDK-created agent to + * quiescence (their session logs are flushed by the awaited agent-handle + * disposes), the fiber's effect disposer re-runs the idempotent shutdown and + * closes the transport, and the process exit that follows IS the teardown of + * the rest of the tree (the bin's EOF/signal handlers own root-context + * disposal for the process-level exits). + */ +export function apply(ctx: Context, config: JsonRpcConfig): void { + // Capture the fiber handle NOW, during apply(): the shutdown path runs LATER, + // from the transport's read loop, and must dispose exactly this plugin's + // fiber (cf. the injection-scope capture note in the acp bridge). + const fiber = ctx.fiber + /* v8 ignore next -- production stdio wiring; tests always inject the runtime seams */ + const input = config.input ?? process.stdin + /* v8 ignore next -- production stdio wiring; tests always inject the runtime seams */ + const output = config.output ?? process.stdout + /* v8 ignore next -- production exit wiring; tests always inject the runtime seams */ + const exit = config.exit ?? ((code: number): void => { process.exit(code) }) + + const transport = new JsonRpcLineTransport(input, output) + const server = new HarnessSdkServer(ctx, transport) + + // The shutdown-request exit path, exactly once (a second `shutdown` frame + // racing the dispose must not re-enter). `exit(0)` runs even if the dispose + // throws — the client was already answered, so exiting is the honest outcome. + let exiting = false + const disposeAndExit = async (): Promise => { + if (exiting) return + exiting = true + try { + await fiber.dispose() + } finally { + exit(0) + } + } + + transport.onRequest(async (method, params) => { + const result = await server.handleRequest(method, params) + if (method === 'shutdown') { + // Answer the request first (setImmediate lets the response frame + // flush), then dispose this plugin's fiber and exit 0 (see apply's doc). + setImmediate(() => { void disposeAndExit() }) + } + return result + }) + + ctx.effect(() => { + transport.start() + return async () => { + await server.shutdown() + transport.close() + } + }, 'jsonrpc.serve') +} diff --git a/packages/ui/jsonrpc/src/server.ts b/packages/ui/jsonrpc/src/server.ts new file mode 100644 index 0000000000..1b403aac49 --- /dev/null +++ b/packages/ui/jsonrpc/src/server.ts @@ -0,0 +1,236 @@ +/** + * `HarnessSdkServer`: the JSON-RPC method surface the `dsh-jsonrpc` plugin + * serves to out-of-process SDK clients (e.g. the Python `deepseek_harness` + * package). Requests: `initialize` → `session/prompt`* → `shutdown`. + * Notifications pushed to the host: `session.event` (every durable session + * event, verbatim), `session.finished` (per prompt turn settle), + * `subagent.started` / `subagent.finished` (child-session lineage and run + * outcomes). The server owns only the SDK-facing session map — the harness + * itself is the context the plugin mounts in; plugins, persistence, and + * the LLM adapter set all come from the external `cordis.yml`. + * + * @module @deepseek-ai/dsh-jsonrpc/server + */ + +import type { Context } from 'cordis' +import type { ContentBlock } from '@deepseek-ai/dsh-llm' +import type { AgentHandle } from '@deepseek-ai/dsh-agent' +import { AgentId } from '@deepseek-ai/dsh-agent' +import { SessionId, type TurnEndReason } from '@deepseek-ai/dsh-session' +import type { SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent' +import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' +import type { JsonRpcTransportPeer } from './transport.ts' + +/** Parameters of the `initialize` request (once per process, before any prompt). */ +export interface InitializeParams { + /** Working directory recorded on every SDK-created session's header. */ + cwd: string + /** Model name every SDK-created agent runs on (see {@link HarnessSdkServer.initialize} for adapter fallback). */ + model: string + /** Accepted for SDK wire compatibility; unused — persistence roots come from the `cordis.yml`. */ + sessionRoot?: string + /** + * Accepted for SDK wire compatibility; currently NOT applied — the deployment + * persona comes from the `cordis.yml` system-prompt config. TODO(jsonrpc): + * map this onto a per-runtime system-prompt section once a per-agent override + * seam exists. + */ + systemPrompt?: string + /** Accepted for SDK wire compatibility; unused diagnostic client identity. */ + clientInfo?: { name?: string; version?: string } +} + +/** Result of the `initialize` request: the server's identity for the SDK handshake. */ +export interface InitializeResult { + /** Wire-stable server identity (`deepseek-harness-sdk-runtime`) and version. */ + serverInfo: { name: string; version: string } +} + +/** Parameters of a `session/prompt` request (one user turn on one SDK session). */ +export interface SessionPromptParams { + /** The SDK-side session id; an unknown id lazily creates the agent+session pair. */ + sessionId: string + /** The prompt content blocks, sent verbatim as the user message. */ + contentBlocks: ContentBlock[] + /** Accepted for SDK wire compatibility; unused — profiles are not a harness concept. */ + profile?: string +} + +/** Result of a `session/prompt` request: the prompt ran to turn settle (outcome rides on `session.finished`). */ +export interface SessionPromptResult { + /** Always `true`; the turn outcome is the paired `session.finished` notification. */ + accepted: true +} + +interface SessionRecord { + handle: AgentHandle + lastTurnEnd: TurnEndReason | undefined +} + +interface SubagentRecord { + childSessionId: string + parentSessionId: string | undefined +} + +/** + * The SDK server over a booted harness context. Constructing it subscribes to + * the context's `session/event`, `session/created`, `agent/created`, and + * `subagent/end` events and forwards them to the host as notifications; the + * subscriptions live until {@link shutdown}. One instance serves one transport + * peer for the process lifetime — there is no re-`initialize`. + */ +export class HarnessSdkServer { + private cwd = process.cwd() + private model = 'deepseek' + private llmFiber: { dispose(): Promise } | undefined + private readonly sessions = new Map() + private readonly subagentSessions = new Map() + private readonly disposers: (() => void)[] = [] + + constructor( + private readonly ctx: Context, + private readonly transport: JsonRpcTransportPeer, + ) { + this.disposers.push(ctx.on('session/event', (session, event) => { + if (event.type === 'turn/end') { + const rec = this.sessions.get(String(session.id)) + if (rec) rec.lastTurnEnd = event.data.reason + } + this.transport.notify('session.event', { sessionId: String(session.id), event }) + })) + this.disposers.push(ctx.on('session/created', (session) => { + const parentSession = session.header.parentSession + if (parentSession === undefined) return + this.transport.notify('subagent.started', { + parentSessionId: String(parentSession), + childSessionId: String(session.id), + }) + })) + // Cache agent → session lineage on creation: by the time `subagent/end` + // fires the child agent may already be disposed and gone from the registry. + this.disposers.push(ctx.on('agent/created', (agent) => { + this.subagentSessions.set(String(agent.id), { + childSessionId: String(agent.session.id), + parentSessionId: agent.session.header.parentSession === undefined + ? undefined + : String(agent.session.header.parentSession), + }) + })) + this.disposers.push(ctx.on('subagent/end', (info: SubagentRunEndInfo) => { + const rec = this.subagentSessions.get(String(info.id)) + const agent = this.ctx.agents.get(info.id) + const childSessionId = rec?.childSessionId ?? (agent === undefined ? undefined : String(agent.session.id)) + const parentSessionId = rec?.parentSessionId ?? ( + agent?.session.header.parentSession === undefined ? undefined : String(agent.session.header.parentSession) + ) + if (childSessionId === undefined) return + this.transport.notify('subagent.finished', { + provider: info.provider, + agentId: String(info.id), + ...(parentSessionId === undefined ? {} : { parentSessionId }), + childSessionId, + status: info.stopReason === 'completed' || info.stopReason === 'max-tokens' ? 'ok' : 'error', + stopReason: info.stopReason, + ...(info.lastAssistantMessage === undefined ? {} : { lastAssistantMessage: info.lastAssistantMessage }), + }) + })) + } + + /** + * Handle `initialize`: record the SDK deployment facts (cwd, model) and, when + * no registered adapter serves `params.model`, mount the DeepSeek adapter for + * it (credentials from `$DEEPSEEK_API_KEY`/`$DEEPSEEK_BASE_URL`) — a config + * that already registered an adapter for the model wins. + * @param params - the SDK handshake parameters. + * @returns the server identity for the handshake. + */ + async initialize(params: InitializeParams): Promise { + this.cwd = params.cwd + this.model = params.model + if (!this.llmFiber && !this.hasAdapterFor(this.model)) { + this.llmFiber = await this.ctx.plugin(LlmDeepSeek, { models: [this.model] }) + } + return { serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } } + } + + /** + * Handle `session/prompt`: get-or-create the session's agent, send the + * content as the user message, await turn settle (quiescence), then notify + * `session.finished` with the settled turn's outcome. + * @param params - the target session id and prompt content. + * @returns `{ accepted: true }` after the turn settled. + */ + async prompt(params: SessionPromptParams): Promise { + const rec = this.getOrCreateSession(params.sessionId) + rec.lastTurnEnd = undefined + rec.handle.agent.send(params.contentBlocks) + await rec.handle.agent.whenIdle() + const status = this.finishedStatus(rec.lastTurnEnd) + this.transport.notify('session.finished', { + sessionId: params.sessionId, + status, + reason: rec.lastTurnEnd, + }) + return { accepted: true } + } + + /** + * Handle `shutdown`: dispose every SDK-created agent handle (awaiting loop + * quiescence), unmount the adapter fiber this server mounted (if any), and + * detach the event subscriptions. The CONTEXT stays up — the bin disposes it + * as part of process exit. + * @returns an empty object (the JSON-RPC result). + */ + async shutdown(): Promise> { + const records = [...this.sessions.values()] + this.sessions.clear() + await Promise.all(records.map(rec => rec.handle.dispose())) + await this.llmFiber?.dispose() + this.llmFiber = undefined + while (this.disposers.length > 0) this.disposers.pop()?.() + return {} + } + + /** + * Dispatch one incoming JSON-RPC request to its typed handler. Throws (→ a + * JSON-RPC error response) on an unknown method. + * @param method - the JSON-RPC method name. + * @param params - the raw params object from the wire. + * @returns the handler's result, to be serialized as the response. + */ + async handleRequest(method: string, params: Record | undefined): Promise { + switch (method) { + case 'initialize': + return this.initialize(params as unknown as InitializeParams) + case 'session/prompt': + return this.prompt(params as unknown as SessionPromptParams) + case 'shutdown': + return this.shutdown() + default: + throw new Error(`unknown DeepSeek Harness SDK runtime method: ${method}`) + } + } + + private getOrCreateSession(sessionId: string): SessionRecord { + const existing = this.sessions.get(sessionId) + if (existing) return existing + const handle = this.ctx.agents.create({ + agentId: AgentId(sessionId), + sessionId: SessionId(sessionId), + meta: { cwd: this.cwd }, + agentOptions: { model: this.model }, + }) + const rec: SessionRecord = { handle, lastTurnEnd: undefined } + this.sessions.set(sessionId, rec) + return rec + } + + private finishedStatus(reason: TurnEndReason | undefined): 'ok' | 'error' { + if (!reason) return 'error' + return reason.kind === 'completed' || reason.kind === 'max-tokens' ? 'ok' : 'error' + } + + private hasAdapterFor(model: string): boolean { + return this.ctx.get('llm')?.models().includes(model) ?? false + } +} diff --git a/packages/ui/jsonrpc/src/transport.ts b/packages/ui/jsonrpc/src/transport.ts new file mode 100644 index 0000000000..003c78f4e5 --- /dev/null +++ b/packages/ui/jsonrpc/src/transport.ts @@ -0,0 +1,215 @@ +/** + * Newline-delimited JSON-RPC 2.0 transport over a byte stream pair (the SDK + * server's stdio channel). One JSON frame per line; a frame with `id`+`method` + * is an incoming request, `id` alone matches a pending outgoing request, and + * `method` alone is a notification. Malformed lines are ignored (a resilient + * wire reader, not a validator); handler failures become JSON-RPC error + * responses, never a crashed transport. + * + * @module @deepseek-ai/dsh-jsonrpc/transport + */ + +import { randomUUID } from 'node:crypto' +import type { Readable, Writable } from 'node:stream' + +type JsonRpcId = string | number +type RequestHandler = (method: string, params: Record) => Promise +type NotificationHandler = (method: string, params: Record) => void + +/** + * The outbound half of a JSON-RPC peer — what {@link HarnessSdkServer} needs + * to talk BACK to the host: awaited `request`s and fire-and-forget `notify`s. + * Narrow on purpose so tests substitute a recording fake without a stream pair. + */ +export interface JsonRpcTransportPeer { + /** + * Send a request to the remote peer and await its response. + * @param method - the JSON-RPC method name. + * @param params - the request parameters object. + * @returns the remote peer's `result`; rejects on a JSON-RPC `error` + * response, a write failure, or transport/input closure. + */ + request(method: string, params: Record): Promise + /** + * Send a notification (no response expected). An omitted `params` sends no + * `params` member at all. + * @param method - the JSON-RPC method name. + * @param params - the optional notification parameters object. + */ + notify(method: string, params?: Record): void +} + +interface PendingRequest { + resolve: (value: unknown) => void + reject: (error: Error) => void +} + +/** + * Line-delimited JSON-RPC 2.0 endpoint over a `Readable`/`Writable` pair. + * Inert until {@link start} attaches the input listeners; {@link close} + * detaches them and rejects every pending outgoing request (dispose-safe: the + * streams themselves are not destroyed — the caller owns them). Incoming + * requests are dispatched to the single {@link onRequest} handler (a missing + * handler answers `-32601 method not found`; a throwing handler answers + * `-32603` with the message); incoming notifications go to {@link + * onNotification} and are dropped without one. + */ +export class JsonRpcLineTransport implements JsonRpcTransportPeer { + private buffer = '' + private started = false + private requestHandler: RequestHandler | undefined + private notificationHandler: NotificationHandler | undefined + private readonly pending = new Map() + + constructor( + private readonly input: Readable, + private readonly output: Writable, + ) {} + + /** Attach the input listeners and begin reading frames. Idempotent. */ + start(): void { + if (this.started) return + this.started = true + this.input.on('data', this.onData) + this.input.on('error', this.onInputError) + this.input.on('end', this.onInputEnd) + } + + /** + * Detach the input listeners and reject every pending outgoing request with + * "JSON-RPC transport closed". Safe to call without a prior {@link start}. + */ + close(): void { + this.input.off('data', this.onData) + this.input.off('error', this.onInputError) + this.input.off('end', this.onInputEnd) + this.failPending(new Error('JSON-RPC transport closed')) + } + + /** + * Install THE handler for incoming requests (a later call replaces it). + * @param handler - resolves to the response `result`; a rejection becomes a + * `-32603` error response carrying the message. + */ + onRequest(handler: RequestHandler): void { + this.requestHandler = handler + } + + /** + * Install THE handler for incoming notifications (a later call replaces it). + * @param handler - invoked per notification with the method and normalized + * params object. + */ + onNotification(handler: NotificationHandler): void { + this.notificationHandler = handler + } + + request(method: string, params: Record): Promise { + const id = `req_${randomUUID().replaceAll('-', '')}` + const message = { jsonrpc: '2.0', id, method, params } + return new Promise((resolve, reject) => { + this.pending.set(id, { resolve, reject }) + try { + this.write(message) + } catch (error) { + this.pending.delete(id) + reject(error instanceof Error ? error : new Error(String(error))) + } + }) + } + + notify(method: string, params?: Record): void { + this.write(params === undefined ? { jsonrpc: '2.0', method } : { jsonrpc: '2.0', method, params }) + } + + private readonly onData = (chunk: Buffer | string): void => { + this.buffer += typeof chunk === 'string' ? chunk : chunk.toString('utf8') + for (;;) { + const newline = this.buffer.indexOf('\n') + if (newline < 0) break + const line = this.buffer.slice(0, newline).trim() + this.buffer = this.buffer.slice(newline + 1) + if (!line) continue + void this.handleLine(line) + } + } + + private readonly onInputError = (error: Error): void => { + this.failPending(error) + } + + private readonly onInputEnd = (): void => { + this.failPending(new Error('JSON-RPC input closed')) + } + + private async handleLine(line: string): Promise { + let message: unknown + try { + message = JSON.parse(line) + } catch { + // Swallows ONLY JSON.parse syntax errors: a malformed wire line is a + // peer bug this resilient reader skips; nothing else runs in the try. + return + } + if (!message || typeof message !== 'object') return + const frame = message as Record + const id = frame.id + const method = frame.method + if ((typeof id === 'string' || typeof id === 'number') && typeof method === 'string') { + await this.handleIncomingRequest(id, method, objectParams(frame.params)) + return + } + if (typeof id === 'string' || typeof id === 'number') { + this.handleIncomingResponse(id, frame) + return + } + if (typeof method === 'string') { + this.notificationHandler?.(method, objectParams(frame.params)) + } + } + + private async handleIncomingRequest(id: JsonRpcId, method: string, params: Record): Promise { + const handler = this.requestHandler + if (!handler) { + this.writeError(id, -32601, `method not found: ${method}`) + return + } + try { + const result = await handler(method, params) + this.write({ jsonrpc: '2.0', id, result }) + } catch (error) { + this.writeError(id, -32603, error instanceof Error ? error.message : String(error)) + } + } + + private handleIncomingResponse(id: JsonRpcId, frame: Record): void { + const pending = this.pending.get(id) + if (!pending) return + this.pending.delete(id) + if (frame.error && typeof frame.error === 'object') { + const error = frame.error as Record + pending.reject(new Error(typeof error.message === 'string' ? error.message : 'JSON-RPC error')) + return + } + pending.resolve(frame.result) + } + + private writeError(id: JsonRpcId, code: number, message: string): void { + this.write({ jsonrpc: '2.0', id, error: { code, message } }) + } + + private write(message: Record): void { + this.output.write(`${JSON.stringify(message)}\n`) + } + + private failPending(error: Error): void { + const pending = [...this.pending.values()] + this.pending.clear() + for (const waiter of pending) waiter.reject(error) + } +} + +/** Normalize JSON-RPC `params` to a plain object (arrays and scalars collapse to `{}`). */ +function objectParams(params: unknown): Record { + return params && typeof params === 'object' && !Array.isArray(params) ? params as Record : {} +} diff --git a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts new file mode 100644 index 0000000000..50821767f9 --- /dev/null +++ b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts @@ -0,0 +1,266 @@ +import { createServer } from 'node:http' +import type { IncomingMessage, Server, ServerResponse } from 'node:http' +import { mkdtemp, rm } from 'node:fs/promises' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { PassThrough, Writable } from 'node:stream' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import * as agentCore from '@deepseek-ai/dsh-agent-core' +import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' +import * as jsonrpc from '../src/index.ts' + +/** + * apply()-level lifecycle coverage for the @deepseek-ai/dsh-jsonrpc plugin: + * the plugin is mounted through the REAL namespace mount path — + * `ctx.plugin(jsonrpc, config)` over the module namespace object, exactly what + * the Loader hands cordis after `unwrapExports` (plugin-shape.spec pins that + * identity) — with the runtime-only `input`/`output`/`exit` seams from + * {@link jsonrpc.JsonRpcConfig} replacing the process stdio, so the whole + * pipeline (line transport → HarnessSdkServer → notifications back onto the + * wire) runs in-process. The scenarios pin the plugin's exit-lifecycle split: + * a `shutdown` REQUEST answers first, then disposes the plugin's own fiber and + * calls `exit(0)` exactly once (a racing second `shutdown` must not re-exit); + * a bare fiber dispose (HMR-style unload, no request) only stops serving and + * never touches `exit`. + */ + +/** One ordered observation on the plugin's outward-facing seams: a JSON-RPC frame written to `output`, or an `exit(code)` call. */ +type WireEvent = + | { kind: 'frame'; frame: Record } + | { kind: 'exit'; code: number } + +interface ApplyHarness { + ctx: Context + /** The jsonrpc plugin's own fiber (NOT the root), for the HMR-style dispose scenario. */ + fiber: Awaited> + /** Every output frame and exit call, in observation order — ordering assertions read this. */ + events: WireEvent[] + send(frame: Record): void + sendRaw(text: string): void + frames(): Record[] + exits(): number[] + waitForFrame(predicate: (frame: Record) => boolean, description: string): Promise> + dispose(): Promise +} + +/** Poll `get` until it yields a value (5s cap) — the output side is fed asynchronously from the transport's read loop. */ +async function waitFor(get: () => T | undefined, description: string): Promise { + const deadline = Date.now() + 5000 + for (;;) { + const value = get() + if (value !== undefined) return value + if (Date.now() > deadline) throw new Error(`timed out waiting for ${description}`) + await new Promise(resolve => setTimeout(resolve, 5)) + } +} + +/** Let pending microtasks, setImmediate callbacks, and stream events drain — for asserting that something did NOT happen. */ +async function settle(): Promise { + await new Promise(resolve => setTimeout(resolve, 25)) +} + +/** + * Boot a minimal harness context (agent-core bundle + JSONL persistence, the + * server.spec recipe) and mount the jsonrpc plugin on it through the real + * namespace mount path, with in-memory seams standing in for stdio/exit. + */ +async function mountPlugin(storageDir: string): Promise { + const ctx = new Context() + await ctx.plugin(agentCore) + await ctx.plugin(SessionPersistenceJsonl, { root: storageDir }) + await new Promise(resolve => setTimeout(resolve, 50)) + + const input = new PassThrough() + const events: WireEvent[] = [] + let pendingOutput = '' + // A hand-rolled Writable (not a PassThrough): _write records each decoded + // frame synchronously, so `events` preserves the true frame-vs-exit order. + const output = new Writable({ + write(chunk: Buffer, _encoding, callback) { + pendingOutput += chunk.toString('utf8') + for (;;) { + const newline = pendingOutput.indexOf('\n') + if (newline < 0) break + const line = pendingOutput.slice(0, newline).trim() + pendingOutput = pendingOutput.slice(newline + 1) + if (line) events.push({ kind: 'frame', frame: JSON.parse(line) as Record }) + } + callback() + }, + }) + const exit = (code: number): void => { events.push({ kind: 'exit', code }) } + + const fiber = await ctx.plugin(jsonrpc, { input, output, exit }) + + const frames = (): Record[] => + events.flatMap(event => event.kind === 'frame' ? [event.frame] : []) + return { + ctx, + fiber, + events, + send: (frame) => { input.write(`${JSON.stringify(frame)}\n`) }, + sendRaw: (text) => { input.write(text) }, + frames, + exits: () => events.flatMap(event => event.kind === 'exit' ? [event.code] : []), + waitForFrame: (predicate, description) => waitFor(() => frames().find(predicate), description), + dispose: async () => { await ctx.fiber.dispose() }, + } +} + +const servers: Server[] = [] + +afterEach(async () => { + await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve)))) + vi.unstubAllEnvs() +}) + +/** The server.spec mock OpenAI-compatible SSE endpoint, so a prompt turn completes without a real key. */ +async function mockCompletionServer(): Promise<{ url: string; requests: unknown[] }> { + const requests: unknown[] = [] + const server = createServer((request: IncomingMessage, response: ServerResponse) => { + let body = '' + request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') }) + request.on('end', () => { + requests.push(JSON.parse(body)) + response.writeHead(200, { 'content-type': 'text/event-stream' }) + response.write('data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}\n\n') + response.write('data: {"choices":[{"delta":{"content":"done"}}]}\n\n') + response.write('data: {"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}\n\n') + response.write('data: [DONE]\n\n') + response.end() + }) + }) + servers.push(server) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + if (address === null || typeof address === 'string') throw new Error('no port') + return { url: `http://127.0.0.1:${address.port}`, requests } +} + +describe('dsh-jsonrpc plugin apply', () => { + it('serves initialize over the injected stdio pair', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-init-')) + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + const harness = await mountPlugin(storageDir) + try { + harness.send({ jsonrpc: '2.0', id: 'init-1', method: 'initialize', params: { cwd: storageDir, model: 'apply-model' } }) + + const response = await harness.waitForFrame(frame => frame.id === 'init-1', 'initialize response') + expect(response).toEqual({ + jsonrpc: '2.0', + id: 'init-1', + result: { serverInfo: { name: 'deepseek-harness-sdk-runtime', version: '0.0.1' } }, + }) + expect(harness.exits()).toEqual([]) + } finally { + await harness.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('drives a session/prompt turn end-to-end and forwards session notifications as output frames', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-prompt-')) + const llmServer = await mockCompletionServer() + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url) + const harness = await mountPlugin(storageDir) + try { + harness.send({ jsonrpc: '2.0', id: 1, method: 'initialize', params: { cwd: storageDir, model: 'dsagent-model' } }) + await harness.waitForFrame(frame => frame.id === 1, 'initialize response') + + harness.send({ + jsonrpc: '2.0', + id: 2, + method: 'session/prompt', + params: { sessionId: 'main', contentBlocks: [{ type: 'text', text: 'fix it' }] }, + }) + const response = await harness.waitForFrame(frame => frame.id === 2, 'prompt response') + expect(response.result).toEqual({ accepted: true }) + + expect(llmServer.requests).toHaveLength(1) + const body = llmServer.requests[0] as { model: string; messages: { role: string }[] } + expect(body.model).toBe('dsagent-model') + expect(body.messages.at(-1)?.role).toBe('user') + + // The server's notify() path rides the SAME transport apply() built: + // session.event / session.finished arrive as id-less frames on output. + const notifications = harness.frames().filter(frame => frame.id === undefined) + expect(notifications.some(frame => frame.method === 'session.event')).toBe(true) + expect(notifications.find(frame => frame.method === 'session.finished')).toMatchObject({ + jsonrpc: '2.0', + params: { sessionId: 'main', status: 'ok' }, + }) + } finally { + await harness.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('answers shutdown before exiting 0 exactly once, even against a racing second shutdown', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-shutdown-')) + const harness = await mountPlugin(storageDir) + try { + // Two shutdown frames in ONE chunk: both are dispatched from the same + // read-loop pass, so both setImmediate exit callbacks get scheduled and + // the second must hit the `exiting` guard instead of re-entering. + const first = { jsonrpc: '2.0', id: 'sd-1', method: 'shutdown' } + const second = { jsonrpc: '2.0', id: 'sd-2', method: 'shutdown' } + harness.sendRaw(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`) + + await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit recorder call') + expect(harness.exits()).toEqual([0]) + + // Response-then-exit ordering: both shutdown responses were flushed to + // output BEFORE exit(0) ran (the setImmediate in the request handler). + const exitIndex = harness.events.findIndex(event => event.kind === 'exit') + const firstResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-1') + const secondResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-2') + expect(firstResponse).toBeGreaterThanOrEqual(0) + expect(secondResponse).toBeGreaterThanOrEqual(0) + expect(exitIndex).toBeGreaterThan(firstResponse) + expect(exitIndex).toBeGreaterThan(secondResponse) + + // Idempotent: the racing second shutdown never produces a second exit. + await settle() + expect(harness.exits()).toEqual([0]) + + // The plugin fiber is disposed: the transport reads no further frames. + const before = harness.frames().length + harness.send({ jsonrpc: '2.0', id: 'after-exit', method: 'initialize', params: { cwd: storageDir, model: 'x' } }) + await settle() + expect(harness.frames().length).toBe(before) + } finally { + await harness.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('stops serving on a bare fiber dispose (HMR-style unload) without calling exit', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-dispose-')) + const harness = await mountPlugin(storageDir) + try { + // Prove the pipeline is live first (an unknown method still answers, as + // a JSON-RPC error frame — the transport's handler-rejection path). + harness.send({ jsonrpc: '2.0', id: 'probe-1', method: 'nope/unknown' }) + const error = await harness.waitForFrame(frame => frame.id === 'probe-1', 'error response for unknown method') + expect(error.error).toMatchObject({ + code: -32603, + message: 'unknown DeepSeek Harness SDK runtime method: nope/unknown', + }) + + await harness.fiber.dispose() + + // The effect disposer shut the server and closed the transport — later + // frames are never read — and the exit seam was never touched. + const before = harness.frames().length + harness.send({ jsonrpc: '2.0', id: 'probe-2', method: 'initialize', params: { cwd: storageDir, model: 'x' } }) + await settle() + expect(harness.frames().length).toBe(before) + expect(harness.exits()).toEqual([]) + } finally { + await harness.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) +}) diff --git a/packages/ui/jsonrpc/tests/plugin-shape.spec.ts b/packages/ui/jsonrpc/tests/plugin-shape.spec.ts new file mode 100644 index 0000000000..3edbf1514f --- /dev/null +++ b/packages/ui/jsonrpc/tests/plugin-shape.spec.ts @@ -0,0 +1,32 @@ +import { describe, expect, it } from 'vitest' +import Loader from '@cordisjs/plugin-loader' +import * as jsonrpc from '../src/index.ts' + +/** + * REAL-export-path guard for the @deepseek-ai/dsh-jsonrpc namespace plugin + * (the packages/AGENTS.md red line: a plugin shipped via `cordis.yml` needs a + * test through the real Loader/export path). A hand-built `ctx.plugin({...})` + * mount bypasses `unwrapExports` — the exact path that once collapsed a + * namespace plugin with a stray `export default` and silently dropped its + * `inject` (docs/postmortem/0001) — so this spec drives the REAL + * `Loader.unwrapExports` over the module namespace and asserts the + * `name`/`inject`/`Config`/`apply` shape survives it intact. + */ +describe('dsh-jsonrpc plugin export shape', () => { + it('has the namespace-plugin export shape (no stray default) so the Loader keeps name/inject/Config/apply', () => { + // A stray `export default` would make `unwrapExports` (`exports.default ?? + // exports`) collapse the module to the bare default, dropping `inject` — + // the plugin would then throw "cannot get property … without inject" at + // its first `ctx.agents` read. Adding `export default` fails this test. + expect('default' in jsonrpc).toBe(false) + expect(typeof jsonrpc.apply).toBe('function') + + const loader = Object.create(Loader.prototype) as Loader + const unwrapped = loader.unwrapExports(jsonrpc) as Record + expect(unwrapped).toBe(jsonrpc) + expect(unwrapped.name).toBe('jsonrpc') + expect(unwrapped.inject).toEqual(['agents']) + expect(unwrapped.Config).toBeDefined() + expect(typeof unwrapped.apply).toBe('function') + }) +}) diff --git a/packages/ui/jsonrpc/tests/server.spec.ts b/packages/ui/jsonrpc/tests/server.spec.ts new file mode 100644 index 0000000000..a3ea6e3c1a --- /dev/null +++ b/packages/ui/jsonrpc/tests/server.spec.ts @@ -0,0 +1,378 @@ +import { createServer } from 'node:http' +import type { IncomingMessage, Server, ServerResponse } from 'node:http' +import { mkdtemp, rm } from 'node:fs/promises' +import { join } from 'node:path' +import { tmpdir } from 'node:os' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { Context } from 'cordis' +import { AgentId } from '@deepseek-ai/dsh-agent' +import { SessionId } from '@deepseek-ai/dsh-session' +import * as agentCore from '@deepseek-ai/dsh-agent-core' +import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl' +import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek' +import { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts' + +class FakeTransport implements JsonRpcTransportPeer { + notifications: { method: string; params?: Record }[] = [] + + async request(method: string, params: Record): Promise { + throw new Error(`the SDK server should not call host JSON-RPC method ${method} with ${JSON.stringify(params)}`) + } + + notify(method: string, params?: Record): void { + this.notifications.push(params === undefined ? { method } : { method, params }) + } +} + +const servers: Server[] = [] + +afterEach(async () => { + await Promise.all(servers.splice(0).map(server => new Promise(resolve => server.close(resolve)))) + vi.unstubAllEnvs() +}) + +async function mockCompletionServer(): Promise<{ url: string; requests: unknown[]; headers: IncomingMessage['headers'][] }> { + const requests: unknown[] = [] + const headers: IncomingMessage['headers'][] = [] + const server = createServer((request: IncomingMessage, response: ServerResponse) => { + let body = '' + request.on('data', (chunk: Buffer) => { body += chunk.toString('utf8') }) + request.on('end', () => { + requests.push(JSON.parse(body)) + headers.push(request.headers) + response.writeHead(200, { 'content-type': 'text/event-stream' }) + response.write('data: {"choices":[{"delta":{"role":"assistant","content":null,"reasoning_content":""}}]}\n\n') + response.write('data: {"choices":[{"delta":{"content":"done"}}]}\n\n') + response.write('data: {"choices":[{"delta":{"content":""},"finish_reason":"stop"}],"usage":{"prompt_tokens":3,"completion_tokens":1}}\n\n') + response.write('data: [DONE]\n\n') + response.end() + }) + }) + servers.push(server) + await new Promise(resolve => server.listen(0, '127.0.0.1', resolve)) + const address = server.address() + if (address === null || typeof address === 'string') throw new Error('no port') + return { url: `http://127.0.0.1:${address.port}`, requests, headers } +} + +async function makeHarness(storageDir: string) { + const ctx = new Context() + await ctx.plugin(agentCore) + await ctx.plugin(SessionPersistenceJsonl, { root: storageDir }) + await new Promise(resolve => setTimeout(resolve, 50)) + return ctx +} + +describe('HarnessSdkServer', () => { + it('creates a harness agent and calls the configured OpenAI-compatible endpoint', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-')) + const llmServer = await mockCompletionServer() + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url) + const ctx = await makeHarness(storageDir) + try { + const transport = new FakeTransport() + const server = new HarnessSdkServer(ctx, transport) + + const init = await server.handleRequest('initialize', { + cwd: storageDir, + model: 'dsagent-model', + sessionRoot: storageDir, + systemPrompt: 'Custom SDK instructions.', + }) as { serverInfo: { name: string } } + expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime') + + await server.handleRequest('session/prompt', { + sessionId: 'main', + contentBlocks: [{ type: 'text', text: 'fix it' }], + profile: 'build', + }) + + expect(llmServer.requests).toHaveLength(1) + const body = llmServer.requests[0] as { model: string; messages: { role: string }[] } + expect(body.model).toBe('dsagent-model') + expect(body.messages[0]?.role).toBe('system') + expect(body.messages.at(-1)?.role).toBe('user') + expect(llmServer.headers[0]?.authorization).toBe('Bearer test-key') + expect(transport.notifications.some(n => n.method === 'session.event')).toBe(true) + expect(transport.notifications.at(-1)).toMatchObject({ + method: 'session.finished', + params: { sessionId: 'main', status: 'ok' }, + }) + + await server.handleRequest('session/prompt', { + sessionId: 'main', + contentBlocks: [{ type: 'text', text: 'again' }], + }) + expect(llmServer.requests).toHaveLength(2) + + const orphanHandle = ctx.agents.create({ + agentId: AgentId('orphan-agent'), + sessionId: SessionId('orphan-session'), + meta: { cwd: storageDir }, + agentOptions: { model: 'dsagent-model' }, + }) + orphanHandle.agent.send([{ type: 'text', text: 'outside the sdk session map' }]) + await orphanHandle.agent.whenIdle() + await orphanHandle.dispose() + expect(llmServer.requests).toHaveLength(3) + + await server.handleRequest('shutdown', undefined) + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('notifies the host when a child session is created with parent lineage', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-')) + const ctx = await makeHarness(storageDir) + try { + const transport = new FakeTransport() + const server = new HarnessSdkServer(ctx, transport) + + ctx.sessions.create(SessionId('root-session'), { + meta: { cwd: storageDir }, + }) + ctx.sessions.create(SessionId('child-session'), { + meta: { cwd: storageDir, parentSession: SessionId('main') }, + }) + + expect(transport.notifications).toContainEqual({ + method: 'subagent.started', + params: { + parentSessionId: 'main', + childSessionId: 'child-session', + }, + }) + + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('creates an SDK session without an optional system prompt', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-no-system-')) + const llmServer = await mockCompletionServer() + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url) + const ctx = await makeHarness(storageDir) + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) + + await server.initialize({ cwd: storageDir, model: 'plain-model' }) + await server.prompt({ + sessionId: 'plain', + contentBlocks: [{ type: 'text', text: 'hello' }], + }) + + expect(llmServer.requests).toHaveLength(1) + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('notifies the host when a subagent run settles', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-end-')) + const ctx = await makeHarness(storageDir) + try { + const transport = new FakeTransport() + const server = new HarnessSdkServer(ctx, transport) + + const handle = ctx.agents.create({ + agentId: AgentId('child-agent'), + sessionId: SessionId('child-session'), + meta: { cwd: storageDir, parentSession: SessionId('main') }, + }) + ctx.emit('subagent/end', { + provider: 'spawn', + id: AgentId('child-agent'), + stopReason: 'completed', + lastAssistantMessage: [{ type: 'text', text: 'child done' }], + }) + + expect(transport.notifications).toContainEqual({ + method: 'subagent.finished', + params: { + provider: 'spawn', + agentId: 'child-agent', + parentSessionId: 'main', + childSessionId: 'child-session', + status: 'ok', + stopReason: 'completed', + lastAssistantMessage: [{ type: 'text', text: 'child done' }], + }, + }) + + await handle.dispose() + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('falls back to live agent lineage for uncached subagent end events', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-fallback-')) + const ctx = await makeHarness(storageDir) + let handle: ReturnType | undefined + let failedHandle: ReturnType | undefined + try { + handle = ctx.agents.create({ + agentId: AgentId('fallback-child-agent'), + sessionId: SessionId('fallback-child-session'), + meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') }, + }) + failedHandle = ctx.agents.create({ + agentId: AgentId('failed-child-agent'), + sessionId: SessionId('failed-child-session'), + meta: { cwd: storageDir }, + }) + const transport = new FakeTransport() + const server = new HarnessSdkServer(ctx, transport) + + ctx.emit('subagent/end', { + provider: 'fork', + id: AgentId('fallback-child-agent'), + stopReason: 'max-tokens', + }) + ctx.emit('subagent/end', { + provider: 'fork', + id: AgentId('failed-child-agent'), + stopReason: 'error', + }) + ctx.emit('subagent/end', { + provider: 'fork', + id: AgentId('missing-child-agent'), + stopReason: 'error', + }) + + expect(transport.notifications).toContainEqual({ + method: 'subagent.finished', + params: { + provider: 'fork', + agentId: 'fallback-child-agent', + parentSessionId: 'fallback-parent', + childSessionId: 'fallback-child-session', + status: 'ok', + stopReason: 'max-tokens', + }, + }) + expect(transport.notifications).toContainEqual({ + method: 'subagent.finished', + params: { + provider: 'fork', + agentId: 'failed-child-agent', + childSessionId: 'failed-child-session', + status: 'error', + stopReason: 'error', + }, + }) + expect(transport.notifications.some(n => + n.method === 'subagent.finished' + && n.params?.agentId === 'missing-child-agent', + )).toBe(false) + + await server.shutdown() + } finally { + await handle?.dispose() + await failedHandle?.dispose() + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('does not re-register an LLM adapter that already exists', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-existing-llm-')) + const ctx = await makeHarness(storageDir) + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + await ctx.plugin(LlmDeepSeek, { models: ['preinstalled-model'] }) + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) + const inspect = server as unknown as { hasAdapterFor(model: string): boolean } + + expect(inspect.hasAdapterFor('preinstalled-model')).toBe(true) + expect(inspect.hasAdapterFor('missing-model')).toBe(false) + await server.initialize({ cwd: storageDir, model: 'preinstalled-model' }) + + expect(ctx.get('llm')?.models().filter(model => model === 'preinstalled-model')).toEqual(['preinstalled-model']) + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('registers a missing model when an LLM service already exists', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-new-llm-')) + const ctx = await makeHarness(storageDir) + vi.stubEnv('DEEPSEEK_API_KEY', 'test-key') + await ctx.plugin(LlmDeepSeek, { models: ['other-model'] }) + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) + + await server.initialize({ cwd: storageDir, model: 'new-model' }) + + expect(ctx.get('llm')?.models()).toEqual(expect.arrayContaining(['other-model', 'new-model'])) + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('classifies defensive finish states', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-finish-states-')) + const ctx = await makeHarness(storageDir) + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as { + finishedStatus(reason: unknown): 'ok' | 'error' + shutdown(): Promise> + } + + expect(server.finishedStatus(undefined)).toBe('error') + expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('ok') + expect(server.finishedStatus({ kind: 'error' })).toBe('error') + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) + + it('reports no adapter when the LLM service is absent', async () => { + const ctx = new Context() + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as { + hasAdapterFor(model: string): boolean + shutdown(): Promise> + } + + expect(server.hasAdapterFor('missing-model')).toBe(false) + await server.shutdown() + } finally { + await ctx.fiber.dispose() + } + }) + + it('rejects unknown JSON-RPC runtime methods', async () => { + const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-unknown-')) + const ctx = await makeHarness(storageDir) + try { + const server = new HarnessSdkServer(ctx, new FakeTransport()) + + await expect(server.handleRequest('does/not/exist', {})) + .rejects + .toThrow('unknown DeepSeek Harness SDK runtime method: does/not/exist') + + await server.shutdown() + } finally { + await ctx.fiber.dispose() + await rm(storageDir, { recursive: true, force: true }) + } + }) +}) diff --git a/packages/ui/jsonrpc/tests/transport.spec.ts b/packages/ui/jsonrpc/tests/transport.spec.ts new file mode 100644 index 0000000000..0e98a4af10 --- /dev/null +++ b/packages/ui/jsonrpc/tests/transport.spec.ts @@ -0,0 +1,202 @@ +import { once } from 'node:events' +import { PassThrough } from 'node:stream' +import { describe, expect, it } from 'vitest' +import { JsonRpcLineTransport } from '../src/index.ts' + +function transportPair() { + const aToB = new PassThrough() + const bToA = new PassThrough() + const a = new JsonRpcLineTransport(bToA, aToB) + const b = new JsonRpcLineTransport(aToB, bToA) + return { a, b, aToB, bToA } +} + +describe('JsonRpcLineTransport', () => { + it('supports bidirectional requests and notifications over newline-delimited JSON-RPC', async () => { + const { a, b } = transportPair() + const notifications: Record[] = [] + + a.onRequest(async (method, params) => { + expect(method).toBe('echo') + return { echoed: params } + }) + b.onNotification((method, params) => { + notifications.push({ method, params }) + }) + a.start() + b.start() + + const response = await b.request('echo', { value: 42 }) + expect(response).toEqual({ echoed: { value: 42 } }) + + a.notify('session.finished', { sessionId: 'main', status: 'ok' }) + a.notify('heartbeat') + await new Promise(resolve => setTimeout(resolve, 10)) + expect(notifications).toEqual([ + { method: 'session.finished', params: { sessionId: 'main', status: 'ok' } }, + { method: 'heartbeat', params: {} }, + ]) + + a.close() + b.close() + }) + + it('reports JSON-RPC request errors from the remote peer', async () => { + const { a, b } = transportPair() + a.onRequest(async () => { + throw new Error('handler boom') + }) + a.start() + b.start() + + await expect(b.request('explode', {})).rejects.toThrow('handler boom') + + a.close() + b.close() + }) + + it('stringifies non-Error request handler failures', async () => { + const { a, b } = transportPair() + a.onRequest(async () => { + throw 'string boom' + }) + a.start() + b.start() + + await expect(b.request('explode-string', {})).rejects.toThrow('string boom') + + a.close() + b.close() + }) + + it('reports method-not-found when no request handler is installed', async () => { + const { a, b } = transportPair() + a.start() + b.start() + + await expect(b.request('missing', {})).rejects.toThrow('method not found: missing') + + a.close() + b.close() + }) + + it('normalizes non-object request params and ignores notifications without a handler', async () => { + const { aToB, bToA, b } = transportPair() + const seen: Record[] = [] + b.onRequest(async (method, params) => { + seen.push({ method, params }) + return { ok: true } + }) + b.start() + + aToB.write('{"jsonrpc":"2.0","method":"ignored"}\n') + aToB.write('{"jsonrpc":"2.0","id":7,"method":"array-params","params":[]}\n') + const chunk = (await once(bToA, 'data'))[0] as Buffer | string + + expect(seen).toEqual([{ method: 'array-params', params: {} }]) + expect(JSON.parse(String(chunk))).toEqual({ jsonrpc: '2.0', id: 7, result: { ok: true } }) + b.close() + }) + + it('ignores malformed frames and accepts notifications without params', async () => { + const { aToB, b } = transportPair() + const notifications: Record[] = [] + b.onNotification((method, params) => { + notifications.push({ method, params }) + }) + b.start() + b.start() + + aToB.write('not json\n') + aToB.write('\n') + aToB.write('null\n') + aToB.write('{"jsonrpc":"2.0","params":{}}\n') + aToB.write('{"jsonrpc":"2.0","method":"tick"}\n') + aToB.emit('data', '{"jsonrpc":"2.0","method":"string-chunk"}\n') + await new Promise(resolve => setTimeout(resolve, 10)) + + expect(notifications).toEqual([ + { method: 'tick', params: {} }, + { method: 'string-chunk', params: {} }, + ]) + b.close() + }) + + it('rejects pending requests when the input closes', async () => { + const { aToB, b } = transportPair() + b.start() + + const pending = b.request('never-replies', {}) + aToB.end() + + await expect(pending).rejects.toThrow('JSON-RPC input closed') + b.close() + }) + + it('rejects pending requests when the input errors', async () => { + const { aToB, b } = transportPair() + b.start() + + const pending = b.request('never-replies', {}) + aToB.emit('error', new Error('input broke')) + + await expect(pending).rejects.toThrow('input broke') + b.close() + }) + + it('rejects pending requests when the transport closes', async () => { + const { b } = transportPair() + + const pending = b.request('never-replies', {}) + b.close() + + await expect(pending).rejects.toThrow('JSON-RPC transport closed') + }) + + it('rejects a request when writing the frame throws', async () => { + const input = new PassThrough() + const output = { + write() { + throw new Error('write exploded') + }, + } + const transport = new JsonRpcLineTransport(input, output as never) + + await expect(transport.request('write-fails', {})).rejects.toThrow('write exploded') + }) + + it('stringifies non-Error write failures', async () => { + const input = new PassThrough() + const output = { + write() { + throw 'write string' + }, + } + const transport = new JsonRpcLineTransport(input, output as never) + + await expect(transport.request('write-fails', {})).rejects.toThrow('write string') + }) + + it('uses a fallback message for malformed JSON-RPC error responses', async () => { + const { aToB, bToA, b } = transportPair() + b.start() + + const pending = b.request('remote-error', {}) + const requestChunk = (await once(bToA, 'data'))[0] as Buffer | string + const request = JSON.parse(String(requestChunk)) as { id: string } + aToB.write(`${JSON.stringify({ jsonrpc: '2.0', id: request.id, error: {} })}\n`) + + await expect(pending).rejects.toThrow('JSON-RPC error') + b.close() + }) + + it('ignores responses that do not match a pending request', async () => { + const { aToB, b } = transportPair() + b.start() + + aToB.write('{"jsonrpc":"2.0","id":"unknown","result":{"ignored":true}}\n') + await new Promise(resolve => setTimeout(resolve, 10)) + + b.close() + }) +}) diff --git a/packages/ui/jsonrpc/tsconfig.json b/packages/ui/jsonrpc/tsconfig.json new file mode 100644 index 0000000000..dcd57ef9af --- /dev/null +++ b/packages/ui/jsonrpc/tsconfig.json @@ -0,0 +1,33 @@ +{ + "extends": "../../../tsconfig.base.json", + "compilerOptions": { + "rootDir": "src", + "outDir": "lib/types" + }, + "include": [ + "src" + ], + "references": [ + { + "path": "../../../vendor/cordis" + }, + { + "path": "../../../vendor/schemastery" + }, + { + "path": "../../llm/llm" + }, + { + "path": "../../llm/llm-deepseek" + }, + { + "path": "../../core/agent" + }, + { + "path": "../../core/session" + }, + { + "path": "../../subagent/subagent" + } + ] +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 649a986385..1d868ae049 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -1252,6 +1252,50 @@ importers: specifier: ^4.0.0-rc.6 version: 4.0.0-rc.6(@cordisjs/plugin-include@vendor+include)(@cordisjs/plugin-loader@vendor+loader) + packages/ui/jsonrpc: + dependencies: + schemastery: + specifier: ^3.17.0 + version: 3.18.0 + devDependencies: + '@cordisjs/plugin-loader': + specifier: workspace:^ + version: link:../../../vendor/loader + '@deepseek-ai/dsh-agent': + specifier: workspace:^ + version: link:../../core/agent + '@deepseek-ai/dsh-agent-core': + specifier: workspace:^ + version: link:../../core/agent-core + '@deepseek-ai/dsh-llm': + specifier: workspace:^ + version: link:../../llm/llm + '@deepseek-ai/dsh-llm-deepseek': + specifier: workspace:^ + version: link:../../llm/llm-deepseek + '@deepseek-ai/dsh-session': + specifier: workspace:^ + version: link:../../core/session + '@deepseek-ai/dsh-session-persistence-jsonl': + specifier: workspace:^ + version: link:../../session-persistence/session-persistence-jsonl + '@deepseek-ai/dsh-subagent': + specifier: workspace:^ + version: link:../../subagent/subagent + cordis: + specifier: ^4.0.0-rc.6 + version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@vendor+loader) + + packages/ui/jsonrpc-agent: + dependencies: + '@deepseek-ai/dsh-app-boot': + specifier: workspace:^ + version: link:../app-boot + devDependencies: + cordis: + specifier: ^4.0.0-rc.6 + version: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4) + packages/ui/stdio-agent: devDependencies: '@cordisjs/plugin-include': @@ -4869,6 +4913,14 @@ snapshots: cosmokit: 1.8.1 js-yaml: 4.2.0 + '@cordisjs/plugin-include@1.0.4(@cordisjs/plugin-loader@vendor+loader)(cordis@4.0.0-rc.6)': + dependencies: + '@cordisjs/plugin-loader': link:vendor/loader + cordis: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@vendor+loader) + cosmokit: 1.8.1 + js-yaml: 4.2.0 + optional: true + '@cordisjs/plugin-loader@1.0.0-rc.4(cordis@4.0.0-rc.6)': dependencies: cordis: 4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@1.0.0-rc.4) @@ -5881,6 +5933,14 @@ snapshots: '@cordisjs/plugin-include': 1.0.4(@cordisjs/plugin-loader@1.0.0-rc.4)(cordis@4.0.0-rc.6) '@cordisjs/plugin-loader': 1.0.0-rc.4(cordis@4.0.0-rc.6) + cordis@4.0.0-rc.6(@cordisjs/plugin-include@1.0.4)(@cordisjs/plugin-loader@vendor+loader): + dependencies: + '@standard-schema/spec': 1.1.0 + cosmokit: 1.8.1 + optionalDependencies: + '@cordisjs/plugin-include': 1.0.4(@cordisjs/plugin-loader@vendor+loader)(cordis@4.0.0-rc.6) + '@cordisjs/plugin-loader': link:vendor/loader + cordis@4.0.0-rc.6(@cordisjs/plugin-include@vendor+include)(@cordisjs/plugin-loader@vendor+loader): dependencies: '@standard-schema/spec': 1.1.0 diff --git a/tsconfig.build.json b/tsconfig.build.json index 9c236fe02a..b3d904aa9b 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -56,6 +56,8 @@ { "path": "./packages/ui/acp" }, { "path": "./packages/ui/acp-agent" }, { "path": "./packages/ui/app-boot" }, + { "path": "./packages/ui/jsonrpc" }, + { "path": "./packages/ui/jsonrpc-agent" }, { "path": "./packages/ui/stdio-agent" }, { "path": "./packages/support/llm-replay" }, { "path": "./packages/support/acp-snapshot" }, diff --git a/tsconfig.json b/tsconfig.json index 2ab1a5da30..a52f21a86e 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -67,6 +67,8 @@ { "path": "./packages/ui/acp" }, { "path": "./packages/ui/acp-agent" }, { "path": "./packages/ui/app-boot" }, + { "path": "./packages/ui/jsonrpc" }, + { "path": "./packages/ui/jsonrpc-agent" }, { "path": "./packages/ui/stdio-agent" }, { "path": "./packages/support/llm-replay" }, { "path": "./packages/support/acp-snapshot" },