diff --git a/packages/ui/jsonrpc/README.md b/packages/ui/jsonrpc/README.md
index 95ebdb0b00..007bd1a5f1 100644
--- a/packages/ui/jsonrpc/README.md
+++ b/packages/ui/jsonrpc/README.md
@@ -20,4 +20,4 @@ The plugin owns the PROTOCOL-level exit: a `shutdown` request is answered first
 
 ## Wire notes
 
-`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (SDK clients key on it, independent of this package's name). Persistence roots and the deployment persona come from `cordis.yml`; the wire exposes only parameters the server applies.
+`initialize.serverInfo.name` is the wire-stable `deepseek-harness-sdk-runtime` (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.
diff --git a/packages/ui/jsonrpc/src/index.ts b/packages/ui/jsonrpc/src/index.ts
index 9147f63e34..624c8db824 100644
--- a/packages/ui/jsonrpc/src/index.ts
+++ b/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
diff --git a/packages/ui/jsonrpc/src/server.ts b/packages/ui/jsonrpc/src/server.ts
index 1b193aaf1e..86af6645f3 100644
--- a/packages/ui/jsonrpc/src/server.ts
+++ b/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
   }
diff --git a/packages/ui/jsonrpc/src/transport.ts b/packages/ui/jsonrpc/src/transport.ts
index 003c78f4e5..1f3011f2eb 100644
--- a/packages/ui/jsonrpc/src/transport.ts
+++ b/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'))
   }
 
diff --git a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
index 50821767f9..8f4a504264 100644
--- a/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
+++ b/packages/ui/jsonrpc/tests/plugin-apply.spec.ts
@@ -28,6 +28,7 @@ import * as jsonrpc from '../src/index.ts'
 /** 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 {
@@ -36,6 +37,7 @@ interface ApplyHarness {
   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>[]
@@ -65,7 +67,10 @@ async function settle(): Promise<void> {
  * 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> {
+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 })
@@ -73,22 +78,39 @@ async function mountPlugin(storageDir: string): Promise<ApplyHarness> {
 
   const input = new PassThrough()
   const events: WireEvent[] = []
+  const outputErrors: Error[] = []
   let pendingOutput = ''
-  // A hand-rolled Writable (not a PassThrough): _write records each decoded
-  // frame synchronously, so `events` preserves the true frame-vs-exit order.
+  // 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) events.push({ kind: 'frame', frame: JSON.parse(line) as Record<string, unknown> })
+        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)
+        }
       }
-      callback()
+      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 })
@@ -99,6 +121,7 @@ async function mountPlugin(storageDir: string): Promise<ApplyHarness> {
     ctx,
     fiber,
     events,
+    outputErrors,
     send: (frame) => { input.write(`${JSON.stringify(frame)}\n`) },
     sendRaw: (text) => { input.write(text) },
     frames,
@@ -199,7 +222,7 @@ describe('dsh-jsonrpc plugin apply', () => {
 
   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)
+    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
@@ -211,15 +234,21 @@ describe('dsh-jsonrpc plugin apply', () => {
       await waitFor(() => harness.exits().length > 0 ? true : undefined, 'exit recorder call')
       expect(harness.exits()).toEqual([0])
 
-      // Response-then-exit ordering: both shutdown responses were flushed to
-      // output BEFORE exit(0) ran (the setImmediate in the request handler).
+      // 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(exitIndex).toBeGreaterThan(firstResponse)
-      expect(exitIndex).toBeGreaterThan(secondResponse)
+      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()
@@ -236,6 +265,27 @@ describe('dsh-jsonrpc plugin apply', () => {
     }
   })
 
+  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)
diff --git a/packages/ui/jsonrpc/tests/server.spec.ts b/packages/ui/jsonrpc/tests/server.spec.ts
index 8d4d791841..1cde550f2e 100644
--- a/packages/ui/jsonrpc/tests/server.spec.ts
+++ b/packages/ui/jsonrpc/tests/server.spec.ts
@@ -152,6 +152,57 @@ describe('HarnessSdkServer', () => {
     }
   })
 
+  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)
@@ -490,11 +541,11 @@ describe('HarnessSdkServer', () => {
       get: () => undefined,
     } as unknown as Context
     const server = new HarnessSdkServer(ctx, new FakeTransport()) as unknown as {
-      sessions: Map<string, { handle: AgentHandle; lastTurnEnd: undefined }>
+      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 })
-    server.sessions.set('second', { handle: { agent: {} as Agent, dispose: secondDispose }, lastTurnEnd: undefined })
+    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()
diff --git a/packages/ui/jsonrpc/tests/transport.spec.ts b/packages/ui/jsonrpc/tests/transport.spec.ts
index 0e98a4af10..2f3173c7c9 100644
--- a/packages/ui/jsonrpc/tests/transport.spec.ts
+++ b/packages/ui/jsonrpc/tests/transport.spec.ts
@@ -1,5 +1,5 @@
 import { once } from 'node:events'
-import { PassThrough } from 'node:stream'
+import { PassThrough, Writable } from 'node:stream'
 import { describe, expect, it } from 'vitest'
 import { JsonRpcLineTransport } from '../src/index.ts'
 
@@ -122,6 +122,64 @@ describe('JsonRpcLineTransport', () => {
     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()
diff --git a/python/README.i18n.yaml b/python/README.i18n.yaml
index 1ea214996e..061c15aa92 100644
--- a/python/README.i18n.yaml
+++ b/python/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write
-README.md: d9403733b7daff9b3e8dbf21e1bc0c68d61b88a7
-README.zh.md: 28859e43b7332a8ef4c3cf199b97860447a011a8
+README.md: 35ed645ce0e4d5e4c88c8aae2fe33d94aea79b3e
+README.zh.md: bddb709114df2e46bbea6e6faeb660bffa30df2b
diff --git a/python/README.md b/python/README.md
index d9403733b7..35ed645ce0 100644
--- a/python/README.md
+++ b/python/README.md
@@ -37,7 +37,8 @@ For an interactive check (needs `DEEPSEEK_API_KEY` in the environment or the rep
 
 ```python
 from deepseek_harness import DeepSeekHarness
-print(DeepSeekHarness().run("say hi").final_response)   # auto-resolution picks the bundled exe
+with DeepSeekHarness() as harness:
+    print(harness.run("say hi").final_response)   # auto-resolution picks the bundled exe
 ```
 
 ## Running the SDK against the Node source (no executable)
diff --git a/python/README.zh.md b/python/README.zh.md
index 28859e43b7..bddb709114 100644
--- a/python/README.zh.md
+++ b/python/README.zh.md
@@ -37,7 +37,8 @@ uv run --project python/sdk pytest                                            #
 
 ```python
 from deepseek_harness import DeepSeekHarness
-print(DeepSeekHarness().run("say hi").final_response)   # auto-resolution picks the bundled exe
+with DeepSeekHarness() as harness:
+    print(harness.run("say hi").final_response)   # auto-resolution picks the bundled exe
 ```
 
 ## 对着 Node 源码运行 SDK（不用可执行文件）
diff --git a/python/sdk/README.i18n.yaml b/python/sdk/README.i18n.yaml
index b312edc1b0..99aa49e0cc 100644
--- a/python/sdk/README.i18n.yaml
+++ b/python/sdk/README.i18n.yaml
@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write
-README.md: 1916086324e18ce79ea572f3c85115fda3b4ab91
-README.zh.md: 380ed1726f2b541cd66403edb68a2629071e18f0
+README.md: 60540376c5fd85b0852e204bc8bad3f01c849de5
+README.zh.md: 7148a45c8f4bd22aaa28fc15ac9cd589cba13dfb
diff --git a/python/sdk/README.md b/python/sdk/README.md
index 1916086324..60540376c5 100644
--- a/python/sdk/README.md
+++ b/python/sdk/README.md
@@ -13,9 +13,12 @@ Installing `deepseek-harness` installs the exact same-version `deepseek-harness-
 ```py
 from deepseek_harness import DeepSeekHarness
 
-result = DeepSeekHarness().run("Say hi.")
+with DeepSeekHarness() as harness:
+    result = harness.run("Say hi.")
 ```
 
+`DeepSeekHarness` keeps its lazily started runtime subprocess for reuse across calls. Use it as a context manager, as above, or call `close()` explicitly when finished.
+
 By default, the SDK launches the bundled single-file `dsh-jsonrpc-agent` executable from the `deepseek-harness-runtime-bin` package and injects that package's default configuration (the stdio JSON-RPC server, agent core, preloaded DeepSeek adapter, JSONL session persistence, local bash) via `DSH_CORDIS_CONFIG`. To run a plugin composition of your own, keep the `@deepseek-ai/dsh-jsonrpc` entry in the config and pass the Cordis config path.
 
 ```py
diff --git a/python/sdk/README.zh.md b/python/sdk/README.zh.md
index 380ed1726f..7148a45c8f 100644
--- a/python/sdk/README.zh.md
+++ b/python/sdk/README.zh.md
@@ -9,9 +9,12 @@
 ```py
 from deepseek_harness import DeepSeekHarness
 
-result = DeepSeekHarness().run("Say hi.")
+with DeepSeekHarness() as harness:
+    result = harness.run("Say hi.")
 ```
 
+`DeepSeekHarness` 会保留延迟启动的运行时子进程，以供多次调用复用。请像上例一样将其用作上下文管理器，或在用完后显式调用 `close()`。
+
 默认情况下，SDK 启动 `deepseek-harness-runtime-bin` 包内置的单文件 `dsh-jsonrpc-agent` 可执行程序，并通过 `DSH_CORDIS_CONFIG` 注入该包的默认配置（stdio JSON-RPC 服务器、agent core、预载的 DeepSeek 适配器、JSONL 会话持久化、本地 bash）。要运行自己用插件组合需要在配置里保留 `@deepseek-ai/dsh-jsonrpc` 条目，并传入 Cordis 配置路径。
 
 ```py
diff --git a/python/sdk/src/deepseek_harness/api.py b/python/sdk/src/deepseek_harness/api.py
index c45a70ee33..3743371cea 100644
--- a/python/sdk/src/deepseek_harness/api.py
+++ b/python/sdk/src/deepseek_harness/api.py
@@ -43,7 +43,12 @@ class TurnResult:
 
 
 class DeepSeekHarness:
-    """Synchronous high-level SDK for running DeepSeek Harness agent turns."""
+    """Reusable synchronous SDK for running DeepSeek Harness agent turns.
+
+    The runtime subprocess starts lazily and remains owned by this instance
+    across calls to :meth:`run`. Use the instance as a context manager, or call
+    :meth:`close` explicitly when finished, so the subprocess is always reaped.
+    """
 
     def __init__(self, config: DeepSeekHarnessConfig | None = None, **kwargs: object) -> None:
         if config is not None and kwargs:
diff --git a/python/sdk/src/deepseek_harness/client.py b/python/sdk/src/deepseek_harness/client.py
index e81dd6f30a..be457c538a 100644
--- a/python/sdk/src/deepseek_harness/client.py
+++ b/python/sdk/src/deepseek_harness/client.py
@@ -350,10 +350,19 @@ class HarnessClient:
             params = message.get("params")
             notification = Notification(method=method, payload=params if isinstance(params, dict) else {})
             with self._lock:
-                subscribers = list(self._notification_subscribers.values())
+                subscribers = list(self._notification_subscribers.items())
             delivered = False
-            for subscriber, predicate in subscribers:
-                if predicate is None or predicate(notification):
+            for subscription_id, (subscriber, predicate) in subscribers:
+                try:
+                    matches = predicate is None or predicate(notification)
+                except BaseException as exc:
+                    with self._lock:
+                        current = self._notification_subscribers.get(subscription_id)
+                        if current is not None and current[0] is subscriber:
+                            self._notification_subscribers.pop(subscription_id, None)
+                    subscriber.put(exc)
+                    continue
+                if matches:
                     subscriber.put(notification)
                     delivered = True
             if not delivered:
diff --git a/python/sdk/tests/test_client.py b/python/sdk/tests/test_client.py
index 0f66ee92e8..327e5f5c39 100644
--- a/python/sdk/tests/test_client.py
+++ b/python/sdk/tests/test_client.py
@@ -356,6 +356,48 @@ def test_client_keeps_unmatched_notifications_available_globally_while_subscribe
         assert notification.payload["sessionId"] == "other"
 
 
+def test_client_contains_notification_filter_failure_to_its_subscription(tmp_path: Path) -> None:
+    script = tmp_path / "fake_bridge.py"
+    script.write_text(
+        """
+import json
+import sys
+
+for line in sys.stdin:
+    msg = json.loads(line)
+    method = msg.get("method")
+    if method == "initialize":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-dsh"}}}), flush=True)
+    elif method in {"emit-first", "emit-second"}:
+        print(json.dumps({"jsonrpc": "2.0", "method": "tick", "params": {"source": method}}), flush=True)
+    elif method == "session/prompt":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+    elif method == "shutdown":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
+        break
+""".strip()
+    )
+
+    def broken_filter(_notification: object) -> bool:
+        raise RuntimeError("bad notification filter")
+
+    with HarnessClient(HarnessConfig(launch_args_override=(sys.executable, str(script)))) as client:
+        client.initialize(cwd="/workspace", model="dsagent")
+        with (
+            client.subscribe_notifications(broken_filter) as broken,
+            client.subscribe_notifications(lambda notification: notification.method == "tick") as healthy,
+        ):
+            client.notify("emit-first")
+            with pytest.raises(RuntimeError, match="bad notification filter"):
+                broken.next()
+            assert healthy.next().payload == {"source": "emit-first"}
+            assert client._notifications.qsize() == 0
+
+            client.session_prompt("main", [{"type": "text", "text": "reader still works"}])
+            client.notify("emit-second")
+            assert healthy.next().payload == {"source": "emit-second"}
+
+
 def test_client_rejects_unaccepted_session_prompt_response(tmp_path: Path) -> None:
     script = tmp_path / "fake_bridge.py"
     script.write_text(