Merge origin/master into feat/sdk-approval-rpc

python/development.i18n.yaml

@@ -12,8 +12,8 @@
   en: 37884fe76e167a0e
   zh: 4b6e50768c4bb945
 /python-contributor-workflows/run-against-node-source:
-  en: 7430d9e6b3d02ddc
-  zh: 9044e6be91d0ad01
+  en: 940c941d189a8248
+  zh: 321f5183383d4a12
 /python-contributor-workflows/build-distributions:
   en: 4aa941562dba5482
   zh: 50543a92b3d0cf04

python/development.md

@@ -66,6 +66,14 @@ Repository contributors can select either development route; both execute the no
 
 `python/sdk/tests/manual_sdk_agent_smoke.py` uses the internal `_launch_args` test adapter to exercise the unbuilt TypeScript CLI under tsx. Arbitrary argv replacement is intentionally absent from the public SDK.
 
+The manual [shutdown example](sdk/tests/manual_sdk_shutdown_smoke.py) uses ordinary SDK calls: run one short real-model turn, close immediately with the default 1-second budget, then open another runtime and restore the same session. Install the SDK/runtime and set `DEEPSEEK_API_KEY` before running:
+
+```sh
+python python/sdk/tests/manual_sdk_shutdown_smoke.py
+```
+
+The example prints the close duration and history files, retains its temporary directory, and lets restoration failures raise normally. Recovery uses a separate 30-second shutdown budget. Reproduction depends on the host and first-turn timing; successful restoration does not establish a shutdown latency bound. This manual example is not collected by pytest.
+
 ## Build distributions
 
 The root `package.json` version is authoritative for both Python distributions. The staging script injects that version into both wheels and pins the SDK to the same `deepseek-harness-runtime-bin` version.

python/development.zh.md

@@ -66,6 +66,14 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/test-dsh-home") as harness:
 
 `python/sdk/tests/manual_sdk_agent_smoke.py` 使用内部 `_launch_args` 测试适配器，通过 tsx 验证未构建的 TypeScript CLI。公开 SDK 刻意不提供任意 argv 替换。
 
+手动[关闭示例](sdk/tests/manual_sdk_shutdown_smoke.py)只使用普通 SDK 调用：运行一轮真实模型短对话，以默认 1 秒预算立即关闭，再启动另一个运行时恢复同一会话。安装 SDK／运行时并设置 `DEEPSEEK_API_KEY` 后运行：
+
+```sh
+python python/sdk/tests/manual_sdk_shutdown_smoke.py
+```
+
+示例打印关闭耗时和历史文件列表，保留临时目录，并让恢复失败正常抛出异常。恢复进程使用独立的 30 秒关闭预算。能否复现取决于主机和首轮耗时，恢复成功不能确定关闭耗时上限。pytest 不收集此手动示例。
+
 ## 构建分发包
 
 根目录 `package.json` 的版本是两个 Python 分发包的权威版本。暂存脚本会将该版本注入两个 wheel 包，并将 SDK 固定到同版本的 `deepseek-harness-runtime-bin`。

python/sdk/README.i18n.yaml

@@ -17,6 +17,9 @@
 /deepseek-harness-python-sdk/customize-plugins:
   en: 69a6a6745c6e89c1
   zh: 85e82a4cc31e16e5
+/deepseek-harness-python-sdk/native-user-questions:
+  en: d9c60c5f43755782
+  zh: 78c82635209ed22d
 /deepseek-harness-python-sdk/results-and-notifications:
   en: d4f7366c2d260639
   zh: 7de99aed2b1bdd0a

python/sdk/README.md

@@ -108,6 +108,12 @@ with DeepSeekHarness(
 
 The shipped `sdk-minimal` profile is a standalone explicit tree rather than an overlay on `dsh-base`. Select it with `profile="sdk-minimal"`; the ordinary `model` argument is the sole runtime model selection, including for model ids outside the adapter's advisory catalog. It provides a platform-selected persistent shell, `read_image`, `working_directory`, local execution, JSONL sessions, and automatic compaction. Text file tools, settings, managed credentials, telemetry, Web tools, and the full default tool roster remain available through the separate full `sdk` and `web` profiles. SDK prompt text does not execute `/compact`.
 
+## Native user questions
+
+Pass `on_user_question` to `DeepSeekHarnessConfig`, `DeepSeekHarness`, or `HarnessConfig` to answer the built-in `ask_user_question` tool through the native SDK protocol. The callback receives a `UserQuestionRequest` (`request_id`, `session_id`, and the original `questions`) and a `threading.Event` signalling cancellation. Return `UserQuestionAnswer` or a JSON object shaped as `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}`. Each original question id must appear exactly once. The application supplies its UI or simulated-user policy; the SDK does not choose answers.
+
+Callbacks run on separate threads so the reader can process cancellation and other requests. They may run concurrently and may make further client requests. Release callback-owned work when the cancellation Event is set. Closing the client signals outstanding callbacks and gives their workers a shared 0.5-second join window; callbacks that ignore cancellation may continue, but their late answers cannot reach the runtime or a later process. A callback exception becomes a protocol error, and an absent callback retains existing native answerers or reports `NO_PROVIDER`.
+
 ## Results and notifications
 
 `Session.run()` owns an activity interval from its prompt's durable inbox receipt through the next whole-agent idle and returns `RunResult(session_id, final_response, finish_reason, events, notifications)`. `final_response` is the last committed root-session assistant text in the interval. `finish_reason` is the `kind` of the last root-session `turn/end`, such as `completed`, `max-tokens`, or `error`, and is `None` when no turn ended. A `turn/end` without a string `data.reason.kind` violates the protocol and raises `SdkProtocolError`.

python/sdk/README.zh.md

@@ -108,6 +108,12 @@ with DeepSeekHarness(
 
 随附的 `sdk-minimal` profile 是独立显式配置树，而不是 `dsh-base` 上的 overlay。使用 `profile="sdk-minimal"` 选择它；普通 `model` 参数是唯一运行时模型选择，也适用于不在适配器建议目录中的模型 ID。它提供按平台选择的持久 shell、`read_image`、`working_directory`、本地执行、JSONL 会话和自动压缩。文本文件工具、设置、托管凭据、遥测、Web 工具与完整默认工具清单仍由独立的完整 `sdk` 与 `web` profile 提供。SDK 提示词文本不会执行 `/compact`。
 
+## 原生用户问答
+
+向 `DeepSeekHarnessConfig`、`DeepSeekHarness` 或 `HarnessConfig` 传入 `on_user_question`，即可通过原生 SDK 协议回答内建的 `ask_user_question` 工具。回调接收 `UserQuestionRequest`（包含 `request_id`、`session_id` 和原始 `questions`）以及表示取消的 `threading.Event`。返回 `UserQuestionAnswer`，或形如 `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}` 的 JSON 对象。每个原问题 id 必须恰好出现一次。应用自行提供 UI 或模拟用户策略；SDK 不选择答案。
+
+回调在独立线程执行，读取线程仍可处理取消和其他请求。回调可以并发执行，也可以再次调用客户端请求。取消 Event 置位时应释放回调拥有的工作。客户端关闭会通知所有待决回调，并给其工作线程总计 0.5 秒的 join 窗口；忽略取消的回调可能继续执行，但晚到答案不会发送到运行时或后续新进程。回调异常转为协议错误；未提供回调时保留原生回答方，或报告 `NO_PROVIDER`。
+
 ## 结果与通知
 
 `Session.run()` 的活动区间从提示词被持久 inbox 接收时开始，到整个 agent 下一次进入空闲状态时结束，并返回 `RunResult(session_id, final_response, finish_reason, events, notifications)`。`final_response` 是该区间内根会话最后提交的助手文本。`finish_reason` 是最后一个根会话 `turn/end` 的 `kind`，例如 `completed`、`max-tokens` 或 `error`；没有轮次结束时为 `None`。缺少字符串 `data.reason.kind` 的 `turn/end` 违反协议，并会抛出 `SdkProtocolError`。

python/sdk/src/deepseek_harness/__init__.py

@@ -1,7 +1,10 @@
 from .api import DeepSeekHarness, DeepSeekHarnessConfig, RunResult, Session
-from .client import HarnessClient, HarnessConfig
+from .client import HarnessClient, HarnessConfig, UserQuestionHandler
 from .errors import SdkProtocolError
-from .models import ApprovalRequestParams, ApprovalRequestResult, IncomingRequest, InitializeResponse, JsonObject, Notification, ServerInfo
+from .models import (
+    ApprovalRequestParams, ApprovalRequestResult, IncomingRequest, InitializeResponse,
+    JsonObject, Notification, ServerInfo, UserQuestionAnswer, UserQuestionAnswerItem, UserQuestionRequest,
+)
 
 __all__ = [
     "DeepSeekHarness",
@@ -10,6 +13,10 @@ __all__ = [
     "RunResult",
     "HarnessClient",
     "HarnessConfig",
+    "UserQuestionHandler",
+    "UserQuestionRequest",
+    "UserQuestionAnswer",
+    "UserQuestionAnswerItem",
     "SdkProtocolError",
     "ApprovalRequestParams",
     "ApprovalRequestResult",

python/sdk/src/deepseek_harness/api.py

@@ -5,7 +5,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Callable
 
-from .client import HarnessClient, HarnessConfig
+from .client import HarnessClient, HarnessConfig, UserQuestionHandler
 from .errors import SdkProtocolError
 from .models import JsonObject, Notification
 
@@ -35,6 +35,7 @@ class DeepSeekHarnessConfig:
     shutdown_timeout_seconds: float | None = 1.0
     base_url: str | None = None
     api_key: str | None = None
+    on_user_question: UserQuestionHandler | None = None
     approval_requests: bool = False
 
 
@@ -85,6 +86,7 @@ class DeepSeekHarness:
                 initialize_timeout_seconds=self.config.initialize_timeout_seconds,
                 request_timeout_seconds=self.config.request_timeout_seconds,
                 shutdown_timeout_seconds=self.config.shutdown_timeout_seconds,
+                on_user_question=self.config.on_user_question,
             ),
             _launch_args=_launch_args,
         )

python/sdk/src/deepseek_harness/client.py

@@ -15,10 +15,11 @@ from typing import Callable, TypeAlias, TypeVar
 from pydantic import BaseModel
 
 from .errors import JsonRpcError, SdkProtocolError, TransportClosedError
-from .models import IncomingRequest, InitializeResponse, JsonObject, JsonValue, Notification
+from .models import IncomingRequest, InitializeResponse, JsonObject, JsonValue, Notification, UserQuestionAnswer, UserQuestionRequest
 
 ModelT = TypeVar("ModelT", bound=BaseModel)
 NotificationFilter: TypeAlias = Callable[[Notification], bool]
+UserQuestionHandler: TypeAlias = Callable[[UserQuestionRequest, threading.Event], UserQuestionAnswer | JsonObject]
 
 
 @dataclass(slots=True)
@@ -34,6 +35,8 @@ class HarnessConfig:
     initialize_timeout_seconds: float = 30.0
     request_timeout_seconds: float | None = None
     shutdown_timeout_seconds: float | None = 1.0
+    # Callbacks run outside the reader; observe the Event to release work on cancellation.
+    on_user_question: UserQuestionHandler | None = None
 
 
 class HarnessClient:
@@ -60,6 +63,9 @@ class HarnessClient:
         self._stderr_lines: deque[str] = deque(maxlen=400)
         self._reader_thread: threading.Thread | None = None
         self._stderr_thread: threading.Thread | None = None
+        self._question_tasks: dict[str, tuple[threading.Event, threading.Thread]] = {}
+        self._closing = False
+        self._user_question_handler: UserQuestionHandler | None = None
 
     def __enter__(self) -> "HarnessClient":
         self.start()
@@ -73,6 +79,9 @@ class HarnessClient:
             return
         with self._lock:
             self._session_parents.clear()
+            self._question_tasks.clear()
+            self._closing = False
+            self._user_question_handler = self.config.on_user_question
         env = os.environ.copy()
         if self.config.env:
             env.update(self.config.env)
@@ -93,6 +102,11 @@ class HarnessClient:
 
     def close(self) -> None:
         """Close the runtime after a bounded opportunity to flush durable state."""
+        with self._lock:
+            self._closing = True
+            question_tasks = list(self._question_tasks.values())
+            for cancelled, _thread in question_tasks:
+                cancelled.set()
         proc = self._proc
         if proc is None:
             return
@@ -129,6 +143,10 @@ class HarnessClient:
             self._reader_thread.join(timeout=0.5)
         if self._stderr_thread and self._stderr_thread.is_alive():
             self._stderr_thread.join(timeout=0.5)
+        deadline = time.monotonic() + 0.5
+        for _cancelled, thread in question_tasks:
+            if thread is not threading.current_thread():
+                thread.join(timeout=max(0, deadline - time.monotonic()))
 
     def initialize(
         self,
@@ -157,6 +175,8 @@ class HarnessClient:
             payload["reasoningEffort"] = reasoning_effort
         if max_tokens is not None:
             payload["maxTokens"] = max_tokens
+        if self._user_question_handler is not None:
+            payload["capabilities"] = {"userQuestions": True}
         try:
             return self.request(
                 "initialize",
@@ -416,7 +436,11 @@ class HarnessClient:
         method = message.get("method")
         if isinstance(msg_id, (str, int)) and isinstance(method, str):
             params = message.get("params")
-            self._requests.put(IncomingRequest(id=msg_id, method=method, payload=params if isinstance(params, dict) else {}))
+            request = IncomingRequest(id=msg_id, method=method, payload=params if isinstance(params, dict) else {})
+            if method == "user-question/ask" and self._user_question_handler is not None:
+                self._start_user_question(request)
+            else:
+                self._requests.put(request)
             return
         if isinstance(msg_id, (str, int)):
             with self._lock:
@@ -433,6 +457,10 @@ class HarnessClient:
             params = message.get("params")
             notification = Notification(method=method, payload=params if isinstance(params, dict) else {})
             with self._lock:
+                if method == "user-question.cancelled":
+                    task = self._question_tasks.get(str(notification.payload.get("requestId")))
+                    if task is not None:
+                        task[0].set()
                 self._record_session_relationship_locked(notification)
                 subscribers = list(self._notification_subscribers.items())
             delivered = False
@@ -454,6 +482,8 @@ class HarnessClient:
 
     def _fail_waiters(self, exc: BaseException) -> None:
         with self._lock:
+            for cancelled, _thread in self._question_tasks.values():
+                cancelled.set()
             waiters = list(self._responses.values())
             self._responses.clear()
             subscribers = list(self._notification_subscribers.values())
@@ -465,6 +495,56 @@ class HarnessClient:
         self._notifications.put(exc)
         self._requests.put(exc)
 
+    def _start_user_question(self, incoming: IncomingRequest) -> None:
+        """Keep the reader free to receive cancellation, other responses, and shutdown."""
+        handler = self._user_question_handler
+        proc = self._proc
+        try:
+            request = UserQuestionRequest.model_validate(incoming.payload)
+            if any(not isinstance(item.get("id"), str) or not isinstance(item.get("question"), str) for item in request.questions):
+                raise ValueError("questions require id and question strings")
+        except ValueError:
+            self.respond_error(incoming.id, code=-32602, message="invalid user-question/ask parameters")
+            return
+        cancelled = threading.Event()
+
+        def answer() -> None:
+            try:
+                if cancelled.is_set() or handler is None:
+                    return
+                result = handler(request, cancelled)
+                wire_result = result.model_dump(exclude_none=True) if isinstance(result, UserQuestionAnswer) else result
+                with self._lock:
+                    if not cancelled.is_set() and self._proc is proc:
+                        self.respond(incoming.id, wire_result)
+            except Exception as exc:
+                with self._lock:
+                    if not cancelled.is_set() and self._proc is proc:
+                        try:
+                            self.respond_error(incoming.id, code=-32603, message=str(exc))
+                        except TransportClosedError:
+                            cancelled.set()
+            finally:
+                with self._lock:
+                    current = self._question_tasks.get(request.request_id)
+                    if current is not None and current[0] is cancelled:
+                        self._question_tasks.pop(request.request_id, None)
+
+        thread = threading.Thread(target=answer, name="dsh-user-question", daemon=True)
+        with self._lock:
+            if self._closing:
+                return
+            if request.request_id in self._question_tasks:
+                self.respond_error(incoming.id, code=-32602, message="duplicate user-question requestId")
+                return
+            self._question_tasks[request.request_id] = (cancelled, thread)
+            try:
+                thread.start()
+            except BaseException:
+                self._question_tasks.pop(request.request_id, None)
+                cancelled.set()
+                raise
+
     def _runtime_closed_error(self, reason: str) -> TransportClosedError:
         diagnostics = self._runtime_diagnostics()
         return TransportClosedError(f"{reason}\n{diagnostics}" if diagnostics else reason)

python/sdk/src/deepseek_harness/models.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 from dataclasses import dataclass
 from typing import Literal, TypeAlias
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, JsonValue as PydanticJsonValue
 
 JsonScalar: TypeAlias = str | int | float | bool | None
 JsonValue: TypeAlias = JsonScalar | dict[str, "JsonValue"] | list["JsonValue"]
@@ -45,3 +45,24 @@ class ApprovalRequestResult(BaseModel):
     """One-operation decision; allowed-once is the only grant."""
 
     outcome: Literal["allowed-once", "rejected", "cancelled"]
+
+class UserQuestionRequest(BaseModel):
+    """Runtime question batch; questions retain their options, detail, and presentation intent."""
+
+    request_id: str = Field(alias="requestId", min_length=1)
+    session_id: str = Field(alias="sessionId")
+    questions: list[dict[str, PydanticJsonValue]] = Field(min_length=1)
+
+
+class UserQuestionAnswerItem(BaseModel):
+    """Answer to one original question id, using selected labels or custom text."""
+
+    id: str
+    selected: list[str]
+    custom: str | None = None
+
+
+class UserQuestionAnswer(BaseModel):
+    """One answer per original question; an empty batch is not an implicit approval."""
+
+    answers: list[UserQuestionAnswerItem]

python/sdk/tests/manual_sdk_shutdown_smoke.py

@@ -0,0 +1,36 @@
+"""Run a short SDK session, close immediately, and try restoring its history.
+
+Requires an installed SDK/runtime and DEEPSEEK_API_KEY. Run this file directly.
+The temporary Harness home is retained for inspection.
+"""
+
+from pathlib import Path
+from tempfile import mkdtemp
+from time import monotonic
+
+from deepseek_harness import DeepSeekHarness, DeepSeekHarnessConfig
+
+root = Path(mkdtemp(prefix="dsh-shutdown-"))
+config = DeepSeekHarnessConfig(
+    profile="sdk-minimal",
+    dsh_home=str(root / "home"),
+    cwd=str(root),
+    shutdown_timeout_seconds=1.0,
+)
+print("Test directory:", root)
+
+with DeepSeekHarness(config) as harness:
+    result = harness.run("只回复 OK，不要调用工具。")
+    started = monotonic()
+
+print(f"close() took {monotonic() - started:.3f}s")
+print("Reply:", result.final_response)
+if result.finish_reason != "completed":
+    raise RuntimeError(f"Model turn failed ({result.finish_reason}); check credentials and the session log.")
+print("History files:", list((root / "home" / "sessions").rglob("*.jsonl")))
+
+# Recovery gets a separate budget so its own shutdown does not repeat the test.
+config.shutdown_timeout_seconds = 30.0
+with DeepSeekHarness(config) as harness:
+    harness.resume_session(result.session_id)
+    print("History restored successfully:", result.session_id)

python/sdk/tests/test_user_questions.py

@@ -0,0 +1,233 @@
+from __future__ import annotations
+
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel
+
+from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig, UserQuestionRequest
+from deepseek_harness.errors import TransportClosedError
+
+
+class Response(BaseModel):
+    answer: dict | None = None
+    error: dict | None = None
+    cancelled: bool = False
+    replies: int = 0
+    capabilities: dict | None = None
+
+
+@pytest.fixture
+def runtime(tmp_path: Path) -> Path:
+    path = tmp_path / "questions.py"
+    path.write_text('''
+import json
+import sys
+
+pending = {}
+replies = 0
+capabilities = None
+serial = 0
+
+def send(frame):
+    print(json.dumps({'jsonrpc': '2.0', **frame}), flush=True)
+
+for line in sys.stdin:
+    frame = json.loads(line)
+    method = frame.get('method')
+    params = frame.get('params') or {}
+    if method is None:
+        replies += 1
+        original = pending.pop(frame['id'], None)
+        if original is not None:
+            send({'id': original, 'result': {'answer': frame.get('result'), 'error': frame.get('error')}})
+    elif method == 'initialize':
+        capabilities = params.get('capabilities')
+        send({'id': frame['id'], 'result': {'serverInfo': {'name': 'question-peer', 'version': '1'}}})
+    elif method == 'ask':
+        serial += 1
+        wire_id = 'host-' + str(serial)
+        pending[wire_id] = frame['id']
+        send({'id': wire_id, 'method': 'user-question/ask', 'params': {
+            'requestId': params.get('requestId', wire_id), 'sessionId': 'main',
+            'questions': params.get('questions', [{'id': 'format', 'question': 'Which format?', 'detail': 'Plan to review',
+                'options': [{'label': 'CSV'}], 'intent': {'kind': 'plan-review', 'approve': 'CSV'}}])}})
+    elif method == 'cancel':
+        send({'method': 'user-question.cancelled', 'params': {'requestId': params['requestId'], 'sessionId': 'main'}})
+        for original in pending.values():
+            send({'id': original, 'result': {'cancelled': True}})
+        pending.clear()
+        send({'id': frame['id'], 'result': {}})
+    elif method == 'disconnect':
+        break
+    elif method == 'stats':
+        send({'id': frame['id'], 'result': {'replies': replies, 'capabilities': capabilities}})
+    elif method == 'shutdown':
+        send({'id': frame['id'], 'result': {}})
+        break
+''')
+    return path
+
+
+def config(callback=None) -> HarnessConfig:
+    return HarnessConfig(on_user_question=callback, request_timeout_seconds=5)
+
+
+def initialize(client: HarnessClient, root: Path) -> None:
+    client.initialize(cwd=str(root), provider="mock", model="mock")
+
+
+def test_callback_preserves_question_metadata_and_answers_without_blocking_reader(runtime: Path, tmp_path: Path) -> None:
+    def answer(request: UserQuestionRequest, cancelled: threading.Event):
+        assert request.session_id == "main"
+        assert request.questions[0]["intent"] == {"kind": "plan-review", "approve": "CSV"}
+        assert request.questions[0]["detail"] == "Plan to review"
+        assert not cancelled.is_set()
+        # Reentrant RPC proves the reader keeps processing while the host decides.
+        assert client.request("stats", {}, response_model=Response).capabilities == {"userQuestions": True}
+        return {"answers": [{"id": "format", "selected": ["CSV"]}]}
+
+    with HarnessClient(config(answer), _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+        result = client.request("ask", {}, response_model=Response)
+        assert result.answer == {"answers": [{"id": "format", "selected": ["CSV"]}]}
+        assert result.error is None
+
+
+def test_missing_callback_does_not_advertise_question_support(runtime: Path, tmp_path: Path) -> None:
+    with HarnessClient(config(), _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+        assert client.request("stats", {}, response_model=Response).capabilities is None
+
+
+@pytest.mark.parametrize("action", ["cancel", "disconnect", "close"])
+def test_cancellation_stops_cooperative_host_work_and_suppresses_late_answer(runtime: Path, tmp_path: Path, action: str) -> None:
+    entered = threading.Event()
+    stopped = threading.Event()
+    errors: list[BaseException] = []
+
+    def answer(request: UserQuestionRequest, cancelled: threading.Event):
+        entered.set()
+        assert cancelled.wait(5), "callback did not receive cancellation"
+        stopped.set()
+        return {"answers": [{"id": "format", "selected": ["late"]}]}
+
+    with HarnessClient(config(answer), _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+
+        def ask():
+            try:
+                client.request("ask", {"requestId": "pending"}, response_model=Response)
+            except BaseException as error:
+                errors.append(error)
+
+        caller = threading.Thread(target=ask, daemon=True)
+        caller.start()
+        try:
+            assert entered.wait(5)
+            with client._lock:
+                workers = [thread for _cancelled, thread in client._question_tasks.values()]
+            if action == "close":
+                client.close()
+            else:
+                try:
+                    client.request(action, {"requestId": "pending"}, response_model=Response)
+                except TransportClosedError:
+                    assert action == "disconnect"
+            assert stopped.wait(5)
+            for worker in workers:
+                worker.join(5)
+                assert not worker.is_alive()
+            caller.join(5)
+            assert not caller.is_alive()
+            if action == "cancel":
+                assert client.request("stats", {}, response_model=Response).replies == 0
+                assert errors == []
+            else:
+                assert len(errors) == 1 and isinstance(errors[0], TransportClosedError)
+        finally:
+            client.close()
+            caller.join(5)
+
+
+def test_callback_failure_is_returned_to_runtime_and_reader_remains_usable(runtime: Path, tmp_path: Path) -> None:
+    def answer(_request: UserQuestionRequest, _cancelled: threading.Event):
+        raise ValueError("answer service unavailable")
+
+    with HarnessClient(config(answer), _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+        result = client.request("ask", {}, response_model=Response)
+        assert result.error == {"code": -32603, "message": "answer service unavailable"}
+        assert client.request("stats", {}, response_model=Response).replies == 1
+
+
+def test_high_level_config_installs_the_same_native_callback(runtime: Path, tmp_path: Path) -> None:
+    with DeepSeekHarness(
+        cwd=str(tmp_path), _launch_args=(sys.executable, str(runtime)),
+        on_user_question=lambda _request, _cancelled: {"answers": [{"id": "format", "selected": [], "custom": "JSON"}]},
+    ) as harness:
+        result = harness.client.request("ask", {}, response_model=Response)
+        assert result.answer == {"answers": [{"id": "format", "selected": [], "custom": "JSON"}]}
+
+
+def test_late_callback_cannot_answer_or_remove_a_new_process_question(runtime: Path, tmp_path: Path) -> None:
+    old_entered = threading.Event()
+    release_old = threading.Event()
+    new_entered = threading.Event()
+    stopped = threading.Event()
+
+    def old_answer(_request, cancelled):
+        old_entered.set()
+        assert release_old.wait(5)
+        assert cancelled.is_set()
+        return {"answers": [{"id": "format", "selected": ["old"]}]}
+
+    def new_answer(_request, cancelled):
+        new_entered.set()
+        assert cancelled.wait(5)
+        stopped.set()
+        return {"answers": [{"id": "format", "selected": ["new"]}]}
+
+    failures = []
+    settings = config(old_answer)
+    with HarnessClient(settings, _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+
+        def ask():
+            try:
+                client.request("ask", {"requestId": "reused"}, response_model=Response)
+            except TransportClosedError as error:
+                failures.append(error)
+
+        old_caller = threading.Thread(target=ask, daemon=True)
+        new_caller = threading.Thread(target=ask, daemon=True)
+        old_caller.start()
+        try:
+            assert old_entered.wait(5)
+            with client._lock:
+                old_worker = client._question_tasks["reused"][1]
+            client.close()
+            old_caller.join(5)
+            assert not old_caller.is_alive()
+            settings.on_user_question = new_answer
+            client.start()
+            initialize(client, tmp_path)
+            new_caller.start()
+            assert new_entered.wait(5)
+            release_old.set()
+            old_worker.join(5)
+            assert not old_worker.is_alive()
+            assert client.request("stats", {}, response_model=Response).replies == 0
+            client.request("cancel", {"requestId": "reused"}, response_model=Response)
+            assert stopped.wait(5)
+            new_caller.join(5)
+            assert not new_caller.is_alive()
+            assert len(failures) == 1
+        finally:
+            release_old.set()
+            client.close()
+            old_caller.join(5)
+            if new_caller.ident is not None:
+                new_caller.join(5)