feat(sdk): support compact through unified query input

python/sdk/README.i18n.yaml

@@ -9,8 +9,11 @@
   en: ddf4dabea526cef1
   zh: 6e31cab10d2f3ed8
 /deepseek-harness-python-sdk/start-a-runtime:
-  en: 1ae09bebf0f9e19b
-  zh: 55a0d557a2cae4d8
+  en: c4c30913fe2642ef
+  zh: 0ed42bca37fba60b
+/deepseek-harness-python-sdk/query-and-response:
+  en: b3ad6ea82350b467
+  zh: 85dacc783fd12992
 /deepseek-harness-python-sdk/client-approvals:
   en: fc69be1beb8abff1
   zh: 91b56781931a6f68

python/sdk/README.md

@@ -58,6 +58,17 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
     result = session.run("Continue the task.")
 ```
 
+<a id="query-and-response"></a>
+## Query and response
+
+Call `session.query(text)` to send text, then iterate `session.receive_response()` to observe its result. `query()` returns `None` after sending, without waiting for the runtime's response. The runtime intercepts only `/compact` as a command; all other text, including `/plan` and other slash-prefixed input, remains an ordinary prompt. This input API accepts text strings only. `delivery="queue"` is the default; `delivery="steer"` directs an ordinary prompt to the next step. Command timing remains owned by the command.
+
+`receive_response()` claims the next query in send order. It first waits for the routing response, then yields `Notification` objects followed by one terminal value: `RunResult` for the prompt's inbox-receipt-to-idle interval, or `CommandExecution` after the command settles. Notifications can accumulate before receiving begins. Separate receivers can observe their queries concurrently, and another `query()` can be sent while a receiver waits. Close an unfinished iterator to release only its local observation; this does not cancel or roll back runtime work. Resources are released before yielding the terminal value.
+
+`CommandExecution` contains `command_id` and a `result` discriminated by `kind`. Success may include `text` and `source_event_seq`; an error includes `text`. Handler errors return `kind="error"`; routing, protocol and transport failures raise during receiving. Sending failures raise from `query()`. `request_timeout_seconds` bounds waiting for the routing response, not later prompt activity; an already-arrived response remains receivable after that deadline.
+
+The full `sdk` profile provides the intercepted `/compact` command. It can make an auxiliary model request and completes without requiring a normal turn or idle notification; empty history can succeed without a summary. The default `sdk-minimal` profile does not provide command execution. `session.run("/compact")` remains ordinary prompt text. The lower-level blocking `HarnessClient.session_command(session_id, line)` remains available for strict command execution and rejects command names other than `/compact`. See the [full SDK compaction example](examples/README.md#compact-a-full-sdk-session).
+
 <a id="client-approvals"></a>
 ## Client approvals
 

python/sdk/README.zh.md

@@ -58,6 +58,17 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
     result = session.run("Continue the task.")
 ```
 
+<a id="query-and-response"></a>
+## 查询与接收
+
+调用 `session.query(text)` 发送文本，再迭代 `session.receive_response()` 观察结果。`query()` 发送后即返回 `None`，不等待运行时响应。运行时只将 `/compact` 拦截为命令；包括 `/plan` 和其他以斜杠开头的输入在内，其余文本仍作为普通提示词。此输入 API 只接受文本字符串。`delivery="queue"` 是默认值；`delivery="steer"` 将普通提示词交给下一个 step。命令执行时机仍由命令决定。
+
+`receive_response()` 按发送顺序领取下一次 query。它先等待路由响应，再逐个产出 `Notification`，最后产出一个终态值：普通提示词从收件箱回执到 idle 区间的 `RunResult`，或命令处理结束后的 `CommandExecution`。通知可以在开始接收前积累。不同接收器可以并行观察各自的 query，也可以在一个接收器等待期间发送另一个 `query()`。关闭未完成的迭代器只释放它自己的本地观察资源，不取消或回滚运行时操作。终态值产出前会释放资源。
+
+`CommandExecution` 包含 `command_id` 和以 `kind` 区分的 `result`。成功结果可以包含 `text` 和 `source_event_seq`；错误结果包含 `text`。处理器错误返回 `kind="error"`；路由、协议和传输失败在接收时抛出。发送失败从 `query()` 抛出。`request_timeout_seconds` 限制路由响应的等待时间，不限制后续提示词活动；已经到达的响应在期限过后仍可接收。
+
+完整 `sdk` profile 提供被拦截的 `/compact` 命令。它可能发起辅助模型请求，完成时不要求普通轮次或 idle 通知；空历史可以不生成摘要而成功返回。默认 `sdk-minimal` profile 不提供命令执行能力。`session.run("/compact")` 仍按普通提示词文本处理。底层阻塞式 `HarnessClient.session_command(session_id, line)` 仍可用于严格命令执行，并拒绝 `/compact` 之外的命令名。另见[完整 SDK 压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
+
 <a id="client-approvals"></a>
 ## 客户端审批
 

python/sdk/examples/README.i18n.yaml

@@ -3,11 +3,14 @@
 # After editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write python/sdk/examples/README.md
 /python-sdk-example:
-  en: d9dfc996ea2ace40
-  zh: 33de241bad6c4ee1
+  en: c97ffc5ce0f7db05
+  zh: bdd91202075498a3
 /python-sdk-example/run-the-minimal-agent:
-  en: f0d785582fe72a91
-  zh: ee563fb17c775ed3
+  en: 4b79fd6576e2759d
+  zh: 6258540acaf4bc37
+/python-sdk-example/compact-a-full-sdk-session:
+  en: 0bada22014bf5e6c
+  zh: 4308ea1a14b48f57
 /python-sdk-example/add-plugins:
   en: 5e1e30dac13f6c9e
   zh: 0af98ebb75c88f43

python/sdk/examples/README.md

@@ -2,7 +2,7 @@
 
 English | [中文](README.zh.md)
 
-Runnable Python SDK example over the sole application launcher, `dsh --profile sdk-minimal`. The Python client owns JSON-RPC stdio; the profile owns the agent composition, persistence, execution policy, and plugins.
+Runnable Python SDK examples over the sole application launcher, `dsh`. The Python client owns JSON-RPC stdio; the selected profile owns the agent composition, persistence, execution policy, and plugins.
 
 ## Run the minimal agent
 
@@ -29,6 +29,19 @@ The bundle does not include `dsh-base`, so every additional row is an explicit p
 
 The persistent PTY can modify any path available to the runtime process, so use a disposable checkout or container.
 
+<a id="compact-a-full-sdk-session"></a>
+## Compact a full SDK Session
+
+The [compaction example](compact.py) uses the full `sdk` profile. It reads `README.md` from the selected workspace into a `Session.query()` prompt, sends `/compact` through the same method, then asks a follow-up using the retained context. Each query is paired with `Session.receive_response()`. It uses the same model credentials and `DSH_MODEL` default as the minimal example, and requires an explicit home:
+
+```sh
+python python/sdk/examples/compact.py \
+  --dsh-home /absolute/path/to/compact-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace
+```
+
+The script consumes notifications and prints the command id and outcome separately from the model's responses. A failed turn or command stops the example. See [Query and response](../README.md#query-and-response) for completion and error behavior; `minimal.py` retains its `run()` entry and does not enable command execution.
+
 ## Add plugins
 
 Use the runtime wheel's `dsh` command against the same explicit home for persistent profile changes:

python/sdk/examples/README.zh.md

@@ -2,7 +2,7 @@
 
 [English](README.md) | 中文
 
-基于唯一应用启动器 `dsh --profile sdk-minimal` 的可运行 Python SDK 示例。Python 客户端负责 JSON-RPC stdio；profile 负责 agent（智能体）组合、持久化、执行策略与插件。
+基于唯一应用启动器 `dsh` 的可运行 Python SDK 示例。Python 客户端负责 JSON-RPC stdio；所选 profile 负责 agent（智能体）组合、持久化、执行策略与插件。
 
 ## 运行极简 agent
 
@@ -29,6 +29,19 @@ python python/sdk/examples/minimal.py \
 
 持久 PTY 可以修改运行时进程可访问的任何路径，因此只应在一次性 checkout 或容器中使用。
 
+<a id="compact-a-full-sdk-session"></a>
+## 压缩完整 SDK Session
+
+[压缩示例](compact.py)使用完整 `sdk` profile。它将所选工作区的 `README.md` 读入 `Session.query()` 输入，通过同一方法发送 `/compact`，随后使用保留的上下文继续提问。每次 query 都配对调用 `Session.receive_response()`。模型凭据和 `DSH_MODEL` 默认值与极简示例一致，并且要求显式指定 home：
+
+```sh
+python python/sdk/examples/compact.py \
+  --dsh-home /absolute/path/to/compact-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace
+```
+
+脚本消费通知，并将命令 id 和结果与模型回复分开打印。轮次或命令失败时，示例会停止。完成和错误行为见[查询与接收](../README.zh.md#query-and-response)；`minimal.py` 保留其 `run()` 入口，不启用命令执行。
+
 ## 添加插件
 
 对同一个显式 home 使用运行时 wheel 包提供的 `dsh` 命令，以进行持久 profile 变更：

python/sdk/examples/compact.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Compact a full SDK Session and continue with its retained context."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from pathlib import Path
+
+from deepseek_harness import CommandExecution, DeepSeekHarness, RunResult, Session
+
+
+def receive(session: Session) -> None:
+    """Consume notifications and print the query's command or activity result."""
+    for result in session.receive_response():
+        if isinstance(result, RunResult):
+            if result.finish_reason != "completed":
+                raise RuntimeError(f"Agent turn ended with {result.finish_reason}: {result.final_response}")
+            print(result.final_response)
+        elif isinstance(result, CommandExecution):
+            print(f"Command {result.command_id}: {result.result.kind}")
+            if result.result.kind == "error":
+                raise RuntimeError(result.result.text)
+            if result.result.text is not None:
+                print(result.result.text)
+
+
+def main() -> None:
+    """Create conversation history, compact it, and ask a follow-up question."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dsh-home", required=True, type=Path)
+    parser.add_argument("--workspace", type=Path, default=Path.cwd())
+    parser.add_argument("--session-id")
+    parser.add_argument("--provider", default="deepseek-official")
+    parser.add_argument("--model", default=os.environ.get("DSH_MODEL", "deepseek-v4-flash"))
+    parser.add_argument("--max-tokens", type=int)
+    args = parser.parse_args()
+
+    workspace = args.workspace.resolve()
+    readme = workspace / "README.md"
+    if not readme.is_file():
+        parser.error("--workspace must contain README.md to provide conversation history")
+
+    with DeepSeekHarness(
+        dsh_home=str(args.dsh_home.resolve()),
+        cwd=str(workspace),
+        profile="sdk",
+        provider=args.provider,
+        model=args.model,
+        max_tokens=args.max_tokens,
+    ) as harness:
+        session = harness.start_session(args.session_id)
+        session.query(
+            "Read this project README and summarize its purpose and development commands without using tools.\n\n"
+            + readme.read_text(encoding="utf-8")
+        )
+        receive(session)
+        session.query("/compact")
+        receive(session)
+        session.query("What are the project's purpose and development commands? Use the earlier README without tools.")
+        receive(session)
+
+
+if __name__ == "__main__":
+    main()

python/sdk/src/deepseek_harness/__init__.py

@@ -2,7 +2,8 @@ from .api import DeepSeekHarness, DeepSeekHarnessConfig, RunResult, Session
 from .client import HarnessClient, HarnessConfig, UserQuestionHandler
 from .errors import SdkProtocolError
 from .models import (
-    ApprovalRequestParams, ApprovalRequestResult, IncomingRequest, InitializeResponse,
+    ApprovalRequestParams, ApprovalRequestResult, CommandErrorResult, CommandExecution,
+    CommandResult, CommandSuccessResult, IncomingRequest, InitializeResponse,
     JsonObject, Notification, ServerInfo, UserQuestionAnswer, UserQuestionAnswerItem, UserQuestionRequest,
 )
 
@@ -11,6 +12,10 @@ __all__ = [
     "DeepSeekHarnessConfig",
     "Session",
     "RunResult",
+    "CommandExecution",
+    "CommandResult",
+    "CommandSuccessResult",
+    "CommandErrorResult",
     "HarnessClient",
     "HarnessConfig",
     "UserQuestionHandler",

python/sdk/src/deepseek_harness/api.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
 import uuid
+import threading
+from collections import deque
+from collections.abc import Iterator
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Callable
+from typing import Callable, Literal
 
-from .client import HarnessClient, HarnessConfig, UserQuestionHandler
-from .errors import SdkProtocolError
-from .models import JsonObject, Notification
+from .client import HarnessClient, HarnessConfig, NotificationSubscription, UserQuestionHandler, _PendingRequest
+from .errors import SdkProtocolError, TransportClosedError
+from .models import CommandExecution, JsonObject, Notification
 
 
 @dataclass(slots=True)
@@ -48,6 +51,113 @@ class RunResult:
     notifications: list[Notification]
 
 
+class _QueryResponseIterator(Iterator[Notification | RunResult | CommandExecution]):
+    def __init__(
+        self, client: HarnessClient, session_id: str, request: _PendingRequest,
+        subscription: NotificationSubscription,
+    ) -> None:
+        self._client = client
+        self._session_id = session_id
+        self._request = request
+        self._subscription = subscription
+        self._lock = threading.Lock()
+        self._closed = False
+        self._released = False
+        self._routed = False
+        self._received = False
+        self._message_id = ""
+        self._notifications: list[Notification] = []
+        self._events: list[JsonObject] = []
+        self._buffer: deque[Notification] = deque()
+        self._terminal: RunResult | CommandExecution | None = None
+
+    def _release(self, error: BaseException | None = None) -> None:
+        with self._lock:
+            if self._released:
+                return
+            self._released = True
+        self._client._discard_request(self._request, error)
+        if error is None:
+            self._subscription.close()
+        else:
+            self._subscription._cancel(error)
+
+    def close(self) -> None:
+        """Stop observing this query; the runtime operation is not cancelled."""
+        with self._lock:
+            if self._closed:
+                return
+            self._closed = True
+        self._release(TransportClosedError("query response iterator closed"))
+
+    def _ensure_open(self) -> None:
+        if self._closed:
+            raise StopIteration
+
+    def __next__(self) -> Notification | RunResult | CommandExecution:
+        try:
+            self._ensure_open()
+            if not self._routed:
+                response = self._client._wait_request(self._request)
+                self._ensure_open()
+                self._routed = True
+                if not isinstance(response, dict):
+                    raise SdkProtocolError("session/query response must be a JSON object")
+                if response.get("kind") == "command":
+                    self._terminal = CommandExecution.model_validate(response.get("execution"))
+                    self._release()
+                    while True:
+                        try:
+                            notification = self._subscription._next_queued()
+                        except TransportClosedError:
+                            break
+                        if notification is None:
+                            break
+                        self._buffer.append(notification)
+                        if _is_command_done(notification, self._session_id, self._terminal.command_id):
+                            break
+                elif response.get("kind") == "prompt" and isinstance(response.get("messageId"), str) and response["messageId"]:
+                    self._message_id = response["messageId"]
+                else:
+                    raise SdkProtocolError("session/query requires a prompt messageId or command execution result")
+            self._ensure_open()
+            if self._buffer:
+                return self._buffer.popleft()
+            if self._terminal is not None:
+                terminal = self._terminal
+                self._terminal = None
+                self._release()
+                self._closed = True
+                return terminal
+            while True:
+                notification = self._subscription.next()
+                self._ensure_open()
+                if not self._received:
+                    if not _is_inbox_receipt(notification, self._session_id, self._message_id):
+                        continue
+                    self._received = True
+                self._notifications.append(notification)
+                if notification.method == "session.event" and notification.payload.get("sessionId") == self._session_id:
+                    event = notification.payload.get("event")
+                    if isinstance(event, dict):
+                        self._events.append(event)
+                if (notification.method == "session.status"
+                    and notification.payload.get("sessionId") == self._session_id
+                    and notification.payload.get("status") == "idle"):
+                    self._terminal = RunResult(
+                        session_id=self._session_id, final_response=final_response(self._events),
+                        finish_reason=finish_reason(self._events), events=self._events, notifications=self._notifications,
+                    )
+                    self._release()
+                return notification
+        except BaseException as error:
+            locally_closed = self._closed
+            self.close()
+            if locally_closed and isinstance(error, (TransportClosedError, StopIteration)):
+                raise StopIteration
+            raise
+
+
 class DeepSeekHarness:
     """Reusable synchronous SDK for running DeepSeek Harness agent turns.
 
@@ -148,6 +258,41 @@ class Session:
     def __init__(self, harness: DeepSeekHarness, session_id: str) -> None:
         self.harness = harness
         self.id = session_id
+        self._queries: deque[_QueryResponseIterator] = deque()
+        self._query_lock = threading.Lock()
+
+    def query(self, text: str, *, delivery: Literal["queue", "steer"] = "queue") -> None:
+        """Send text for runtime prompt or command routing without waiting for the response.
+
+        Pair each call with receive_response(). Delivery selects where ordinary
+        prompts enter the inbox; commands retain their native execution timing.
+        """
+        if not isinstance(text, str):
+            raise TypeError("query text must be a string")
+        if delivery not in ("queue", "steer"):
+            raise ValueError("query delivery must be queue or steer")
+        self.harness.start()
+        client = self.harness.client
+        with self._query_lock:
+            subscription = client.subscribe_session_notifications(self.id)
+            try:
+                request = client._begin_request("session/query", {"sessionId": self.id, "text": text, "delivery": delivery})
+            except BaseException:
+                subscription.close()
+                raise
+            self._queries.append(_QueryResponseIterator(client, self.id, request, subscription))
+
+    def receive_response(self) -> _QueryResponseIterator:
+        """Claim the next query in send order and iterate its notifications and terminal result.
+
+        Prompts collect the inbox-receipt-to-idle interval used by run(); commands
+        finish on their RPC result. Close an unfinished iterator to release its
+        local observation resources without cancelling the runtime operation.
+        """
+        with self._query_lock:
+            if not self._queries:
+                raise RuntimeError("receive_response requires a preceding query")
+            return self._queries.popleft()
 
     def get_working_directory(self) -> str:
         """Read the effective directory, recovering a missing directory to the Session origin."""
@@ -225,6 +370,16 @@ def _is_inbox_receipt(notification: Notification, session_id: str, message_id: s
     )
 
 
+def _is_command_done(notification: Notification, session_id: str, command_id: str) -> bool:
+    if notification.method != "session.event" or notification.payload.get("sessionId") != session_id:
+        return False
+    event = notification.payload.get("event")
+    if not isinstance(event, dict) or event.get("type") != "command/done":
+        return False
+    data = event.get("data")
+    return isinstance(data, dict) and data.get("commandId") == command_id
+
+
 def normalize_input(input: str | list[JsonObject]) -> list[JsonObject]:
     if isinstance(input, str):
         return [{"type": "text", "text": input}]

python/sdk/src/deepseek_harness/client.py

@@ -15,7 +15,7 @@ from typing import Callable, TypeAlias, TypeVar
 from pydantic import BaseModel
 
 from .errors import JsonRpcError, SdkProtocolError, TransportClosedError
-from .models import IncomingRequest, InitializeResponse, JsonObject, JsonValue, Notification, UserQuestionAnswer, UserQuestionRequest
+from .models import CommandExecution, IncomingRequest, InitializeResponse, JsonObject, JsonValue, Notification, UserQuestionAnswer, UserQuestionRequest
 
 ModelT = TypeVar("ModelT", bound=BaseModel)
 NotificationFilter: TypeAlias = Callable[[Notification], bool]
@@ -39,6 +39,14 @@ class HarnessConfig:
     on_user_question: UserQuestionHandler | None = None
 
 
+@dataclass(slots=True)
+class _PendingRequest:
+    request_id: str
+    method: str
+    waiter: queue.Queue[JsonValue | BaseException]
+    deadline: float | None
+
+
 class HarnessClient:
     """Synchronous JSON-RPC client for the DeepSeek Harness SDK runtime over stdio."""
 
@@ -225,6 +233,14 @@ class HarnessClient:
             response_model=_EmptyResponse,
         )
 
+    def session_command(self, session_id: str, line: str) -> CommandExecution:
+        """Execute an explicit command and return its handler result without waiting for a turn."""
+        return self.request(
+            "session/command",
+            {"sessionId": session_id, "line": line},
+            response_model=CommandExecution,
+        )
+
     def get_working_directory(self, session_id: str) -> str:
         """Read the effective directory, recovering a missing directory to the Session origin."""
         response = self.request(
@@ -324,15 +340,29 @@ class HarnessClient:
         notification_filter: NotificationFilter | None = None,
         notification_subscription: "NotificationSubscription | None" = None,
     ) -> JsonValue:
-        request_id = str(uuid.uuid4())
-        waiter: queue.Queue[JsonValue | BaseException] = queue.Queue(maxsize=1)
         temp_subscription: NotificationSubscription | None = None
         subscription = notification_subscription
-        with self._lock:
-            self._responses[request_id] = waiter
         if on_notification is not None and subscription is None:
             temp_subscription = self.subscribe_notifications(notification_filter)
             subscription = temp_subscription
+        try:
+            pending = self._begin_request(method, params, timeout_seconds=timeout_seconds)
+            return self._wait_request(pending, on_notification=on_notification, notification_subscription=subscription)
+        finally:
+            if temp_subscription is not None:
+                temp_subscription.close()
+
+    def _begin_request(
+        self,
+        method: str,
+        params: JsonObject | None = None,
+        *,
+        timeout_seconds: float | None = None,
+    ) -> _PendingRequest:
+        request_id = str(uuid.uuid4())
+        waiter: queue.Queue[JsonValue | BaseException] = queue.Queue(maxsize=1)
+        with self._lock:
+            self._responses[request_id] = waiter
         try:
             message: JsonObject = {"jsonrpc": "2.0", "id": request_id, "method": method}
             if params is not None:
@@ -341,48 +371,59 @@ class HarnessClient:
         except BaseException:
             with self._lock:
                 self._responses.pop(request_id, None)
-            if temp_subscription is not None:
-                temp_subscription.close()
             raise
         timeout = self.config.request_timeout_seconds if timeout_seconds is None else timeout_seconds
         deadline = None if timeout is None else time.monotonic() + timeout
+        return _PendingRequest(request_id, method, waiter, deadline)
+
+    def _discard_request(self, pending: _PendingRequest, error: BaseException | None = None) -> None:
+        with self._lock:
+            if self._responses.get(pending.request_id) is pending.waiter:
+                self._responses.pop(pending.request_id)
+                if error is not None:
+                    pending.waiter.put_nowait(error)
+
+    def _wait_request(
+        self,
+        pending: _PendingRequest,
+        *,
+        on_notification: Callable[[Notification], None] | None = None,
+        notification_subscription: "NotificationSubscription | None" = None,
+    ) -> JsonValue:
+        subscription = notification_subscription
         try:
             while True:
                 if on_notification is not None and subscription is not None:
                     subscription.drain(on_notification)
+                try:
+                    item = pending.waiter.get_nowait()
+                    break
+                except queue.Empty:
+                    pass
                 wait_timeout = None
                 if on_notification is not None:
                     wait_timeout = 0.05
-                if deadline is not None:
-                    remaining = deadline - time.monotonic()
+                if pending.deadline is not None:
+                    remaining = pending.deadline - time.monotonic()
                     if remaining <= 0:
-                        with self._lock:
-                            self._responses.pop(request_id, None)
                         diagnostics = self._runtime_diagnostics()
                         suffix = f"\n{diagnostics}" if diagnostics else ""
                         raise TimeoutError(
-                            f"{method} timed out waiting for DeepSeek Harness runtime{suffix}"
+                            f"{pending.method} timed out waiting for DeepSeek Harness runtime{suffix}"
                         )
                     wait_timeout = remaining if wait_timeout is None else min(wait_timeout, remaining)
                 try:
-                    item = waiter.get(timeout=wait_timeout)
-                    if on_notification is not None and subscription is not None:
-                        subscription.drain(on_notification)
+                    item = pending.waiter.get(timeout=wait_timeout)
                     break
                 except queue.Empty:
                     continue
-        except BaseException:
-            with self._lock:
-                self._responses.pop(request_id, None)
-            if temp_subscription is not None:
-                temp_subscription.close()
-            raise
+            if on_notification is not None and subscription is not None:
+                subscription.drain(on_notification)
+            if isinstance(item, BaseException):
+                raise item
+            return item
         finally:
-            if temp_subscription is not None:
-                temp_subscription.close()
-        if isinstance(item, BaseException):
-            raise item
-        return item
+            self._discard_request(pending)
 
     def _write_message(self, message: JsonObject) -> None:
         proc = self._proc
@@ -675,12 +716,25 @@ class NotificationSubscription:
         self._closed = True
         self._client._unsubscribe_notifications(self._subscription_id)
 
+    def _cancel(self, error: BaseException) -> None:
+        self.close()
+        self._notifications.put(error)
+
     def next(self) -> Notification:
         item = self._notifications.get()
         if isinstance(item, BaseException):
             raise item
         return item
 
+    def _next_queued(self) -> Notification | None:
+        try:
+            item = self._notifications.get_nowait()
+        except queue.Empty:
+            return None
+        if isinstance(item, BaseException):
+            raise item
+        return item
+
     def drain(self, on_notification: Callable[[Notification], None]) -> None:
         while True:
             try:

python/sdk/src/deepseek_harness/models.py

@@ -1,9 +1,9 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Literal, TypeAlias
+from typing import Annotated, Literal, TypeAlias
 
-from pydantic import BaseModel, Field, JsonValue as PydanticJsonValue
+from pydantic import BaseModel, Field, JsonValue as PydanticJsonValue, field_validator
 
 JsonScalar: TypeAlias = str | int | float | bool | None
 JsonValue: TypeAlias = JsonScalar | dict[str, "JsonValue"] | list["JsonValue"]
@@ -32,6 +32,45 @@ class InitializeResponse(BaseModel):
     serverInfo: ServerInfo | None = None
 
 
+class CommandSuccessResult(BaseModel):
+    """Successful handler result, optionally linked to a Session event."""
+
+    kind: Literal["success"]
+    text: str | None = None
+    source_event_seq: int | None = Field(default=None, alias="sourceEventSeq", strict=True, ge=0, le=2**53 - 1)
+
+    @field_validator("text", "source_event_seq", mode="before")
+    @classmethod
+    def _reject_null(cls, value: object) -> object:
+        if value is None:
+            raise ValueError("optional command result fields must be omitted instead of null")
+        return value
+
+
+class CommandErrorResult(BaseModel):
+    """A command handler's reported failure."""
+
+    kind: Literal["error"]
+    text: str
+
+    @field_validator("text")
+    @classmethod
+    def _non_blank_text(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("command error text must not be blank")
+        return value
+
+
+CommandResult: TypeAlias = Annotated[CommandSuccessResult | CommandErrorResult, Field(discriminator="kind")]
+
+
+class CommandExecution(BaseModel):
+    """Settled command result and the id pairing its command/run and command/done events."""
+
+    command_id: str = Field(alias="commandId", min_length=1)
+    result: CommandResult
+
+
 class ApprovalRequestParams(BaseModel):
     """Server-to-client approval/request payload after explicit handshake opt-in."""
 

python/sdk/tests/test_commands.py

@@ -0,0 +1,139 @@
+"""Explicit command transport, results, and ordinary prompt compatibility."""
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+import pytest
+from pydantic import ValidationError
+
+from deepseek_harness import CommandErrorResult, CommandExecution, CommandSuccessResult, DeepSeekHarness
+from deepseek_harness.errors import JsonRpcError
+
+
+@pytest.fixture
+def peer(tmp_path: Path) -> Path:
+    script = tmp_path / "commands.py"
+    script.write_text('''import json
+import os
+import sys
+
+def send(frame):
+    print(json.dumps({"jsonrpc": "2.0", **frame}), flush=True)
+
+for line in sys.stdin:
+    frame = json.loads(line)
+    with open(os.environ["REQUEST_LOG"], "a", encoding="utf-8") as log:
+        log.write(line)
+    method = frame["method"]
+    params = frame.get("params") or {}
+    if method == "initialize":
+        send({"id": frame["id"], "result": {"serverInfo": {"name": "command-peer"}}})
+    elif method == "session/command":
+        for event_type in ("command/run", "command/done"):
+            send({"method": "session.event", "params": {"sessionId": params["sessionId"],
+                "event": {"type": event_type, "data": {"commandId": "command-1"}}}})
+        send({"id": frame["id"], **json.loads(os.environ["COMMAND_RESPONSE"])})
+    elif method == "session/prompt":
+        session = params["sessionId"]
+        send({"id": frame["id"], "result": {"messageId": "input-1"}})
+        events = [
+            {"type": "agent/inbox/spliced", "data": {"inserted": [{"id": "input-1"}]}},
+            {"type": "assistant/message", "data": {"message": {"content": [{"type": "text", "text": "ordinary prompt"}]}}},
+            {"type": "turn/end", "data": {"turn": 1, "reason": {"kind": "completed"}}},
+        ]
+        for event in events:
+            send({"method": "session.event", "params": {"sessionId": session, "event": event}})
+        send({"method": "session.status", "params": {"sessionId": session, "status": "idle"}})
+    elif method == "shutdown":
+        send({"id": frame["id"], "result": {}})
+        break
+    else:
+        raise AssertionError(method)
+''', encoding="utf-8")
+    return script
+
+
+def harness(peer: Path, tmp_path: Path, response: dict) -> DeepSeekHarness:
+    return DeepSeekHarness(
+        cwd=str(tmp_path),
+        dsh_home=str(tmp_path / "home"),
+        request_timeout_seconds=5,
+        _launch_args=(sys.executable, str(peer)),
+        env={"REQUEST_LOG": str(tmp_path / "requests.jsonl"), "COMMAND_RESPONSE": json.dumps(response)},
+    )
+
+
+def frames(tmp_path: Path) -> list[dict]:
+    return [json.loads(line) for line in (tmp_path / "requests.jsonl").read_text(encoding="utf-8").splitlines()]
+
+
+@pytest.mark.parametrize("result", [
+    {"kind": "success"},
+    {"kind": "success", "text": "Compacted.", "sourceEventSeq": 0},
+    {"kind": "success", "sourceEventSeq": 2**53 - 1},
+    {"kind": "error", "text": "The agent is busy."},
+    {"kind": "error", "text": "  The agent is busy.\n"},
+])
+def test_command_returns_without_turn_or_idle_and_preserves_events(peer: Path, tmp_path: Path, result: dict) -> None:
+    response = {"commandId": "command-1", "result": result}
+    with harness(peer, tmp_path, {"result": response}) as sdk:
+        session = sdk.start_session("main")
+        with sdk.client.subscribe_session_notifications(session.id) as subscription:
+            execution = sdk.client.session_command(session.id, "/compact")
+            notifications = []
+            subscription.drain(notifications.append)
+        assert isinstance(execution, CommandExecution)
+        assert execution.command_id == "command-1"
+        assert isinstance(execution.result, CommandSuccessResult if result["kind"] == "success" else CommandErrorResult)
+        assert execution.model_dump(by_alias=True, exclude_none=True) == response
+        assert [item.payload["event"]["type"] for item in notifications] == ["command/run", "command/done"]
+    sent = frames(tmp_path)
+    assert [frame["method"] for frame in sent] == ["initialize", "session/command", "shutdown"]
+    assert sent[1]["params"] == {"sessionId": "main", "line": "/compact"}
+
+
+def test_low_level_command_preserves_the_line(peer: Path, tmp_path: Path) -> None:
+    with harness(peer, tmp_path, {"result": {"commandId": "command-1", "result": {"kind": "success"}}}) as sdk:
+        assert sdk.client.session_command("other", "/compact  \t").result.kind == "success"
+    assert frames(tmp_path)[1]["params"] == {"sessionId": "other", "line": "/compact  \t"}
+
+
+def test_command_rpc_rejection_does_not_fall_back_to_prompt(peer: Path, tmp_path: Path) -> None:
+    with harness(peer, tmp_path, {"error": {"code": -32602, "message": "unsupported command"}}) as sdk:
+        with pytest.raises(JsonRpcError, match="unsupported command") as raised:
+            sdk.client.session_command("main", "/plan")
+        assert raised.value.code == -32602
+    assert [frame["method"] for frame in frames(tmp_path)] == ["initialize", "session/command", "shutdown"]
+
+
+@pytest.mark.parametrize("response", [
+    {},
+    {"commandId": "command-1"},
+    {"commandId": "", "result": {"kind": "success"}},
+    {"commandId": "command-1", "result": {"kind": "unknown"}},
+    {"commandId": "command-1", "result": {"kind": "error"}},
+    {"commandId": "command-1", "result": {"kind": "error", "text": ""}},
+    {"commandId": "command-1", "result": {"kind": "error", "text": " \t\n"}},
+    {"commandId": "command-1", "result": {"kind": "success", "text": None}},
+    {"commandId": "command-1", "result": {"kind": "success", "sourceEventSeq": None}},
+    {"commandId": "command-1", "result": {"kind": "success", "sourceEventSeq": "4"}},
+    {"commandId": "command-1", "result": {"kind": "success", "sourceEventSeq": True}},
+    {"commandId": "command-1", "result": {"kind": "success", "sourceEventSeq": -1}},
+    {"commandId": "command-1", "result": {"kind": "success", "sourceEventSeq": 2**53}},
+])
+def test_command_rejects_malformed_results(peer: Path, tmp_path: Path, response: dict) -> None:
+    with harness(peer, tmp_path, {"result": response}) as sdk:
+        with pytest.raises(ValidationError):
+            sdk.client.session_command("main", "/compact")
+
+
+def test_run_keeps_slash_text_as_an_ordinary_prompt(peer: Path, tmp_path: Path) -> None:
+    with harness(peer, tmp_path, {"result": {}}) as sdk:
+        result = sdk.start_session("main").run("/compact")
+        assert result.final_response == "ordinary prompt"
+        assert result.finish_reason == "completed"
+    sent = frames(tmp_path)
+    assert [frame["method"] for frame in sent] == ["initialize", "session/prompt", "shutdown"]
+    assert sent[1]["params"] == {"sessionId": "main", "contentBlocks": [{"type": "text", "text": "/compact"}]}

python/sdk/tests/test_commands_runtime.py

@@ -0,0 +1,465 @@
+"""Exercise /compact through Python and the real sdk profile with a local model.
+
+The default carrier is the repository's built dsh CLI. Set
+DSH_COMMANDS_TEST_RUNTIME=installed to run these same cases against the imported
+SDK wheel and its bundled runtime, including from outside the repository.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import sys
+import threading
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing, contextmanager
+from dataclasses import dataclass, field
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Callable, Iterator, TypeVar
+
+import pytest
+
+from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, RunResult, Session
+
+
+_WAIT_SECONDS = 30
+_SESSION_ID = "commands-runtime"
+_MODEL = "deepseek-v4-flash"
+_HISTORY_MARKER = "DSH_COMMANDS_OLD_USER_INPUT"
+_FIRST_REPLY = "SDK_COMPACT_READY"
+_CHECKPOINT = "## Primary Request and Intent\n- DSH_COMMANDS_CHECKPOINT: retain the earlier task."
+_CONTINUE_PROMPT = "DSH_COMMANDS_CONTINUE: answer using the retained checkpoint."
+_CONTINUE_REPLY = "SDK_COMPACT_OK"
+_T = TypeVar("_T")
+
+
+@dataclass
+class _Completion:
+    text: str
+    blocked: bool = False
+    entered: threading.Event = field(default_factory=threading.Event)
+    release: threading.Event = field(default_factory=threading.Event)
+
+    def __post_init__(self) -> None:
+        if not self.blocked:
+            self.release.set()
+
+
+def _messages_response(text: str, model: str, ordinal: int) -> bytes:
+    events = [
+        {"type": "message_start", "message": {
+            "id": f"commands-response-{ordinal}", "type": "message", "role": "assistant",
+            "model": model, "content": [], "stop_reason": None, "stop_sequence": None,
+            "usage": {"input_tokens": 10000, "output_tokens": 0},
+        }},
+        {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
+        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
+        {"type": "content_block_stop", "index": 0},
+        {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+         "usage": {"output_tokens": 32}},
+        {"type": "message_stop"},
+    ]
+    return "".join(f"event: {event['type']}\ndata: {json.dumps(event)}\n\n" for event in events).encode()
+
+
+class _ModelServer:
+    def __init__(self, completions: list[_Completion]) -> None:
+        self.completions = completions
+        self.requests: list[dict] = []
+        self.failures: list[BaseException] = []
+        self.lock = threading.Lock()
+        owner = self
+
+        class Handler(BaseHTTPRequestHandler):
+            def setup(self) -> None:
+                self.request.settimeout(_WAIT_SECONDS)
+                super().setup()
+
+            def do_POST(self) -> None:
+                assert self.path == "/v1/messages", f"unexpected model endpoint: {self.path}"
+                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+                assert isinstance(body, dict) and isinstance(body.get("model"), str)
+                with owner.lock:
+                    ordinal = len(owner.requests)
+                    owner.requests.append(body)
+                    assert ordinal < len(owner.completions), "unexpected additional model request"
+                    completion = owner.completions[ordinal]
+                completion.entered.set()
+                # The test owns release. This larger bound only contains a failed test.
+                assert completion.release.wait(_WAIT_SECONDS * 2), "model barrier was not released"
+                payload = _messages_response(completion.text, body["model"], ordinal)
+                self.send_response(200)
+                self.send_header("Content-Type", "text/event-stream")
+                self.send_header("Content-Length", str(len(payload)))
+                self.end_headers()
+                self.wfile.write(payload)
+                self.wfile.flush()
+
+            def log_message(self, _format: str, *_args: object) -> None:
+                pass
+
+        class Server(ThreadingHTTPServer):
+            # server_close joins request handlers after their barriers are released.
+            daemon_threads = False
+
+            def handle_error(self, _request: object, _client_address: object) -> None:
+                error = sys.exc_info()[1]
+                assert error is not None
+                with owner.lock:
+                    owner.failures.append(error)
+
+        self.server = Server(("127.0.0.1", 0), Handler)
+        self.thread = threading.Thread(target=self.server.serve_forever, name="commands-model", daemon=True)
+        host, port = self.server.server_address
+        self.url = f"http://{host}:{port}"
+
+    def __enter__(self) -> _ModelServer:
+        try:
+            self.thread.start()
+        except BaseException:
+            self.server.server_close()
+            raise
+        return self
+
+    def release_all(self) -> None:
+        for completion in self.completions:
+            completion.release.set()
+
+    def captured(self) -> list[dict]:
+        with self.lock:
+            return list(self.requests)
+
+    def __exit__(self, exc_type: object, _exc: object, _tb: object) -> None:
+        self.release_all()
+        self.server.shutdown()
+        try:
+            self.server.server_close()
+        finally:
+            self.thread.join(_WAIT_SECONDS)
+        assert not self.thread.is_alive(), "model listener did not stop"
+        if exc_type is None:
+            assert not self.failures, [repr(error) for error in self.failures]
+
+
+def _dsh_bin() -> str | None:
+    mode = os.environ.get("DSH_COMMANDS_TEST_RUNTIME", "built")
+    if mode == "installed":
+        return None
+    if mode != "built":
+        raise ValueError("DSH_COMMANDS_TEST_RUNTIME must be built or installed")
+    repo = Path(__file__).resolve().parents[3]
+    node = shutil.which("node")
+    binary = repo / "apps/cli/lib/bin.js"
+    if node is None or not binary.is_file():
+        pytest.skip("requires Node and built dsh CLI, or DSH_COMMANDS_TEST_RUNTIME=installed")
+    return str(binary)
+
+
+@dataclass
+class _Runtime:
+    harness: DeepSeekHarness
+    model: _ModelServer
+    workers: ThreadPoolExecutor
+    logs: Path
+
+    def call(self, operation: Callable[..., _T], *args: object, **kwargs: object) -> _T:
+        return self.workers.submit(operation, *args, **kwargs).result(timeout=_WAIT_SECONDS)
+
+
+@contextmanager
+def _runtime(
+    tmp_path: Path, completions: list[_Completion], *, capture_patch: Path | None = None,
+) -> Iterator[_Runtime]:
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    home = tmp_path / "home"
+    logs = tmp_path / "sessions"
+    patch = tmp_path / "commands.patch.yml"
+    # Raw JSONL permits an independent read at the compaction durability boundary.
+    patch.write_text(
+        "- id: session-persistence-jsonl\n  config:\n    root: " + json.dumps(str(logs))
+        + "\n    compression: none\n",
+        encoding="utf-8",
+    )
+    patches = (patch,) if capture_patch is None else (patch, capture_patch.resolve())
+    dsh_bin = _dsh_bin()
+    with _ModelServer(completions) as model, ThreadPoolExecutor(max_workers=2) as workers:
+        harness = DeepSeekHarness(
+            cwd=str(workspace), runtime_cwd=str(workspace), dsh_home=str(home),
+            dsh_bin=dsh_bin, profile="sdk", patches=tuple(str(item) for item in patches), model=_MODEL,
+            base_url=model.url, api_key="keyless-commands-runtime",
+            initialize_timeout_seconds=60, request_timeout_seconds=20, shutdown_timeout_seconds=20,
+            env={"DSH_PERMISSION_MODE": "danger-full-access",
+                 "DSH_TELEMETRY_DISABLED": "1", "DSH_TOOLS_MODE": "native"},
+        )
+        try:
+            harness.start()
+            yield _Runtime(harness, model, workers, logs)
+        finally:
+            # Also covers failed initialization: release HTTP, close the client, then join callers.
+            model.release_all()
+            harness.close()
+
+
+def _read_events(root: Path, *, live: bool = False) -> list[dict]:
+    paths = list(root.rglob("session*.jsonl"))
+    assert len(paths) == 1, f"expected one fresh Session log, found {paths}"
+    content = paths[0].read_bytes()
+    if not live:
+        assert content.endswith(b"\n"), "closed Session log has an incomplete final row"
+    # A live writer may append an unflushed suffix after the checkpoint we inspect.
+    complete = content[:content.rfind(b"\n") + 1]
+    records = [json.loads(line) for line in complete.splitlines()]
+    assert records and records[0]["type"] == "session"
+    assert records[0]["id"] == _SESSION_ID
+    events = records[1:]
+    for seq, event in enumerate(events):
+        assert isinstance(event, dict)
+        assert event.setdefault("seq", seq) == seq
+    return events
+
+
+def _of_type(events: list[dict], name: str) -> list[dict]:
+    return [event for event in events if event["type"] == name]
+
+
+def _command_pair(events: list[dict], execution: CommandExecution) -> None:
+    runs = [event for event in _of_type(events, "command/run")
+            if event["data"]["commandId"] == execution.command_id]
+    done = [event for event in _of_type(events, "command/done")
+            if event["data"]["commandId"] == execution.command_id]
+    assert len(runs) == len(done) == 1
+    assert runs[0]["data"]["name"] == "compact"
+    assert runs[0]["seq"] < done[0]["seq"]
+    assert done[0]["data"]["kind"] == execution.result.kind
+    assert done[0]["data"].get("text") == execution.result.text
+
+
+def _text_blocks(content: object) -> list[str]:
+    if isinstance(content, str):
+        return [content]
+    assert isinstance(content, list)
+    return [block["text"] for block in content if block.get("type") == "text"]
+
+
+def _request_texts(request: dict) -> list[str]:
+    return [text for message in request["messages"] for text in _text_blocks(message["content"])]
+
+
+def _receive_result(
+    session: Session, response_claimed: threading.Event | None = None,
+) -> RunResult | CommandExecution:
+    result: RunResult | CommandExecution | None = None
+    with closing(session.receive_response()) as responses:
+        if response_claimed is not None:
+            response_claimed.set()
+        for item in responses:
+            assert result is None, "receive_response yielded another item after its terminal result"
+            if not isinstance(item, Notification):
+                assert isinstance(item, (RunResult, CommandExecution)), type(item)
+                result = item
+    assert result is not None, "receive_response ended without a terminal result"
+    return result
+
+
+def _wait_command_done(next_notification: Callable[[], Notification]) -> dict:
+    while True:
+        notification = next_notification()
+        if notification.method != "session.event" or notification.payload.get("sessionId") != _SESSION_ID:
+            continue
+        event = notification.payload.get("event")
+        if isinstance(event, dict) and event.get("type") == "command/done":
+            return event["data"]
+
+
+def test_compact_without_history_returns_without_a_model_turn(tmp_path: Path) -> None:
+    with _runtime(tmp_path, []) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        observer = runtime.harness.client.subscribe_session_notifications(_SESSION_ID)
+        try:
+            assert runtime.call(session.query, " \n/压缩\t ", delivery="queue") is None
+            # Wait independently for completion before starting the query's receiver.
+            done = runtime.call(_wait_command_done, observer.next)
+            execution = runtime.call(_receive_result, session)
+        except BaseException:
+            # Wake a timed-out observer before unregistering its otherwise blocking queue.
+            runtime.harness.close()
+            raise
+        finally:
+            observer.close()
+        assert isinstance(execution, CommandExecution)
+        assert done["commandId"] == execution.command_id
+        assert execution.result.kind == "success"
+        assert execution.result.source_event_seq is None
+        assert runtime.model.captured() == []
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution)
+    assert not {event["type"] for event in events} & {
+        "turn/start", "step/start", "request/header", "compaction/start", "user/message",
+    }
+
+
+def test_compact_rejects_busy_agent_before_the_model_is_released(tmp_path: Path) -> None:
+    response = _Completion("DSH_COMMANDS_BUSY_REPLY", blocked=True)
+    with _runtime(tmp_path, [response]) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        sending = runtime.workers.submit(
+            session.query, "Hold this request while compaction is attempted.", delivery="queue",
+        )
+        assert response.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        assert sending.result(timeout=_WAIT_SECONDS) is None
+        assert not response.release.is_set(), "query waited for the model before returning"
+
+        # Claim the prompt's FIFO slot before issuing the command for another receiver.
+        receiving = threading.Event()
+        running = runtime.workers.submit(_receive_result, session, receiving)
+        assert receiving.wait(_WAIT_SECONDS), runtime.model.failures
+        assert not running.done()
+        assert runtime.call(session.query, "/compact", delivery="queue") is None
+        execution = runtime.call(_receive_result, session)
+        assert isinstance(execution, CommandExecution)
+        assert execution.result.kind == "error"
+        assert not response.release.is_set()
+        assert not running.done()
+        assert len(runtime.model.captured()) == 1
+        response.release.set()
+        completed = running.result(timeout=_WAIT_SECONDS)
+        assert isinstance(completed, RunResult)
+        assert completed.final_response == response.text
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution)
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 1
+    assert not _of_type(events, "compaction/start")
+
+
+def run_compaction_scenario(tmp_path: Path, *, capture_patch: Path | None = None) -> Path:
+    """Run the complete scenario and retain its real Session log under the caller's directory."""
+    summary_response = _Completion(_CHECKPOINT, blocked=True)
+    with _runtime(
+        tmp_path, [_Completion(_FIRST_REPLY), summary_response, _Completion(_CONTINUE_REPLY)],
+        capture_patch=capture_patch,
+    ) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        # The replaced span is deliberately larger than the framed summary.
+        prompt = _HISTORY_MARKER + "\n" + "Keep the original request only in the stored history. " * 256
+        assert runtime.call(session.query, prompt, delivery="queue") is None
+        first_result = runtime.call(_receive_result, session)
+        assert isinstance(first_result, RunResult)
+        assert first_result.final_response == _FIRST_REPLY
+
+        sending = runtime.workers.submit(session.query, "/compact", delivery="queue")
+        assert summary_response.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        assert sending.result(timeout=_WAIT_SECONDS) is None
+        assert not summary_response.release.is_set(), "query waited for the summary before returning"
+        receiving = threading.Event()
+        compacting = runtime.workers.submit(_receive_result, session, receiving)
+        assert receiving.wait(_WAIT_SECONDS), runtime.model.failures
+        assert not compacting.done(), "command returned before its summary request completed"
+        summary_response.release.set()
+        execution = compacting.result(timeout=_WAIT_SECONDS)
+        assert isinstance(execution, CommandExecution)
+        assert execution.result.kind == "success"
+        assert execution.result.source_event_seq is not None
+
+        # The native compactor promises this prefix before returning, unlike command/done.
+        checkpointed = _read_events(runtime.logs, live=True)
+        summaries = _of_type(checkpointed, "compaction/summary")
+        starts = _of_type(checkpointed, "compaction/start")
+        ends = _of_type(checkpointed, "compaction/end")
+        assert len(starts) == len(summaries) == len(ends) == 1
+        summary = summaries[0]
+        assert summary["seq"] == execution.result.source_event_seq
+        assert summary["data"]["llmStreamCall"] is True
+        assert summary["data"]["provider"] == "deepseek-official"
+        assert summary["data"]["model"] == _MODEL
+        for event in [starts[0], summary, ends[0]]:
+            assert event["data"]["sourceCommandId"] == execution.command_id
+            assert event["data"]["compactionId"] == summary["data"]["compactionId"]
+        assert starts[0]["data"]["turn"] is None and ends[0]["data"]["turn"] is None
+        assert "error" not in ends[0]["data"]
+        checkpoint = checkpointed[summary["seq"] + 1]
+        assert checkpoint["type"] == "user/message"
+        assert checkpoint["data"]["source"]["kind"] == "compact-checkpoint"
+        assert checkpoint["surfaceOp"]["op"] == "replace"
+        checkpoint_text = "\n".join(_text_blocks(checkpoint["data"]["content"]))
+        assert _CHECKPOINT in checkpoint_text
+        assert "<compacted-summary>" in checkpoint_text and "</compacted-summary>" in checkpoint_text
+        assert len(_of_type(checkpointed, "turn/start")) == len(_of_type(checkpointed, "turn/end")) == 1
+
+        assert runtime.call(session.query, _CONTINUE_PROMPT, delivery="queue") is None
+        continued = runtime.call(_receive_result, session)
+        assert isinstance(continued, RunResult)
+        assert continued.final_response == _CONTINUE_REPLY
+        first, compaction, following = runtime.model.captured()
+        assert first["system"] == compaction["system"]
+        assert first["tools"] == compaction["tools"]
+        initial_texts = _request_texts(first)
+        summary_texts = _request_texts(compaction)
+        assert summary_texts[:len(initial_texts)] == initial_texts
+        assert "You are now acting as a compaction engine" in summary_texts[-1]
+        following_texts = _request_texts(following)
+        following_text = "\n".join(following_texts)
+        assert _CHECKPOINT in following_text and _CONTINUE_PROMPT in following_text
+        assert _FIRST_REPLY in following_text
+        assert _HISTORY_MARKER not in following_text
+        assert "/compact" not in following_texts
+        assert execution.result.text not in following_text
+
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution)
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2
+    assert len(_of_type(events, "step/start")) == len(_of_type(events, "step/end")) == 2
+    assert _HISTORY_MARKER in json.dumps(events), "compaction erased the original stored input"
+    return next(runtime.logs.rglob("session*.jsonl"))
+
+
+def test_compact_checkpoints_history_before_continuing_the_same_session(tmp_path: Path) -> None:
+    run_compaction_scenario(tmp_path)
+
+
+def test_unknown_slash_and_compact_arguments_follow_the_plain_prompt_path(tmp_path: Path) -> None:
+    prompts = ["/sdk-unregistered-command-fixture", "/compact extra"]
+    replies = ["SDK_UNKNOWN_SLASH_REPLY", "SDK_COMPACT_ARGUMENTS_REPLY"]
+    with _runtime(tmp_path, [_Completion(text) for text in replies]) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        for prompt, reply in zip(prompts, replies):
+            assert runtime.call(session.query, prompt, delivery="queue") is None
+            result = runtime.call(_receive_result, session)
+            assert isinstance(result, RunResult)
+            assert result.final_response == reply
+        requests = runtime.model.captured()
+        assert len(requests) == len(prompts)
+        for prompt, request in zip(prompts, requests):
+            assert prompt in _request_texts(request)
+    events = _read_events(runtime.logs)
+    assert not _of_type(events, "command/run")
+    assert not _of_type(events, "compaction/start")
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2
+
+
+def test_unimplemented_slash_remains_plain_text_without_changing_plan_mode(tmp_path: Path) -> None:
+    plan_text = "/plan DSH_PLAN_TEXT_INPUT"
+    following = "DSH_AFTER_PLAN_TEXT_INPUT"
+    plan_reply = "DSH_PLAN_TEXT_REPLY"
+    reply = "DSH_AFTER_PLAN_TEXT_REPLY"
+    with _runtime(tmp_path, [_Completion(plan_reply), _Completion(reply)]) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        assert runtime.call(session.query, plan_text, delivery="queue") is None
+        first = runtime.call(_receive_result, session)
+        assert isinstance(first, RunResult)
+        assert first.final_response == plan_reply
+
+        assert runtime.call(session.query, following, delivery="queue") is None
+        result = runtime.call(_receive_result, session)
+        assert isinstance(result, RunResult)
+        assert result.final_response == reply
+        initial, continued = runtime.model.captured()
+        assert plan_text in _request_texts(initial)
+        assert following in _request_texts(continued)
+        assert initial["tools"] == continued["tools"]
+    events = _read_events(runtime.logs)
+    assert not _of_type(events, "command/run")
+    assert not any(event["data"].get("active") is True for event in _of_type(events, "plan/mode"))
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2

python/sdk/tests/test_queries.py

@@ -0,0 +1,391 @@
+"""Split query submission and per-request response observation."""
+from __future__ import annotations
+
+import sys
+import threading
+from pathlib import Path
+
+import pytest
+from pydantic import BaseModel, ValidationError
+
+from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, RunResult, SdkProtocolError
+from deepseek_harness.errors import JsonRpcError, TransportClosedError
+
+
+class Ack(BaseModel):
+    pass
+
+
+@pytest.fixture
+def peer(tmp_path: Path) -> Path:
+    script = tmp_path / "queries.py"
+    script.write_text('''import json
+import sys
+
+serial = 0
+pending_commands = []
+pending_prompts = []
+
+def send(frame):
+    print(json.dumps({"jsonrpc": "2.0", **frame}), flush=True)
+
+def event(session, type, data):
+    send({"method": "session.event", "params": {"sessionId": session, "event": {"type": type, "data": data}}})
+
+def command(frame, number):
+    session = frame["params"]["sessionId"]
+    identity = "command-" + str(number)
+    event(session, "command/run", {"commandId": identity})
+    event(session, "command/done", {"commandId": identity, "kind": "success"})
+    send({"id": frame["id"], "result": {"kind": "command", "execution": {
+        "commandId": identity, "result": {"kind": "success", "text": "compacted"}}}})
+
+def receipt(frame, number):
+    session = frame["params"]["sessionId"]
+    send({"method": "session.status", "params": {"sessionId": session, "status": "idle"}})
+    event(session, "agent/inbox/spliced", {"inserted": [{"id": "input-" + str(number)}]})
+
+def finish(frame):
+    session = frame["params"]["sessionId"]
+    event(session, "assistant/message", {"message": {"content": [
+        {"type": "text", "text": "answer for " + frame["params"]["text"]}]}})
+    event(session, "turn/end", {"turn": 1, "reason": {"kind": "completed"}})
+    send({"method": "session.status", "params": {"sessionId": session, "status": "idle"}})
+
+for line in sys.stdin:
+    frame = json.loads(line)
+    method = frame["method"]
+    if method == "initialize":
+        send({"id": frame["id"], "result": {"serverInfo": {"name": "query-peer"}}})
+    elif method == "session/query":
+        serial += 1
+        params = frame["params"]
+        assert set(params) == {"sessionId", "text", "delivery"}
+        assert params["delivery"] in ("queue", "steer")
+        text = params["text"]
+        if text == "hold-command":
+            pending_commands.append((frame, serial))
+        elif text == "/compact":
+            command(frame, serial)
+        elif text == "silent-command":
+            send({"id": frame["id"], "result": {"kind": "command", "execution": {
+                "commandId": "silent", "result": {"kind": "success"}}}})
+        elif text == "rpc-error":
+            send({"id": frame["id"], "error": {"code": -32602, "message": "query rejected"}})
+        elif text == "bad-route":
+            send({"id": frame["id"], "result": {"kind": "unknown"}})
+        elif text == "bad-execution":
+            send({"id": frame["id"], "result": {"kind": "command", "execution": {"commandId": "c"}}})
+        else:
+            result = {"kind": "prompt", "messageId": "input-" + str(serial)}
+            if text == "fast-prompt":
+                receipt(frame, serial)
+                finish(frame)
+                send({"id": frame["id"], "result": result})
+            else:
+                send({"id": frame["id"], "result": result})
+                receipt(frame, serial)
+                if text == "hold-prompt":
+                    pending_prompts.append(frame)
+                else:
+                    finish(frame)
+    elif method == "release":
+        for original, number in pending_commands:
+            command(original, number)
+        pending_commands.clear()
+        for original in pending_prompts:
+            finish(original)
+        pending_prompts.clear()
+        send({"id": frame["id"], "result": {}})
+    elif method == "barrier":
+        send({"id": frame["id"], "result": {}})
+    elif method == "disconnect":
+        send({"id": frame["id"], "result": {}})
+        break
+    elif method == "shutdown":
+        send({"id": frame["id"], "result": {}})
+        break
+    else:
+        raise AssertionError(method)
+''', encoding="utf-8")
+    return script
+
+
+@pytest.fixture
+def sdk(peer: Path, tmp_path: Path):
+    with DeepSeekHarness(
+        cwd=str(tmp_path), dsh_home=str(tmp_path / "home"), request_timeout_seconds=5,
+        _launch_args=(sys.executable, str(peer)),
+    ) as harness:
+        yield harness
+
+
+def release(sdk: DeepSeekHarness) -> None:
+    sdk.client.request("release", {}, response_model=Ack, timeout_seconds=5)
+
+
+def assert_released(sdk: DeepSeekHarness) -> None:
+    assert sdk.client._responses == {}
+    assert sdk.client._notification_subscribers == {}
+
+
+def test_query_only_sends_then_receive_returns_command(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    assert session.query("hold-command") is None
+    assert len(sdk.client._responses) == 1
+    release(sdk)
+    responses = list(session.receive_response())
+    assert [item.payload["event"]["type"] for item in responses[:-1]] == ["command/run", "command/done"]
+    assert isinstance(responses[-1], CommandExecution)
+    assert responses[-1].command_id == "command-1"
+    assert_released(sdk)
+
+
+@pytest.mark.parametrize("text", ["ordinary", "fast-prompt", "/plan"])
+@pytest.mark.parametrize("delivery", ["queue", "steer"])
+def test_prompt_yields_receipt_through_idle_then_run_result(sdk: DeepSeekHarness, text: str, delivery: str) -> None:
+    session = sdk.start_session("main")
+    session.query(text, delivery=delivery)
+    responses = list(session.receive_response())
+    assert all(isinstance(item, Notification) for item in responses[:-1])
+    assert responses[0].payload["event"]["type"] == "agent/inbox/spliced"
+    result = responses[-1]
+    assert isinstance(result, RunResult)
+    assert result.final_response == "answer for " + text
+    assert result.finish_reason == "completed"
+    assert result.notifications == responses[:-1]
+    assert len(result.events) == 3
+    assert_released(sdk)
+
+
+def test_late_receive_uses_an_arrived_response_before_expired_deadline(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    sdk.client.config.request_timeout_seconds = 0
+    session.query("/compact")
+    sdk.client.request("barrier", {}, response_model=Ack, timeout_seconds=5)
+    assert isinstance(list(session.receive_response())[-1], CommandExecution)
+    assert_released(sdk)
+
+
+def test_pending_response_timeout_releases_only_its_observer(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    sdk.client.config.request_timeout_seconds = 0
+    session.query("hold-command")
+    with pytest.raises(TimeoutError, match="session/query timed out"):
+        list(session.receive_response())
+    assert_released(sdk)
+    release(sdk)
+
+
+@pytest.mark.parametrize("text,error", [
+    ("rpc-error", JsonRpcError), ("bad-route", SdkProtocolError), ("bad-execution", ValidationError),
+])
+def test_query_errors_arrive_during_receive_and_leave_later_queries_usable(sdk: DeepSeekHarness, text: str, error: type) -> None:
+    session = sdk.start_session("main")
+    session.query(text)
+    with pytest.raises(error):
+        list(session.receive_response())
+    assert_released(sdk)
+    session.query("/compact")
+    assert isinstance(list(session.receive_response())[-1], CommandExecution)
+
+
+def test_receivers_claim_fifo_even_when_responses_arrive_out_of_order(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-command")
+    session.query("/compact")
+    release(sdk)
+    first = session.receive_response()
+    second = session.receive_response()
+    assert list(second)[-1].command_id == "command-2"
+    assert list(first)[-1].command_id == "command-1"
+    assert_released(sdk)
+    with pytest.raises(RuntimeError, match="preceding query"):
+        session.receive_response()
+
+
+def test_command_receiver_stops_at_its_done_before_a_later_prompt(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("/compact")
+    session.query("ordinary")
+    sdk.client.request("barrier", {}, response_model=Ack)
+    command = list(session.receive_response())
+    assert [item.payload["event"]["type"] for item in command[:-1]] == ["command/run", "command/done"]
+    assert command[-1].command_id == "command-1"
+    prompt = list(session.receive_response())
+    assert isinstance(prompt[-1], RunResult)
+    assert prompt[-1].final_response == "answer for ordinary"
+    assert_released(sdk)
+
+
+def test_command_result_without_notifications_does_not_wait_for_done(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("silent-command")
+    responses = list(session.receive_response())
+    assert len(responses) == 1
+    assert responses[0].command_id == "silent"
+    assert_released(sdk)
+
+
+@pytest.mark.parametrize("text", ["/compact", "silent-command"])
+def test_late_command_receive_preserves_completed_result_after_eof(sdk: DeepSeekHarness, text: str) -> None:
+    session = sdk.start_session("main")
+    session.query(text)
+    sdk.client.request("disconnect", {}, response_model=Ack)
+    sdk.client._reader_thread.join(5)
+    assert not sdk.client._reader_thread.is_alive()
+    responses = list(session.receive_response())
+    assert isinstance(responses[-1], CommandExecution)
+    assert responses[-1].result.kind == "success"
+    if text == "/compact":
+        assert [item.payload["event"]["type"] for item in responses[:-1]] == ["command/run", "command/done"]
+    else:
+        assert len(responses) == 1
+    assert_released(sdk)
+
+
+def test_command_receive_does_not_hide_other_subscription_errors(sdk: DeepSeekHarness, monkeypatch) -> None:
+    session = sdk.start_session("main")
+    session.query("/compact")
+    sdk.client.request("barrier", {}, response_model=Ack)
+    receiver = session.receive_response()
+
+    def fail():
+        raise SdkProtocolError("invalid notification")
+
+    monkeypatch.setattr(receiver._subscription, "_next_queued", fail)
+    with pytest.raises(SdkProtocolError, match="invalid notification"):
+        list(receiver)
+    assert_released(sdk)
+
+
+def test_query_and_another_receiver_can_run_while_a_prompt_receiver_waits(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-prompt")
+    receiver = session.receive_response()
+    entered = threading.Event()
+    responses = []
+    errors = []
+
+    def receive() -> None:
+        try:
+            for item in receiver:
+                responses.append(item)
+                entered.set()
+        except BaseException as error:
+            errors.append(error)
+
+    thread = threading.Thread(target=receive)
+    thread.start()
+    try:
+        assert entered.wait(5)
+        session.query("/compact")
+        assert list(session.receive_response())[-1].command_id == "command-2"
+        assert thread.is_alive()
+        release(sdk)
+        thread.join(5)
+        assert not thread.is_alive()
+        assert not errors
+        assert isinstance(responses[-1], RunResult)
+        assert responses[-1].final_response == "answer for hold-prompt"
+        assert_released(sdk)
+    finally:
+        receiver.close()
+        thread.join(5)
+
+
+def test_iterator_close_before_first_next_releases_request_and_subscription(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-command")
+    receiver = session.receive_response()
+    receiver.close()
+    receiver.close()
+    assert list(receiver) == []
+    assert_released(sdk)
+    release(sdk)
+    session.query("/compact")
+    assert list(session.receive_response())[-1].command_id == "command-2"
+
+
+def test_iterator_close_after_a_prompt_notification_releases_its_subscription(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-prompt")
+    receiver = session.receive_response()
+    assert isinstance(next(receiver), Notification)
+    receiver.close()
+    assert list(receiver) == []
+    assert_released(sdk)
+    release(sdk)
+
+
+def test_iterator_close_wakes_its_blocked_request_receiver(sdk: DeepSeekHarness, monkeypatch) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-command")
+    receiver = session.receive_response()
+    entered = threading.Event()
+    errors = []
+    original = sdk.client._wait_request
+
+    def wait(request, **kwargs):
+        entered.set()
+        return original(request, **kwargs)
+
+    monkeypatch.setattr(sdk.client, "_wait_request", wait)
+
+    def receive() -> None:
+        try:
+            assert list(receiver) == []
+        except BaseException as error:
+            errors.append(error)
+
+    thread = threading.Thread(target=receive)
+    thread.start()
+    try:
+        assert entered.wait(5)
+        receiver.close()
+        thread.join(5)
+        assert not thread.is_alive()
+        assert not errors
+        assert_released(sdk)
+    finally:
+        receiver.close()
+        thread.join(5)
+
+
+def test_terminal_result_releases_resources_before_caller_sends_another_query(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("/compact")
+    receiver = session.receive_response()
+    for item in receiver:
+        if isinstance(item, CommandExecution):
+            assert_released(sdk)
+            session.query("/compact")
+            receiver.close()
+    assert list(session.receive_response())[-1].command_id == "command-2"
+    assert_released(sdk)
+
+
+def test_runtime_close_settles_unreceived_queries(sdk: DeepSeekHarness) -> None:
+    session = sdk.start_session("main")
+    session.query("hold-command")
+    sdk.close()
+    with pytest.raises(TransportClosedError):
+        list(session.receive_response())
+    assert_released(sdk)
+
+
+def test_send_failure_releases_subscription_without_queueing_a_receiver(sdk: DeepSeekHarness, monkeypatch) -> None:
+    session = sdk.start_session("main")
+    original = sdk.client._write_message
+
+    def write(message):
+        if message.get("method") == "session/query":
+            raise TransportClosedError("write failed")
+        original(message)
+
+    monkeypatch.setattr(sdk.client, "_write_message", write)
+    with pytest.raises(TransportClosedError, match="write failed"):
+        session.query("ordinary")
+    assert_released(sdk)
+    with pytest.raises(RuntimeError, match="preceding query"):
+        session.receive_response()