jsonrpc: SDK serving surface as plugins (dsh-jsonrpc + dsh-jsonrpc-agent)

packages/ui/jsonrpc/tests/plugin-apply.spec.ts

@@ -0,0 +1,266 @@
+import { createServer } from 'node:http'
+import type { IncomingMessage, Server, ServerResponse } from 'node:http'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import { PassThrough, Writable } from 'node:stream'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { Context } from 'cordis'
+import * as agentCore from '@deepseek-ai/dsh-agent-core'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as jsonrpc from '../src/index.ts'
+
+/**
+ * apply()-level lifecycle coverage for the @deepseek-ai/dsh-jsonrpc plugin:
+ * the plugin is mounted through the REAL namespace mount path —
+ * `ctx.plugin(jsonrpc, config)` over the module namespace object, exactly what
+ * the Loader hands cordis after `unwrapExports` (plugin-shape.spec pins that
+ * identity) — with the runtime-only `input`/`output`/`exit` seams from
+ * {@link jsonrpc.JsonRpcConfig} replacing the process stdio, so the whole
+ * pipeline (line transport → HarnessSdkServer → notifications back onto the
+ * wire) runs in-process. The scenarios pin the plugin's exit-lifecycle split:
+ * a `shutdown` REQUEST answers first, then disposes the plugin's own fiber and
+ * calls `exit(0)` exactly once (a racing second `shutdown` must not re-exit);
+ * a bare fiber dispose (HMR-style unload, no request) only stops serving and
+ * never touches `exit`.
+ */
+
+/** One ordered observation on the plugin's outward-facing seams: a JSON-RPC frame written to `output`, or an `exit(code)` call. */
+type WireEvent =
+  | { kind: 'frame'; frame: Record<string, unknown> }
+  | { 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[]
+  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): 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[] = []
+  let pendingOutput = ''
+  // A hand-rolled Writable (not a PassThrough): _write records each decoded
+  // frame synchronously, so `events` preserves the true frame-vs-exit order.
+  const output = new Writable({
+    write(chunk: Buffer, _encoding, callback) {
+      pendingOutput += chunk.toString('utf8')
+      for (;;) {
+        const newline = pendingOutput.indexOf('\n')
+        if (newline < 0) break
+        const line = pendingOutput.slice(0, newline).trim()
+        pendingOutput = pendingOutput.slice(newline + 1)
+        if (line) events.push({ kind: 'frame', frame: JSON.parse(line) as Record<string, unknown> })
+      }
+      callback()
+    },
+  })
+  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,
+    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)
+    try {
+      // Two shutdown frames in ONE chunk: both are dispatched from the same
+      // read-loop pass, so both setImmediate exit callbacks get scheduled and
+      // the second must hit the `exiting` guard instead of re-entering.
+      const first = { jsonrpc: '2.0', id: 'sd-1', method: 'shutdown' }
+      const second = { jsonrpc: '2.0', id: 'sd-2', method: 'shutdown' }
+      harness.sendRaw(`${JSON.stringify(first)}\n${JSON.stringify(second)}\n`)
+
+      await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit recorder call')
+      expect(harness.exits()).toEqual([0])
+
+      // Response-then-exit ordering: both shutdown responses were flushed to
+      // output BEFORE exit(0) ran (the setImmediate in the request handler).
+      const exitIndex = harness.events.findIndex(event => event.kind === 'exit')
+      const firstResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-1')
+      const secondResponse = harness.events.findIndex(event => event.kind === 'frame' && event.frame.id === 'sd-2')
+      expect(firstResponse).toBeGreaterThanOrEqual(0)
+      expect(secondResponse).toBeGreaterThanOrEqual(0)
+      expect(exitIndex).toBeGreaterThan(firstResponse)
+      expect(exitIndex).toBeGreaterThan(secondResponse)
+
+      // Idempotent: the racing second shutdown never produces a second exit.
+      await settle()
+      expect(harness.exits()).toEqual([0])
+
+      // The plugin fiber is disposed: the transport reads no further frames.
+      const before = harness.frames().length
+      harness.send({ jsonrpc: '2.0', id: 'after-exit', method: 'initialize', params: { cwd: storageDir, model: 'x' } })
+      await settle()
+      expect(harness.frames().length).toBe(before)
+    } finally {
+      await harness.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('stops serving on a bare fiber dispose (HMR-style unload) without calling exit', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-apply-dispose-'))
+    const harness = await mountPlugin(storageDir)
+    try {
+      // Prove the pipeline is live first (an unknown method still answers, as
+      // a JSON-RPC error frame — the transport's handler-rejection path).
+      harness.send({ jsonrpc: '2.0', id: 'probe-1', method: 'nope/unknown' })
+      const error = await harness.waitForFrame(frame => frame.id === 'probe-1', 'error response for unknown method')
+      expect(error.error).toMatchObject({
+        code: -32603,
+        message: 'unknown DeepSeek Harness SDK runtime method: nope/unknown',
+      })
+
+      await harness.fiber.dispose()
+
+      // The effect disposer shut the server and closed the transport — later
+      // frames are never read — and the exit seam was never touched.
+      const before = harness.frames().length
+      harness.send({ jsonrpc: '2.0', id: 'probe-2', method: 'initialize', params: { cwd: storageDir, model: 'x' } })
+      await settle()
+      expect(harness.frames().length).toBe(before)
+      expect(harness.exits()).toEqual([])
+    } finally {
+      await harness.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+})

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')
+  })
+})

packages/ui/jsonrpc/tests/server.spec.ts

@@ -0,0 +1,378 @@
+import { createServer } from 'node:http'
+import type { IncomingMessage, Server, ServerResponse } from 'node:http'
+import { mkdtemp, rm } from 'node:fs/promises'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import { afterEach, describe, expect, it, vi } from 'vitest'
+import { Context } from 'cordis'
+import { AgentId } from '@deepseek-ai/dsh-agent'
+import { SessionId } from '@deepseek-ai/dsh-session'
+import * as agentCore from '@deepseek-ai/dsh-agent-core'
+import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
+import * as LlmDeepSeek from '@deepseek-ai/dsh-llm-deepseek'
+import { HarnessSdkServer, type JsonRpcTransportPeer } from '../src/index.ts'
+
+class FakeTransport implements JsonRpcTransportPeer {
+  notifications: { method: string; params?: Record<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(SessionPersistenceJsonl, { root: storageDir })
+  await new Promise(resolve => setTimeout(resolve, 50))
+  return ctx
+}
+
+describe('HarnessSdkServer', () => {
+  it('creates a harness agent and calls the configured OpenAI-compatible endpoint', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-'))
+    const llmServer = await mockCompletionServer()
+    vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
+    vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url)
+    const ctx = await makeHarness(storageDir)
+    try {
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+
+      const init = await server.handleRequest('initialize', {
+        cwd: storageDir,
+        model: 'dsagent-model',
+        sessionRoot: storageDir,
+        systemPrompt: 'Custom SDK instructions.',
+      }) as { serverInfo: { name: string } }
+      expect(init.serverInfo.name).toBe('deepseek-harness-sdk-runtime')
+
+      await server.handleRequest('session/prompt', {
+        sessionId: 'main',
+        contentBlocks: [{ type: 'text', text: 'fix it' }],
+        profile: 'build',
+      })
+
+      expect(llmServer.requests).toHaveLength(1)
+      const body = llmServer.requests[0] as { model: string; messages: { role: string }[] }
+      expect(body.model).toBe('dsagent-model')
+      expect(body.messages[0]?.role).toBe('system')
+      expect(body.messages.at(-1)?.role).toBe('user')
+      expect(llmServer.headers[0]?.authorization).toBe('Bearer test-key')
+      expect(transport.notifications.some(n => n.method === 'session.event')).toBe(true)
+      expect(transport.notifications.at(-1)).toMatchObject({
+        method: 'session.finished',
+        params: { sessionId: 'main', status: 'ok' },
+      })
+
+      await server.handleRequest('session/prompt', {
+        sessionId: 'main',
+        contentBlocks: [{ type: 'text', text: 'again' }],
+      })
+      expect(llmServer.requests).toHaveLength(2)
+
+      const orphanHandle = ctx.agents.create({
+        agentId: AgentId('orphan-agent'),
+        sessionId: SessionId('orphan-session'),
+        meta: { cwd: storageDir },
+        agentOptions: { model: 'dsagent-model' },
+      })
+      orphanHandle.agent.send([{ type: 'text', text: 'outside the sdk session map' }])
+      await orphanHandle.agent.whenIdle()
+      await orphanHandle.dispose()
+      expect(llmServer.requests).toHaveLength(3)
+
+      await server.handleRequest('shutdown', undefined)
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('notifies the host when a child session is created with parent lineage', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-'))
+    const ctx = await makeHarness(storageDir)
+    try {
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+
+      ctx.sessions.create(SessionId('root-session'), {
+        meta: { cwd: storageDir },
+      })
+      ctx.sessions.create(SessionId('child-session'), {
+        meta: { cwd: storageDir, parentSession: SessionId('main') },
+      })
+
+      expect(transport.notifications).toContainEqual({
+        method: 'subagent.started',
+        params: {
+          parentSessionId: 'main',
+          childSessionId: 'child-session',
+        },
+      })
+
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('creates an SDK session without an optional system prompt', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-no-system-'))
+    const llmServer = await mockCompletionServer()
+    vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
+    vi.stubEnv('DEEPSEEK_BASE_URL', llmServer.url)
+    const ctx = await makeHarness(storageDir)
+    try {
+      const server = new HarnessSdkServer(ctx, new FakeTransport())
+
+      await server.initialize({ cwd: storageDir, model: 'plain-model' })
+      await server.prompt({
+        sessionId: 'plain',
+        contentBlocks: [{ type: 'text', text: 'hello' }],
+      })
+
+      expect(llmServer.requests).toHaveLength(1)
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('notifies the host when a subagent run settles', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-end-'))
+    const ctx = await makeHarness(storageDir)
+    try {
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+
+      const handle = ctx.agents.create({
+        agentId: AgentId('child-agent'),
+        sessionId: SessionId('child-session'),
+        meta: { cwd: storageDir, parentSession: SessionId('main') },
+      })
+      ctx.emit('subagent/end', {
+        provider: 'spawn',
+        id: AgentId('child-agent'),
+        stopReason: 'completed',
+        lastAssistantMessage: [{ type: 'text', text: 'child done' }],
+      })
+
+      expect(transport.notifications).toContainEqual({
+        method: 'subagent.finished',
+        params: {
+          provider: 'spawn',
+          agentId: 'child-agent',
+          parentSessionId: 'main',
+          childSessionId: 'child-session',
+          status: 'ok',
+          stopReason: 'completed',
+          lastAssistantMessage: [{ type: 'text', text: 'child done' }],
+        },
+      })
+
+      await handle.dispose()
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('falls back to live agent lineage for uncached subagent end events', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-subagent-fallback-'))
+    const ctx = await makeHarness(storageDir)
+    let handle: ReturnType<typeof ctx.agents.create> | undefined
+    let failedHandle: ReturnType<typeof ctx.agents.create> | undefined
+    try {
+      handle = ctx.agents.create({
+        agentId: AgentId('fallback-child-agent'),
+        sessionId: SessionId('fallback-child-session'),
+        meta: { cwd: storageDir, parentSession: SessionId('fallback-parent') },
+      })
+      failedHandle = ctx.agents.create({
+        agentId: AgentId('failed-child-agent'),
+        sessionId: SessionId('failed-child-session'),
+        meta: { cwd: storageDir },
+      })
+      const transport = new FakeTransport()
+      const server = new HarnessSdkServer(ctx, transport)
+
+      ctx.emit('subagent/end', {
+        provider: 'fork',
+        id: AgentId('fallback-child-agent'),
+        stopReason: 'max-tokens',
+      })
+      ctx.emit('subagent/end', {
+        provider: 'fork',
+        id: AgentId('failed-child-agent'),
+        stopReason: 'error',
+      })
+      ctx.emit('subagent/end', {
+        provider: 'fork',
+        id: AgentId('missing-child-agent'),
+        stopReason: 'error',
+      })
+
+      expect(transport.notifications).toContainEqual({
+        method: 'subagent.finished',
+        params: {
+          provider: 'fork',
+          agentId: 'fallback-child-agent',
+          parentSessionId: 'fallback-parent',
+          childSessionId: 'fallback-child-session',
+          status: 'ok',
+          stopReason: 'max-tokens',
+        },
+      })
+      expect(transport.notifications).toContainEqual({
+        method: 'subagent.finished',
+        params: {
+          provider: 'fork',
+          agentId: 'failed-child-agent',
+          childSessionId: 'failed-child-session',
+          status: 'error',
+          stopReason: 'error',
+        },
+      })
+      expect(transport.notifications.some(n =>
+        n.method === 'subagent.finished'
+        && n.params?.agentId === 'missing-child-agent',
+      )).toBe(false)
+
+      await server.shutdown()
+    } finally {
+      await handle?.dispose()
+      await failedHandle?.dispose()
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('does not re-register an LLM adapter that already exists', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-existing-llm-'))
+    const ctx = await makeHarness(storageDir)
+    vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
+    await ctx.plugin(LlmDeepSeek, { models: ['preinstalled-model'] })
+    try {
+      const server = new HarnessSdkServer(ctx, new FakeTransport())
+      const inspect = server as unknown as { hasAdapterFor(model: string): boolean }
+
+      expect(inspect.hasAdapterFor('preinstalled-model')).toBe(true)
+      expect(inspect.hasAdapterFor('missing-model')).toBe(false)
+      await server.initialize({ cwd: storageDir, model: 'preinstalled-model' })
+
+      expect(ctx.get('llm')?.models().filter(model => model === 'preinstalled-model')).toEqual(['preinstalled-model'])
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('registers a missing model when an LLM service already exists', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-new-llm-'))
+    const ctx = await makeHarness(storageDir)
+    vi.stubEnv('DEEPSEEK_API_KEY', 'test-key')
+    await ctx.plugin(LlmDeepSeek, { models: ['other-model'] })
+    try {
+      const server = new HarnessSdkServer(ctx, new FakeTransport())
+
+      await server.initialize({ cwd: storageDir, model: 'new-model' })
+
+      expect(ctx.get('llm')?.models()).toEqual(expect.arrayContaining(['other-model', 'new-model']))
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('classifies defensive finish states', async () => {
+    const storageDir = await mkdtemp(join(tmpdir(), 'dsh-jsonrpc-finish-states-'))
+    const ctx = await makeHarness(storageDir)
+    try {
+      const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
+        finishedStatus(reason: unknown): 'ok' | 'error'
+        shutdown(): Promise<Record<string, never>>
+      }
+
+      expect(server.finishedStatus(undefined)).toBe('error')
+      expect(server.finishedStatus({ kind: 'max-tokens' })).toBe('ok')
+      expect(server.finishedStatus({ kind: 'error' })).toBe('error')
+      await server.shutdown()
+    } finally {
+      await ctx.fiber.dispose()
+      await rm(storageDir, { recursive: true, force: true })
+    }
+  })
+
+  it('reports no adapter when the LLM service is absent', async () => {
+    const ctx = new Context()
+    try {
+      const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
+        hasAdapterFor(model: string): boolean
+        shutdown(): Promise<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 })
+    }
+  })
+})

packages/ui/jsonrpc/tests/transport.spec.ts

@@ -0,0 +1,202 @@
+import { once } from 'node:events'
+import { PassThrough } from 'node:stream'
+import { describe, expect, it } from 'vitest'
+import { JsonRpcLineTransport } from '../src/index.ts'
+
+function transportPair() {
+  const aToB = new PassThrough()
+  const bToA = new PassThrough()
+  const a = new JsonRpcLineTransport(bToA, aToB)
+  const b = new JsonRpcLineTransport(aToB, bToA)
+  return { a, b, aToB, bToA }
+}
+
+describe('JsonRpcLineTransport', () => {
+  it('supports bidirectional requests and notifications over newline-delimited JSON-RPC', async () => {
+    const { a, b } = transportPair()
+    const notifications: Record<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('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()
+  })
+})