fix(sdk): return RunResult after native command activity

python/sdk/README.i18n.yaml

@@ -12,8 +12,8 @@
   en: ac793c588b13ce87
   zh: 502015cda46de3ac
 /deepseek-harness-python-sdk/run-text-and-commands:
-  en: 477bb559cbf13f84
-  zh: c5825be4ae2f1508
+  en: 945bbaded33412ab
+  zh: b9905a5225de1f86
 /deepseek-harness-python-sdk/client-approvals:
   en: fc69be1beb8abff1
   zh: 91b56781931a6f68

python/sdk/README.md

@@ -33,7 +33,7 @@ The Python SDK has no separate application entrypoint. It launches the bundled `
 Every launch requires an explicit Harness home. Pass `dsh_home` or provide a non-empty `DSH_HOME` in the child environment. The SDK deliberately never discovers `~/.dsh`.
 
 ```py
-from deepseek_harness import DeepSeekHarness, RunResult
+from deepseek_harness import DeepSeekHarness
 
 with DeepSeekHarness(
     dsh_home="/absolute/path/to/isolated-dsh-home",
@@ -45,7 +45,7 @@ with DeepSeekHarness(
 ) as harness:
     result = harness.run("Say hi.", session_id="example-001")
 
-print(result.final_response if isinstance(result, RunResult) else result.result.text)
+print(result.final_response)
 ```
 
 `DeepSeekHarness` starts lazily and reuses its runtime until `close()` or context-manager exit. The initial profile handshake has an independent 30-second default bound through `initialize_timeout_seconds`; ordinary activity remains unbounded after its protocol response, whose wait can be limited with `request_timeout_seconds`. A timeout names the selected profile and includes retained runtime diagnostics. `cwd` is the agent workspace; `runtime_cwd` independently selects the subprocess working directory. Both become absolute before launch. `provider`, `model`, optional `reasoning_effort`, and optional positive `max_tokens` are sent during JSON-RPC initialization. `base_url` and `api_key` explicitly override `DEEPSEEK_BASE_URL` and `DEEPSEEK_API_KEY` in the child environment.
@@ -63,21 +63,21 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 
 `session.run(input, on_notification=..., delivery="queue")` is the public prompt-or-command 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 `RunCommandResult` after a command settles. Both results include `session_id`, including when `harness.run()` creates a fresh Session. `RunCommandResult` extends the native `CommandExecution` with that Session identity. 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.
+Every call returns `RunResult`, including the receiving `session_id`, root events and Session-tree notifications. `final_response` contains the final assistant text, or an empty string when the activity emits none. State-only commands return the native handler's result text. `finish_reason` reports the terminal outcome. Native command identities and outcomes remain in `command/run` and `command/done` events.
 
-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 `RunCommandResult`. 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.
+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 `RunResult`. To bypass native command routing, call the low-level `client.session_prompt(session_id, blocks)` and manage subsequent activity; 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 `RunCommandResult.result.kind="error"`; routing, protocol, transport and observer failures raise. Malformed native-command responses from `run()` or `session_command()` raise `SdkProtocolError`, with the model validation error retained as the cause. 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.
+Each call subscribes before sending and releases its subscription before returning or raising. `on_notification` receives the input's Session-tree activity in wire order; the interval can include overlapping work. Calls may overlap without consuming each other's results. Native refusals return `finish_reason="error"`; routing, protocol, transport and observer failures raise. Malformed native-command responses raise `SdkProtocolError`, retaining their validation cause. 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 input 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)` accepts ASCII registered names or canonical English aliases of supported builtin definitions, without prompt fallback or attachments, and returns the native `CommandExecution`. Use `run()` for Chinese aliases. 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).
 
-`/plan` enters plan mode, `/plan off` leaves, and `/plan <message>` enters then submits the trimmed message through native steer. Its `RunCommandResult` confirms command completion, not completion of planning or review. Enter with `/plan` then run ordinary text when a `RunResult` is needed. For combined input, subscribe before `run()` and use `session/wait` while keeping the question handler alive, as in the [plan example](examples/README.md#review-a-plan). Plan guidance and `exit_plan_mode` follow their native lifecycle without changing permission settings.
+`/plan` enters plan mode, `/plan off` leaves, and `/plan <message>` enters then submits the trimmed message through native steer. `run()` waits for submitted model activity, including native review and approved continuation; no separate subscription or `session/wait` call is required. Keep `on_user_question` available for review, as in the [plan example](examples/README.md#review-a-plan). Plans submitted through `exit_plan_mode` arrive in the review question's `detail`; `final_response` contains the final assistant text. Plan mode does not change permission settings.
 
-`/goal` shows the current goal; `/goal <objective>`, `edit <objective>`, `pause`, `resume`, and `clear` use the native handler. Creation and resume can start automatic rounds after `run()` returns. Subscribe before sending, observe terminal `goal/change` operations, then call `session/wait` for any closing model reply. Idle between rounds is not goal completion. A restored active goal stays disarmed until explicit resume. See the [goal example](examples/README.md#run-a-goal).
+`/goal` shows the current goal. Creation and resume through `run()` wait across automatic rounds until the goal stops and its closing activity finishes. Completion returns `completed`, pause or clear returns `aborted`, and blocking or lost activation reports failure. State-only controls return without waiting for unrelated activity. A restored active goal stays disarmed until explicit resume. See the [goal example](examples/README.md#run-a-goal).
 
 <a id="client-approvals"></a>
 ## Client approvals

python/sdk/README.zh.md

@@ -33,7 +33,7 @@ Python SDK 没有独立的应用入口。它以 `--profile sdk` 启动内置的
 每次启动都必须显式指定 Harness home。请传入 `dsh_home`，或在子进程环境中提供非空的 `DSH_HOME`。SDK 刻意不会发现 `~/.dsh`。
 
 ```py
-from deepseek_harness import DeepSeekHarness, RunResult
+from deepseek_harness import DeepSeekHarness
 
 with DeepSeekHarness(
     dsh_home="/absolute/path/to/isolated-dsh-home",
@@ -45,7 +45,7 @@ with DeepSeekHarness(
 ) as harness:
     result = harness.run("Say hi.", session_id="example-001")
 
-print(result.final_response if isinstance(result, RunResult) else result.result.text)
+print(result.final_response)
 ```
 
 `DeepSeekHarness` 延迟启动运行时，并在调用 `close()` 或退出上下文管理器前复用该进程。首次 profile 握手通过 `initialize_timeout_seconds` 使用独立的 30 秒默认上限；普通活动在协议响应后仍不设时限；`request_timeout_seconds` 可以限制协议响应的等待。超时诊断会指明所选 profile，并包含保留的运行时诊断。`cwd` 是 agent workspace；`runtime_cwd` 独立选择子进程工作目录。两者都会在启动前转成绝对路径。`provider`、`model`、可选的 `reasoning_effort` 和可选的正整数 `max_tokens` 通过 JSON-RPC 初始化发送。`base_url` 与 `api_key` 会显式覆盖子进程环境中的 `DEEPSEEK_BASE_URL` 与 `DEEPSEEK_API_KEY`。
@@ -63,21 +63,21 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 
 `session.run(input, on_notification=..., delivery="queue")` 是字符串和内容块数组共用的公开提示词／命令输入入口。两者都使用运行时的[命令形式与 Desktop 别名](../../packages/sdk/server/README.zh.md#command-input)识别规则；未识别的输入保持普通提示词及原始内容块顺序。对普通输入，包括图片，`delivery="steer"` 指向下一个可用步骤。命令的接纳规则与执行时序仍由原生处理器负责。
 
-普通提示词返回其入队回执到下一次 idle 区间的 `RunResult`，命令则在执行结束后返回 `RunCommandResult`。两种结果都包含 `session_id`，包括 `harness.run()` 创建新 Session 的情况。`RunCommandResult` 在原生 `CommandExecution` 上增加该 Session 标识。读取 `final_response` 前使用 `isinstance(result, RunResult)` 检查类型，否则检查命令的 `result.kind`。命令保留自己的 `command_id` 和可选 `source_event_seq`，不会选取更早的助手回复作为结果。
+每次调用都返回 `RunResult`，包含接收输入的 `session_id`、根会话事件和会话树通知。`final_response` 包含最终助手文本；活动未输出助手文本时为空字符串。仅改变状态的命令返回原生 handler 的结果文本。`finish_reason` 表示结束结果。原生命令身份与结果保留在 `command/run` 和 `command/done` 事件中。
 
-内容块数组与字符串使用相同的命令路由。`session.run([{"type": "text", "text": "/compact"}])` 和 `session.run("/compact")` 都调用原生压缩并返回 `RunCommandResult`。要绕过原生命令路由，调用低层 `client.session_prompt(session_id, blocks)` 并自行管理后续订阅与等待；运行时 skill（技能）识别仍然适用。
+内容块数组与字符串使用相同的命令路由。`session.run([{"type": "text", "text": "/compact"}])` 和 `session.run("/compact")` 都调用原生压缩并返回 `RunResult`。要绕过原生命令路由，调用底层 `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 树通知，因此该区间可能包含重叠工作的通知。调用可以重叠，不会消耗彼此的结果。处理器拒绝返回 `RunCommandResult.result.kind="error"`，路由、协议、传输和观察回调失败则抛出异常。`run()` 或 `session_command()` 收到格式错误的原生命令响应时抛出 `SdkProtocolError`，并保留模型校验错误作为 cause。超时或观察回调失败不回滚运行时工作，也不授权自动重试。`request_timeout_seconds` 限制协议响应等待，不限制之后的普通提示词活动。
+每次调用在发送前订阅，并在返回或抛出异常前释放订阅。`on_notification` 按传输顺序接收本次输入的会话树活动；区间可以包含重叠工作。调用可以重叠，不会消耗彼此的结果。原生拒绝返回 `finish_reason="error"`；路由、协议、传输和观察回调失败则抛出异常。格式错误的原生命令响应抛出 `SdkProtocolError`，保留校验原因。超时或观察回调失败不回滚运行时工作，也不授权自动重试。`request_timeout_seconds` 限制协议响应等待，不限制之后的输入活动。
 
 默认的完整 `sdk` profile 提供已支持的原生命令；`sdk-minimal` 将未支持的输入保留在普通提示词路径。`/compact` 可以发起辅助模型请求，其完成不需要普通 turn 或 idle 通知；空历史可以在不生成摘要时成功。底层 `HarnessClient.session_command(session_id, line)` 接受受支持内置定义的 ASCII 注册名或英文规范别名，不回退为提示词，也不接受附件，返回原生 `CommandExecution`。中文别名使用 `run()`。见[压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
 
 `/permission <preset>` 只修改接收命令的 Session。未知 preset 返回命令错误，不改变策略。后续输入与恢复保留成功的修改，其他 Session 和调用方的 `DeepSeekHarnessConfig` 保留自己的值。运行中的审批策略变化可以注入通知并触发下一模型步骤。见[权限示例](examples/README.zh.md#change-session-permissions)。
 
-`/plan` 进入计划模式，`/plan off` 退出，`/plan <message>` 进入后通过原生 steer 提交去除首尾空白的正文。其 `RunCommandResult` 确认命令执行结束，不代表计划生成或审阅结束。需要 `RunResult` 时，先用 `/plan` 进入，再运行普通文本。组合输入应在 `run()` 前订阅，并保持问答处理器存活，通过 `session/wait` 观察后续工作，见[计划示例](examples/README.zh.md#review-a-plan)。计划指导和 `exit_plan_mode` 遵循原生生命周期，不修改权限设置。
+`/plan` 进入计划模式，`/plan off` 退出，`/plan <message>` 进入后通过原生 steer 提交去除首尾空白的正文。`run()` 等待提交的模型活动，包括原生审阅及获批后的后续执行；无需单独订阅或调用 `session/wait`。审阅期间保持 `on_user_question` 可用，见[计划示例](examples/README.zh.md#review-a-plan)。通过 `exit_plan_mode` 提交的计划位于审阅问题的 `detail` 中；`final_response` 包含最终助手文本。计划模式不修改权限设置。
 
-`/goal` 查看当前目标；`/goal <objective>`、`edit <objective>`、`pause`、`resume` 和 `clear` 使用原生处理器。创建和恢复可以在 `run()` 返回后启动自动轮次。发送前订阅，观察终态 `goal/change` 操作后，再调用 `session/wait` 等待可能的模型收尾。轮次之间的 idle 不代表目标完成。恢复的活动目标在显式 resume 前保持 disarmed。见[目标示例](examples/README.zh.md#run-a-goal)。
+`/goal` 显示当前目标。通过 `run()` 创建或恢复目标时，会跨越自动轮次等待，直到目标停止及最后的活动结束。完成返回 `completed`，暂停或清除返回 `aborted`，阻塞或失去激活状态报告失败。仅改变状态的控制命令不等待无关活动。恢复的活跃目标保持未激活状态，直到显式 resume。见[目标示例](examples/README.zh.md#run-a-goal)。
 
 <a id="client-approvals"></a>
 ## 客户端审批

python/sdk/examples/README.i18n.yaml

@@ -9,17 +9,17 @@
   en: ac09bb6bc3ab76a0
   zh: e69d1e2bd796f3bf
 /python-sdk-example/compact-a-full-sdk-session:
-  en: 7749773f97211ac1
-  zh: 594348ae66a2ef6c
+  en: 6325ac9aaf9b9b67
+  zh: 004a0ba5c2ff341b
 /python-sdk-example/change-session-permissions:
   en: 1fe5e06c6842a53b
   zh: ff0836d8cffaeef7
 /python-sdk-example/review-a-plan:
-  en: ce1851c8373d0f98
-  zh: 6bd4a5099f2b29e3
+  en: 7a6d2d7929724240
+  zh: 8a938c71a827f09b
 /python-sdk-example/run-a-goal:
-  en: f4748b8e074f3497
-  zh: 271aa49d20d83a73
+  en: 59f62e70dc508bd6
+  zh: bb10e2db241a2e92
 /python-sdk-example/add-plugins:
   en: 5e1e30dac13f6c9e
   zh: 0af98ebb75c88f43

python/sdk/examples/README.md

@@ -40,7 +40,7 @@ python python/sdk/examples/compact.py \
   --workspace /absolute/path/to/disposable-workspace
 ```
 
-The script prints command ids and outcomes separately from model responses. A failed turn or command stops the example. See [Run text and commands](../README.md#run-text-and-commands) for completion and error behavior; `minimal.py` uses the same `run()` entry, with capabilities supplied by its selected profile.
+The script prints `RunResult.final_response` for prompts and commands. An unsuccessful `finish_reason` stops the example. See [Run text and commands](../README.md#run-text-and-commands) for completion and error behavior; `minimal.py` uses the same `run()` entry, with capabilities supplied by its selected profile.
 
 <a id="change-session-permissions"></a>
 ## Change Session permissions
@@ -59,7 +59,7 @@ The command changes the Session's policy while the printed launch setting remain
 <a id="review-a-plan"></a>
 ## Review a plan
 
-The [plan example](plan.py) sends `/plan <task>` through `run()` and keeps the Session notification subscription and question handler alive after command completion. Enter `a` to approve, `k` to give feedback, or `d` to dismiss; cancellation or EOF before an answer dismisses the review. Approval allows implementation from a later model step, so choose an isolated workspace.
+The [plan example](plan.py) sends `/plan <task>` through `run()` and keeps the Session notification subscription and question handler alive within that call until plan activity finishes. Enter `a` to approve, `k` to give feedback, or `d` to dismiss; cancellation or EOF before an answer dismisses the review. Approval allows implementation from a later model step, so choose an isolated workspace.
 
 ```sh
 python python/sdk/examples/plan.py \
@@ -68,7 +68,7 @@ python python/sdk/examples/plan.py \
   "Plan the requested change and present it for review."
 ```
 
-The native plan is carried in `exit_plan_mode`, not written to a workspace document. Permissions remain workspace-write. Dismissal leaves plan mode active and ends the current turn. The example waits through `session/wait` and rejects truncated or failed root turns; its question callback is independent of the protocol reader. Ctrl+C stops the example even while it is waiting for input.
+The native plan is carried in `exit_plan_mode`, not written to a workspace document. Permissions remain workspace-write. Dismissal leaves plan mode active and ends the current turn. The example waits through `run()` and rejects truncated or failed root turns; its question callback is independent of the protocol reader. Ctrl+C stops the example even while it is waiting for input.
 
 Piped answers use the parent process's stdin encoding and leave later lines for subsequent questions. UTF-16 and UTF-32 input requires an explicit little-endian or big-endian encoding. Truncated characters at EOF are reported as input errors.
 
@@ -84,7 +84,7 @@ python python/sdk/examples/goal.py \
   "Calculate 17 * 19, verify the result, and complete the goal without changing files."
 ```
 
-`--timeout-seconds` requires a finite positive value and bounds the initial command wait and subsequent observation (300 seconds by default). A timeout reports failure and closes the runtime; it does not mark the goal complete. The persisted goal can be inspected after restoration and explicitly resumed. `session/wait` alone is insufficient because automatic rounds have idle gaps.
+`--timeout-seconds` requires a finite positive value and bounds the initial command wait and subsequent observation (300 seconds by default). A timeout reports failure and closes the runtime; it does not mark the goal complete. The persisted goal can be inspected after restoration and explicitly resumed. `run()` includes the automatic rounds and final reply.
 
 ## Add plugins
 

python/sdk/examples/README.zh.md

@@ -40,7 +40,7 @@ python python/sdk/examples/compact.py \
   --workspace /absolute/path/to/disposable-workspace
 ```
 
-脚本将命令 id 和结果与模型回复分开打印。轮次或命令失败时，示例会停止。完成和错误行为见[运行文本与命令](../README.zh.md#run-text-and-commands)；`minimal.py` 使用同一个 `run()` 入口，能力由所选 profile 提供。
+脚本对提示词和命令均打印 `RunResult.final_response`。不成功的 `finish_reason` 会停止示例。完成和错误行为见[运行文本与命令](../README.zh.md#run-text-and-commands)；`minimal.py` 使用同一个 `run()` 入口，能力由所选 profile 提供。
 
 <a id="change-session-permissions"></a>
 ## 修改 Session 权限
@@ -59,7 +59,7 @@ python python/sdk/examples/permission.py \
 <a id="review-a-plan"></a>
 ## 审查计划
 
-[计划示例](plan.py) 通过 `run()` 发送 `/plan <task>`，并在命令完成后保持 Session 通知订阅和问答处理器运行。输入 `a` 批准、`k` 提供反馈、`d` 关闭；取消或作答前遇到 EOF 会关闭审查。批准后模型可以从后续步骤开始实施，因此应选择隔离的工作区。
+[计划示例](plan.py) 通过 `run()` 发送 `/plan <task>`，并在这次调用中保持 Session 通知订阅和问答处理器运行，直到计划活动结束。输入 `a` 批准、`k` 提供反馈、`d` 关闭；取消或作答前遇到 EOF 会关闭审查。批准后模型可以从后续步骤开始实施，因此应选择隔离的工作区。
 
 ```sh
 python python/sdk/examples/plan.py \
@@ -68,7 +68,7 @@ python python/sdk/examples/plan.py \
   "Plan the requested change and present it for review."
 ```
 
-原生计划通过 `exit_plan_mode` 携带，而不是写入工作区文档。权限仍为 workspace-write。关闭审查会保留计划模式并结束当前轮次。示例通过 `session/wait` 等待，并报告根轮次截断或失败；问答回调独立于协议读取线程。等待输入时也可以通过 Ctrl+C 结束示例。
+原生计划通过 `exit_plan_mode` 携带，而不是写入工作区文档。权限仍为 workspace-write。关闭审查会保留计划模式并结束当前轮次。示例通过 `run()` 等待，并报告根轮次截断或失败；问答回调独立于协议读取线程。等待输入时也可以通过 Ctrl+C 结束示例。
 
 管道答案使用父进程的 stdin 编码，后续行保留给后续问题。UTF-16 和 UTF-32 输入须明确指定小端或大端编码。EOF 处截断的字符会作为输入错误报告。
 
@@ -84,7 +84,7 @@ python python/sdk/examples/goal.py \
   "Calculate 17 * 19, verify the result, and complete the goal without changing files."
 ```
 
-`--timeout-seconds` 必须是有限正数，用于限制首次命令等待及之后的观察时间，默认为 300 秒。超时会报告失败并关闭运行时，不会将目标标记为完成。恢复会话后可查看持久化目标，再显式恢复执行。自动轮次之间存在 idle 间隙，因此仅调用 `session/wait` 不足以等待目标结束。
+`--timeout-seconds` 必须是有限正数，用于限制首次命令等待及之后的观察时间，默认为 300 秒。超时会报告失败并关闭运行时，不会将目标标记为完成。恢复会话后可查看持久化目标，再显式恢复执行。`run()` 包含自动轮次及最后回复。
 
 ## 添加插件
 

python/sdk/examples/compact.py

@@ -7,21 +7,14 @@ import argparse
 import os
 from pathlib import Path
 
-from deepseek_harness import RunCommandResult, DeepSeekHarness, RunResult
+from deepseek_harness import DeepSeekHarness, RunResult
 
 
-def display(result: RunResult | RunCommandResult) -> None:
+def display(result: RunResult) -> None:
     """Print the input's result and stop on a reported failure."""
-    if isinstance(result, RunCommandResult):
-        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)
-    else:
-        if result.finish_reason != "completed":
-            raise RuntimeError(f"Agent turn ended with {result.finish_reason}: {result.final_response}")
-        print(result.final_response)
+    if result.finish_reason != "completed":
+        raise RuntimeError(f"Input ended with {result.finish_reason}: {result.final_response}")
+    print(result.final_response)
 
 
 def main() -> None:

python/sdk/examples/goal.py

@@ -6,19 +6,13 @@ from __future__ import annotations
 import argparse
 import math
 import os
-import time
 from pathlib import Path
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as FutureTimeoutError
 
-from pydantic import BaseModel
-
-from deepseek_harness import RunCommandResult, DeepSeekHarness, Notification
+from deepseek_harness import DeepSeekHarness, Notification
 from deepseek_harness.api import final_response
 
 
-class WaitResult(BaseModel):
-    """The empty session/wait result."""
-
-
 def display(notification: Notification) -> dict | None:
     """Print durable assistant output and return the observed root Session event."""
     if notification.method != "session.event":
@@ -49,43 +43,22 @@ def main() -> None:
     with DeepSeekHarness(
         dsh_home=str(args.dsh_home.resolve()), cwd=str(args.workspace.resolve()),
         profile="sdk", model=args.model, env={"DSH_PERMISSION_MODE": "workspace-write"},
-        request_timeout_seconds=args.timeout_seconds,
     ) as harness:
         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, RunCommandResult):
-                raise RuntimeError("The SDK profile did not execute the requested command")
-            print(execution.result.text, flush=True)
-            if execution.result.kind == "error":
-                raise RuntimeError(execution.result.text)
-            while True:
-                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",
-                }:
-                    break
-            # A terminal goal event can precede the model's closing reply.
-            harness.client.request(
-                "session/wait", {"sessionId": session.id}, response_model=WaitResult,
-                timeout_seconds=remaining_seconds(),
-                on_notification=observe, notification_subscription=notifications,
+        with ThreadPoolExecutor(max_workers=1) as worker:
+            running = worker.submit(
+                session.run, f"/goal {args.objective}",
+                on_notification=lambda notice: display(notice) if notice.payload.get("sessionId") == session.id else None,
             )
-            remaining_seconds()
+            try:
+                result = running.result(timeout=args.timeout_seconds)
+            except FutureTimeoutError as error:
+                raise TimeoutError("Goal observation timed out") from error
+            finally:
+                if not running.done():
+                    harness.close()
+        if result.finish_reason != "completed":
+            raise RuntimeError(f"Goal activity ended with {result.finish_reason}: {result.final_response}")
 
 
 if __name__ == "__main__":

python/sdk/examples/minimal.py

@@ -7,7 +7,7 @@ import argparse
 import os
 from pathlib import Path
 
-from deepseek_harness import RunCommandResult, DeepSeekHarness
+from deepseek_harness import DeepSeekHarness
 
 
 def main() -> None:
@@ -41,12 +41,9 @@ def main() -> None:
         profile=args.profile,
     ) as harness:
         result = harness.run(args.prompt, session_id=args.session_id)
-    if isinstance(result, RunCommandResult):
-        if result.result.kind == "error":
-            raise RuntimeError(result.result.text)
-        print(result.result.text or "")
-    else:
-        print(result.final_response)
+    if result.finish_reason not in (None, "completed"):
+        raise RuntimeError(f"Input ended with {result.finish_reason}: {result.final_response}")
+    print(result.final_response)
 
 
 if __name__ == "__main__":

python/sdk/examples/permission.py

@@ -7,21 +7,14 @@ import argparse
 import os
 from pathlib import Path
 
-from deepseek_harness import RunCommandResult, DeepSeekHarness, RunResult
+from deepseek_harness import DeepSeekHarness, RunResult
 
 
-def display(result: RunResult | RunCommandResult) -> None:
+def display(result: RunResult) -> None:
     """Print the input's result and stop on a reported failure."""
-    if isinstance(result, RunCommandResult):
-        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)
-    else:
-        if result.finish_reason != "completed":
-            raise RuntimeError(f"Agent turn ended with {result.finish_reason}: {result.final_response}")
-        print(result.final_response)
+    if result.finish_reason != "completed":
+        raise RuntimeError(f"Input ended with {result.finish_reason}: {result.final_response}")
+    print(result.final_response)
 
 
 def main() -> None:

python/sdk/examples/plan.py

@@ -8,19 +8,12 @@ import os
 import subprocess
 import sys
 import threading
-from contextlib import closing
 from pathlib import Path
 
-from pydantic import BaseModel
-
-from deepseek_harness import RunCommandResult, DeepSeekHarness, Notification, UserQuestionRequest
+from deepseek_harness import DeepSeekHarness, Notification, UserQuestionRequest
 from deepseek_harness.api import final_response
 
 
-class WaitResult(BaseModel):
-    """The empty session/wait result."""
-
-
 class _PlanReviews:
     """Own cancellable readers with the parent's stdin encoding and join review callbacks before exit."""
 
@@ -156,7 +149,7 @@ def display(notification: Notification, session_id: str) -> None:
 
 
 def main() -> None:
-    """Keep the runtime and question handler alive after the plan command completes."""
+    """Keep the runtime and question handler alive until the plan activity finishes."""
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--dsh-home", required=True, type=Path)
     parser.add_argument("--workspace", type=Path, default=Path.cwd())
@@ -171,17 +164,9 @@ def main() -> None:
         env={"DSH_PERMISSION_MODE": "workspace-write"},
     ) as harness:
         session = harness.start_session()
-        with closing(harness.client.subscribe_session_notifications(session.id)) as notifications:
-            execution = session.run(f"/plan {args.task}")
-            if not isinstance(execution, RunCommandResult):
-                raise RuntimeError("The SDK profile did not execute the requested command")
-            print(execution.result.text, flush=True)
-            if execution.result.kind == "error":
-                raise RuntimeError(execution.result.text)
-            harness.client.request(
-                "session/wait", {"sessionId": session.id}, response_model=WaitResult,
-                on_notification=lambda notice: display(notice, session.id), notification_subscription=notifications,
-            )
+        result = session.run(f"/plan {args.task}", on_notification=lambda notice: display(notice, session.id))
+        if result.finish_reason != "completed":
+            raise RuntimeError(f"Plan activity ended with {result.finish_reason}: {result.final_response}")
 
 
 if __name__ == "__main__":

python/sdk/src/deepseek_harness/__init__.py

@@ -1,4 +1,4 @@
-from .api import RunCommandResult, DeepSeekHarness, DeepSeekHarnessConfig, RunResult, Session
+from .api import DeepSeekHarness, DeepSeekHarnessConfig, RunResult, Session
 from .client import HarnessClient, HarnessConfig, UserQuestionHandler
 from .errors import SdkProtocolError
 from .models import (
@@ -12,7 +12,6 @@ __all__ = [
     "DeepSeekHarnessConfig",
     "Session",
     "RunResult",
-    "RunCommandResult",
     "CommandExecution",
     "CommandResult",
     "CommandSuccessResult",

python/sdk/src/deepseek_harness/api.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from typing import Callable, Literal
 
 from .client import HarnessClient, HarnessConfig, UserQuestionHandler
-from .errors import SdkProtocolError, TransportClosedError
+from .errors import SdkProtocolError
 from .models import CommandExecution, JsonObject, Notification
 
 
@@ -43,32 +43,35 @@ class DeepSeekHarnessConfig:
 
 @dataclass(slots=True)
 class RunResult:
+    """Completed input with result text, outcome, root events and Session-tree notifications."""
+
     session_id: str
     final_response: str
+    """Final assistant text (empty when absent), native state-only result text, or a failure diagnostic."""
     finish_reason: str | None
     events: list[JsonObject]
     notifications: list[Notification]
 
 
-class RunCommandResult(CommandExecution):
-    """Settled native command result with the Session id for continued interaction."""
+class _RunActivity:
+    """Collect one input's root events and Session-tree notifications from its admission."""
 
-    session_id: str
-
-
-class _PromptActivity:
-    """Collect an ordinary prompt's notifications from its inbox receipt through idle."""
-
-    def __init__(self, session_id: str, message_id: str) -> None:
+    def __init__(self, session_id: str, receipt_kind: Literal["prompt", "command"], receipt_id: str) -> None:
         self.session_id = session_id
-        self.message_id = message_id
+        self.receipt_kind = receipt_kind
+        self.receipt_id = receipt_id
         self.received = False
         self.events: list[JsonObject] = []
         self.notifications: list[Notification] = []
 
     def record(self, notification: Notification, on_notification: Callable[[Notification], None] | None = None) -> bool:
+        if notification.method == "session.query.completed":
+            return False
         if not self.received:
-            if not _is_inbox_receipt(notification, self.session_id, self.message_id):
+            if self.receipt_kind == "prompt":
+                if not _is_inbox_receipt(notification, self.session_id, self.receipt_id):
+                    return False
+            elif not _is_command_event(notification, self.session_id, self.receipt_id, "command/run"):
                 return False
             self.received = True
         self.notifications.append(notification)
@@ -80,10 +83,12 @@ class _PromptActivity:
                 self.events.append(event)
         return True
 
-    def result(self) -> RunResult:
+    def result(self, completed: JsonObject | None = None) -> RunResult:
         return RunResult(
-            session_id=self.session_id, final_response=final_response(self.events),
-            finish_reason=finish_reason(self.events), events=self.events, notifications=self.notifications,
+            session_id=self.session_id,
+            final_response=completed["finalResponse"] if completed is not None else final_response(self.events),
+            finish_reason=completed["finishReason"] if completed is not None else finish_reason(self.events),
+            events=self.events, notifications=self.notifications,
         )
 
 
@@ -180,8 +185,8 @@ class DeepSeekHarness:
         session_id: str | None = None,
         on_notification: Callable[[Notification], None] | None = None,
         delivery: Literal["queue", "steer"] = "queue",
-    ) -> RunResult | RunCommandResult:
-        """Run input on a fresh or named Session, returning its id with either result kind."""
+    ) -> RunResult:
+        """Run input to completion on a fresh or named Session and return its RunResult."""
         return self.start_session(session_id).run(input, on_notification=on_notification, delivery=delivery)
 
 
@@ -206,20 +211,21 @@ class Session:
         *,
         on_notification: Callable[[Notification], None] | None = None,
         delivery: Literal["queue", "steer"] = "queue",
-    ) -> RunResult | RunCommandResult:
+    ) -> RunResult:
         """Run text or content blocks through native input routing.
 
-        Input returns the prompt's receipt-to-idle RunResult or a settled
-        RunCommandResult. Both include this Session's id. Command completion does
-        not wait for later model work.
+        Ordinary prompts finish at the next idle after their inbox receipt.
+        State-only commands finish with their native outcome. Commands that
+        submit model input wait for that activity; goal creation and resume wait
+        through automatic rounds and the last reply after the goal stops.
         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.
+        wire order. Every result retains this Session id, its root events and
+        Session-tree notifications, including overlapping work. Native refusals
+        return finish_reason="error". 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")
@@ -227,20 +233,21 @@ class Session:
         with self.harness.client.subscribe_session_notifications(self.id) as subscription:
             response = self.harness.client._query(self.id, input, delivery)
             if isinstance(response, CommandExecution):
-                execution = response
+                activity = _RunActivity(self.id, "command", response.command_id)
                 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 RunCommandResult(session_id=self.id, commandId=execution.command_id, result=execution.result)
-            activity = _PromptActivity(self.id, response)
+                    notification = subscription.next()
+                    payload = notification.payload
+                    if notification.method == "session.query.completed" and payload.get("sessionId") == self.id:
+                        if not isinstance(payload.get("commandId"), str) or not payload["commandId"]:
+                            raise SdkProtocolError("Invalid session.query.completed command identity")
+                        if payload["commandId"] != response.command_id:
+                            continue
+                        if (not isinstance(payload.get("finalResponse"), str)
+                                or not isinstance(payload.get("finishReason"), str) or not payload["finishReason"]):
+                            raise SdkProtocolError("Invalid session.query.completed result")
+                        return activity.result(payload)
+                    activity.record(notification, on_notification)
+            activity = _RunActivity(self.id, "prompt", response)
             while True:
                 notification = subscription.next()
                 if not activity.record(notification, on_notification):
@@ -268,11 +275,11 @@ 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:
+def _is_command_event(notification: Notification, session_id: str, command_id: str, event_type: 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":
+    if not isinstance(event, dict) or event.get("type") != event_type:
         return False
     data = event.get("data")
     return isinstance(data, dict) and data.get("commandId") == command_id

python/sdk/src/deepseek_harness/client.py

@@ -705,15 +705,6 @@ class NotificationSubscription:
             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/tests/run_installed_commands.py

@@ -15,6 +15,7 @@ TEST_FILES = (
     "test_commands_runtime.py",
     "test_command_attachments_runtime.py",
     "test_goal_runtime.py",
+    "test_goal_result_text_runtime.py",
     "test_plan_runtime.py",
     "test_plan_example.py",
     "test_plan_input.py",

python/sdk/tests/test_command_attachments_runtime.py

@@ -12,7 +12,7 @@ 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,
+    _Completion, _SESSION_ID, _WAIT_SECONDS, _command_pair, _native_result, _of_type,
     _read_events, _request_texts, _runtime, _wait_activity,
 )
 from test_goal_runtime import _command, _until
@@ -66,8 +66,8 @@ def test_plan_image_reaches_the_model_and_persists_as_a_reference(
     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
+        assert isinstance(execution, RunResult)
+        assert execution.finish_reason == "completed" and execution.final_response == "IMAGE_PLAN_READY"
         _wait_activity(runtime, session)
         [request] = runtime.model.captured()
         _assert_request_images(request, picture)
@@ -95,8 +95,9 @@ def test_plan_off_compact_and_permission_reject_images_without_applying_the_comm
         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 isinstance(execution, RunResult)
+            assert execution.finish_reason == "error"
+            execution = _native_result(execution)
             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
@@ -121,10 +122,10 @@ def test_goal_create_and_edit_images_reach_following_requests_as_durable_referen
     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."}])
+        created = runtime.call(runtime.harness.client._query, session.id, [original, {"type": "text", "text": "/goal Inspect the original picture."}], "queue")
         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."}])
+        edited = runtime.call(runtime.harness.client._query, session.id, [revised, {"type": "text", "text": "/goal edit Inspect the revised picture."}], "queue")
         assert isinstance(edited, CommandExecution) and edited.result.kind == "success"
         _command(runtime, session, "/goal clear")
         assert not first.release.is_set()
@@ -152,6 +153,35 @@ def test_goal_create_and_edit_images_reach_following_requests_as_durable_referen
         }]
 
 
+def test_cleared_goal_waits_for_its_image_input_queued_during_compaction(tmp_path: Path) -> None:
+    picture = _image()
+    summary = _Completion("Retain the earlier task.", blocked=True)
+    reply = _Completion("REFERENCE_INPUT_REPLY", blocked=True)
+    with _runtime(tmp_path, [_Completion("EARLIER_REPLY"), summary, reply], model=_IMAGE_MODEL) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        runtime.call(session.run, "Retain this earlier task. " * 256)
+        compacting = runtime.workers.submit(session.run, "/compact")
+        assert summary.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        with runtime.harness.client.subscribe_session_notifications(session.id) as notices:
+            running = runtime.workers.submit(session.run, [
+                {"type": "text", "text": "/goal Inspect the reference."}, picture,
+            ])
+            _until(notices, "goal/change", "create")
+            cleared = session.run("/goal clear")
+            assert cleared.finish_reason == "completed" and cleared.final_response == "Goal cleared."
+            summary.release.set()
+            assert compacting.result(timeout=_WAIT_SECONDS).finish_reason == "completed"
+            assert reply.entered.wait(_WAIT_SECONDS), runtime.model.failures
+            assert not running.done()
+            _assert_request_images(runtime.model.captured()[-1], picture)
+            reply.release.set()
+            result = running.result(timeout=_WAIT_SECONDS)
+            assert result.finish_reason == "aborted" and result.final_response == "REFERENCE_INPUT_REPLY"
+    events = _read_events(runtime.logs)
+    _command_pair(events, result, "goal")
+    assert [event["data"]["operation"] for event in _of_type(events, "goal/change")] == ["create", "clear"]
+
+
 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:
@@ -162,7 +192,8 @@ def test_goal_controls_with_images_leave_the_active_request_and_goal_unchanged(t
             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"
+                assert isinstance(execution, RunResult) and execution.finish_reason == "error"
+                execution = _native_result(execution)
                 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
@@ -238,7 +269,8 @@ def test_invalid_image_batch_cannot_enter_plan_mode_or_deliver_a_partial_prompt(
             ])
         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 isinstance(rejected, RunResult) and rejected.finish_reason == "error"
+        rejected = _native_result(rejected)
         assert "base64" in rejected.result.text
         with pytest.raises(JsonRpcError, match="base64"):
             runtime.call(session.run, [*images, {"type": "text", "text": "Inspect the pictures."}])

python/sdk/tests/test_commands_runtime.py

@@ -302,7 +302,20 @@ 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, name: str = "compact") -> None:
+def _native_result(result: RunResult) -> CommandExecution:
+    """Read the native outcome retained in a high-level result's command lifecycle events."""
+    started = next(event for event in result.events if event["type"] == "command/run")
+    identity = started["data"]["commandId"]
+    done = next(event["data"] for event in result.events
+                if event["type"] == "command/done" and event["data"]["commandId"] == identity)
+    return CommandExecution.model_validate({"commandId": identity, "result": {
+        key: value for key, value in done.items() if key in ("kind", "text", "sourceEventSeq")
+    }})
+
+
+def _command_pair(events: list[dict], execution: CommandExecution | RunResult, name: str = "compact") -> None:
+    if isinstance(execution, RunResult):
+        execution = _native_result(execution)
     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")
@@ -348,7 +361,8 @@ def test_compact_without_history_returns_without_a_model_turn(tmp_path: Path) ->
             raise
         finally:
             observer.close()
-        assert isinstance(execution, CommandExecution)
+        assert isinstance(execution, RunResult)
+        execution = _native_result(execution)
         assert done["commandId"] == execution.command_id
         assert execution.result.kind == "success"
         assert execution.result.source_event_seq is None
@@ -370,7 +384,8 @@ def test_compact_rejects_busy_agent_before_the_model_is_released(tmp_path: Path)
         assert response.entered.wait(_WAIT_SECONDS), runtime.model.failures
         assert not running.done() and not response.release.is_set()
         execution = runtime.call(session.run, "/compact", delivery="queue")
-        assert isinstance(execution, CommandExecution)
+        assert isinstance(execution, RunResult)
+        execution = _native_result(execution)
         assert execution.result.kind == "error"
         assert not response.release.is_set()
         assert not running.done()
@@ -405,7 +420,8 @@ def run_compaction_scenario(tmp_path: Path, *, capture_patch: Path | None = None
         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 isinstance(execution, RunResult)
+        execution = _native_result(execution)
         assert execution.result.kind == "success"
         assert execution.result.source_event_seq is not None
 
@@ -526,8 +542,8 @@ def _permission_facts(events: list[dict]) -> list[dict]:
 
 def _permission_command(runtime: _Runtime, session: Session, preset: str) -> CommandExecution:
     result = runtime.call(session.run, f"/permission {preset}")
-    assert isinstance(result, CommandExecution)
-    return result
+    assert isinstance(result, RunResult)
+    return _native_result(result)
 
 
 def test_permission_changes_only_the_selected_session_and_retains_prompt_semantics(tmp_path: Path) -> None:

python/sdk/tests/test_goal_example.py

@@ -5,9 +5,9 @@ from __future__ import annotations
 import json
 import runpy
 import sys
+import threading
 from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
-from types import SimpleNamespace
 
 import pytest
 
@@ -17,8 +17,8 @@ from deepseek_harness import DeepSeekHarness
 _EXAMPLE = Path(__file__).resolve().parents[1] / "examples/goal.py"
 
 
-@pytest.mark.parametrize("pending", ["command", "goal-event", "wrapup", "complete", "goal-queue", "wrapup-queue"])
-def test_goal_example_bounds_pending_work_and_reaps_runtime(
+@pytest.mark.parametrize("pending", ["command", "activity", "complete", "error", "interrupt"])
+def test_goal_example_bounds_the_whole_run_and_reaps_runtime(
     tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str], pending: str,
 ) -> None:
     peer = tmp_path / "goal-peer.py"
@@ -41,26 +41,25 @@ for line in sys.stdin:
         send({"id": frame["id"], "result": {"serverInfo": {"name": "goal-example-peer"}}})
     elif method == "session/query":
         assert frame["params"]["contentBlocks"] == [{"type": "text", "text": "/goal Finish this task"}]
-        session = frame["params"]["sessionId"]
         if pending == "command":
             continue
-        if pending in ("wrapup", "complete", "wrapup-queue"):
+        session = frame["params"]["sessionId"]
+        for kind in ("command/run", "command/done"):
             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}}}})
+                "type": kind, "data": {"commandId": "goal-command"}}}})
         send({"id": frame["id"], "result": {"kind": "command", "execution": {
             "commandId": "goal-command", "result": {"kind": "success", "text": "Goal set"}}}})
-    elif method == "session/wait":
-        if pending == "wrapup":
+        if pending in ("activity", "interrupt"):
             continue
-        for _ in range(2 if pending == "wrapup-queue" else 1):
+        if pending == "complete":
+            send({"method": "session.event", "params": {"sessionId": session, "event": {
+                "type": "goal/change", "data": {"operation": "complete"}}}})
             send({"method": "session.event", "params": {"sessionId": session, "event": {
                 "type": "assistant/message", "data": {"message": {
                     "content": [{"type": "text", "text": "Closing reply"}]}}}}})
-        send({"id": frame["id"], "result": {}})
+        send({"method": "session.query.completed", "params": {"sessionId": session,
+            "commandId": "goal-command", "finalResponse": "Closing reply" if pending == "complete" else "Goal stopped",
+            "finishReason": "completed" if pending == "complete" else "error"}})
     elif method == "shutdown":
         send({"id": frame["id"], "result": {}})
         break
@@ -79,49 +78,47 @@ for line in sys.stdin:
         return harness
 
     monkeypatch.setitem(example["main"].__globals__, "DeepSeekHarness", configured_harness)
-    queued = pending in {"goal-queue", "wrapup-queue"}
-    displayed = []
-    if queued:
-        clock = [0.0]
+    if pending == "interrupt":
+        entered = threading.Event()
         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"
+        def observe(notice):
+            if notice.method == "session.event" and notice.payload["event"]["type"] == "command/done":
+                entered.set()
+            return display(notice)
+        class InterruptedExecutor(ThreadPoolExecutor):
+            def submit(self, *args, **kwargs):
+                future = super().submit(*args, **kwargs)
+                def interrupt(timeout=None):
+                    assert entered.wait(10), "command must be observed before interrupting its wait"
+                    raise KeyboardInterrupt
+                monkeypatch.setattr(future, "result", interrupt)
+                return future
+        monkeypatch.setitem(example["main"].__globals__, "display", observe)
+        monkeypatch.setitem(example["main"].__globals__, "ThreadPoolExecutor", InterruptedExecutor)
+    timeout = "1" if pending in {"command", "activity"} else "10"
     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.
     with ThreadPoolExecutor(max_workers=1) as workers:
         execution = workers.submit(example["main"])
         try:
-            if pending == "complete":
-                execution.result(timeout=30)
-            else:
-                expected = {"command": "session/query timed out", "goal-event": "SDK notification",
-                            "wrapup": "session/wait timed out", "goal-queue": "Goal observation timed out",
-                            "wrapup-queue": "Goal observation timed out"}[pending]
-                with pytest.raises(TimeoutError, match=expected):
+            if pending in {"command", "activity"}:
+                with pytest.raises(TimeoutError, match="Goal observation timed out"):
                     execution.result(timeout=30)
+            elif pending == "interrupt":
+                with pytest.raises(KeyboardInterrupt):
+                    execution.result(timeout=30)
+            elif pending == "error":
+                with pytest.raises(RuntimeError, match="Goal stopped"):
+                    execution.result(timeout=30)
+            else:
+                execution.result(timeout=30)
         finally:
             for harness in harnesses:
                 harness.close()
     [harness] = harnesses
     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", "wrapup-queue"} else []), "shutdown"]
-    if queued:
-        assert displayed == (["goal/change"] if pending == "goal-queue" else ["goal/change", "assistant/message"])
+    assert [frame["method"] for frame in frames] == ["initialize", "session/query", "shutdown"]
     if pending == "complete":
         output = capsys.readouterr().out
         assert "Goal: complete" in output

python/sdk/tests/test_goal_result_text_runtime.py

@@ -0,0 +1,29 @@
+"""Goal run results preserve empty assistant text through the shipped SDK profile."""
+
+from pathlib import Path
+
+from test_commands_runtime import _Completion, _runtime, _SESSION_ID
+from test_goal_runtime import _completion_steps
+
+
+def test_completed_goal_without_assistant_text_returns_empty_result(tmp_path: Path) -> None:
+    with _runtime(tmp_path, _completion_steps(_Completion("")),
+                  initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        result = runtime.call(session.run, "/goal Finish the task.")
+        assert result.final_response == "" and result.finish_reason == "completed"
+        assert any(event["type"] == "goal/change" and event["data"]["operation"] == "complete"
+                   for event in result.events)
+        cleared = runtime.call(session.run, "/goal clear")
+        assert cleared.final_response == "Goal cleared." and cleared.finish_reason == "completed"
+        assert len(runtime.model.captured()) == 3
+
+
+def test_blocked_goal_without_assistant_text_returns_empty_result(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [_Completion("")], max_goal_rounds=1) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        result = runtime.call(session.run, "/goal Finish the task.")
+        assert result.final_response == "" and result.finish_reason == "error"
+        assert any(event["type"] == "goal/change" and event["data"]["goal"]["phase"] == "blocked"
+                   for event in result.events)
+        assert len(runtime.model.captured()) == 1

python/sdk/tests/test_goal_runtime.py

@@ -9,7 +9,7 @@ import threading
 import pytest
 from pathlib import Path
 
-from deepseek_harness import CommandExecution, Session
+from deepseek_harness import CommandExecution, RunResult, Session
 from deepseek_harness.client import NotificationSubscription
 from test_commands_runtime import (
     _Completion, _Runtime, _SESSION_ID, _WAIT_SECONDS, _WaitResult, _assert_request_permission,
@@ -19,7 +19,7 @@ from test_commands_runtime import (
 
 
 def _command(runtime: _Runtime, session: Session, text: str, *, kind: str = "success") -> CommandExecution:
-    result = runtime.call(session.run, text)
+    result = runtime.call(runtime.harness.client._query, session.id, text, "queue")
     assert isinstance(result, CommandExecution)
     assert result.result.kind == kind, result.result.text
     return result
@@ -64,61 +64,31 @@ def test_goal_reads_and_invalid_controls_do_not_start_model_work(tmp_path: Path)
         _command_pair(events, execution, "goal")
 
 
-def test_goal_continues_rounds_and_finishes_wrapup_after_completion_event(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+@pytest.mark.parametrize("clear_before_reply", [False, True])
+def test_goal_continues_rounds_and_finishes_wrapup_after_completion_event(
+    tmp_path: Path, clear_before_reply: bool,
+) -> None:
     final = _Completion("GOAL_DONE", blocked=True)
     with _runtime(tmp_path, [_Completion("PROGRESS"), *_completion_steps(final)],
                   initial_permission="workspace-write", max_goal_rounds=3) as runtime:
         session = runtime.harness.start_session(_SESSION_ID)
         with runtime.harness.client.subscribe_session_notifications(session.id) as events:
-            execution = _command(runtime, session, "/目标 Verify the small change.")
+            running = runtime.workers.submit(session.run, "/目标 Verify the small change.")
             completed = _until(events, "goal/change", "complete")
             assert completed["goal"]["phase"] == "complete" and completed["roundsStarted"] == 2
             assert final.entered.wait(_WAIT_SECONDS)
-            client = runtime.harness.client
-            sent = threading.Event()
-            replied = threading.Event()
-            wait_ids = set()
-            write = client._write_message
-            handle = client._handle_message
-
-            def record_send(message: dict) -> None:
-                is_wait = message.get("method") == "session/wait"
-                if is_wait:
-                    wait_ids.add(message["id"])
-                write(message)
-                if is_wait:
-                    sent.set()
-
-            def record_response(message: object) -> None:
-                if isinstance(message, dict) and "method" not in message and message.get("id") in wait_ids:
-                    replied.set()
-                handle(message)
-
-            monkeypatch.setattr(client, "_write_message", record_send)
-            monkeypatch.setattr(client, "_handle_message", record_response)
-            waiting = runtime.workers.submit(client.request, "session/wait", {"sessionId": session.id}, response_model=_WaitResult)
-            assert sent.wait(_WAIT_SECONDS)
-            # Observe the reader directly so caller scheduling cannot hide an early response.
-            runtime.call(client.get_working_directory, session.id)
-            assert not replied.is_set() and not final.release.is_set()
+            assert not running.done()
+            if clear_before_reply:
+                cleared = runtime.call(session.run, "/goal clear")
+                assert cleared.final_response == "Goal cleared." and cleared.finish_reason == "completed"
+                assert not running.done()
             final.release.set()
-            assert waiting.result(timeout=_WAIT_SECONDS).model_dump() == {}
-            assert _until(events, "turn/end")["turn"] == 2
-        requests = runtime.model.captured()
-        assert len(requests) == 4
-        assert 'Objective: "Verify the small change."\nRound: 1/3' in "\n".join(_request_texts(requests[0]))
-        assert 'Round: 2/3' in "\n".join(_request_texts(requests[1]))
-        assert '<goal_complete>' in "\n".join(_request_texts(requests[-1]))
-        for request in requests:
-            _assert_request_permission(request, "workspace-write", "ask")
-        assert runtime.harness.config.env["DSH_PERMISSION_MODE"] == "workspace-write"
-    events = _read_events(runtime.logs)
-    _command_pair(events, execution, "goal")
-    assert [e["data"]["turn"] for e in _of_type(events, "turn/start")] == [1, 2]
-    assert [e["data"] for e in _of_type(events, "step/start")] == [
-        {"turn": 1, "step": 1}, {"turn": 2, "step": 1}, {"turn": 2, "step": 2}, {"turn": 2, "step": 3},
-    ]
-    assert len(_permission_facts(events)) == 3
+            result = running.result(timeout=_WAIT_SECONDS)
+            assert isinstance(result, RunResult)
+            assert result.final_response == "GOAL_DONE" and result.finish_reason == "completed"
+            operations = ["create", "complete", *(["clear"] if clear_before_reply else [])]
+            assert [event["data"]["operation"] for event in result.events if event["type"] == "goal/change"] == operations
+    _command_pair(_read_events(runtime.logs), result, "goal")
 
 
 def test_pause_aborts_then_edit_and_resume_use_revised_goal_without_changing_other_sessions(tmp_path: Path) -> None:

python/sdk/tests/test_plan_runtime.py

@@ -7,10 +7,10 @@ from pathlib import Path
 
 import pytest
 
-from deepseek_harness import CommandExecution, RunCommandResult, RunResult, Session, UserQuestionRequest
+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,
+    _command_pair, _native_result, _of_type, _permission_facts, _read_events,
     _request_texts, _runtime, _text_blocks, _wait_activity,
 )
 
@@ -20,7 +20,7 @@ _GUIDANCE = "You are in plan mode."
 
 
 def _command(runtime: _Runtime, session: Session, text: str) -> CommandExecution:
-    result = runtime.call(session.run, text)
+    result = runtime.call(runtime.harness.client._query, session.id, text, "queue")
     assert isinstance(result, CommandExecution)
     assert result.result.kind == "success", result.result.text
     return result
@@ -217,14 +217,14 @@ def test_plan_state_survives_resume_without_changing_permissions(tmp_path: Path)
 def test_top_level_plan_result_identifies_the_session_for_wait_continue_and_resume(tmp_path: Path) -> None:
     first = _Completion("PLAN_READY", blocked=True)
     with _runtime(tmp_path, [first, _Completion("CONTINUED_PLAN")], initial_permission="workspace-write") as runtime:
-        execution = runtime.call(runtime.harness.run, "/plan Design this change without tools.")
-        assert isinstance(execution, RunCommandResult)
-        assert execution.result.kind == "success"
+        running = runtime.workers.submit(runtime.harness.run, "/plan Design this change without tools.")
         assert first.entered.wait(_WAIT_SECONDS), runtime.model.failures
-        assert not first.release.is_set()
-        session = runtime.harness.start_session(execution.session_id)
+        assert not running.done()
         first.release.set()
-        _wait_activity(runtime, session)
+        execution = running.result(timeout=_WAIT_SECONDS)
+        assert isinstance(execution, RunResult)
+        assert execution.finish_reason == "completed" and execution.final_response == "PLAN_READY"
+        session = runtime.harness.start_session(execution.session_id)
         continued = runtime.call(runtime.harness.run, "Continue this plan without tools.", session_id=execution.session_id)
         assert isinstance(continued, RunResult)
         assert continued.session_id == execution.session_id

python/sdk/tests/test_run.py

@@ -10,7 +10,7 @@ from concurrent.futures import ThreadPoolExecutor
 import pytest
 from pydantic import BaseModel, ValidationError
 
-from deepseek_harness import CommandExecution, RunCommandResult, DeepSeekHarness, Notification, RunResult, SdkProtocolError
+from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, RunResult, SdkProtocolError
 from deepseek_harness.errors import JsonRpcError
 
 
@@ -28,6 +28,7 @@ import sys
 serial = 0
 pending_commands = []
 pending_prompts = []
+pending_plans = []
 
 def send(frame):
     print(json.dumps({"jsonrpc": "2.0", **frame}), flush=True)
@@ -43,6 +44,8 @@ def command(frame, number):
     identity = "command-" + str(number)
     event(session, "command/run", {"commandId": identity})
     event(session, "command/done", {"commandId": identity, "kind": "success"})
+    send({"method": "session.query.completed", "params": {"sessionId": session, "commandId": identity,
+        "finalResponse": "command completed", "finishReason": "completed"}})
     send({"id": frame["id"], "result": {"kind": "command", "execution": {
         "commandId": identity, "result": {"kind": "success", "text": "command completed"}}}})
 
@@ -71,17 +74,33 @@ for line in sys.stdin:
         assert set(params) == {"sessionId", "contentBlocks", "delivery"}
         assert params["delivery"] in ("queue", "steer")
         text = input_text(frame)
-        if text == "hold-command":
+        if text == "/plan hold":
+            identity = "plan-" + str(serial)
+            event(params["sessionId"], "command/run", {"commandId": identity})
+            event(params["sessionId"], "command/done", {"commandId": identity, "kind": "success"})
+            send({"id": frame["id"], "result": {"kind": "command", "execution": {
+                "commandId": identity, "result": {"kind": "success", "text": "Plan enabled"}}}})
+            pending_plans.append((frame, identity))
+        elif text == "hold-command":
             pending_commands.append((frame, serial))
         elif text in ("/compact", "/plan Describe this image."):
             command(frame, serial)
         elif text == "silent-command":
+            send({"method": "session.query.completed", "params": {"sessionId": params["sessionId"],
+                "commandId": "silent", "finalResponse": "", "finishReason": "completed"}})
             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 == "custom-response":
-            send({"id": frame["id"], "result": json.loads(os.environ["QUERY_RESPONSE"])})
+            response = json.loads(os.environ["QUERY_RESPONSE"])
+            execution = response.get("execution") if isinstance(response, dict) else None
+            if isinstance(execution, dict) and isinstance(execution.get("result"), dict):
+                send({"method": "session.query.completed", "params": {"sessionId": params["sessionId"],
+                    "commandId": execution.get("commandId"), "finalResponse": execution["result"].get("text", ""),
+                    "finishReason": "error" if execution["result"].get("kind") == "error" else "completed",
+                    **json.loads(os.environ.get("QUERY_COMPLETION", "{}"))}})
+            send({"id": frame["id"], "result": response})
         elif text == "bad-route":
             send({"id": frame["id"], "result": {"kind": "unknown"}})
         elif text == "bad-execution":
@@ -100,6 +119,11 @@ for line in sys.stdin:
                 else:
                     finish(frame)
     elif method == "release":
+        for original, identity in pending_plans:
+            finish(original)
+            send({"method": "session.query.completed", "params": {"sessionId": original["params"]["sessionId"],
+                "commandId": identity, "finalResponse": "Generated plan", "finishReason": "completed"}})
+        pending_plans.clear()
         for original, number in pending_commands:
             command(original, number)
         pending_commands.clear()
@@ -145,11 +169,11 @@ def test_run_returns_command_without_waiting_for_idle(sdk: DeepSeekHarness, entr
     notifications = []
     result = (sdk.run("/compact", session_id="main", on_notification=notifications.append)
               if entry == "harness" else sdk.start_session("main").run("/compact", on_notification=notifications.append))
-    assert isinstance(result, CommandExecution)
-    assert isinstance(result, RunCommandResult)
+    assert isinstance(result, RunResult)
+    assert isinstance(result, RunResult)
     assert result.session_id == "main"
-    assert result.result.kind == "success"
-    assert [notice.payload["event"]["type"] for notice in notifications] == ["command/run", "command/done"]
+    assert result.finish_reason == "completed"
+    assert [notice.payload["event"]["type"] for notice in notifications if notice.method == "session.event"] == ["command/run", "command/done"]
     assert_released(sdk)
 
 
@@ -158,7 +182,7 @@ def test_top_level_command_results_identify_distinct_sessions_and_allow_continui
 ) -> None:
     first = sdk.run("/compact")
     second = sdk.run("/compact")
-    assert isinstance(first, RunCommandResult) and isinstance(second, RunCommandResult)
+    assert isinstance(first, RunResult) and isinstance(second, RunResult)
     assert first.session_id and second.session_id and first.session_id != second.session_id
     continued = sdk.run("continue the first session", session_id=first.session_id)
     assert isinstance(continued, RunResult)
@@ -185,11 +209,10 @@ def test_run_adds_session_identity_without_changing_command_results(peer: Path,
         })},
     ) as sdk:
         returned = sdk.run("custom-response")
-        assert isinstance(returned, RunCommandResult)
+        assert isinstance(returned, RunResult)
         assert returned.session_id
-        assert returned.model_dump(by_alias=True, exclude_none=True) == {
-            **execution, "session_id": returned.session_id,
-        }
+        assert returned.final_response == result.get("text", "")
+        assert returned.finish_reason == ("error" if result["kind"] == "error" else "completed")
         low_level = sdk.client._query(returned.session_id, "custom-response", "queue")
         assert isinstance(low_level, CommandExecution)
         assert low_level.model_dump(by_alias=True, exclude_none=True) == execution
@@ -236,8 +259,8 @@ def test_run_forwards_content_blocks_with_the_routed_result_and_notifications(
         )
     )
     if text == "/plan Describe this image.":
-        assert isinstance(result, CommandExecution)
-        assert [notice.payload["event"]["type"] for notice in notifications] == ["command/run", "command/done"]
+        assert isinstance(result, RunResult)
+        assert [notice.payload["event"]["type"] for notice in notifications if notice.method == "session.event"] == ["command/run", "command/done"]
     else:
         assert isinstance(result, RunResult)
         assert result.final_response == "answer for " + text
@@ -267,7 +290,7 @@ def test_concurrent_runs_keep_their_own_results(sdk: DeepSeekHarness) -> None:
         try:
             assert entered.wait(5), "the first run must observe its receipt"
             second = workers.submit(session.run, "/compact")
-            assert isinstance(second.result(timeout=5), CommandExecution)
+            assert isinstance(second.result(timeout=5), RunResult)
         finally:
             release(sdk)
         result = first.result(timeout=5)
@@ -279,7 +302,7 @@ def test_concurrent_runs_keep_their_own_results(sdk: DeepSeekHarness) -> None:
 def test_command_result_needs_no_notifications(sdk: DeepSeekHarness) -> None:
     notices = []
     result = sdk.start_session("main").run("silent-command", on_notification=notices.append)
-    assert isinstance(result, CommandExecution)
+    assert isinstance(result, RunResult)
     assert notices == []
     assert_released(sdk)
 
@@ -290,7 +313,7 @@ def test_run_reports_routing_failures_without_retry(sdk: DeepSeekHarness, text:
     with pytest.raises(error):
         session.run(text)
     assert_released(sdk)
-    assert isinstance(session.run("/compact"), CommandExecution)
+    assert isinstance(session.run("/compact"), RunResult)
 
 
 @pytest.mark.parametrize("text", ["ordinary", "/compact"])
@@ -310,7 +333,7 @@ def test_command_timeout_keeps_runtime_responsive(sdk: DeepSeekHarness) -> None:
     sdk.client.config.request_timeout_seconds = 5
     assert_released(sdk)
     release(sdk)
-    assert isinstance(sdk.start_session("main").run("/compact"), CommandExecution)
+    assert isinstance(sdk.start_session("main").run("/compact"), RunResult)
 
 
 def test_run_requires_supported_delivery(sdk: DeepSeekHarness) -> None:
@@ -338,4 +361,40 @@ def test_run_rejects_invalid_routing_responses_as_protocol_errors(peer: Path, tm
         if isinstance(response, dict) and response.get("kind") == "command":
             assert isinstance(raised.value.__cause__, ValidationError)
         assert_released(sdk)
-        assert isinstance(sdk.start_session("main").run("/compact"), CommandExecution)
+        assert isinstance(sdk.start_session("main").run("/compact"), RunResult)
+
+
+def test_plan_run_waits_for_model_output_after_the_command_receipt(sdk: DeepSeekHarness) -> None:
+    admitted = threading.Event()
+    session = sdk.start_session("plan")
+    def observe(notice: Notification) -> None:
+        if notice.method == "session.event" and notice.payload["event"]["type"] == "command/done":
+            admitted.set()
+    with ThreadPoolExecutor(max_workers=1) as workers:
+        running = workers.submit(session.run, "/plan hold", on_notification=observe)
+        try:
+            assert admitted.wait(5)
+            assert not running.done()
+        finally:
+            release(sdk)
+        result = running.result(timeout=5)
+    assert isinstance(result, RunResult)
+    assert result.final_response == "Generated plan" and result.finish_reason == "completed"
+    assert_released(sdk)
+
+
+@pytest.mark.parametrize("completion", [{"commandId": None}, {"finalResponse": 1}, {"finishReason": None}, {"finishReason": ""}])
+def test_run_rejects_malformed_completion_and_releases_its_subscription(
+    peer: Path, tmp_path: Path, completion: dict,
+) -> None:
+    with DeepSeekHarness(
+        cwd=str(tmp_path), dsh_home=str(tmp_path / "home"),
+        _launch_args=(sys.executable, str(peer)),
+        env={"REQUEST_LOG": str(tmp_path / "requests.jsonl"),
+             "QUERY_COMPLETION": json.dumps(completion),
+             "QUERY_RESPONSE": json.dumps({"kind": "command", "execution": {
+                 "commandId": "c", "result": {"kind": "success"}}})},
+    ) as sdk:
+        with pytest.raises(SdkProtocolError):
+            sdk.run("custom-response")
+        assert_released(sdk)