Merge remote-tracking branch 'origin/master' into unify-acp-example

# Conflicts: # docs/config-catalog.md # docs/module-graph.md # pnpm-lock.yaml
2026-07-14 00:56:28 +08:00
parent 95635dfa66 fdc5dea100
commit cdc3522e48
131 changed files with 11181 additions and 188 deletions
--- a/packages/ui/README.md
+++ b/packages/ui/README.md
@@ -11,10 +11,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.
--- a/packages/ui/acp-agent/src/index.ts
+++ b/packages/ui/acp-agent/src/index.ts
@@ -72,6 +72,8 @@ export const Config: z<Config> = z.object({
  // schemastery's native [] default would read as an invalid configured list.
  toolOrder: z.array(z.string()).default(undefined as unknown as string[]),
  tools: ToolRegistry.Config,
+  // TODO(single-default-literal): share this schema default and the defensive
+  // apply() fallback through one named constant while retaining both boundaries.
  persistenceRoot: z.string().default('./.sessions'),
  skills: agentCore.SkillConfigSchema,
 })
--- a/packages/ui/acp/src/index.ts
+++ b/packages/ui/acp/src/index.ts
@@ -105,6 +105,8 @@ export const name = 'acp'
 // because `initialize` advertises `loadSession: true`. `tools` lets a tool own
 // how its calls render (`presentCall`/`presentResult`); the bridge looks up the
 // definition by name and falls back to a generic presentation when absent.
+// TODO(acp-session-inject): drop `sessions`; this bridge never reads
+// ctx.sessions, and agent/session ownership is already behind ctx.agents.
 export const inject = ['agents', 'sessions', 'sessionPersistence', 'tools', 'userInteraction']

 /**
@@ -350,10 +352,13 @@ export function apply(ctx: Context, config: AcpConfig): void {
  // this warn sink so a throwing tool presenter is logged, not propagated.
  const makePresenter = (agent?: Agent): ToolPresenter => new ToolPresenter(tools, (message) => { logger.warn(message) }, agent)

+  // TODO(derive-acp-session-id): derive an event's id from agent.session and
+  // verify sessions.get(id)?.agent === agent; then remove this reverse map and
+  // SessionRecord.sessionId, whose sole read duplicates the same identity.
  // Live sessions keyed by id (RFC 011 multi-session), plus an agent→sessionId
  // reverse map so `agent/*` events (which carry only the Agent) demux in O(1).
-  // The two stay in lockstep: a record is added to `sessions` and the agent to
-  // `bySession` together, and removed together.
+  // The forward record and weak reverse entry are installed together; removing
+  // the record releases its strong Agent reference, so the WeakMap entry expires.
  const sessions = new Map<SessionId, SessionRecord>()
  const bySession = new WeakMap<Agent, SessionId>()
  // Session ids whose `session/load` is mid-`resume()` (the slot is reserved
--- a/packages/ui/acp/tests/stream-update.spec.ts
+++ b/packages/ui/acp/tests/stream-update.spec.ts
@@ -30,6 +30,21 @@ function registryOf(...tools: ToolDefinition[]): Pick<ToolRegistryType, 'get'> {
  return { get: name => map.get(name) }
 }

+function updatesWith(presenter: ToolPresenter, ...events: SessionEvent[]): SessionNotification['update'][] {
+  const out: SessionNotification['update'][] = []
+  for (const event of events) streamSessionEventUpdate(SessionId('s1'), event, n => out.push(n.update), presenter)
+  return out
+}
+
+async function fsCtx(): Promise<Context> {
+  const ctx = new Context()
+  await ctx.plugin(SystemPrompt)
+  await ctx.plugin(ToolRegistry)
+  await ctx.plugin(FsLocal)
+  await ctx.plugin(ToolFs)
+  return ctx
+}
+
 function evt<T extends SessionEvent['type']>(type: T, data: Extract<SessionEvent, { type: T }>['data']): SessionEvent {
  return { type, seq: 0, time: 0, data } as SessionEvent
 }
@@ -181,12 +196,6 @@ describe('ToolPresenter (tool-owned presentation via the tool registry)', () =>
    }),
  }

-  function updatesWith(presenter: ToolPresenter, ...events: SessionEvent[]): SessionNotification['update'][] {
-    const out: SessionNotification['update'][] = []
-    for (const event of events) streamSessionEventUpdate(SessionId('s1'), event, n => out.push(n.update), presenter)
-    return out
-  }
-
  it('tool/call uses the tool: description→title, command→rawInput, tool kind', () => {
    const presenter = new ToolPresenter(registryOf(bashLike))
    const [update] = updatesWith(presenter, evt('tool/call', {
@@ -627,21 +636,6 @@ describe('result-time diff card (REAL fs edit tool → tool_call_update diff blo
  // result card the bridge forwards as `{ type: 'diff' }` content blocks. Uses
  // the REAL tool (not a stand-in) per the anti-mock convention, mirroring the
  // call-side diff test above.
-  async function fsCtx(): Promise<Context> {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRegistry)
-    await ctx.plugin(FsLocal)
-    await ctx.plugin(ToolFs)
-    return ctx
-  }
-
-  function updatesWith(presenter: ToolPresenter, ...events: SessionEvent[]): SessionNotification['update'][] {
-    const out: SessionNotification['update'][] = []
-    for (const event of events) streamSessionEventUpdate(SessionId('s1'), event, n => out.push(n.update), presenter)
-    return out
-  }
-
  it('forwards the applied-hunk meta onto the wire as tool_call_update diff content', async () => {
    const ctx = await fsCtx()
    const presenter = new ToolPresenter(ctx.tools)
@@ -739,14 +733,6 @@ describe('relative-path display titles (bridge relativizes the title against the
  // diff paths RAW. Drive it with the REAL fs tools so the title/locations come
  // from the shipping presentCall, and pass an ABSOLUTE file path (which a real
  // editor forwards). The presenter is pure/args-only; the cwd is known only here.
-  async function fsCtx(): Promise<Context> {
-    const ctx = new Context()
-    await ctx.plugin(SystemPrompt)
-    await ctx.plugin(ToolRegistry)
-    await ctx.plugin(FsLocal)
-    await ctx.plugin(ToolFs)
-    return ctx
-  }
  function callUpdate(ctx: Context, sessionCwd: string | undefined, name: string, args: unknown): SessionNotification['update'] {
    const presenter = new ToolPresenter(ctx.tools)
    const out: SessionNotification['update'][] = []
--- a/packages/ui/jsonrpc-agent/README.md
+++ 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 <path/to/cordis.yml>`, 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).
--- a/packages/ui/jsonrpc-agent/package.json
+++ 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"
+  }
+}
--- a/packages/ui/jsonrpc-agent/src/bin.ts
+++ 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} <path/to/cordis.yml> (or set DSH_CORDIS_CONFIG=<path>, 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<void> {
+  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 */
--- a/packages/ui/jsonrpc-agent/src/index.ts
+++ 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 {}
--- a/packages/ui/jsonrpc-agent/tsconfig.json
+++ 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"
+    }
+  ]
+}
--- a/packages/ui/jsonrpc-agent/tsdown.config.ts
+++ 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,
+})
--- a/packages/ui/jsonrpc/README.md
+++ 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). 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.
--- a/packages/ui/jsonrpc/package.json
+++ 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"
+  }
+}
--- a/packages/ui/jsonrpc/src/index.ts
+++ b/packages/ui/jsonrpc/src/index.ts
@@ -0,0 +1,144 @@
+/**
+ * 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<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).
+ */
+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 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.
+  let exitTask: Promise<void> | undefined
+  const disposeAndExit = (): Promise<void> => {
+    exitTask ??= (async () => {
+      await Promise.allSettled([Promise.resolve().then(() => transport.flush())])
+      await Promise.allSettled([Promise.resolve().then(() => fiber.dispose())])
+      exit(0)
+    })()
+    return exitTask
+  }
+
+  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).
+      setImmediate(() => { void disposeAndExit() })
+    }
+    return result
+  })
+
+  ctx.effect(() => {
+    transport.start()
+    return async () => {
+      await server.shutdown()
+      transport.close()
+    }
+  }, 'jsonrpc.serve')
+}
--- a/packages/ui/jsonrpc/src/server.ts
+++ b/packages/ui/jsonrpc/src/server.ts
@@ -0,0 +1,276 @@
+/**
+ * `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 { resolve } from 'node:path'
+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
+}
+
+/** 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,
+ * with at most one in flight per 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[]
+}
+
+/** 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
+  activePrompt: boolean
+}
+
+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<void> } | undefined
+  private readonly sessions = new Map<string, SessionRecord>()
+  private readonly sessionCreations = new Map<string, Promise<SessionRecord>>()
+  private readonly subagentSessions = new Map<string, SubagentRecord>()
+  private readonly disposers: (() => void)[] = []
+  private shutdownTask: Promise<Record<string, never>> | undefined
+  private shuttingDown = false
+
+  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' ? '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<InitializeResult> {
+    this.cwd = resolve(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. A session accepts at
+   * most one prompt at a time; an overlapping request fails immediately while
+   * other sessions remain independent.
+   * @param params - the target session id and prompt content.
+   * @returns `{ accepted: true }` after the turn settled.
+   */
+  async prompt(params: SessionPromptParams): Promise<SessionPromptResult> {
+    const rec = await this.getOrCreateSession(params.sessionId)
+    if (rec.activePrompt) throw new Error(`session already has an active prompt: ${params.sessionId}`)
+    rec.activePrompt = true
+    try {
+      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 }
+    } finally {
+      rec.activePrompt = false
+    }
+  }
+
+  /**
+   * 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).
+   */
+  shutdown(): Promise<Record<string, never>> {
+    this.shutdownTask ??= this.performShutdown()
+    return this.shutdownTask
+  }
+
+  private async performShutdown(): Promise<Record<string, never>> {
+    this.shuttingDown = true
+    const pendingCreations = [...this.sessionCreations.values()]
+    await Promise.allSettled(pendingCreations)
+    this.sessionCreations.clear()
+    const records = [...this.sessions.values()]
+    this.sessions.clear()
+    this.subagentSessions.clear()
+    const failures: unknown[] = []
+    while (this.disposers.length > 0) {
+      try {
+        this.disposers.pop()?.()
+      } catch (error) {
+        failures.push(error)
+      }
+    }
+    const teardownResults = await Promise.allSettled([
+      ...records.map(rec => Promise.resolve().then(() => rec.handle.dispose())),
+      ...(this.llmFiber === undefined ? [] : [Promise.resolve().then(() => this.llmFiber?.dispose())]),
+    ])
+    this.llmFiber = undefined
+    failures.push(...teardownResults
+      .filter((result): result is PromiseRejectedResult => result.status === 'rejected')
+      .map(result => result.reason as unknown))
+    if (failures.length === 1) throw failures[0]
+    if (failures.length > 1) throw new AggregateError(failures, 'SDK server teardown failed')
+    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<string, unknown> | undefined): Promise<unknown> {
+    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 async getOrCreateSession(sessionId: string): Promise<SessionRecord> {
+    if (this.shuttingDown) throw new Error('SDK server is shutting down')
+    const existing = this.sessions.get(sessionId)
+    if (existing) return existing
+    const pending = this.sessionCreations.get(sessionId)
+    if (pending) return pending
+    const creation = this.createSession(sessionId)
+    this.sessionCreations.set(sessionId, creation)
+    void creation.then(
+      () => { this.sessionCreations.delete(sessionId) },
+      () => { this.sessionCreations.delete(sessionId) },
+    )
+    return creation
+  }
+
+  private async createSession(sessionId: string): Promise<SessionRecord> {
+    const handle = await this.ctx.agents.create({
+      agentId: AgentId(sessionId),
+      sessionId: SessionId(sessionId),
+      meta: { cwd: this.cwd },
+      agentOptions: { model: this.model },
+    })
+    const rec: SessionRecord = { handle, lastTurnEnd: undefined, activePrompt: false }
+    this.sessions.set(sessionId, rec)
+    return rec
+  }
+
+  private finishedStatus(reason: TurnEndReason | undefined): 'ok' | 'error' {
+    if (!reason) return 'error'
+    return reason.kind === 'completed' ? 'ok' : 'error'
+  }
+
+  private hasAdapterFor(model: string): boolean {
+    return this.ctx.get('llm')?.models().includes(model) ?? false
+  }
+}
--- a/packages/ui/jsonrpc/src/transport.ts
+++ b/packages/ui/jsonrpc/src/transport.ts
@@ -0,0 +1,238 @@
+/**
+ * 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'
+import { StringDecoder } from 'node:string_decoder'
+
+type JsonRpcId = string | number
+type RequestHandler = (method: string, params: Record<string, unknown>) => Promise<unknown>
+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.
+ */
+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<string, unknown>): Promise<unknown>
+  /**
+   * 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<string, unknown>): 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 readonly decoder = new StringDecoder('utf8')
+  private started = false
+  private requestHandler: RequestHandler | undefined
+  private notificationHandler: NotificationHandler | undefined
+  private readonly pending = new Map<JsonRpcId, PendingRequest>()
+
+  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<string, unknown>): Promise<unknown> {
+    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<string, unknown>): void {
+    this.write(params === undefined ? { jsonrpc: '2.0', method } : { jsonrpc: '2.0', method, params })
+  }
+
+  /**
+   * 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.
+   * @returns a promise that settles with the output write callback.
+   */
+  flush(): Promise<void> {
+    return new Promise<void>((resolve, reject) => {
+      this.output.write('', (error) => {
+        if (error) reject(error)
+        else resolve()
+      })
+    })
+  }
+
+  private readonly onData = (chunk: Buffer | string): void => {
+    this.buffer += typeof chunk === 'string' ? chunk : this.decoder.write(chunk)
+    this.drainLines()
+  }
+
+  private drainLines(): void {
+    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.buffer += this.decoder.end()
+    this.drainLines()
+    this.failPending(new Error('JSON-RPC input closed'))
+  }
+
+  private async handleLine(line: string): Promise<void> {
+    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<string, unknown>
+    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<string, unknown>): Promise<void> {
+    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<string, unknown>): 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<string, unknown>
+      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<string, unknown>): 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<string, unknown> {
+  return params && typeof params === 'object' && !Array.isArray(params) ? params as Record<string, unknown> : {}
+}
--- a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
+++ b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
@@ -0,0 +1,316 @@
+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<string, unknown> }
+  | { kind: 'write-complete'; ids: (string | number)[] }
+  | { 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<ReturnType<Context['plugin']>>
+  /** Every output frame and exit call, in observation order — ordering assertions read this. */
+  events: WireEvent[]
+  outputErrors: Error[]
+  send(frame: Record<string, unknown>): void
+  sendRaw(text: string): void
+  frames(): Record<string, unknown>[]
+  exits(): number[]
+  waitForFrame(predicate: (frame: Record<string, unknown>) => boolean, description: string): Promise<Record<string, unknown>>
+  dispose(): Promise<void>
+}
+
+/** Poll `get` until it yields a value (5s cap) — the output side is fed asynchronously from the transport's read loop. */
+async function waitFor<T>(get: () => T | undefined, description: string): Promise<T> {
+  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<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.
+ */
+async function mountPlugin(
+  storageDir: string,
+  options: { writeDelayMs?: number; failFlush?: boolean } = {},
+): Promise<ApplyHarness> {
+  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[] = []
+  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.
+  const output = new Writable({
+    write(chunk: Buffer, _encoding, callback) {
+      const ids: (string | number)[] = []
+      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) {
+          const frame = JSON.parse(line) as Record<string, unknown>
+          events.push({ kind: 'frame', frame })
+          if (typeof frame.id === 'string' || typeof frame.id === 'number') ids.push(frame.id)
+        }
+      }
+      const complete = (): void => {
+        if (options.failFlush === true && chunk.length === 0) {
+          callback(new Error('flush callback failed'))
+          return
+        }
+        events.push({ kind: 'write-complete', ids })
+        callback()
+      }
+      if ((options.writeDelayMs ?? 0) > 0) setTimeout(complete, options.writeDelayMs)
+      else complete()
+    },
+  })
+  output.on('error', (error: Error) => { outputErrors.push(error) })
+  const exit = (code: number): void => { events.push({ kind: 'exit', code }) }
+
+  const fiber = await ctx.plugin(jsonrpc, { input, output, exit })
+
+  const frames = (): Record<string, unknown>[] =>
+    events.flatMap(event => event.kind === 'frame' ? [event.frame] : [])
+  return {
+    ctx,
+    fiber,
+    events,
+    outputErrors,
+    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<void>(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, { 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.
+      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 response write callbacks and the
+      // empty flush barrier complete before exit(0), even on delayed output.
+      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')
+      const firstComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.includes('sd-1'))
+      const secondComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.includes('sd-2'))
+      const flushComplete = harness.events.findIndex(event => event.kind === 'write-complete' && event.ids.length === 0)
+      expect(firstResponse).toBeGreaterThanOrEqual(0)
+      expect(secondResponse).toBeGreaterThanOrEqual(0)
+      expect(firstComplete).toBeGreaterThan(firstResponse)
+      expect(secondComplete).toBeGreaterThan(secondResponse)
+      expect(flushComplete).toBeGreaterThan(firstComplete)
+      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, model: 'x' } })
+      await settle()
+      expect(harness.frames().length).toBe(before)
+    } finally {
+      await harness.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('still disposes and exits once when the flush callback fails', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-flush-failure-'))
+    const harness = await mountPlugin(storageDir, { failFlush: true })
+    try {
+      harness.send({ jsonrpc: '2.0', id: 'sd-fail', method: 'shutdown' })
+
+      await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit after flush failure')
+      await settle()
+      expect(harness.exits()).toEqual([0])
+      expect(harness.outputErrors.map(error => error.message)).toEqual(['flush callback failed'])
+
+      const before = harness.frames().length
+      harness.send({ jsonrpc: '2.0', id: 'after-flush-failure', 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 })
+    }
+  })
+})
--- a/packages/ui/jsonrpc/tests/plugin-shape.spec.ts
+++ 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<string, unknown>
+    expect(unwrapped).toBe(jsonrpc)
+    expect(unwrapped.name).toBe('jsonrpc')
+    expect(unwrapped.inject).toEqual(['agents'])
+    expect(unwrapped.Config).toBeDefined()
+    expect(typeof unwrapped.apply).toBe('function')
+  })
+})
--- a/packages/ui/jsonrpc/tests/server.spec.ts
+++ b/packages/ui/jsonrpc/tests/server.spec.ts
@@ -0,0 +1,572 @@
+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, type Agent, type AgentHandle } 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 SubagentService, { type SubagentRunEndInfo } from '@deepseek-ai/dsh-subagent'
+import { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts'
+
+class FakeTransport implements JsonRpcTransportPeer {
+  notifications: { method: string; params?: Record<string, unknown> }[] = []
+
+  async request(method: string, params: Record<string, unknown>): Promise<unknown> {
+    throw new Error(`the SDK server should not call host JSON-RPC method ${method} with ${JSON.stringify(params)}`)
+  }
+
+  notify(method: string, params?: Record<string, unknown>): 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<void>(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(SubagentService)
+  await ctx.plugin(SessionPersistenceJsonl, { root: storageDir })
+  await new Promise(resolve => setTimeout(resolve, 50))
+  return ctx
+}
+
+/** Drive the owning service so test lifecycle events carry the real parent scope. */
+async function settleSubagent(ctx: Context, parent: Agent, info: SubagentRunEndInfo): Promise<void> {
+  const disposeProvider = ctx.subagents.registerProvider({
+    name: info.provider,
+    capabilities: { outputSchema: false, depthLimit: false, toolFilter: false, persona: false },
+    inheritsParentContext: false,
+    async start() {
+      return {
+        id: info.id,
+        result: info.lastAssistantMessage === undefined
+          ? Promise.reject(new Error('synthetic infrastructure failure'))
+          : Promise.resolve({ output: info.lastAssistantMessage, stopReason: info.stopReason }),
+        dispose: () => Promise.resolve(),
+      }
+    },
+  })
+  try {
+    const run = await ctx.subagents.start(info.provider, {
+      parent,
+      prompt: [],
+      signal: new AbortController().signal,
+    })
+    await run.result.then(() => undefined, () => undefined)
+    await run.dispose()
+  } finally {
+    disposeProvider()
+  }
+}
+
+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',
+      }) 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' }],
+      })
+
+      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 = await 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('rejects overlapping prompts for one session without serializing other sessions', async () => {
+    let releaseMain: (() => void) | undefined
+    const firstMainIdle = new Promise<void>((resolve) => { releaseMain = resolve })
+    const mainWhenIdle = vi.fn<() => Promise<void>>()
+      .mockReturnValueOnce(firstMainIdle)
+      .mockResolvedValue(undefined)
+    const mainSend = vi.fn()
+    const mainAgent = {
+      send: mainSend,
+      whenIdle: mainWhenIdle,
+    } as unknown as Agent
+    const otherSend = vi.fn()
+    const otherAgent = {
+      send: otherSend,
+      whenIdle: vi.fn(() => Promise.resolve()),
+    } as unknown as Agent
+    const mainHandle = { agent: mainAgent, dispose: vi.fn(() => Promise.resolve()) }
+    const otherHandle = { agent: otherAgent, dispose: vi.fn(() => Promise.resolve()) }
+    const create = vi.fn(async (options: { agentId: AgentId }) =>
+      String(options.agentId) === 'main' ? mainHandle : otherHandle)
+    const ctx = {
+      on: vi.fn(() => () => undefined),
+      agents: { create, get: () => undefined },
+      get: () => undefined,
+    } as unknown as Context
+    const server = new HarnessSdkServer(ctx, new FakeTransport())
+    const prompt = (sessionId: string, text: string) => server.prompt({
+      sessionId,
+      contentBlocks: [{ type: 'text', text }],
+    })
+
+    const first = prompt('main', 'first')
+    await vi.waitFor(() => { expect(mainSend).toHaveBeenCalledOnce() })
+
+    await expect(prompt('main', 'overlap')).rejects.toThrow('session already has an active prompt: main')
+    await expect(prompt('other', 'independent')).resolves.toEqual({ accepted: true })
+    releaseMain?.()
+    await expect(first).resolves.toEqual({ accepted: true })
+    await expect(prompt('main', 'sequential')).resolves.toEqual({ accepted: true })
+
+    mainWhenIdle.mockRejectedValueOnce(new Error('turn wait failed'))
+    await expect(prompt('main', 'failing')).rejects.toThrow('turn wait failed')
+    await expect(prompt('main', 'after failure')).resolves.toEqual({ accepted: true })
+
+    expect(mainSend).toHaveBeenCalledTimes(4)
+    expect(otherSend).toHaveBeenCalledOnce()
+    await server.shutdown()
+    expect(mainHandle.dispose).toHaveBeenCalledOnce()
+    expect(otherHandle.dispose).toHaveBeenCalledOnce()
+  })
+
+  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 parentHandle = await ctx.agents.create({
+        agentId: AgentId('parent-agent'),
+        sessionId: SessionId('main'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'deepseek' },
+      })
+      const handle = await ctx.agents.create({
+        agentId: AgentId('child-agent'),
+        sessionId: SessionId('child-session'),
+        meta: { cwd: storageDir, parentSession: SessionId('main') },
+        agentOptions: { model: 'deepseek' },
+      })
+      await settleSubagent(ctx, parentHandle.agent, {
+        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 parentHandle.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 parentHandle: AgentHandle | undefined
+    let handle: AgentHandle | undefined
+    let failedHandle: AgentHandle | undefined
+    try {
+      parentHandle = await ctx.agents.create({
+        agentId: AgentId('fallback-parent-agent'),
+        sessionId: SessionId('fallback-parent'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'deepseek' },
+      })
+      handle = await ctx.agents.create({
+        agentId: AgentId('fallback-child-agent'),
+        sessionId: SessionId('fallback-child-session'),
+        meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
+        agentOptions: { model: 'deepseek' },
+      })
+      failedHandle = await ctx.agents.create({
+        agentId: AgentId('failed-child-agent'),
+        sessionId: SessionId('failed-child-session'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'deepseek' },
+      })
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+
+      await settleSubagent(ctx, parentHandle.agent, {
+        provider: 'fork',
+        id: AgentId('fallback-child-agent'),
+        stopReason: 'max-tokens',
+        lastAssistantMessage: [],
+      })
+      await settleSubagent(ctx, parentHandle.agent, {
+        provider: 'fork',
+        id: AgentId('failed-child-agent'),
+        stopReason: 'error',
+      })
+      await settleSubagent(ctx, parentHandle.agent, {
+        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: 'error',
+          stopReason: 'max-tokens',
+          lastAssistantMessage: [],
+        },
+      })
+      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 parentHandle?.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<Record<string, never>>
+      }
+
+      expect(server.finishedStatus(undefined)).toBe('error')
+      expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('error')
+      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<Record<string, never>>
+      }
+
+      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 })
+    }
+  })
+
+  it('coalesces concurrent session creation and retries a failed creation', async () => {
+    let resolveShared: ((handle: AgentHandle) => void) | undefined
+    const sharedCreation = new Promise<AgentHandle>((resolve) => { resolveShared = resolve })
+    const sharedHandle = { agent: {} as Agent, dispose: vi.fn(() => Promise.resolve()) }
+    const retryHandle = { agent: {} as Agent, dispose: vi.fn(() => Promise.resolve()) }
+    const create = vi.fn<(options: unknown) => Promise<AgentHandle>>()
+      .mockReturnValueOnce(sharedCreation)
+      .mockRejectedValueOnce(new Error('creation failed'))
+      .mockResolvedValueOnce(retryHandle)
+    const ctx = {
+      on: vi.fn(() => () => undefined),
+      agents: { create, get: () => undefined },
+      get: () => undefined,
+    } as unknown as Context
+    const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
+      getOrCreateSession(sessionId: string): Promise<{ handle: AgentHandle }>
+      shutdown(): Promise<Record<string, never>>
+    }
+
+    const first = server.getOrCreateSession('shared')
+    const second = server.getOrCreateSession('shared')
+    expect(create).toHaveBeenCalledTimes(1)
+    resolveShared?.(sharedHandle)
+    const [firstRecord, secondRecord] = await Promise.all([first, second])
+    expect(firstRecord).toBe(secondRecord)
+
+    await expect(server.getOrCreateSession('retry')).rejects.toThrow('creation failed')
+    await expect(server.getOrCreateSession('retry')).resolves.toMatchObject({ handle: retryHandle })
+    expect(create).toHaveBeenCalledTimes(3)
+
+    await server.shutdown()
+    expect(sharedHandle.dispose).toHaveBeenCalledOnce()
+    expect(retryHandle.dispose).toHaveBeenCalledOnce()
+    await expect(server.getOrCreateSession('after-shutdown')).rejects.toThrow('SDK server is shutting down')
+  })
+
+  it('resolves a relative cwd before creating the session', async () => {
+    const create = vi.fn<(options: unknown) => Promise<AgentHandle>>()
+      .mockResolvedValue({ agent: {} as Agent, dispose: () => Promise.resolve() })
+    const ctx = {
+      on: vi.fn(() => () => undefined),
+      agents: { create, get: () => undefined },
+      get: () => ({ models: () => ['model'] }),
+    } as unknown as Context
+    const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
+      initialize(params: { cwd: string; model: string }): Promise<unknown>
+      getOrCreateSession(sessionId: string): Promise<unknown>
+      shutdown(): Promise<Record<string, never>>
+    }
+
+    await server.initialize({ cwd: '.', model: 'model' })
+    await server.getOrCreateSession('relative')
+
+    expect(create).toHaveBeenCalledWith(expect.objectContaining({ meta: { cwd: process.cwd() } }))
+    await server.shutdown()
+  })
+
+  it('settles every teardown and aggregates multiple failures', async () => {
+    const firstDispose = vi.fn(() => { throw new Error('first teardown failed') })
+    const secondDispose = vi.fn(() => Promise.reject(new Error('second teardown failed')))
+    const ctx = {
+      on: vi.fn(() => () => undefined),
+      agents: { create: vi.fn(), get: () => undefined },
+      get: () => undefined,
+    } as unknown as Context
+    const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
+      sessions: Map<string, { handle: AgentHandle; lastTurnEnd: undefined; activePrompt: boolean }>
+      shutdown(): Promise<Record<string, never>>
+    }
+    server.sessions.set('first', { handle: { agent: {} as Agent, dispose: firstDispose }, lastTurnEnd: undefined, activePrompt: false })
+    server.sessions.set('second', { handle: { agent: {} as Agent, dispose: secondDispose }, lastTurnEnd: undefined, activePrompt: false })
+
+    await expect(server.shutdown()).rejects.toThrow('SDK server teardown failed')
+    expect(firstDispose).toHaveBeenCalledOnce()
+    expect(secondDispose).toHaveBeenCalledOnce()
+  })
+
+  it('continues teardown after a subscription disposer fails', async () => {
+    let subscription = 0
+    const listenerFailure = new Error('listener teardown failed')
+    const on = vi.fn(() => {
+      subscription += 1
+      return subscription === 1 ? () => { throw listenerFailure } : () => undefined
+    })
+    const ctx = {
+      on,
+      agents: { create: vi.fn(), get: () => undefined },
+      get: () => undefined,
+    } as unknown as Context
+    const server = new HarnessSdkServer(ctx, new FakeTransport())
+
+    await expect(server.shutdown()).rejects.toBe(listenerFailure)
+    expect(on).toHaveBeenCalledTimes(4)
+  })
+})
--- a/packages/ui/jsonrpc/tests/transport.spec.ts
+++ b/packages/ui/jsonrpc/tests/transport.spec.ts
@@ -0,0 +1,260 @@
+import { once } from 'node:events'
+import { PassThrough, Writable } 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<string, unknown>[] = []
+
+    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<string, unknown>[] = []
+    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<string, unknown>[] = []
+    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('preserves multibyte UTF-8 characters split across Buffer chunks', async () => {
+    const input = new PassThrough()
+    const output = new PassThrough()
+    const transport = new JsonRpcLineTransport(input, output)
+    const notifications: Record<string, unknown>[] = []
+    transport.onNotification((method, params) => { notifications.push({ method, params }) })
+    transport.start()
+
+    const frame = Buffer.from(`${JSON.stringify({ jsonrpc: '2.0', method: 'message', params: { text: '你好' } })}\n`)
+    const character = Buffer.from('你')
+    const characterStart = frame.indexOf(character)
+    expect(characterStart).toBeGreaterThanOrEqual(0)
+    input.write(frame.subarray(0, characterStart + 1))
+    input.write(frame.subarray(characterStart + 1))
+    await new Promise(resolve => setTimeout(resolve, 10))
+
+    expect(notifications).toEqual([{ method: 'message', params: { text: '你好' } }])
+    transport.close()
+  })
+
+  it('flush waits for all earlier output writes', async () => {
+    const events: string[] = []
+    const output = new Writable({
+      write(chunk: Buffer, _encoding, callback) {
+        const label = chunk.length === 0 ? 'barrier' : 'frame'
+        events.push(`start:${label}`)
+        setTimeout(() => {
+          events.push(`finish:${label}`)
+          callback()
+        }, 5)
+      },
+    })
+    const transport = new JsonRpcLineTransport(new PassThrough(), output)
+
+    transport.notify('tick')
+    await transport.flush()
+
+    expect(events).toEqual([
+      'start:frame',
+      'finish:frame',
+      'start:barrier',
+      'finish:barrier',
+    ])
+    transport.close()
+  })
+
+  it('reports an output callback failure from flush', async () => {
+    const output = {
+      write(_chunk: string, callback?: (error?: Error) => void) {
+        callback?.(new Error('flush failed'))
+        return true
+      },
+    }
+    const transport = new JsonRpcLineTransport(new PassThrough(), output as never)
+
+    await expect(transport.flush()).rejects.toThrow('flush failed')
+  })
+
+  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()
+  })
+})
--- a/packages/ui/jsonrpc/tsconfig.json
+++ 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"
+    }
+  ]
+}
--- a/packages/ui/stdio-agent/src/index.ts
+++ b/packages/ui/stdio-agent/src/index.ts
@@ -93,6 +93,8 @@ export const Config: z<Config> = z.object({
  // schemastery's native [] default would read as an invalid configured list.
  toolOrder: z.array(z.string()).default(undefined as unknown as string[]),
  tools: ToolRegistry.Config,
+  // TODO(single-default-literal): share these schema defaults and defensive
+  // apply() fallbacks through named constants while retaining both boundaries.
  persistenceRoot: z.string().default('./.sessions'),
  welcome: z.string().default('ready.'),
  skills: agentCore.SkillConfigSchema,
--- a/packages/ui/stdio-agent/src/stdio-chat.ts
+++ b/packages/ui/stdio-agent/src/stdio-chat.ts
@@ -36,6 +36,8 @@ export const inject = ['agents', 'userInteraction']
 export interface Config {
  /** Banner printed once on start, before the first `> ` prompt. */
  welcome?: string
+  // TODO(fixed-stdio-agent): this app-internal plugin is mounted only for the
+  // precreated `main` agent; remove configurability and its config-only test.
  /** Id of the agent stdin drives (`send`/`steer`) and whose status gates the EOF exit; rendering is global. Defaults to `'main'`. */
  agent?: string
 }