fix(sdk): harden runtime lifecycle and JSON-RPC

Keep DeepSeekHarness.run() reusable, but make ownership of its lazy
runtime process explicit. Document the context-manager/close contract and
update every construction example to use a context manager so repeated runs
remain valid without encouraging leaked subprocesses.

Contain notification predicate failures at the subscription boundary. Remove
only the subscriber whose callback raised, deliver that exception through its
queue, and continue dispatching to healthy subscribers so arbitrary callback
code cannot terminate the shared reader thread or strand later requests.

Enforce one in-flight prompt per server session with an atomic activePrompt
guard. Route overlap through the existing -32603 handler-error response and
clear the guard in finally, preserving parallel prompts across sessions and
sequential reuse without changing JSON-RPC request or notification shapes.

Use StringDecoder for line framing so a UTF-8 code point split across Buffer
chunks is not corrupted. Add a queued-write flush barrier, and make memoized
shutdown await it before disposal and exit while retaining exactly-once
cleanup when shutdown calls race or flushing fails.

Cover callback isolation, same-session exclusion, cross-session concurrency,
split multibyte input, delayed writes, racing shutdown, and flush failure with
deterministic tests.

packages/ui/jsonrpc/src/index.ts

@@ -83,8 +83,9 @@ export const Config: Schema<JsonRpcConfig> = Schema.object({})
  * detaches the event subscriptions) and `transport.close()`.
  *
  * The `shutdown` request's process-exit semantics live HERE, because the
- * plugin owns the server and transport: the request is answered first
- * (`setImmediate` lets the response frame flush), then the plugin disposes its
+ * 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
@@ -109,24 +110,25 @@ export function apply(ctx: Context, config: JsonRpcConfig): void {
   const server = new HarnessSdkServer(ctx, transport)
 
   // The shutdown-request exit path, exactly once (a second `shutdown` frame
-  // racing the dispose must not re-enter). `exit(0)` runs even if the dispose
-  // throws — the client was already answered, so exiting is the honest outcome.
-  let exiting = false
-  const disposeAndExit = async (): Promise<void> => {
-    if (exiting) return
-    exiting = true
-    try {
-      await fiber.dispose()
-    } finally {
+  // 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') {
-      // Answer the request first (setImmediate lets the response frame
-      // flush), then dispose this plugin's fiber and exit 0 (see apply's doc).
+      // 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

packages/ui/jsonrpc/src/server.ts

@@ -36,7 +36,10 @@ export interface InitializeResult {
   serverInfo: { name: string; version: string }
 }
 
-/** Parameters of a `session/prompt` request (one user turn on one SDK session). */
+/**
+ * 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
@@ -53,6 +56,7 @@ export interface SessionPromptResult {
 interface SessionRecord {
   handle: AgentHandle
   lastTurnEnd: TurnEndReason | undefined
+  activePrompt: boolean
 }
 
 interface SubagentRecord {
@@ -147,22 +151,30 @@ export class HarnessSdkServer {
   /**
    * Handle `session/prompt`: get-or-create the session's agent, send the
    * content as the user message, await turn settle (quiescence), then notify
-   * `session.finished` with the settled turn's outcome.
+   * `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)
-    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 }
+    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
+    }
   }
 
   /**
@@ -248,7 +260,7 @@ export class HarnessSdkServer {
       meta: { cwd: this.cwd },
       agentOptions: { model: this.model },
     })
-    const rec: SessionRecord = { handle, lastTurnEnd: undefined }
+    const rec: SessionRecord = { handle, lastTurnEnd: undefined, activePrompt: false }
     this.sessions.set(sessionId, rec)
     return rec
   }

packages/ui/jsonrpc/src/transport.ts

@@ -11,6 +11,7 @@
 
 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>
@@ -56,6 +57,7 @@ interface PendingRequest {
  */
 export class JsonRpcLineTransport implements JsonRpcTransportPeer {
   private buffer = ''
+  private readonly decoder = new StringDecoder('utf8')
   private started = false
   private requestHandler: RequestHandler | undefined
   private notificationHandler: NotificationHandler | undefined
@@ -122,8 +124,27 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
     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 : chunk.toString('utf8')
+    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
@@ -139,6 +160,8 @@ export class JsonRpcLineTransport implements JsonRpcTransportPeer {
   }
 
   private readonly onInputEnd = (): void => {
+    this.buffer += this.decoder.end()
+    this.drainLines()
     this.failPending(new Error('JSON-RPC input closed'))
   }