Merge remote-tracking branch 'origin/master' into worktree/provider-routed-llm-adapters
# Conflicts: # docs/config-catalog.md # docs/cordis-catalog/events.md # docs/cordis-catalog/services.md # docs/core-data-structures/core.md # docs/event-producer-consumer.md # docs/persistence-catalog.md # examples/acp-agent/tests/snapshots/advanced-toolchain/session.1.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/session.2.jsonl # examples/acp-agent/tests/snapshots/advanced-toolchain/session.jsonl # examples/acp-agent/tests/snapshots/both-mode-turn/session.jsonl # examples/acp-agent/tests/snapshots/hook-cc-pretool-ask/session.jsonl # examples/acp-agent/tests/snapshots/skill-load/session.jsonl # examples/acp-agent/tests/snapshots/text-turn/session.jsonl # examples/sandbox-acp-agent/cordis.yml # examples/sandbox-acp-agent/tests/snapshots/escalation-approved/session.jsonl # examples/sandbox-acp-agent/tests/snapshots/escalation-rejected/session.jsonl # examples/sandbox-acp-agent/tests/snapshots/mode-switching/session.jsonl # packages/compact/compact-basic/README.md # packages/compact/compact-basic/src/index.ts # packages/compact/compact-basic/tests/compact-basic.spec.ts # packages/core/agent-loop/README.md # packages/core/agent-loop/src/loop.ts # packages/core/agent-loop/tests/properties.spec.ts # packages/core/session/README.md # packages/core/session/src/types.ts # packages/core/session/tests/derived-cache.spec.ts # packages/llm/llm-deepseek/src/index.ts # packages/llm/llm-pi-ai/README.md # packages/llm/llm-pi-ai/src/adapter.ts # packages/llm/llm-pi-ai/src/convert.ts # packages/llm/llm-pi-ai/tests/adapter.spec.ts # packages/llm/llm/README.md # packages/llm/llm/src/call-config.ts # packages/llm/llm/src/index.ts # packages/ui/acp-agent/src/index.ts # packages/ui/acp/tests/harness.ts # packages/ui/jsonrpc/README.md # packages/ui/jsonrpc/src/server.ts # packages/ui/stdio-agent/README.md # packages/ui/stdio-agent/src/index.ts # python/sdk/README.i18n.yaml
This commit is contained in:
@@ -1,23 +1,37 @@
|
||||
# @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).
|
||||
Stdio JSON-RPC plugin for out-of-process SDK clients such as Python `deepseek_harness`. [`HarnessSdkServer`](src/server.ts) handles `initialize` → `session/prompt` → `shutdown` plus session and subagent notifications over [`JsonRpcLineTransport`](src/transport.ts). This package owns the protocol; [`jsonrpc-agent`](../jsonrpc-agent/README.md) boots the external `cordis.yml` that chooses the surrounding runtime. See the [single-executable RFC](../../../docs/rfc/implemented/architecture/2026-07-10-single-file-executable-sdk-runtime-distribution.md) for the distribution design.
|
||||
|
||||
## Wiring
|
||||
|
||||
`inject: ['agents']` — the server creates one agent per SDK `sessionId` (get-or-create on `session/prompt`) from the `initialize.provider`+`initialize.model` pair and demuxes `subagent/end` through the registry. The LLM seam is read opportunistically via `ctx.get('llm')` (not injected): an already registered adapter for the provider route wins; when the route is `deepseek` and unowned, the plugin mounts `dsh-llm-deepseek` (credentials from `$DEEPSEEK_API_KEY` / `$DEEPSEEK_BASE_URL`); any other unowned provider fails initialization. Everything else — persistence, the tool stacks, the adapter set — comes from the surrounding `cordis.yml`.
|
||||
`inject: ['agents']`. The server gets or creates one agent per `sessionId` from the `initialize.provider`/`initialize.model` pair and demuxes `subagent/end` through the registry. A registered owner for the provider route wins; an unowned `deepseek` route mounts `dsh-llm-deepseek` using `$DEEPSEEK_API_KEY` and `$DEEPSEEK_BASE_URL`, while any other unowned provider fails initialization. Persistence, tools, and other adapters come 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`.
|
||||
No `cordis.yml` keys. `JsonRpcConfig.input`, `output`, and `exit` are test-only runtime seams; production uses process stdio and `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.
|
||||
stdout carries only JSON-RPC frames. The loading config must omit stdout loggers; 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.
|
||||
A `shutdown` request flushes its response, disposes the plugin fiber, then exits 0. Disposal idempotently shuts down every SDK-created agent to quiescence, detaches subscriptions, and closes the transport. Bare fiber disposal only stops serving; it does not exit. The app bin owns root disposal for stdin EOF (0), SIGTERM (0), and SIGINT (130).
|
||||
|
||||
## Wire notes
|
||||
|
||||
`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (SDK clients key on it, independent of this package's name). A session accepts at most one in-flight `session/prompt`; an overlapping prompt for the same `sessionId` fails immediately through the standard handler-error response, while other sessions remain independent and the same session can be reused after the active prompt settles. Persistence roots and the deployment persona come from `cordis.yml`; the wire exposes only parameters the server applies.
|
||||
`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime`. Each session permits one in-flight prompt; overlap fails immediately, other sessions remain independent, and the session is reusable after settlement. Persistence roots and deployment persona remain in `cordis.yml`.
|
||||
|
||||
## Model Experience
|
||||
|
||||
### SDK user message
|
||||
|
||||
**What the model sees**: For each accepted `session/prompt`, the conversation model receives the caller-supplied `contentBlocks` verbatim as one user message in that SDK session. This package adds no system-prompt prose or tool schema; those come from the plugins in the surrounding `cordis.yml`.
|
||||
|
||||
**Token effect**: Data-dependent user-message tokens enter retained session history and are resent on later turns until another package compacts them. The JSON-RPC frames, session notifications, and server bookkeeping add zero model-context tokens.
|
||||
|
||||
## Known Limitations and Deferred Work
|
||||
|
||||
- **The wire has no per-session close or prompt-cancel method** — SDK-created agents remain live until process shutdown, and one accepted prompt runs to agent idle before that session accepts another.
|
||||
- **stdout purity is deployment-enforced** — a surrounding config can still load a stdout logger and corrupt the JSON-RPC channel; this plugin does not inspect or veto sibling loggers.
|
||||
- **Automatic adapter mounting is DeepSeek-specific** — `initialize` can reuse any pre-registered model adapter, but its only fallback mounts `dsh-llm-deepseek`.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"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",
|
||||
"description": "Stdio JSON-RPC server plugin for out-of-process DeepSeek Harness SDK clients",
|
||||
"version": "0.0.1",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
|
||||
@@ -1,30 +1,10 @@
|
||||
/**
|
||||
* 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).
|
||||
* SDK-facing JSON-RPC plugin over stdio. An external `cordis.yml` decides
|
||||
* whether to load it; see the single-executable RFC and package README.
|
||||
* Stdout is reserved for protocol frames, so the tree must not load a stdout logger.
|
||||
* This plugin answers `shutdown`, disposes its own fiber, and exits 0; the app bin
|
||||
* owns EOF and signal exits. Keep named plugin exports with no default export so
|
||||
* Loader `unwrapExports` preserves `name`, `inject`, `Config`, and `apply`.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-jsonrpc
|
||||
*/
|
||||
@@ -39,65 +19,29 @@ 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.
|
||||
// Only the agent factory is required; initialize reads the optional LLM seam with ctx.get().
|
||||
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`).
|
||||
*/
|
||||
/** Runtime-only test seams; no field is configurable from `cordis.yml`. */
|
||||
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.
|
||||
*/
|
||||
/** Transport input override; production uses `process.stdin`. */
|
||||
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.
|
||||
*/
|
||||
/** Transport output override; production uses `process.stdout`. */
|
||||
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.
|
||||
*/
|
||||
/** Process-exit override; production uses `process.exit`. */
|
||||
exit?: (code: number) => void
|
||||
}
|
||||
|
||||
export const Config: Schema<JsonRpcConfig> = 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, an
|
||||
* explicit output-write barrier confirms the response frame flushed, 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).
|
||||
* Serve SDK requests over the configured streams. Effect disposal shuts down
|
||||
* SDK-created agents and closes the transport. A `shutdown` response is flushed
|
||||
* before this plugin's fiber is disposed and the process exits 0; the app bin
|
||||
* owns root-context disposal for EOF and signals.
|
||||
*/
|
||||
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).
|
||||
// The later transport callback must dispose this plugin's fiber, not its ambient context.
|
||||
const fiber = ctx.fiber
|
||||
/* v8 ignore next -- production stdio wiring; tests always inject the runtime seams */
|
||||
const input = config.input ?? process.stdin
|
||||
@@ -109,10 +53,7 @@ export function apply(ctx: Context, config: JsonRpcConfig): void {
|
||||
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 shares the same task). Flush and disposal failures are
|
||||
// settled independently: once shutdown was answered, process exit is still
|
||||
// the honest outcome and neither failure may prevent the next teardown step.
|
||||
// Share one exit task and attempt flush and disposal independently before exiting.
|
||||
let exitTask: Promise<void> | undefined
|
||||
const disposeAndExit = (): Promise<void> => {
|
||||
exitTask ??= (async () => {
|
||||
@@ -126,9 +67,7 @@ export function apply(ctx: Context, config: JsonRpcConfig): void {
|
||||
transport.onRequest(async (method, params) => {
|
||||
const result = await server.handleRequest(method, params)
|
||||
if (method === 'shutdown') {
|
||||
// The transport writes the returned result after this handler resolves.
|
||||
// Schedule the explicit flush barrier after that write, then dispose this
|
||||
// plugin's fiber and exit 0 (see apply's doc).
|
||||
// Run after the handler result is written; the task then flushes, disposes, and exits.
|
||||
setImmediate(() => { void disposeAndExit() })
|
||||
}
|
||||
return result
|
||||
|
||||
@@ -1,13 +1,8 @@
|
||||
/**
|
||||
* `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`.
|
||||
* JSON-RPC methods and notifications for SDK clients. Requests are
|
||||
* `initialize`, repeated `session/prompt`, then `shutdown`; notifications carry
|
||||
* durable session events, settled turns, and subagent lineage/outcomes. The
|
||||
* external `cordis.yml` owns plugins, persistence, and the adapter set.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-jsonrpc/server
|
||||
*/
|
||||
@@ -22,7 +17,7 @@ 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). */
|
||||
/** One-time SDK initialization parameters. */
|
||||
export interface InitializeParams {
|
||||
/** Working directory recorded on every SDK-created session's header. */
|
||||
cwd: string
|
||||
@@ -32,7 +27,7 @@ export interface InitializeParams {
|
||||
model: string
|
||||
}
|
||||
|
||||
/** Result of the `initialize` request: the server's identity for the SDK handshake. */
|
||||
/** SDK handshake result. */
|
||||
export interface InitializeResult {
|
||||
/** Wire-stable server identity (`deepseek-harness-sdk-runtime`) and version. */
|
||||
serverInfo: { name: string; version: string }
|
||||
@@ -49,7 +44,7 @@ export interface SessionPromptParams {
|
||||
contentBlocks: ContentBlock[]
|
||||
}
|
||||
|
||||
/** Result of a `session/prompt` request: the prompt ran to turn settle (outcome rides on `session.finished`). */
|
||||
/** Accepted prompt result; the outcome is reported by `session.finished`. */
|
||||
export interface SessionPromptResult {
|
||||
/** Always `true`; the turn outcome is the paired `session.finished` notification. */
|
||||
accepted: true
|
||||
@@ -67,11 +62,9 @@ interface SubagentRecord {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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`.
|
||||
* SDK server over one booted harness context and transport peer. Construction
|
||||
* subscribes to session, agent, and subagent lifecycle events until shutdown;
|
||||
* reinitialization is unsupported.
|
||||
*/
|
||||
export class HarnessSdkServer {
|
||||
private cwd = process.cwd()
|
||||
@@ -104,8 +97,7 @@ export class HarnessSdkServer {
|
||||
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.
|
||||
// Cache lineage before child disposal removes the agent from the registry.
|
||||
this.disposers.push(ctx.on('agent/created', (agent) => {
|
||||
this.subagentSessions.set(String(agent.id), {
|
||||
childSessionId: String(agent.session.id),
|
||||
@@ -135,10 +127,8 @@ export class HarnessSdkServer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle `initialize`: record the SDK deployment facts and, when provider
|
||||
* `deepseek` has no registered owner, mount the native DeepSeek adapter
|
||||
* (credentials from `$DEEPSEEK_API_KEY`/`$DEEPSEEK_BASE_URL`). Other missing
|
||||
* providers fail without guessing an implementation.
|
||||
* Record cwd and provider/model, mounting the DeepSeek adapter only when the
|
||||
* `deepseek` provider route has no configured owner.
|
||||
* @param params - the SDK handshake parameters.
|
||||
* @returns the server identity for the handshake.
|
||||
*/
|
||||
@@ -154,11 +144,9 @@ export class HarnessSdkServer {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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. A session accepts at
|
||||
* most one prompt at a time; an overlapping request fails immediately while
|
||||
* other sessions remain independent.
|
||||
* Get or create the session agent, send the prompt, await quiescence, then
|
||||
* notify `session.finished`. A session accepts one prompt at a time; other
|
||||
* sessions remain independent.
|
||||
* @param params - the target session id and prompt content.
|
||||
* @returns `{ accepted: true }` after the turn settled.
|
||||
*/
|
||||
@@ -183,10 +171,8 @@ export class HarnessSdkServer {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Dispose SDK-created agents to quiescence, unmount the server-mounted adapter,
|
||||
* and detach subscriptions. The surrounding context remains running.
|
||||
* @returns an empty object (the JSON-RPC result).
|
||||
*/
|
||||
shutdown(): Promise<Record<string, never>> {
|
||||
@@ -224,8 +210,8 @@ export class HarnessSdkServer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Dispatch one incoming JSON-RPC request to its typed handler. Throws (→ a
|
||||
* JSON-RPC error response) on an unknown method.
|
||||
* Dispatch an incoming request; unknown methods throw for transport conversion
|
||||
* to a JSON-RPC error response.
|
||||
* @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.
|
||||
|
||||
@@ -1,10 +1,7 @@
|
||||
/**
|
||||
* 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.
|
||||
* Newline-delimited JSON-RPC 2.0 over byte streams. Frames with `id` and
|
||||
* `method` are requests, `id` alone is a response, and `method` alone is a
|
||||
* notification. Malformed lines are ignored; handler failures become error frames.
|
||||
*
|
||||
* @module @deepseek-ai/dsh-jsonrpc/transport
|
||||
*/
|
||||
@@ -18,22 +15,18 @@ type RequestHandler = (method: string, params: Record<string, unknown>) => Promi
|
||||
type NotificationHandler = (method: string, params: Record<string, unknown>) => 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.
|
||||
* Outbound request and notification surface used by {@link HarnessSdkServer}.
|
||||
*/
|
||||
export interface JsonRpcTransportPeer {
|
||||
/**
|
||||
* Send a request to the remote peer and await its response.
|
||||
* Send a request 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.
|
||||
* @returns the result; rejects on an error response, write failure, or closure.
|
||||
*/
|
||||
request(method: string, params: Record<string, unknown>): Promise<unknown>
|
||||
/**
|
||||
* Send a notification (no response expected). An omitted `params` sends no
|
||||
* `params` member at all.
|
||||
* Send a notification; omitted params produce no `params` member.
|
||||
* @param method - the JSON-RPC method name.
|
||||
* @param params - the optional notification parameters object.
|
||||
*/
|
||||
@@ -46,14 +39,10 @@ interface PendingRequest {
|
||||
}
|
||||
|
||||
/**
|
||||
* 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.
|
||||
* Line-delimited endpoint over caller-owned streams. {@link start} attaches
|
||||
* listeners; {@link close} detaches them and rejects pending requests without
|
||||
* destroying the streams. Missing request handlers return `-32601`; handler
|
||||
* failures return `-32603`. Notifications without a handler are dropped.
|
||||
*/
|
||||
export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
private buffer = ''
|
||||
@@ -78,8 +67,7 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Detach the input listeners and reject every pending outgoing request with
|
||||
* "JSON-RPC transport closed". Safe to call without a prior {@link start}.
|
||||
* Detach listeners and reject pending requests. Safe before {@link start}.
|
||||
*/
|
||||
close(): void {
|
||||
this.input.off('data', this.onData)
|
||||
@@ -89,7 +77,7 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Install THE handler for incoming requests (a later call replaces it).
|
||||
* Install the request handler, replacing any prior handler.
|
||||
* @param handler - resolves to the response `result`; a rejection becomes a
|
||||
* `-32603` error response carrying the message.
|
||||
*/
|
||||
@@ -98,7 +86,7 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Install THE handler for incoming notifications (a later call replaces it).
|
||||
* Install the notification handler, replacing any prior handler.
|
||||
* @param handler - invoked per notification with the method and normalized
|
||||
* params object.
|
||||
*/
|
||||
@@ -125,9 +113,7 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
}
|
||||
|
||||
/**
|
||||
* Wait until every frame written before this call has reached the output's
|
||||
* write callback. The empty queued write is a barrier and emits no protocol
|
||||
* bytes.
|
||||
* Wait for prior frame write callbacks. The empty barrier emits no bytes.
|
||||
* @returns a promise that settles with the output write callback.
|
||||
*/
|
||||
flush(): Promise<void> {
|
||||
@@ -170,8 +156,7 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
|
||||
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.
|
||||
// Only JSON syntax errors reach this catch; malformed peer lines are ignored.
|
||||
return
|
||||
}
|
||||
if (!message || typeof message !== 'object') return
|
||||
|
||||
@@ -11,21 +11,12 @@ 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`.
|
||||
* Mount the real namespace plugin with in-memory stdio and exit seams. Covers
|
||||
* the full transport/server path, response-before-exit shutdown exactly once,
|
||||
* and bare-fiber disposal without process exit.
|
||||
*/
|
||||
|
||||
/** One ordered observation on the plugin's outward-facing seams: a JSON-RPC frame written to `output`, or an `exit(code)` call. */
|
||||
/** One ordered frame, write completion, or exit observation. */
|
||||
type WireEvent =
|
||||
| { kind: 'frame'; frame: Record<string, unknown> }
|
||||
| { kind: 'write-complete'; ids: (string | number)[] }
|
||||
@@ -33,9 +24,9 @@ type WireEvent =
|
||||
|
||||
interface ApplyHarness {
|
||||
ctx: Context
|
||||
/** The jsonrpc plugin's own fiber (NOT the root), for the HMR-style dispose scenario. */
|
||||
/** The plugin fiber used by the bare-dispose case. */
|
||||
fiber: Awaited<ReturnType<Context['plugin']>>
|
||||
/** Every output frame and exit call, in observation order — ordering assertions read this. */
|
||||
/** Frames, write completions, and exits in observation order. */
|
||||
events: WireEvent[]
|
||||
outputErrors: Error[]
|
||||
send(frame: Record<string, unknown>): void
|
||||
@@ -46,7 +37,7 @@ interface ApplyHarness {
|
||||
dispose(): Promise<void>
|
||||
}
|
||||
|
||||
/** Poll `get` until it yields a value (5s cap) — the output side is fed asynchronously from the transport's read loop. */
|
||||
/** Poll asynchronous output for up to five seconds. */
|
||||
async function waitFor<T>(get: () => T | undefined, description: string): Promise<T> {
|
||||
const deadline = Date.now() + 5000
|
||||
for (;;) {
|
||||
@@ -57,16 +48,12 @@ async function waitFor<T>(get: () => T | undefined, description: string): Promis
|
||||
}
|
||||
}
|
||||
|
||||
/** Let pending microtasks, setImmediate callbacks, and stream events drain — for asserting that something did NOT happen. */
|
||||
/** Drain asynchronous work before a negative assertion. */
|
||||
async function settle(): Promise<void> {
|
||||
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.
|
||||
*/
|
||||
/** Mount the real plugin on a minimal harness with in-memory stdio and exit. */
|
||||
async function mountPlugin(
|
||||
storageDir: string,
|
||||
options: { writeDelayMs?: number; failFlush?: boolean } = {},
|
||||
@@ -80,9 +67,8 @@ async function mountPlugin(
|
||||
const events: WireEvent[] = []
|
||||
const outputErrors: Error[] = []
|
||||
let pendingOutput = ''
|
||||
// A hand-rolled Writable (not a PassThrough): _write records frames on
|
||||
// admission and write-complete only when its callback fires, so a delayed
|
||||
// output proves exit waits for the transport's flush barrier.
|
||||
// Record frame admission separately from write completion so delayed output
|
||||
// tests the flush barrier.
|
||||
const output = new Writable({
|
||||
write(chunk: Buffer, _encoding, callback) {
|
||||
const ids: (string | number)[] = []
|
||||
@@ -138,7 +124,7 @@ afterEach(async () => {
|
||||
vi.unstubAllEnvs()
|
||||
})
|
||||
|
||||
/** The server.spec mock OpenAI-compatible SSE endpoint, so a prompt turn completes without a real key. */
|
||||
/** Keyless SSE endpoint for completing a prompt turn. */
|
||||
async function mockCompletionServer(): Promise<{ url: string; requests: unknown[] }> {
|
||||
const requests: unknown[] = []
|
||||
const server = createServer((request: IncomingMessage, response: ServerResponse) => {
|
||||
@@ -206,8 +192,7 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
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.
|
||||
// Notifications use the same transport and arrive as id-less frames.
|
||||
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({
|
||||
@@ -224,9 +209,7 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-shutdown-'))
|
||||
const harness = await mountPlugin(storageDir, { writeDelayMs: 10 })
|
||||
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.
|
||||
// One chunk makes the two deferred exit callbacks race.
|
||||
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`)
|
||||
@@ -234,8 +217,7 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit recorder call')
|
||||
expect(harness.exits()).toEqual([0])
|
||||
|
||||
// Response-then-exit ordering: both response write callbacks and the
|
||||
// empty flush barrier complete before exit(0), even on delayed output.
|
||||
// Both response writes and the flush barrier complete before exit.
|
||||
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')
|
||||
@@ -250,11 +232,9 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
expect(flushComplete).toBeGreaterThan(secondComplete)
|
||||
expect(exitIndex).toBeGreaterThan(flushComplete)
|
||||
|
||||
// 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, provider: 'deepseek', model: 'x' } })
|
||||
await settle()
|
||||
@@ -290,8 +270,7 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
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).
|
||||
// Prove the handler-rejection path is live before disposal.
|
||||
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({
|
||||
@@ -301,8 +280,6 @@ describe('dsh-jsonrpc plugin apply', () => {
|
||||
|
||||
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, provider: 'deepseek', model: 'x' } })
|
||||
await settle()
|
||||
|
||||
@@ -3,21 +3,11 @@ 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.
|
||||
* Run the real namespace export through `Loader.unwrapExports`; a stray
|
||||
* default would discard `name`, `inject`, `Config`, and `apply`.
|
||||
*/
|
||||
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')
|
||||
|
||||
|
||||
Reference in New Issue
Block a user