feat(sdk): route content blocks through native commands

Reuse the existing run input for command images and ordinary queue/steer delivery.
Align Unicode command separators, bound goal observation, preserve plan input encoding,
and handle explicitly expected Windows cancellation disconnects in runtime tests.

python/sdk/README.i18n.yaml

@@ -12,8 +12,8 @@
   en: ac793c588b13ce87
   zh: 502015cda46de3ac
 /deepseek-harness-python-sdk/run-text-and-commands:
-  en: 0c5b04951c9b7d32
-  zh: b049925160bec5f5
+  en: bfdb0c48d5a29f15
+  zh: b6164533835279dc
 /deepseek-harness-python-sdk/client-approvals:
   en: fc69be1beb8abff1
   zh: 91b56781931a6f68

python/sdk/README.md

@@ -61,15 +61,17 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 <a id="run-text-and-commands"></a>
 ## Run text and commands
 
-`session.run(input, on_notification=..., delivery="queue")` is the high-level input entry. Non-blank strings use the runtime's [supported command forms and Desktop aliases](../../packages/sdk/server/README.md#command-input); unrecognized input remains ordinary text. `delivery="steer"` targets the next available step for ordinary text. Command admission and timing remain owned by the native handler.
+`session.run(input, on_notification=..., delivery="queue")` is the high-level input entry for strings and content-block arrays. Both use the runtime's [supported command forms and Desktop aliases](../../packages/sdk/server/README.md#command-input); unrecognized input remains an ordinary prompt with its original block order. `delivery="steer"` targets the next available step for ordinary input, including images. Command admission and timing remain owned by the native handler.
 
 The call returns `RunResult` for an ordinary prompt's inbox-receipt-to-idle interval, or `CommandExecution` after a command settles. Use `isinstance(result, RunResult)` before reading `final_response`; otherwise inspect the command's `result.kind`. A command retains its `command_id` and optional `source_event_seq`, rather than selecting an earlier assistant reply.
 
-Content-block arrays stay literal queued prompts and return `RunResult`. For example, `session.run([{"type": "text", "text": "/compact"}])` sends that text to the model, while `session.run("/compact")` invokes native compaction. Content blocks require queue delivery; a steer request raises `ValueError` before sending.
+Content-block arrays use the same command routing as strings. Both `session.run([{"type": "text", "text": "/compact"}])` and `session.run("/compact")` invoke native compaction and return `CommandExecution`. To bypass native command routing, call the low-level `client.session_prompt(session_id, blocks)` and manage the subsequent subscription and wait yourself; runtime skill recognition still applies.
+
+Select a model that supports image input. Submit a command and image in the existing content-block array, for example `session.run([{"type": "text", "text": "/plan Analyze this image."}, {"type": "image", "data": image_base64, "mimeType": "image/png"}])`; `harness.run()` accepts the same input. Plan input and goal creation or editing accept images; `/plan off`, goal controls, compact, and permission commands reject them through the [native attachment rules](../../packages/sdk/server/README.md#command-input). Native command input supports encoded raster images, not general file uploads or client-supplied attachment references.
 
 Each call subscribes before sending and releases its subscription before returning or raising. `on_notification` receives the ordinary activity interval in wire order; command callbacks replay Session-tree notifications from subscription through their own `command/done` after the command response, so that interval can include overlapping work. Calls may overlap without consuming each other's results. Handler refusals return `CommandExecution.result.kind="error"`; routing, protocol, transport and observer failures raise. A timeout or observer failure does not roll back runtime work or authorize an automatic retry. `request_timeout_seconds` bounds the protocol response wait, not later ordinary prompt activity.
 
-The default full `sdk` profile provides the supported native commands; `sdk-minimal` keeps unsupported input on its ordinary prompt path. `/compact` can make an auxiliary model request and completes without a normal turn or idle notification; empty history can succeed without a summary. The lower-level `HarnessClient.session_command(session_id, line)` executes supported command names without prompt fallback. See the [compaction example](examples/README.md#compact-a-full-sdk-session).
+The default full `sdk` profile provides the supported native commands; `sdk-minimal` keeps unsupported input on its ordinary prompt path. `/compact` can make an auxiliary model request and completes without a normal turn or idle notification; empty history can succeed without a summary. The lower-level `HarnessClient.session_command(session_id, line)` executes supported command names without prompt fallback or attachments. See the [compaction example](examples/README.md#compact-a-full-sdk-session).
 
 `/permission <preset>` changes only the receiving Session. An unknown preset returns a command error without changing policy. Later input and restoration retain successful changes; other Sessions and the caller's `DeepSeekHarnessConfig` keep their own values. A running approval-policy change can inject a notice and cause another model step. See the [permission example](examples/README.md#change-session-permissions).
 

python/sdk/README.zh.md

@@ -61,15 +61,17 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 <a id="run-text-and-commands"></a>
 ## 运行文本与命令
 
-`session.run(input, on_notification=..., delivery="queue")` 是高层输入入口。非空白字符串使用运行时的[命令形式与 Desktop 别名](../../packages/sdk/server/README.zh.md#command-input)识别规则；未识别的输入保持普通文本。对普通文本，`delivery="steer"` 指向下一个可用步骤。命令的接纳规则与执行时序仍由原生处理器负责。
+`session.run(input, on_notification=..., delivery="queue")` 是字符串和内容块数组共用的高层输入入口。两者都使用运行时的[命令形式与 Desktop 别名](../../packages/sdk/server/README.zh.md#command-input)识别规则；未识别的输入保持普通提示词及原始内容块顺序。对普通输入，包括图片，`delivery="steer"` 指向下一个可用步骤。命令的接纳规则与执行时序仍由原生处理器负责。
 
 普通提示词返回其入队回执到下一次 idle 区间的 `RunResult`，命令则在执行结束后返回 `CommandExecution`。读取 `final_response` 前使用 `isinstance(result, RunResult)` 检查类型，否则检查命令的 `result.kind`。命令保留自己的 `command_id` 和可选 `source_event_seq`，不会选取更早的助手回复作为结果。
 
-内容块数组保持原样排队输入，并返回 `RunResult`。例如，`session.run([{"type": "text", "text": "/compact"}])` 将该文本发送给模型，而 `session.run("/compact")` 调用原生压缩。内容块要求 queue 投递；请求 steer 会在发送前抛出 `ValueError`。
+内容块数组与字符串使用相同的命令路由。`session.run([{"type": "text", "text": "/compact"}])` 和 `session.run("/compact")` 都调用原生压缩并返回 `CommandExecution`。要绕过原生命令路由，调用低层 `client.session_prompt(session_id, blocks)` 并自行管理后续订阅与等待；运行时 skill（技能）识别仍然适用。
+
+先选择支持图片输入的模型。通过现有内容块数组同时提交命令和图片，例如 `session.run([{"type": "text", "text": "/plan Analyze this image."}, {"type": "image", "data": image_base64, "mimeType": "image/png"}])`；`harness.run()` 接受相同输入。计划输入和目标创建、编辑接受图片；`/plan off`、目标控制、压缩和权限命令按[原生附件规则](../../packages/sdk/server/README.zh.md#command-input)拒绝图片。原生命令输入支持编码栅格图片，不支持一般文件上传或客户端提供的附件引用。
 
 每次调用在发送前订阅，并在返回或抛出异常前释放自己的订阅。`on_notification` 按线上的顺序接收普通活动区间通知；命令回调在命令响应到达后，按顺序交付从订阅时刻到自身 `command/done` 的 Session 树通知，因此该区间可能包含重叠工作的通知。调用可以重叠，不会消耗彼此的结果。处理器拒绝返回 `CommandExecution.result.kind="error"`，路由、协议、传输和观察回调失败则抛出异常。超时或观察回调失败不回滚运行时工作，也不授权自动重试。`request_timeout_seconds` 限制协议响应等待，不限制之后的普通提示词活动。
 
-默认的完整 `sdk` profile 提供已支持的原生命令；`sdk-minimal` 将未支持的输入保留在普通提示词路径。`/compact` 可以发起辅助模型请求，其完成不需要普通 turn 或 idle 通知；空历史可以在不生成摘要时成功。底层 `HarnessClient.session_command(session_id, line)` 严格执行支持的命令名，不回退为提示词。见[压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
+默认的完整 `sdk` profile 提供已支持的原生命令；`sdk-minimal` 将未支持的输入保留在普通提示词路径。`/compact` 可以发起辅助模型请求，其完成不需要普通 turn 或 idle 通知；空历史可以在不生成摘要时成功。底层 `HarnessClient.session_command(session_id, line)` 严格执行支持的命令名，不回退为提示词，也不接受附件。见[压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
 
 `/permission <preset>` 只修改接收命令的 Session。未知 preset 返回命令错误，不改变策略。后续输入与恢复保留成功的修改，其他 Session 和调用方的 `DeepSeekHarnessConfig` 保留自己的值。运行中的审批策略变化可以注入通知并触发下一模型步骤。见[权限示例](examples/README.zh.md#change-session-permissions)。
 

python/sdk/examples/goal.py

@@ -54,6 +54,19 @@ def main() -> None:
         session = harness.start_session()
         with harness.client.subscribe_notifications(lambda notice: notice.payload.get("sessionId") == session.id) as notifications:
             deadline = time.monotonic() + args.timeout_seconds
+
+            def remaining_seconds() -> float:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
+                    raise TimeoutError("Goal observation timed out")
+                return remaining
+
+            def observe(notification: Notification) -> dict | None:
+                remaining_seconds()
+                event = display(notification)
+                remaining_seconds()
+                return event
+
             execution = session.run(f"/goal {args.objective}")
             if not isinstance(execution, CommandExecution):
                 raise RuntimeError("The SDK profile did not execute the requested command")
@@ -61,7 +74,7 @@ def main() -> None:
             if execution.result.kind == "error":
                 raise RuntimeError(execution.result.text)
             while True:
-                event = display(notifications.next(timeout_seconds=max(0, deadline - time.monotonic())))
+                event = observe(notifications.next(timeout_seconds=remaining_seconds()))
                 if event is not None and event["type"] == "goal/change" and event["data"]["operation"] in {
                     "complete", "block", "pause", "clear",
                 }:
@@ -69,9 +82,10 @@ def main() -> None:
             # A terminal goal event can precede the model's closing reply.
             harness.client.request(
                 "session/wait", {"sessionId": session.id}, response_model=WaitResult,
-                timeout_seconds=max(0, deadline - time.monotonic()),
-                on_notification=display, notification_subscription=notifications,
+                timeout_seconds=remaining_seconds(),
+                on_notification=observe, notification_subscription=notifications,
             )
+            remaining_seconds()
 
 
 if __name__ == "__main__":

python/sdk/examples/plan.py

@@ -22,10 +22,11 @@ class WaitResult(BaseModel):
 
 
 class _PlanReviews:
-    """Own cancellable stdin readers and join review callbacks before interpreter exit."""
+    """Own cancellable readers with the parent's stdin encoding and join review callbacks before exit."""
 
     # A buffered reader can consume later answers from a pipe before it exits.
     _READ_LINE = """import os, sys
+sys.stdin.reconfigure(encoding=sys.argv[1])
 if sys.stdin.isatty():
     try:
         text = input() + '\\n'
@@ -81,7 +82,7 @@ sys.stdout.buffer.write(text.encode('utf-8'))
             if self._closed.is_set() or cancelled.is_set():
                 return None
             reader = subprocess.Popen(
-                [sys.executable, "-c", self._READ_LINE], stdin=sys.stdin,
+                [sys.executable, "-c", self._READ_LINE, sys.stdin.encoding or "utf-8"], stdin=sys.stdin,
                 stdout=subprocess.PIPE, stderr=subprocess.PIPE,
             )
             self._readers.append(reader)

python/sdk/src/deepseek_harness/api.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import uuid
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Callable, Literal, overload
+from typing import Callable, Literal
 
 from .client import HarnessClient, HarnessConfig, UserQuestionHandler
 from .errors import SdkProtocolError, TransportClosedError
@@ -167,19 +167,6 @@ class DeepSeekHarness:
         self._client.resume_session(session_id)
         return Session(self, session_id)
 
-    @overload
-    def run(
-        self, input: list[JsonObject], *, session_id: str | None = None,
-        on_notification: Callable[[Notification], None] | None = None,
-    ) -> RunResult: ...
-
-    @overload
-    def run(
-        self, input: str | list[JsonObject], *, session_id: str | None = None,
-        on_notification: Callable[[Notification], None] | None = None,
-        delivery: Literal["queue", "steer"] = "queue",
-    ) -> RunResult | CommandExecution: ...
-
     def run(
         self,
         input: str | list[JsonObject],
@@ -188,7 +175,7 @@ class DeepSeekHarness:
         on_notification: Callable[[Notification], None] | None = None,
         delivery: Literal["queue", "steer"] = "queue",
     ) -> RunResult | CommandExecution:
-        """Run text or a native command on a fresh or named Session."""
+        """Run text or content blocks through native input routing on a fresh or named Session."""
         return self.start_session(session_id).run(input, on_notification=on_notification, delivery=delivery)
 
 
@@ -207,19 +194,6 @@ class Session:
         self.harness.start()
         return self.harness.client.set_working_directory(self.id, path)
 
-    @overload
-    def run(
-        self, input: list[JsonObject], *,
-        on_notification: Callable[[Notification], None] | None = None,
-    ) -> RunResult: ...
-
-    @overload
-    def run(
-        self, input: str | list[JsonObject], *,
-        on_notification: Callable[[Notification], None] | None = None,
-        delivery: Literal["queue", "steer"] = "queue",
-    ) -> RunResult | CommandExecution: ...
-
     def run(
         self,
         input: str | list[JsonObject],
@@ -227,49 +201,45 @@ class Session:
         on_notification: Callable[[Notification], None] | None = None,
         delivery: Literal["queue", "steer"] = "queue",
     ) -> RunResult | CommandExecution:
-        """Run text through native input routing, or queue literal content blocks.
+        """Run text or content blocks through native input routing.
 
-        Text returns the prompt's receipt-to-idle RunResult or a settled
+        Input returns the prompt's receipt-to-idle RunResult or a settled
         CommandExecution; command completion does not wait for later model work.
-        Content blocks stay literal and accept only queue delivery. Notifications
-        retain wire order. Command callbacks replay Session-tree notifications
+        The first text block selects native commands. Matched commands require
+        one text block with only encoded images beside it; native handlers decide
+        which commands accept those images. Ordinary prompts retain block order
+        and use the requested queue or steer delivery. Notifications retain
+        wire order. Command callbacks replay Session-tree notifications
         from subscription through their own command/done after the response,
         including overlapping work. Handler errors remain command results;
         routing, transport and observer failures raise without rolling back work.
         """
         if delivery not in ("queue", "steer"):
             raise ValueError("run delivery must be queue or steer")
-        if not isinstance(input, str) and delivery != "queue":
-            raise ValueError("content-block input requires queue delivery")
         self.harness.start()
         with self.harness.client.subscribe_session_notifications(self.id) as subscription:
-            if isinstance(input, str):
-                response = self.harness.client._request_raw(
-                    "session/query", {"sessionId": self.id, "text": input, "delivery": delivery},
-                )
-                if not isinstance(response, dict):
-                    raise SdkProtocolError("session/query response must be a JSON object")
-                if response.get("kind") == "command":
-                    execution = CommandExecution.model_validate(response.get("execution"))
-                    while True:
-                        try:
-                            notification = subscription._next_queued()
-                        except TransportClosedError:
-                            break
-                        if notification is None:
-                            break
-                        if on_notification is not None:
-                            on_notification(notification)
-                        if _is_command_done(notification, self.id, execution.command_id):
-                            break
-                    return execution
-                if response.get("kind") != "prompt" or not isinstance(response.get("messageId"), str) or not response["messageId"]:
-                    raise SdkProtocolError("session/query requires a prompt messageId or command execution result")
-                message_id = response["messageId"]
-            else:
-                message_id = self.harness.client.session_prompt(
-                    self.id, input, notification_subscription=subscription,
-                )
+            response = self.harness.client._request_raw("session/query", {
+                "sessionId": self.id, "contentBlocks": normalize_input(input), "delivery": delivery,
+            })
+            if not isinstance(response, dict):
+                raise SdkProtocolError("session/query response must be a JSON object")
+            if response.get("kind") == "command":
+                execution = CommandExecution.model_validate(response.get("execution"))
+                while True:
+                    try:
+                        notification = subscription._next_queued()
+                    except TransportClosedError:
+                        break
+                    if notification is None:
+                        break
+                    if on_notification is not None:
+                        on_notification(notification)
+                    if _is_command_done(notification, self.id, execution.command_id):
+                        break
+                return execution
+            if response.get("kind") != "prompt" or not isinstance(response.get("messageId"), str) or not response["messageId"]:
+                raise SdkProtocolError("session/query requires a prompt messageId or command execution result")
+            message_id = response["messageId"]
             activity = _PromptActivity(self.id, message_id)
             while True:
                 notification = subscription.next()

python/sdk/tests/run_installed_commands.py

@@ -13,6 +13,7 @@ import pytest
 
 TEST_FILES = (
     "test_commands_runtime.py",
+    "test_command_attachments_runtime.py",
     "test_goal_runtime.py",
     "test_plan_runtime.py",
     "test_plan_example.py",

python/sdk/tests/test_client.py

@@ -38,11 +38,11 @@ for line in sys.stdin:
     if method == "initialize":
         json.dump(msg.get("params"), open(os.environ["INIT_DUMP"], "w"))
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({
             "jsonrpc": "2.0",
             "method": "session.event",
@@ -233,10 +233,10 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "main", "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "main", "childSessionId": "child"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "idle"}}), flush=True)
     elif method == "shutdown":
@@ -272,10 +272,10 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "turn/end", "data": {"turn": 1, "reason": {}}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "idle"}}), flush=True)
     elif method == "shutdown":
@@ -346,10 +346,10 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "main", "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "main", "childSessionId": "child"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.finished", "params": {"parentSessionId": "main", "childSessionId": "child", "status": "ok", "stopReason": "completed"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "idle"}}), flush=True)
@@ -388,11 +388,11 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         root = (msg.get("params") or {})["sessionId"]
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": root, "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": root, "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": root, "childSessionId": "child"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "child", "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "child response"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "child", "childSessionId": "grandchild"}}), flush=True)
@@ -448,13 +448,13 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "other", "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "wrong session"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "other", "status": "idle"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "right session"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "idle"}}), flush=True)
     elif method == "shutdown":
@@ -485,11 +485,11 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "ok"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "idle"}}), flush=True)
     elif method == "shutdown":
@@ -517,14 +517,14 @@ for line in sys.stdin:
     method = msg.get("method")
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
-    elif method == "session/prompt":
+    elif method == "session/query":
         turn += 1
         params = msg.get("params") or {}
         session_id = params["sessionId"]
         message_id = f"message-{turn}"
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": session_id, "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": message_id}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": session_id, "status": "running"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": message_id}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"kind": "prompt", "messageId": message_id}}), flush=True)
         if turn == 1:
             print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": session_id, "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "first"}]}}}}), flush=True)
             print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": session_id, "status": "idle"}}), flush=True)

python/sdk/tests/test_command_attachments_runtime.py

@@ -0,0 +1,253 @@
+"""Encoded command images retain desktop command rules through the shipped SDK profile."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import threading
+from pathlib import Path
+
+import pytest
+
+from deepseek_harness import CommandExecution, RunResult
+from deepseek_harness.errors import JsonRpcError
+from test_commands_runtime import (
+    _Completion, _SESSION_ID, _WAIT_SECONDS, _command_pair, _of_type,
+    _read_events, _request_texts, _runtime, _wait_activity,
+)
+from test_goal_runtime import _command, _until
+from test_plan_runtime import _GUIDANCE, _effective_system
+
+
+_IMAGE_MODEL = "deepseek-flash"
+_RED = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="
+_BLUE = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGNgYPj/HwADAgH/5ncLrgAAAABJRU5ErkJggg=="
+
+
+def _image(name: str = "reference.png", data: str = _RED) -> dict:
+    return {"type": "image", "data": data, "mimeType": "image/png", "name": name}
+
+
+def _request_images(request: dict) -> list[dict]:
+    return [block for message in request["messages"] for block in message["content"]
+            if block["type"] == "image"]
+
+
+def _assert_request_images(request: dict, *images: dict) -> None:
+    assert _request_images(request) == [
+        {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": image["data"]}}
+        for image in images
+    ]
+
+
+def _image_messages(events: list[dict]) -> list[dict]:
+    return [event for event in _of_type(events, "user/message")
+            if any(block["type"] == "image" for block in event["data"]["content"])]
+
+
+def _assert_durable_image(block: dict, image: dict) -> None:
+    data = base64.b64decode(image["data"])
+    assert block == {"type": "image", "attachment": {
+        "attachmentId": "sha256:" + hashlib.sha256(data).hexdigest(),
+        "mediaType": "image/png", "width": 1, "height": 1,
+        "bytes": len(data), "name": image["name"],
+    }}
+
+
+@pytest.mark.parametrize("line,body,image_first", [
+    ("/plan Describe this picture.", "Describe this picture.", False), ("/plan", "", True),
+])
+def test_plan_image_reaches_the_model_and_persists_as_a_reference(
+    tmp_path: Path, line: str, body: str, image_first: bool,
+) -> None:
+    picture = _image()
+    text = {"type": "text", "text": line}
+    content = [picture, text] if image_first else [text, picture]
+    with _runtime(tmp_path, [_Completion("IMAGE_PLAN_READY")], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        execution = runtime.call(session.run, content)
+        assert isinstance(execution, CommandExecution)
+        assert execution.result.kind == "success", execution.result.text
+        _wait_activity(runtime, session)
+        [request] = runtime.model.captured()
+        _assert_request_images(request, picture)
+        assert _GUIDANCE in _effective_system(request)
+        assert line not in _request_texts(request)
+        if body:
+            assert body in _request_texts(request)
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution, "plan")
+    assert [event["data"] for event in _of_type(events, "plan/mode")] == [{"active": True}]
+    [message] = _image_messages(events)
+    content = message["data"]["content"]
+    _assert_durable_image(content[0], picture)
+    assert content[1:] == ([{"type": "text", "text": body}] if body else [])
+    digest = content[0]["attachment"]["attachmentId"].removeprefix("sha256:")
+    stored = tmp_path / "home" / "attachments" / "v1" / "objects" / digest[:2] / digest
+    assert stored.read_bytes() == base64.b64decode(picture["data"])
+
+
+def test_plan_off_compact_and_permission_reject_images_without_applying_the_command(tmp_path: Path) -> None:
+    picture = _image()
+    with _runtime(tmp_path, [], model=_IMAGE_MODEL, initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        _command(runtime, session, "/plan")
+        executions = []
+        for line in ["/plan off", "/compact", "/permission danger-full-access"]:
+            execution = runtime.call(session.run, [{"type": "text", "text": line}, picture])
+            assert isinstance(execution, CommandExecution)
+            assert execution.result.kind == "error"
+            assert "attachment" in execution.result.text.lower()
+            executions.append((line.split()[0][1:], execution))
+        assert "Plan mode off" in _command(runtime, session, "/plan off").result.text
+        assert runtime.model.captured() == []
+    events = _read_events(runtime.logs)
+    for name, execution in executions:
+        _command_pair(events, execution, name)
+        start = next(event["seq"] for event in _of_type(events, "command/run")
+                     if event["data"]["commandId"] == execution.command_id)
+        done = next(event["seq"] for event in _of_type(events, "command/done")
+                    if event["data"]["commandId"] == execution.command_id)
+        assert events[start + 1:done] == []
+    assert [event["data"] for event in _of_type(events, "plan/mode")] == [{"active": True}, {"active": False}]
+    assert [event["data"] for event in _of_type(events, "permission/preset")] == [{"preset": "workspace-write"}]
+    assert not _image_messages(events) and not _of_type(events, "turn/start")
+
+
+def test_goal_create_and_edit_images_reach_following_requests_as_durable_references(tmp_path: Path) -> None:
+    original = _image("original.png")
+    revised = _image("revised.png", _BLUE)
+    first = _Completion("ORIGINAL_GOAL_IMAGE", blocked=True)
+    second = _Completion("REVISED_GOAL_IMAGE", blocked=True)
+    with _runtime(tmp_path, [first, second], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        created = runtime.call(session.run, [original, {"type": "text", "text": "/goal Inspect the original picture."}])
+        assert isinstance(created, CommandExecution) and created.result.kind == "success"
+        assert first.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        edited = runtime.call(session.run, [revised, {"type": "text", "text": "/goal edit Inspect the revised picture."}])
+        assert isinstance(edited, CommandExecution) and edited.result.kind == "success"
+        _command(runtime, session, "/goal clear")
+        assert not first.release.is_set()
+        first.release.set()
+        assert second.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        second.release.set()
+        _wait_activity(runtime, session)
+        before, after = runtime.model.captured()
+        _assert_request_images(before, original)
+        _assert_request_images(after, original, revised)
+        assert "Reference attachments for the goal objective." in _request_texts(before)
+        assert "Reference attachments for the goal objective." in _request_texts(after)
+    events = _read_events(runtime.logs)
+    _command_pair(events, created, "goal")
+    _command_pair(events, edited, "goal")
+    changes = _of_type(events, "goal/change")
+    assert [event["data"]["operation"] for event in changes] == ["create", "edit", "clear"]
+    assert changes[1]["data"]["goal"]["objective"] == "Inspect the revised picture."
+    messages = _image_messages(events)
+    assert len(messages) == 2
+    for message, picture in zip(messages, [original, revised]):
+        _assert_durable_image(message["data"]["content"][0], picture)
+        assert message["data"]["content"][1:] == [{
+            "type": "text", "text": "Reference attachments for the goal objective.",
+        }]
+
+
+def test_goal_controls_with_images_leave_the_active_request_and_goal_unchanged(tmp_path: Path) -> None:
+    first = _Completion("CURRENT_GOAL_REQUEST_FINISHED", blocked=True)
+    with _runtime(tmp_path, [first], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        with runtime.harness.client.subscribe_session_notifications(session.id) as notices:
+            _command(runtime, session, "/goal Keep the current goal.")
+            assert first.entered.wait(_WAIT_SECONDS), runtime.model.failures
+            rejected = []
+            for line in ["/goal", "/goal pause", "/goal resume", "/goal clear", "/goal edit", "/goal Duplicate goal"]:
+                execution = runtime.call(session.run, [{"type": "text", "text": line}, _image()])
+                assert isinstance(execution, CommandExecution) and execution.result.kind == "error"
+                rejected.append(execution)
+            assert "Status: active" in _command(runtime, session, "/goal").result.text
+            assert not first.release.is_set() and len(runtime.model.captured()) == 1
+            _command(runtime, session, "/goal clear")
+            first.release.set()
+            _wait_activity(runtime, session)
+            assert _until(notices, "turn/end")["reason"]["kind"] == "completed"
+        assert len(runtime.model.captured()) == 1
+    events = _read_events(runtime.logs)
+    for execution in rejected:
+        _command_pair(events, execution, "goal")
+    assert [event["data"]["operation"] for event in _of_type(events, "goal/change")] == ["create", "clear"]
+    assert not _image_messages(events)
+
+
+@pytest.mark.parametrize("delivery", ["queue", "steer"])
+def test_prompt_images_use_the_requested_delivery_during_an_active_turn(tmp_path: Path, delivery: str) -> None:
+    picture = _image()
+    text = {"type": "text", "text": "Inspect the image."}
+    content = [text, picture] if delivery == "queue" else [picture, text]
+    first = _Completion("BEFORE_IMAGE", blocked=True)
+    with _runtime(tmp_path, [first, _Completion("IMAGE_RECEIVED")], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        running = runtime.workers.submit(session.run, "Start without an image.")
+        assert first.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        accepted = threading.Event()
+        following = runtime.workers.submit(
+            session.run, content, delivery=delivery,
+            on_notification=lambda _notification: accepted.set(),
+        )
+        assert accepted.wait(_WAIT_SECONDS), "image prompt did not reach its inbox receipt"
+        assert not first.release.is_set() and not following.done()
+        first.release.set()
+        assert isinstance(running.result(timeout=_WAIT_SECONDS), RunResult)
+        result = following.result(timeout=_WAIT_SECONDS)
+        assert isinstance(result, RunResult) and result.final_response == "IMAGE_RECEIVED"
+        before, after = runtime.model.captured()
+        _assert_request_images(before)
+        _assert_request_images(after, picture)
+    events = _read_events(runtime.logs)
+    [message] = _image_messages(events)
+    recorded = message["data"]["content"]
+    _assert_durable_image(next(block for block in recorded if block["type"] == "image"), picture)
+    assert [block["type"] for block in recorded] == [block["type"] for block in content]
+    assert [block for block in recorded if block["type"] == "text"] == [text]
+    assert not _of_type(events, "command/run")
+    assert [event["data"] for event in _of_type(events, "step/start")] == (
+        [{"turn": 1, "step": 1}, {"turn": 2, "step": 1}] if delivery == "queue"
+        else [{"turn": 1, "step": 1}, {"turn": 1, "step": 2}]
+    )
+
+
+def test_top_level_run_accepts_an_image_without_text(tmp_path: Path) -> None:
+    picture = _image()
+    with _runtime(tmp_path, [_Completion("IMAGE_ONLY_REPLY")], model=_IMAGE_MODEL) as runtime:
+        result = runtime.call(runtime.harness.run, [picture], session_id=_SESSION_ID)
+        assert isinstance(result, RunResult) and result.final_response == "IMAGE_ONLY_REPLY"
+        [request] = runtime.model.captured()
+        _assert_request_images(request, picture)
+    [message] = _image_messages(_read_events(runtime.logs))
+    [block] = message["data"]["content"]
+    _assert_durable_image(block, picture)
+
+
+def test_invalid_image_batch_cannot_enter_plan_mode_or_deliver_a_partial_prompt(tmp_path: Path) -> None:
+    images = [_image(), _image("invalid.png", "not canonical base64")]
+    with _runtime(tmp_path, [], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        with pytest.raises(JsonRpcError) as ambiguous:
+            runtime.call(session.run, [
+                {"type": "text", "text": "/plan off"}, _image(),
+                {"type": "text", "text": "This is separate text, not another command argument."},
+            ])
+        assert "command" in str(ambiguous.value).lower()
+        rejected = runtime.call(session.run, [{"type": "text", "text": "/plan Inspect the pictures."}, *images])
+        assert isinstance(rejected, CommandExecution) and rejected.result.kind == "error"
+        assert "base64" in rejected.result.text
+        with pytest.raises(JsonRpcError, match="base64"):
+            runtime.call(session.run, [*images, {"type": "text", "text": "Inspect the pictures."}])
+        assert "already inactive" in _command(runtime, session, "/plan off").result.text
+        assert runtime.model.captured() == []
+    events = _read_events(runtime.logs)
+    _command_pair(events, rejected, "plan")
+    assert len(_of_type(events, "command/run")) == 2
+    assert not _of_type(events, "plan/mode")
+    assert not _of_type(events, "agent/inbox/spliced")
+    assert not _of_type(events, "turn/start")
+    assert not _image_messages(events)

python/sdk/tests/test_commands.py

@@ -129,11 +129,9 @@ def test_command_rejects_malformed_results(peer: Path, tmp_path: Path, response:
             sdk.client.session_command("main", "/compact")
 
 
-def test_content_blocks_keep_slash_text_as_an_ordinary_prompt(peer: Path, tmp_path: Path) -> None:
+def test_low_level_prompt_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([{"type": "text", "text": "/compact"}])
-        assert result.final_response == "ordinary prompt"
-        assert result.finish_reason == "completed"
+        assert sdk.client.session_prompt("main", [{"type": "text", "text": "/compact"}]) == "input-1"
     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

@@ -13,13 +13,15 @@ import shutil
 import sys
 import threading
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import contextmanager
+from contextlib import closing, contextmanager
 from dataclasses import dataclass, field
+from http.client import HTTPConnection, RemoteDisconnected
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 from typing import Callable, Iterator, TypeVar
 
 import pytest
+from pydantic import BaseModel
 
 from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, RunResult, Session
 from deepseek_harness.client import UserQuestionHandler
@@ -88,8 +90,13 @@ class _ModelServer:
                 super().setup()
 
             def do_POST(self) -> None:
+                payload = self.rfile.read(int(self.headers["Content-Length"]))
+                if self.path == "/v1/files":
+                    # The model fixture accepts inline images through the provider's Files fallback.
+                    self.send_error(404, "Images are accepted inline")
+                    return
                 assert self.path == "/v1/messages", f"unexpected model endpoint: {self.path}"
-                body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
+                body = json.loads(payload)
                 assert isinstance(body, dict) and isinstance(body.get("model"), str)
                 with owner.lock:
                     ordinal = len(owner.requests)
@@ -108,7 +115,7 @@ class _ModelServer:
                     self.end_headers()
                     self.wfile.write(payload)
                     self.wfile.flush()
-                except (BrokenPipeError, ConnectionResetError):
+                except (BrokenPipeError, ConnectionResetError, ConnectionAbortedError):
                     if not completion.allow_disconnect:
                         raise
 
@@ -158,6 +165,39 @@ class _ModelServer:
             assert not self.failures, [repr(error) for error in self.failures]
 
 
+@pytest.mark.parametrize("error,allow_disconnect,accepted", [
+    (ConnectionAbortedError(10053, "response connection aborted"), True, True),
+    (ConnectionAbortedError(10053, "response connection aborted"), False, False),
+    (OSError(5, "unrelated response write failure"), True, False),
+])
+def test_model_server_only_allows_expected_connection_aborts(
+    monkeypatch: pytest.MonkeyPatch, error: OSError, allow_disconnect: bool, accepted: bool,
+) -> None:
+    model = _ModelServer([_Completion("response", allow_disconnect=allow_disconnect)])
+    body = {"model": _MODEL}
+
+    def fail_response(_handler: BaseHTTPRequestHandler) -> None:
+        raise error
+
+    # Keep the real HTTP handler and teardown; only the response write fails deterministically.
+    monkeypatch.setattr(model.server.RequestHandlerClass, "end_headers", fail_response)
+
+    def exchange() -> None:
+        with model, closing(HTTPConnection(*model.server.server_address, timeout=_WAIT_SECONDS)) as connection:
+            connection.request("POST", "/v1/messages", body=json.dumps(body))
+            with pytest.raises(RemoteDisconnected):
+                connection.getresponse()
+
+    if accepted:
+        exchange()
+    else:
+        with pytest.raises(AssertionError, match=type(error).__name__):
+            exchange()
+    assert model.captured() == [body]
+    assert model.failures == ([] if accepted else [error])
+    assert not model.thread.is_alive()
+
+
 def _dsh_bin() -> str | None:
     mode = os.environ.get("DSH_COMMANDS_TEST_RUNTIME", "built")
     if mode == "installed":
@@ -183,12 +223,22 @@ class _Runtime:
         return self.workers.submit(operation, *args, **kwargs).result(timeout=_WAIT_SECONDS)
 
 
+class _WaitResult(BaseModel):
+    pass
+
+
+def _wait_activity(runtime: _Runtime, session: Session) -> None:
+    # Command completion does not own the model activity that its native handler starts.
+    runtime.call(runtime.harness.client.request, "session/wait", {"sessionId": session.id}, response_model=_WaitResult)
+
+
 @contextmanager
 def _runtime(
     tmp_path: Path, completions: list[_Completion], *, capture_patch: Path | None = None,
     initial_permission: str = "danger-full-access",
     on_user_question: UserQuestionHandler | None = None,
     max_goal_rounds: int | None = None,
+    model: str = _MODEL,
 ) -> Iterator[_Runtime]:
     workspace = tmp_path / "workspace"
     workspace.mkdir(exist_ok=True)
@@ -206,11 +256,11 @@ def _runtime(
             extra.write(f"- id: goal\n  config:\n    defaultMaxGoalRounds: {max_goal_rounds}\n")
     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:
+    with _ModelServer(completions) as model_server, 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",
+            dsh_bin=dsh_bin, profile="sdk", patches=tuple(str(item) for item in patches), model=model,
+            base_url=model_server.url, api_key="keyless-commands-runtime",
             initialize_timeout_seconds=60, request_timeout_seconds=20, shutdown_timeout_seconds=20,
             on_user_question=on_user_question,
             env={"DSH_PERMISSION_MODE": initial_permission,
@@ -218,10 +268,10 @@ def _runtime(
         )
         try:
             harness.start()
-            yield _Runtime(harness, model, workers, logs)
+            yield _Runtime(harness, model_server, workers, logs)
         finally:
             # Also covers failed initialization: release HTTP, close the client, then join callers.
-            model.release_all()
+            model_server.release_all()
             harness.close()
 
 
@@ -509,8 +559,8 @@ def test_permission_changes_only_the_selected_session_and_retains_prompt_semanti
 
         for session in [other, runtime.harness.start_session("permission-new")]:
             assert isinstance(runtime.call(session.run, "Report your own policy; do not use tools."), RunResult)
-        ordinary = runtime.call(main.run, [{"type": "text", "text": "/permission danger-full-access"}])
-        assert isinstance(ordinary, RunResult)
+        runtime.call(runtime.harness.client.session_prompt, main.id, [{"type": "text", "text": "/permission danger-full-access"}])
+        _wait_activity(runtime, main)
         assert isinstance(runtime.call(main.run, "/permission"), RunResult)
         assert runtime.harness.config.env == initial_config
 

python/sdk/tests/test_cot_translation_projection.py

@@ -27,13 +27,13 @@ for line in sys.stdin:
     message = json.loads(line)
     if message["method"] == "initialize":
         result = {"serverInfo": {"name": "audit-fixture"}}
-    elif message["method"] == "session/prompt":
+    elif message["method"] == "session/query":
         session_id = message["params"]["sessionId"]
         notify("session.event", {"sessionId": session_id, "event": {
             "type": "agent/inbox/spliced", "seq": 0, "time": 0,
             "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "accepted"}]}}})
         notify("session.status", {"sessionId": session_id, "status": "running"})
-        print(json.dumps({"jsonrpc": "2.0", "id": message["id"], "result": {"messageId": "accepted"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": message["id"], "result": {"kind": "prompt", "messageId": "accepted"}}), flush=True)
         notify("session.event", {"sessionId": session_id, "event": {"type": "turn/start", "seq": 1, "time": 0, "data": {"turn": 0}}})
         notify("session.event", {"sessionId": session_id, "event": {"type": "assistant/message", "seq": 2, "time": 0,
             "data": {"message": {"role": "assistant", "content": [{"type": "text", "text": os.environ["MAIN_RESPONSE"]}]}}}})

python/sdk/tests/test_goal_example.py

@@ -7,6 +7,7 @@ import runpy
 import sys
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
+from types import SimpleNamespace
 
 import pytest
 
@@ -16,7 +17,7 @@ from deepseek_harness import DeepSeekHarness
 _EXAMPLE = Path(__file__).resolve().parents[1] / "examples/goal.py"
 
 
-@pytest.mark.parametrize("pending", ["command", "goal-event", "wrapup", "complete"])
+@pytest.mark.parametrize("pending", ["command", "goal-event", "wrapup", "complete", "goal-queue", "wrapup-queue"])
 def test_goal_example_bounds_pending_work_and_reaps_runtime(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], pending: str,
 ) -> None:
@@ -39,21 +40,26 @@ for line in sys.stdin:
     if method == "initialize":
         send({"id": frame["id"], "result": {"serverInfo": {"name": "goal-example-peer"}}})
     elif method == "session/query":
-        assert frame["params"]["text"] == "/goal Finish this task"
+        assert frame["params"]["contentBlocks"] == [{"type": "text", "text": "/goal Finish this task"}]
         session = frame["params"]["sessionId"]
         if pending == "command":
             continue
-        if pending in ("wrapup", "complete"):
+        if pending in ("wrapup", "complete", "wrapup-queue"):
             send({"method": "session.event", "params": {"sessionId": session, "event": {
                 "type": "goal/change", "data": {"operation": "complete"}}}})
+        elif pending == "goal-queue":
+            for operation in ("create", "edit"):
+                send({"method": "session.event", "params": {"sessionId": session, "event": {
+                    "type": "goal/change", "data": {"operation": operation}}}})
         send({"id": frame["id"], "result": {"kind": "command", "execution": {
             "commandId": "goal-command", "result": {"kind": "success", "text": "Goal set"}}}})
     elif method == "session/wait":
         if pending == "wrapup":
             continue
-        send({"method": "session.event", "params": {"sessionId": session, "event": {
-            "type": "assistant/message", "data": {"message": {
-                "content": [{"type": "text", "text": "Closing reply"}]}}}}})
+        for _ in range(2 if pending == "wrapup-queue" else 1):
+            send({"method": "session.event", "params": {"sessionId": session, "event": {
+                "type": "assistant/message", "data": {"message": {
+                    "content": [{"type": "text", "text": "Closing reply"}]}}}}})
         send({"id": frame["id"], "result": {}})
     elif method == "shutdown":
         send({"id": frame["id"], "result": {}})
@@ -73,7 +79,24 @@ for line in sys.stdin:
         return harness
 
     monkeypatch.setitem(example["main"].__globals__, "DeepSeekHarness", configured_harness)
-    timeout = "10" if pending == "complete" else "1"
+    queued = pending in {"goal-queue", "wrapup-queue"}
+    displayed = []
+    if queued:
+        clock = [0.0]
+        display = example["display"]
+
+        def advance_after_display(notification):
+            event = display(notification)
+            if event is not None:
+                displayed.append(event["type"])
+                if pending == "goal-queue" or event["type"] == "assistant/message":
+                    clock[0] = 11.0
+            return event
+
+        # Advance only the example clock while the real transport retains its own timeout.
+        monkeypatch.setitem(example["main"].__globals__, "time", SimpleNamespace(monotonic=lambda: clock[0]))
+        monkeypatch.setitem(example["main"].__globals__, "display", advance_after_display)
+    timeout = "10" if pending == "complete" or queued else "1"
     monkeypatch.setattr(sys, "argv", [str(_EXAMPLE), "--dsh-home", str(tmp_path / "home"),
                                      "--workspace", str(tmp_path), "--timeout-seconds", timeout, "Finish this task"])
     # The outer bound also releases an implementation that forgets its own deadline.
@@ -84,7 +107,8 @@ for line in sys.stdin:
                 execution.result(timeout=30)
             else:
                 expected = {"command": "session/query timed out", "goal-event": "SDK notification",
-                            "wrapup": "session/wait timed out"}[pending]
+                            "wrapup": "session/wait timed out", "goal-queue": "Goal observation timed out",
+                            "wrapup-queue": "Goal observation timed out"}[pending]
                 with pytest.raises(TimeoutError, match=expected):
                     execution.result(timeout=30)
         finally:
@@ -94,7 +118,10 @@ for line in sys.stdin:
     assert harness.client._proc is None
     frames = [json.loads(line) for line in observed.read_text(encoding="utf-8").splitlines()]
     methods = [frame["method"] for frame in frames]
-    assert methods == ["initialize", "session/query", *(["session/wait"] if pending in {"wrapup", "complete"} else []), "shutdown"]
+    assert methods == ["initialize", "session/query",
+                       *(["session/wait"] if pending in {"wrapup", "complete", "wrapup-queue"} else []), "shutdown"]
+    if queued:
+        assert displayed == (["goal/change"] if pending == "goal-queue" else ["goal/change", "assistant/message"])
     if pending == "complete":
         output = capsys.readouterr().out
         assert "Goal: complete" in output

python/sdk/tests/test_goal_runtime.py

@@ -9,14 +9,13 @@ import threading
 import pytest
 from pathlib import Path
 
-from deepseek_harness import CommandExecution, RunResult, Session
+from deepseek_harness import CommandExecution, Session
 from deepseek_harness.client import NotificationSubscription
 from test_commands_runtime import (
-    _Completion, _Runtime, _SESSION_ID, _WAIT_SECONDS, _assert_request_permission,
+    _Completion, _Runtime, _SESSION_ID, _WAIT_SECONDS, _WaitResult, _assert_request_permission,
     _command_pair, _of_type, _permission_facts, _read_events,
-    _request_texts, _runtime, _text_blocks,
+    _request_texts, _runtime, _text_blocks, _wait_activity,
 )
-from test_plan_runtime import _WaitResult, _wait_activity
 
 
 def _command(runtime: _Runtime, session: Session, text: str, *, kind: str = "success") -> CommandExecution:
@@ -145,8 +144,8 @@ def test_pause_aborts_then_edit_and_resume_use_revised_goal_without_changing_oth
             assert 'Objective: "Revised objective"' in "\n".join(_request_texts(runtime.model.captured()[1]))
             _command(runtime, session, "/goal clear")
             _command(runtime, session, "/goal clear")
-            result = runtime.call(session.run, [{"type": "text", "text": "/goal Plain text objective"}])
-            assert isinstance(result, RunResult) and result.final_response == "RAW_GOAL_TEXT"
+            runtime.call(runtime.harness.client.session_prompt, session.id, [{"type": "text", "text": "/goal Plain text objective"}])
+            _wait_activity(runtime, session)
             assert "/goal Plain text objective" in _request_texts(runtime.model.captured()[-1])
     changes = _of_type(_read_events(runtime.logs), "goal/change")
     assert [e["data"]["operation"] for e in changes] == ["create", "pause", "edit", "resume", "complete", "clear"]

python/sdk/tests/test_plan_input.py

@@ -96,7 +96,9 @@ if sys.platform != "win32":
     import resource
     resource.setrlimit(resource.RLIMIT_CORE, (0, 0))
 
-example_path, peer, root_argument = sys.argv[1:]
+example_path, peer, root_argument, stdin_encoding = sys.argv[1:]
+if stdin_encoding:
+    sys.stdin.reconfigure(encoding=stdin_encoding)
 root = Path(root_argument)
 example = runpy.run_path(example_path)
 readers = []
@@ -147,6 +149,7 @@ def _wait_prompt(process: subprocess.Popen[bytes], output: Path, marker: str) ->
 
 def _run(
     tmp_path: Path, data: bytes, *, stage: str | None = None, action: str | None = None, terminal: bool = False,
+    stdin_encoding: str | None = None, python_args: tuple[str, ...] = (), env: dict[str, str] | None = None,
 ) -> list[dict]:
     peer = tmp_path / "peer.py"
     peer.write_text(_PEER, encoding="utf-8")
@@ -154,7 +157,8 @@ def _run(
     wrapper.write_text(_WRAPPER, encoding="utf-8")
     output = tmp_path / "stdout.txt"
     errors = tmp_path / "stderr.txt"
-    env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
+    if env is None:
+        env = {**os.environ, "PYTHONIOENCODING": "utf-8"}
     with ExitStack() as stack, output.open("wb") as stdout, errors.open("wb") as stderr:
         if terminal:
             master, slave = os.openpty()
@@ -164,7 +168,7 @@ def _run(
         else:
             child_input = subprocess.PIPE
         process = subprocess.Popen(
-            [sys.executable, str(wrapper), str(_EXAMPLE), str(peer), str(tmp_path)],
+            [sys.executable, *python_args, str(wrapper), str(_EXAMPLE), str(peer), str(tmp_path), stdin_encoding or ""],
             stdin=child_input, stdout=stdout, stderr=stderr, env=env, start_new_session=os.name != "nt",
         )
         if not terminal:
@@ -202,6 +206,10 @@ def _run(
     cleanup = json.loads((tmp_path / "cleanup.json").read_text())
     assert cleanup["readers"] and all(reader["returncode"] is not None for reader in cleanup["readers"])
     assert cleanup["callbacks"] == []
+    answers = tmp_path / "answers.jsonl"
+    if answers.is_file():
+        responses = [json.loads(line) for line in answers.read_text().splitlines()]
+        assert not [response for response in responses if "error" in response], responses
     return [json.loads(line) for line in (tmp_path / "callbacks.jsonl").read_text().splitlines()]
 
 
@@ -218,6 +226,13 @@ def test_plan_input_preserves_piped_lines_and_eof(tmp_path: Path, data: bytes, e
     assert _run(tmp_path, data) == expected
 
 
+def test_plan_input_preserves_reconfigured_stdin_feedback(tmp_path: Path) -> None:
+    assert _run(tmp_path, "k\nréviser\nd\n".encode("latin-1"), stdin_encoding="latin-1") == [
+        {"answers": [{"id": "plan-review", "selected": ["Keep planning"], "custom": "réviser"}]},
+        {"cancelled": True},
+    ]
+
+
 def test_plan_input_print_failure_reaps_reader_before_owner_exit(monkeypatch: pytest.MonkeyPatch) -> None:
     example = runpy.run_path(str(_EXAMPLE))
     owner_type = example["_PlanReviews"]

python/sdk/tests/test_plan_input_posix.py

@@ -1,5 +1,6 @@
 """POSIX terminal input and SIGINT cleanup for the plan example."""
 
+import os
 from pathlib import Path
 import sys
 
@@ -19,6 +20,22 @@ def test_plan_input_preserves_terminal_feedback(tmp_path: Path) -> None:
     ]
 
 
+def test_plan_input_preserves_reconfigured_terminal_feedback(tmp_path: Path) -> None:
+    assert _run(tmp_path, "k\nréviser\nd\n".encode("latin-1"), terminal=True, stdin_encoding="latin-1") == [
+        {"answers": [{"id": "plan-review", "selected": ["Keep planning"], "custom": "réviser"}]},
+        {"cancelled": True},
+    ]
+
+
+def test_plan_input_preserves_explicit_utf8_mode_with_ascii_child(tmp_path: Path) -> None:
+    env = {**os.environ, "LC_ALL": "C", "PYTHONUTF8": "0", "PYTHONCOERCECLOCALE": "0"}
+    env.pop("PYTHONIOENCODING", None)
+    assert _run(tmp_path, "k\n继续检查\nd\n".encode("utf-8"), python_args=("-X", "utf8"), env=env) == [
+        {"answers": [{"id": "plan-review", "selected": ["Keep planning"], "custom": "继续检查"}]},
+        {"cancelled": True},
+    ]
+
+
 @pytest.mark.parametrize("stage", ["choice", "feedback"])
 def test_plan_input_interrupt_reaps_open_pipe_reader(tmp_path: Path, stage: str) -> None:
     data = b"k\npartial feedback" if stage == "feedback" else b"partial choice"

python/sdk/tests/test_plan_runtime.py

@@ -6,13 +6,12 @@ import threading
 from pathlib import Path
 
 import pytest
-from pydantic import BaseModel
 
 from deepseek_harness import CommandExecution, RunResult, Session, UserQuestionRequest
 from test_commands_runtime import (
     _Completion, _Runtime, _SESSION_ID, _WAIT_SECONDS, _assert_request_permission,
     _command_pair, _of_type, _permission_facts, _read_events,
-    _request_texts, _runtime, _text_blocks,
+    _request_texts, _runtime, _text_blocks, _wait_activity,
 )
 
 _PLAN = "# Implement the change\n\n- Inspect existing code.\n- Make the smallest change.\n- Run focused tests."
@@ -20,10 +19,6 @@ _PLAN_REVISED = "# Revised plan\n\n- Include a rollback procedure."
 _GUIDANCE = "You are in plan mode."
 
 
-class _WaitResult(BaseModel):
-    pass
-
-
 def _command(runtime: _Runtime, session: Session, text: str) -> CommandExecution:
     result = runtime.call(session.run, text)
     assert isinstance(result, CommandExecution)
@@ -31,11 +26,6 @@ def _command(runtime: _Runtime, session: Session, text: str) -> CommandExecution
     return result
 
 
-def _wait_activity(runtime: _Runtime, session: Session) -> None:
-    # Command completion does not own the model activity that its native handler starts.
-    runtime.call(runtime.harness.client.request, "session/wait", {"sessionId": session.id}, response_model=_WaitResult)
-
-
 def _effective_system(request: dict) -> str:
     system = "\n".join(_text_blocks(request.get("system", "")))
     for message in request["messages"]:
@@ -224,7 +214,7 @@ def test_plan_state_survives_resume_without_changing_permissions(tmp_path: Path)
     assert [e["data"] for e in _of_type(events, "plan/mode")] == [{"active": True}]
 
 
-def test_plan_state_is_local_and_run_keeps_ordinary_text_semantics(tmp_path: Path) -> None:
+def test_plan_state_is_local_and_literal_prompts_do_not_toggle_it(tmp_path: Path) -> None:
     with _runtime(tmp_path, [_Completion("OTHER"), _Completion("PLANNING"), _Completion("RAW_INPUT")],
                   initial_permission="workspace-write") as runtime:
         session = runtime.harness.start_session(_SESSION_ID)
@@ -232,8 +222,8 @@ def test_plan_state_is_local_and_run_keeps_ordinary_text_semantics(tmp_path: Pat
         other = runtime.harness.start_session("plan-other")
         assert isinstance(runtime.call(other.run, "Inspect your own task without tools."), RunResult)
         assert isinstance(runtime.call(session.run, "Inspect the plan without tools."), RunResult)
-        result = runtime.call(session.run, [{"type": "text", "text": "/plan off"}])
-        assert isinstance(result, RunResult)
+        runtime.call(runtime.harness.client.session_prompt, session.id, [{"type": "text", "text": "/plan off"}])
+        _wait_activity(runtime, session)
         first, second, third = runtime.model.captured()
         assert _GUIDANCE not in _effective_system(first)
         assert _GUIDANCE in _effective_system(second)

python/sdk/tests/test_run.py

@@ -1,6 +1,7 @@
 """Unified blocking input, native command results and independent run observation."""
 from __future__ import annotations
 
+import json
 import sys
 import threading
 from pathlib import Path
@@ -21,6 +22,7 @@ class Ack(BaseModel):
 def peer(tmp_path: Path) -> Path:
     script = tmp_path / "queries.py"
     script.write_text('''import json
+import os
 import sys
 
 serial = 0
@@ -33,13 +35,16 @@ def send(frame):
 def event(session, type, data):
     send({"method": "session.event", "params": {"sessionId": session, "event": {"type": type, "data": data}}})
 
+def input_text(frame):
+    return "".join(block["text"] for block in frame["params"]["contentBlocks"] if block["type"] == "text")
+
 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"}}}})
+        "commandId": identity, "result": {"kind": "success", "text": "command completed"}}}})
 
 def receipt(frame, number):
     session = frame["params"]["sessionId"]
@@ -49,24 +54,26 @@ def receipt(frame, number):
 def finish(frame):
     session = frame["params"]["sessionId"]
     event(session, "assistant/message", {"message": {"content": [
-        {"type": "text", "text": "answer for " + frame["params"]["text"]}]}})
+        {"type": "text", "text": "answer for " + input_text(frame)}]}})
     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)
+    with open(os.environ["REQUEST_LOG"], "a", encoding="utf-8") as log:
+        log.write(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 set(params) == {"sessionId", "contentBlocks", "delivery"}
         assert params["delivery"] in ("queue", "steer")
-        text = params["text"]
+        text = input_text(frame)
         if text == "hold-command":
             pending_commands.append((frame, serial))
-        elif text == "/compact":
+        elif text in ("/compact", "/plan Describe this image."):
             command(frame, serial)
         elif text == "silent-command":
             send({"id": frame["id"], "result": {"kind": "command", "execution": {
@@ -117,6 +124,7 @@ 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)),
+        env={"REQUEST_LOG": str(tmp_path / "requests.jsonl")},
     ) as harness:
         yield harness
 
@@ -159,6 +167,51 @@ def test_run_accepts_steer_for_text(sdk: DeepSeekHarness) -> None:
     assert result.final_response == "answer for next step"
 
 
+@pytest.mark.parametrize("entry,text,delivery", [
+    ("harness", "/plan Describe this image.", "queue"),
+    ("session", "/plan Describe this image.", "steer"),
+    ("harness", "Explain this image.", "queue"),
+    ("session", "/unknown-slash", "steer"),
+    ("session", "", "queue"),
+])
+def test_run_forwards_content_blocks_with_the_routed_result_and_notifications(
+    sdk: DeepSeekHarness, tmp_path: Path, entry: str, text: str, delivery: str,
+) -> None:
+    blocks = [
+        {"type": "text", "text": text},
+        {"type": "image", "data": "AQ==", "mimeType": "image/png", "name": "diagram.png"},
+    ]
+    notifications = []
+    result = (
+        sdk.run(blocks, session_id="main", delivery=delivery, on_notification=notifications.append)
+        if entry == "harness" else sdk.start_session("main").run(
+            blocks, delivery=delivery, on_notification=notifications.append,
+        )
+    )
+    if text == "/plan Describe this image.":
+        assert isinstance(result, CommandExecution)
+        assert [notice.payload["event"]["type"] for notice in notifications] == ["command/run", "command/done"]
+    else:
+        assert isinstance(result, RunResult)
+        assert result.final_response == "answer for " + text
+        assert result.notifications == notifications
+    sent = [json.loads(line) for line in (tmp_path / "requests.jsonl").read_text(encoding="utf-8").splitlines()]
+    assert sent[-1]["method"] == "session/query"
+    assert sent[-1]["params"] == {
+        "sessionId": "main", "contentBlocks": blocks, "delivery": delivery,
+    }
+    assert_released(sdk)
+
+
+def test_run_normalizes_strings_without_changing_their_text(sdk: DeepSeekHarness, tmp_path: Path) -> None:
+    result = sdk.run("  preserve whitespace  ", session_id="main")
+    assert isinstance(result, RunResult)
+    sent = [json.loads(line) for line in (tmp_path / "requests.jsonl").read_text(encoding="utf-8").splitlines()]
+    assert sent[-1]["params"] == {
+        "sessionId": "main", "contentBlocks": [{"type": "text", "text": "  preserve whitespace  "}], "delivery": "queue",
+    }
+
+
 def test_concurrent_runs_keep_their_own_results(sdk: DeepSeekHarness) -> None:
     session = sdk.start_session("main")
     entered = threading.Event()
@@ -217,5 +270,5 @@ def test_run_requires_supported_delivery(sdk: DeepSeekHarness) -> None:
     session = sdk.start_session("main")
     with pytest.raises(ValueError, match="delivery"):
         session.run("ordinary", delivery="unknown")
-    with pytest.raises(ValueError, match="content-block"):
-        session.run([{"type": "text", "text": "ordinary"}], delivery="steer")
+    with pytest.raises(ValueError, match="delivery"):
+        session.run([{"type": "text", "text": "ordinary"}], delivery="unknown")