feat(sdk): support native plan commands and review dismissal

python/sdk/README.i18n.yaml

@@ -12,8 +12,8 @@
   en: c4c30913fe2642ef
   zh: 0ed42bca37fba60b
 /deepseek-harness-python-sdk/query-and-response:
-  en: a3f6c4ee867928a8
-  zh: f5af634f4fcdc547
+  en: 1348c26311dee604
+  zh: 9290f1d66da1781c
 /deepseek-harness-python-sdk/client-approvals:
   en: fc69be1beb8abff1
   zh: 91b56781931a6f68
@@ -21,8 +21,8 @@
   en: 69a6a6745c6e89c1
   zh: 85e82a4cc31e16e5
 /deepseek-harness-python-sdk/native-user-questions:
-  en: d9c60c5f43755782
-  zh: 78c82635209ed22d
+  en: 979ad00fe498b2be
+  zh: 4b656bd268292a13
 /deepseek-harness-python-sdk/results-and-notifications:
   en: d4f7366c2d260639
   zh: 7de99aed2b1bdd0a

python/sdk/README.md

@@ -61,16 +61,18 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 <a id="query-and-response"></a>
 ## Query and response
 
-Call `session.query(text)` to send text, then iterate `session.receive_response()` to observe its result. `query()` returns `None` after sending, without waiting for runtime routing or execution. The full SDK intercepts bare `/compact` and `/permission <preset>` with a non-blank preset, including their registered Desktop aliases. Bare `/permission` or `/权限`, `/plan`, `/model` and other input remain ordinary prompts. This API accepts text strings only. `delivery="queue"` is the default; `delivery="steer"` directs an ordinary prompt to the next step. Command timing remains owned by the command.
+Call `session.query(text)` to send text, then iterate `session.receive_response()` to observe its result. `query()` returns `None` after sending, without waiting for runtime routing or execution. The full SDK intercepts bare `/compact`, `/permission <preset>` with a non-blank preset, and `/plan [off|message]`, including their registered Desktop aliases. Bare `/permission` or `/权限`, `/model` and other input remain ordinary prompts. This API accepts text strings only. `delivery="queue"` is the default; `delivery="steer"` directs an ordinary prompt to the next step. Command timing remains owned by the command.
 
 `receive_response()` claims the next query in send order. It first waits for the routing response, then yields `Notification` objects followed by one terminal value: `RunResult` for the prompt's inbox-receipt-to-idle interval, or `CommandExecution` after the command settles. Notifications can accumulate before receiving begins. Separate receivers can observe their queries concurrently, and another `query()` can be sent while a receiver waits. Close an unfinished iterator to release only its local observation; this does not cancel or roll back runtime work. Resources are released before yielding the terminal value.
 
 `CommandExecution` contains `command_id` and a `result` discriminated by `kind`. Success may include `text` and `source_event_seq`; an error includes `text`. Handler errors return `kind="error"`; routing, protocol and transport failures raise during receiving. Sending failures raise from `query()`. `request_timeout_seconds` bounds waiting for the routing response, not later prompt activity; an already-arrived response remains receivable after that deadline.
 
-The full `sdk` profile provides both commands; the default `sdk-minimal` profile does not provide command execution. `/compact` can make an auxiliary model request and completes without requiring a normal turn or idle notification; empty history can succeed without a summary. `run()` retains ordinary prompt semantics for slash text. The lower-level blocking `HarnessClient.session_command(session_id, line)` strictly accepts `/compact` or `/permission <preset>` and rejects bare `/permission`. See the [compaction example](examples/README.md#compact-a-full-sdk-session).
+The full `sdk` profile provides these commands; the default `sdk-minimal` profile does not provide command execution. `/compact` can make an auxiliary model request and completes without requiring a normal turn or idle notification; empty history can succeed without a summary. `run()` retains ordinary prompt semantics for slash text. The lower-level blocking `HarnessClient.session_command(session_id, line)` strictly accepts `/compact`, `/permission <preset>`, or `/plan [off|message]` and rejects bare `/permission`. See the [compaction example](examples/README.md#compact-a-full-sdk-session).
 
 `/permission <preset>` delegates to the native permission handler. An unknown preset returns `CommandExecution.result.kind="error"` without changing permission state. A successful change belongs to that Session: later queries and restoration retain it, other Sessions retain their own state, and new Sessions use configured defaults. `DeepSeekHarnessConfig` remains startup input, not a shared live state mirror; the command does not rewrite it or saved user defaults. 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 `CommandExecution` confirms command completion, not completion of planning or review. To collect a normal activity result, enter with `/plan` then submit a separate ordinary query. For combined input, subscribe before sending and use `session/wait` while keeping the question handler alive, as in the [plan example](examples/README.md#review-a-plan). Plan mode changes guidance and progressively admits `exit_plan_mode`; it does not change permission settings. The default guidance prohibits file edits, including plan documents, while execution remains governed by the sandbox.
+
 <a id="client-approvals"></a>
 ## Client approvals
 
@@ -123,7 +125,7 @@ The shipped `sdk-minimal` profile is a standalone explicit tree rather than an o
 
 ## Native user questions
 
-Pass `on_user_question` to `DeepSeekHarnessConfig`, `DeepSeekHarness`, or `HarnessConfig` to answer the built-in `ask_user_question` tool through the native SDK protocol. The callback receives a `UserQuestionRequest` (`request_id`, `session_id`, and the original `questions`) and a `threading.Event` signalling cancellation. Return `UserQuestionAnswer` or a JSON object shaped as `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}`. Each original question id must appear exactly once. The application supplies its UI or simulated-user policy; the SDK does not choose answers.
+Pass `on_user_question` to `DeepSeekHarnessConfig`, `DeepSeekHarness`, or `HarnessConfig` to answer the built-in `ask_user_question` tool through the native SDK protocol. The callback receives a `UserQuestionRequest` (`request_id`, `session_id`, and the original `questions`) and a `threading.Event` signalling cancellation. Return `UserQuestionAnswer` or a JSON object shaped as `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}`. Each original question id must appear exactly once. Return `{ "cancelled": true }` to dismiss instead of answering; this preserves the native plan-review dismissal, which ends the turn while keeping plan mode active. The application supplies its UI or simulated-user policy; the SDK does not choose answers.
 
 Callbacks run on separate threads so the reader can process cancellation and other requests. They may run concurrently and may make further client requests. Release callback-owned work when the cancellation Event is set. Closing the client signals outstanding callbacks and gives their workers a shared 0.5-second join window; callbacks that ignore cancellation may continue, but their late answers cannot reach the runtime or a later process. A callback exception becomes a protocol error, and an absent callback retains existing native answerers or reports `NO_PROVIDER`.
 

python/sdk/README.zh.md

@@ -61,16 +61,18 @@ with DeepSeekHarness(dsh_home="/absolute/path/to/saved-home") as harness:
 <a id="query-and-response"></a>
 ## 查询与接收
 
-调用 `session.query(text)` 发送文本，再迭代 `session.receive_response()` 观察结果。`query()` 发送后即返回 `None`，不等待运行时路由或执行。完整 SDK 接管裸 `/compact` 和带非空 preset 的 `/permission <preset>`，包括绑定到其定义的 Desktop 别名。裸 `/permission` 或 `/权限`、`/plan`、`/model` 及其余输入仍是普通提示词。此 API 只接受文本字符串。`delivery="queue"` 是默认值；`delivery="steer"` 将普通提示词交给下一个 step。命令执行时机仍由命令决定。
+调用 `session.query(text)` 发送文本，再迭代 `session.receive_response()` 观察结果。`query()` 发送后即返回 `None`，不等待运行时路由或执行。完整 SDK 接管裸 `/compact`、带非空 preset 的 `/permission <preset>` 和 `/plan [off|message]`，包括绑定到其定义的 Desktop 别名。裸 `/permission` 或 `/权限`、`/model` 及其余输入仍是普通提示词。此 API 只接受文本字符串。`delivery="queue"` 是默认值；`delivery="steer"` 将普通提示词交给下一个 step。命令执行时机仍由命令决定。
 
 `receive_response()` 按发送顺序领取下一次 query。它先等待路由响应，再逐个产出 `Notification`，最后产出一个终态值：普通提示词从收件箱回执到 idle 区间的 `RunResult`，或命令处理结束后的 `CommandExecution`。通知可以在开始接收前积累。不同接收器可以并行观察各自的 query，也可以在一个接收器等待期间发送另一个 `query()`。关闭未完成的迭代器只释放它自己的本地观察资源，不取消或回滚运行时操作。终态值产出前会释放资源。
 
 `CommandExecution` 包含 `command_id` 和以 `kind` 区分的 `result`。成功结果可以包含 `text` 和 `source_event_seq`；错误结果包含 `text`。处理器错误返回 `kind="error"`；路由、协议和传输失败在接收时抛出。发送失败从 `query()` 抛出。`request_timeout_seconds` 限制路由响应的等待时间，不限制后续提示词活动；已经到达的响应在期限过后仍可接收。
 
-完整 `sdk` profile 提供这两个命令；默认 `sdk-minimal` profile 不提供命令执行能力。`/compact` 可能发起辅助模型请求，完成时不要求普通轮次或 idle 通知；空历史可以不生成摘要而成功返回。`run()` 对 slash 文本保留普通提示词语义。底层阻塞式 `HarnessClient.session_command(session_id, line)` 严格接受 `/compact` 或 `/permission <preset>`，并拒绝裸 `/permission`。另见[压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
+完整 `sdk` profile 提供这些命令；默认 `sdk-minimal` profile 不提供命令执行能力。`/compact` 可能发起辅助模型请求，完成时不要求普通轮次或 idle 通知；空历史可以不生成摘要而成功返回。`run()` 对 slash 文本保留普通提示词语义。底层阻塞式 `HarnessClient.session_command(session_id, line)` 严格接受 `/compact`、`/permission <preset>` 或 `/plan [off|message]`，并拒绝裸 `/permission`。另见[压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
 
 `/permission <preset>` 委托给原生权限 handler。未知 preset 返回 `CommandExecution.result.kind="error"`，不改变权限状态。成功的变更属于该 Session：后续 query 和恢复保留该状态，其他 Session 保留各自状态，新 Session 使用配置的默认值。`DeepSeekHarnessConfig` 仍是启动输入，不是共享的实时状态镜像；命令不重写它或已保存的用户默认值。运行中改变审批策略可能注入通知并触发额外模型 step。另见[权限示例](examples/README.zh.md#change-session-permissions)。
 
+`/plan` 进入计划模式，`/plan off` 退出，`/plan <message>` 进入后通过原生 steer 提交去除首尾空白的正文。返回的 `CommandExecution` 只表示命令完成，不表示规划或审查结束。需要普通活动结果时，可以先用 `/plan` 进入，再单独提交普通 query。组合输入则应在发送前订阅，并在保持问答处理器运行的同时使用 `session/wait`，见[计划示例](examples/README.zh.md#review-a-plan)。计划模式改变指导并渐进披露 `exit_plan_mode`，不修改权限设置。默认指导禁止编辑文件，包括计划文档；实际执行仍由沙箱控制。
+
 <a id="client-approvals"></a>
 ## 客户端审批
 
@@ -123,7 +125,7 @@ with DeepSeekHarness(
 
 ## 原生用户问答
 
-向 `DeepSeekHarnessConfig`、`DeepSeekHarness` 或 `HarnessConfig` 传入 `on_user_question`，即可通过原生 SDK 协议回答内建的 `ask_user_question` 工具。回调接收 `UserQuestionRequest`（包含 `request_id`、`session_id` 和原始 `questions`）以及表示取消的 `threading.Event`。返回 `UserQuestionAnswer`，或形如 `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}` 的 JSON 对象。每个原问题 id 必须恰好出现一次。应用自行提供 UI 或模拟用户策略；SDK 不选择答案。
+向 `DeepSeekHarnessConfig`、`DeepSeekHarness` 或 `HarnessConfig` 传入 `on_user_question`，即可通过原生 SDK 协议回答内建的 `ask_user_question` 工具。回调接收 `UserQuestionRequest`（包含 `request_id`、`session_id` 和原始 `questions`）以及表示取消的 `threading.Event`。返回 `UserQuestionAnswer`，或形如 `{"answers": [{"id": "question-id", "selected": ["option label"], "custom": "optional free text"}]}` 的 JSON 对象。每个原问题 id 必须恰好出现一次。也可返回 `{ "cancelled": true }` 关闭问题而不作答；这会保留原生计划审查的关闭行为：结束轮次并保持计划模式。应用自行提供 UI 或模拟用户策略；SDK 不选择答案。
 
 回调在独立线程执行，读取线程仍可处理取消和其他请求。回调可以并发执行，也可以再次调用客户端请求。取消 Event 置位时应释放回调拥有的工作。客户端关闭会通知所有待决回调，并给其工作线程总计 0.5 秒的 join 窗口；忽略取消的回调可能继续执行，但晚到答案不会发送到运行时或后续新进程。回调异常转为协议错误；未提供回调时保留原生回答方，或报告 `NO_PROVIDER`。
 

python/sdk/examples/README.i18n.yaml

@@ -12,8 +12,11 @@
   en: ad4d67225570ed01
   zh: 0d59f6dac40129d6
 /python-sdk-example/change-session-permissions:
-  en: c1162e41cd57dec4
-  zh: 96f608fa25e26c77
+  en: 3471642d8444d0e9
+  zh: 7831f28295a6e68a
+/python-sdk-example/review-a-plan:
+  en: b013d5ab5e24bb83
+  zh: 84ed4333970b9dc4
 /python-sdk-example/add-plugins:
   en: 5e1e30dac13f6c9e
   zh: 0af98ebb75c88f43

python/sdk/examples/README.md

@@ -56,6 +56,20 @@ python python/sdk/examples/permission.py \
 
 The command changes the Session's policy while the printed launch setting remains `workspace-write`. The later query uses the Session's current policy. Bare `/permission` is ordinary text, and unknown presets produce a command error. See [Query and response](../README.md#query-and-response) for result and state semantics.
 
+<a id="review-a-plan"></a>
+## Review a plan
+
+The [plan example](plan.py) sends `/plan <task>` through query 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; EOF dismisses without approving. Approval allows implementation from a later model step, so choose an isolated workspace.
+
+```sh
+python python/sdk/examples/plan.py \
+  --dsh-home /absolute/path/to/plan-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace \
+  "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`; its question callback is independent of the protocol reader.
+
 ## Add plugins
 
 Use the runtime wheel's `dsh` command against the same explicit home for persistent profile changes:

python/sdk/examples/README.zh.md

@@ -56,6 +56,20 @@ python python/sdk/examples/permission.py \
 
 命令会修改 Session 的策略，而打印出的启动设置仍为 `workspace-write`。后续 query 使用 Session 的当前策略。裸 `/permission` 是普通文本，未知 preset 返回命令错误。结果和状态语义见[查询与接收](../README.zh.md#query-and-response)。
 
+<a id="review-a-plan"></a>
+## 审查计划
+
+[计划示例](plan.py) 通过 query 发送 `/plan <task>`，并在命令完成后保持 Session 通知订阅和问答处理器运行。输入 `a` 批准、`k` 提供反馈、`d` 关闭；EOF 也会关闭而不批准。批准后模型可以从后续步骤开始实施，因此应选择隔离的工作区。
+
+```sh
+python python/sdk/examples/plan.py \
+  --dsh-home /absolute/path/to/plan-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace \
+  "Plan the requested change and present it for review."
+```
+
+原生计划通过 `exit_plan_mode` 携带，而不是写入工作区文档。权限仍为 workspace-write。关闭审查会保留计划模式并结束当前轮次。示例通过 `session/wait` 等待，问答回调独立于协议读取线程。
+
 ## 添加插件
 
 对同一个显式 home 使用运行时 wheel 包提供的 `dsh` 命令，以进行持久 profile 变更：

python/sdk/examples/plan.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Submit a plan command, review its plan, and observe the Session's continuing work."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import threading
+from contextlib import closing
+from pathlib import Path
+
+from pydantic import BaseModel
+
+from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, UserQuestionRequest
+from deepseek_harness.api import final_response
+
+
+class WaitResult(BaseModel):
+    """The empty session/wait result."""
+
+
+def review(request: UserQuestionRequest, cancelled: threading.Event) -> dict:
+    """Let the user approve, give feedback, or dismiss the native plan review."""
+    if cancelled.is_set():
+        return {"cancelled": True}
+    if len(request.questions) != 1 or request.questions[0].get("intent", {}).get("kind") != "plan-review":
+        raise RuntimeError("This example handles plan reviews only")
+    question = request.questions[0]
+    print(question["detail"])
+    try:
+        while True:
+            choice = input("[a] Approve, [k] Keep planning, [d] Dismiss: ").strip().lower()
+            if choice == "d":
+                return {"cancelled": True}
+            if choice == "a":
+                return {"answers": [{"id": question["id"], "selected": ["Approve"]}]}
+            if choice == "k":
+                return {"answers": [{"id": question["id"], "selected": ["Keep planning"], "custom": input("Feedback: ")}]}
+    except EOFError:
+        return {"cancelled": True}
+
+
+def display(notification: Notification) -> None:
+    """Print completed assistant text while the Session runs."""
+    if notification.method == "session.event":
+        event = notification.payload.get("event")
+        if isinstance(event, dict) and event.get("type") == "assistant/message":
+            text = final_response([event])
+            if text:
+                print(text)
+
+
+def main() -> None:
+    """Keep the runtime and question handler alive after the plan command completes."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--dsh-home", required=True, type=Path)
+    parser.add_argument("--workspace", type=Path, default=Path.cwd())
+    parser.add_argument("--model", default=os.environ.get("DSH_MODEL", "deepseek-v4-flash"))
+    parser.add_argument("task", help="Task to plan; implementation waits for your approval")
+    args = parser.parse_args()
+    if not args.task.strip():
+        parser.error("task must be non-blank")
+    with DeepSeekHarness(
+        dsh_home=str(args.dsh_home.resolve()), cwd=str(args.workspace.resolve()),
+        profile="sdk", model=args.model, on_user_question=review,
+        env={"DSH_PERMISSION_MODE": "workspace-write"},
+    ) as harness:
+        session = harness.start_session()
+        with closing(harness.client.subscribe_session_notifications(session.id)) as notifications:
+            session.query(f"/plan {args.task}")
+            for message in session.receive_response():
+                if isinstance(message, CommandExecution):
+                    print(message.result.text)
+                    if message.result.kind == "error":
+                        raise RuntimeError(message.result.text)
+            harness.client.request(
+                "session/wait", {"sessionId": session.id}, response_model=WaitResult,
+                on_notification=display, notification_subscription=notifications,
+            )
+
+
+if __name__ == "__main__":
+    main()

python/sdk/tests/test_commands.py

@@ -103,7 +103,7 @@ def test_low_level_command_preserves_the_line(peer: Path, tmp_path: Path) -> Non
 def test_command_rpc_rejection_does_not_fall_back_to_prompt(peer: Path, tmp_path: Path) -> None:
     with harness(peer, tmp_path, {"error": {"code": -32602, "message": "unsupported command"}}) as sdk:
         with pytest.raises(JsonRpcError, match="unsupported command") as raised:
-            sdk.client.session_command("main", "/plan")
+            sdk.client.session_command("main", "/goal")
         assert raised.value.code == -32602
     assert [frame["method"] for frame in frames(tmp_path)] == ["initialize", "session/command", "shutdown"]
 

python/sdk/tests/test_commands_runtime.py

@@ -22,6 +22,7 @@ from typing import Callable, Iterator, TypeVar
 import pytest
 
 from deepseek_harness import CommandExecution, DeepSeekHarness, Notification, RunResult, Session
+from deepseek_harness.client import UserQuestionHandler
 
 
 _WAIT_SECONDS = 30
@@ -39,6 +40,7 @@ _T = TypeVar("_T")
 class _Completion:
     text: str
     blocked: bool = False
+    tool_call: tuple[str, dict] | None = None
     entered: threading.Event = field(default_factory=threading.Event)
     release: threading.Event = field(default_factory=threading.Event)
 
@@ -47,17 +49,24 @@ class _Completion:
             self.release.set()
 
 
-def _messages_response(text: str, model: str, ordinal: int) -> bytes:
+def _messages_response(text: str, model: str, ordinal: int, tool_call: tuple[str, dict] | None = None) -> bytes:
+    if tool_call is None:
+        block = {"type": "text", "text": ""}
+        delta = {"type": "text_delta", "text": text}
+    else:
+        name, arguments = tool_call
+        block = {"type": "tool_use", "id": f"commands-tool-{ordinal}", "name": name, "input": {}}
+        delta = {"type": "input_json_delta", "partial_json": json.dumps(arguments)}
     events = [
         {"type": "message_start", "message": {
             "id": f"commands-response-{ordinal}", "type": "message", "role": "assistant",
             "model": model, "content": [], "stop_reason": None, "stop_sequence": None,
             "usage": {"input_tokens": 10000, "output_tokens": 0},
         }},
-        {"type": "content_block_start", "index": 0, "content_block": {"type": "text", "text": ""}},
-        {"type": "content_block_delta", "index": 0, "delta": {"type": "text_delta", "text": text}},
+        {"type": "content_block_start", "index": 0, "content_block": block},
+        {"type": "content_block_delta", "index": 0, "delta": delta},
         {"type": "content_block_stop", "index": 0},
-        {"type": "message_delta", "delta": {"stop_reason": "end_turn", "stop_sequence": None},
+        {"type": "message_delta", "delta": {"stop_reason": "end_turn" if tool_call is None else "tool_use", "stop_sequence": None},
          "usage": {"output_tokens": 32}},
         {"type": "message_stop"},
     ]
@@ -89,7 +98,7 @@ class _ModelServer:
                 completion.entered.set()
                 # The test owns release. This larger bound only contains a failed test.
                 assert completion.release.wait(_WAIT_SECONDS * 2), "model barrier was not released"
-                payload = _messages_response(completion.text, body["model"], ordinal)
+                payload = _messages_response(completion.text, body["model"], ordinal, completion.tool_call)
                 self.send_response(200)
                 self.send_header("Content-Type", "text/event-stream")
                 self.send_header("Content-Length", str(len(payload)))
@@ -172,6 +181,7 @@ class _Runtime:
 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,
 ) -> Iterator[_Runtime]:
     workspace = tmp_path / "workspace"
     workspace.mkdir(exist_ok=True)
@@ -192,6 +202,7 @@ def _runtime(
             dsh_bin=dsh_bin, profile="sdk", patches=tuple(str(item) for item in patches), model=_MODEL,
             base_url=model.url, api_key="keyless-commands-runtime",
             initialize_timeout_seconds=60, request_timeout_seconds=20, shutdown_timeout_seconds=20,
+            on_user_question=on_user_question,
             env={"DSH_PERMISSION_MODE": initial_permission,
                  "DSH_TELEMETRY_DISABLED": "1", "DSH_TOOLS_MODE": "native"},
         )
@@ -445,29 +456,29 @@ def test_unknown_slash_and_compact_arguments_follow_the_plain_prompt_path(tmp_pa
     assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2
 
 
-def test_unimplemented_slash_remains_plain_text_without_changing_plan_mode(tmp_path: Path) -> None:
-    plan_text = "/plan DSH_PLAN_TEXT_INPUT"
+def test_unimplemented_slash_remains_plain_text_without_creating_a_goal(tmp_path: Path) -> None:
+    input_text = "/goal DSH_GOAL_TEXT_INPUT"
     following = "DSH_AFTER_PLAN_TEXT_INPUT"
-    plan_reply = "DSH_PLAN_TEXT_REPLY"
+    reply_before = "DSH_GOAL_TEXT_REPLY"
     reply = "DSH_AFTER_PLAN_TEXT_REPLY"
-    with _runtime(tmp_path, [_Completion(plan_reply), _Completion(reply)]) as runtime:
+    with _runtime(tmp_path, [_Completion(reply_before), _Completion(reply)]) as runtime:
         session = runtime.harness.start_session(_SESSION_ID)
-        assert runtime.call(session.query, plan_text, delivery="queue") is None
+        assert runtime.call(session.query, input_text, delivery="queue") is None
         first = runtime.call(_receive_result, session)
         assert isinstance(first, RunResult)
-        assert first.final_response == plan_reply
+        assert first.final_response == reply_before
 
         assert runtime.call(session.query, following, delivery="queue") is None
         result = runtime.call(_receive_result, session)
         assert isinstance(result, RunResult)
         assert result.final_response == reply
         initial, continued = runtime.model.captured()
-        assert plan_text in _request_texts(initial)
+        assert input_text in _request_texts(initial)
         assert following in _request_texts(continued)
         assert initial["tools"] == continued["tools"]
     events = _read_events(runtime.logs)
     assert not _of_type(events, "command/run")
-    assert not any(event["data"].get("active") is True for event in _of_type(events, "plan/mode"))
+    assert not any(event["type"].startswith("goal/") for event in events)
     assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2
 
 

python/sdk/tests/test_plan_runtime.py

@@ -0,0 +1,255 @@
+"""Native plan guidance, input timing and reviewed exit through the real SDK profile."""
+
+from __future__ import annotations
+
+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, _receive_result,
+    _request_texts, _runtime, _text_blocks,
+)
+
+_PLAN = "# Implement the change\n\n- Inspect existing code.\n- Make the smallest change.\n- Run focused tests."
+_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:
+    assert runtime.call(session.query, text) is None
+    result = runtime.call(_receive_result, session)
+    assert isinstance(result, CommandExecution)
+    assert result.result.kind == "success", result.result.text
+    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"]:
+        if message["role"] == "system":
+            system = "\n".join(_text_blocks(message["content"]))
+    return system
+
+
+def _tool_names(request: dict) -> set[str]:
+    return {tool["name"] for tool in request.get("tools", [])}
+
+
+def _review(plan: str = _PLAN) -> _Completion:
+    return _Completion("", tool_call=("exit_plan_mode", {"plan": plan}))
+
+
+def _answer(request: UserQuestionRequest, choice: str, feedback: str | None = None) -> dict:
+    [question] = request.questions
+    assert question["intent"]["kind"] == "plan-review"
+    item = {"id": question["id"], "selected": [choice]}
+    if feedback is not None:
+        item["custom"] = feedback
+    return {"answers": [item]}
+
+
+def test_bare_plan_and_off_change_only_plan_state_without_a_model_request(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        on = _command(runtime, session, " \n/计划\t ")
+        off = _command(runtime, session, "/plan off")
+        repeated = _command(runtime, session, "/plan off")
+        assert runtime.model.captured() == []
+        assert "already inactive" in repeated.result.text
+        assert runtime.harness.config.env["DSH_PERMISSION_MODE"] == "workspace-write"
+    events = _read_events(runtime.logs)
+    for execution in [on, off, repeated]:
+        _command_pair(events, execution, "plan")
+    assert [e["data"] for e in _of_type(events, "plan/mode")] == [{"active": True}, {"active": False}]
+    assert not _of_type(events, "turn/start")
+    assert [e["data"] for e in _permission_facts(events)] == [
+        {"preset": "workspace-write"}, {"mode": "workspace-write"}, {"policy": "ask"},
+    ]
+
+
+def test_plan_body_uses_native_input_then_off_removes_guidance_but_keeps_exit_tool(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [_Completion("PLAN_READY"), _Completion("AFTER_OFF")], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        execution = _command(runtime, session, "/计划  Design this change without tools. ")
+        _wait_activity(runtime, session)
+        first = runtime.model.captured()[0]
+        assert "Design this change without tools." in _request_texts(first)
+        assert "/计划  Design this change without tools. " not in _request_texts(first)
+        assert _GUIDANCE in _effective_system(first)
+        assert "Do not edit or write files" in _effective_system(first)
+        assert {"exit_plan_mode", "write", "edit"} <= _tool_names(first)
+        _assert_request_permission(first, "workspace-write", "ask")
+        _command(runtime, session, "/plan off")
+        runtime.call(session.query, "Continue after the user changed the mode.")
+        result = runtime.call(_receive_result, session)
+        assert isinstance(result, RunResult) and result.final_response == "AFTER_OFF"
+        second = runtime.model.captured()[1]
+        assert _GUIDANCE not in _effective_system(second)
+        assert _tool_names(first) == _tool_names(second)
+        assert "The user switched this session back to the default mode." in _request_texts(second)
+        _assert_request_permission(second, "workspace-write", "ask")
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution, "plan")
+    assert len(_of_type(events, "turn/start")) == 2
+    assert len(_permission_facts(events)) == 3
+
+
+def test_plan_body_steers_the_next_step_of_a_running_turn(tmp_path: Path) -> None:
+    first = _Completion("FIRST_REPLY", blocked=True)
+    with _runtime(tmp_path, [first, _Completion("PLANNING_REPLY")], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        runtime.call(session.query, "Start this task without tools.")
+        assert first.entered.wait(_WAIT_SECONDS)
+        claimed = threading.Event()
+        observing = runtime.workers.submit(_receive_result, session, claimed)
+        assert claimed.wait(_WAIT_SECONDS)
+        execution = _command(runtime, session, "/plan Design a smaller change.")
+        assert not first.release.is_set() and not observing.done()
+        first.release.set()
+        result = observing.result(timeout=_WAIT_SECONDS)
+        assert isinstance(result, RunResult) and result.final_response == "PLANNING_REPLY"
+        a, b = runtime.model.captured()
+        assert _GUIDANCE not in _effective_system(a)
+        assert _GUIDANCE in _effective_system(b)
+        assert "Design a smaller change." in _request_texts(b)
+        _assert_request_permission(b, "workspace-write", "ask")
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution, "plan")
+    assert [e["data"] for e in _of_type(events, "step/start")] == [{"turn": 1, "step": 1}, {"turn": 1, "step": 2}]
+    assert len(_of_type(events, "turn/start")) == 1
+
+
+@pytest.mark.parametrize("cancel", [False, True])
+def test_running_bare_selection_does_not_force_a_step_and_can_be_cancelled(tmp_path: Path, cancel: bool) -> None:
+    first = _Completion("FIRST_REPLY", blocked=True)
+    with _runtime(tmp_path, [first, _Completion("NEXT_REPLY")], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        runtime.call(session.query, "Start without tools.")
+        assert first.entered.wait(_WAIT_SECONDS)
+        claimed = threading.Event()
+        observing = runtime.workers.submit(_receive_result, session, claimed)
+        assert claimed.wait(_WAIT_SECONDS)
+        _command(runtime, session, "/plan")
+        if cancel:
+            cancelled = _command(runtime, session, "/plan off")
+            assert "cancelled" in cancelled.result.text
+        assert len(runtime.model.captured()) == 1
+        first.release.set()
+        assert isinstance(observing.result(timeout=_WAIT_SECONDS), RunResult)
+        assert len(runtime.model.captured()) == 1
+        runtime.call(session.query, "Inspect the next task without tools.")
+        assert isinstance(runtime.call(_receive_result, session), RunResult)
+        assert (_GUIDANCE in _effective_system(runtime.model.captured()[1])) is (not cancel)
+    events = _read_events(runtime.logs)
+    assert len(_of_type(events, "turn/start")) == 2
+    assert len(_of_type(events, "step/start")) == 2
+    assert [e["data"] for e in _of_type(events, "plan/mode")] == ([] if cancel else [{"active": True}])
+
+
+@pytest.mark.parametrize("first_choice", ["Approve", "Keep planning", "dismiss"])
+def test_review_approval_feedback_and_dismissal_keep_native_model_behavior(tmp_path: Path, first_choice: str) -> None:
+    seen: list[str] = []
+    def answer(request: UserQuestionRequest, cancelled: threading.Event):
+        assert not cancelled.is_set()
+        seen.append(request.questions[0]["detail"])
+        if len(seen) == 1:
+            if first_choice == "dismiss":
+                return {"cancelled": True}
+            return _answer(request, first_choice, "Include rollback." if first_choice == "Keep planning" else None)
+        return _answer(request, "Approve")
+    completions = [_review()]
+    if first_choice == "Keep planning":
+        completions.append(_review(_PLAN_REVISED))
+    if first_choice != "dismiss":
+        completions.append(_Completion("AFTER_APPROVAL"))
+    with _runtime(tmp_path, completions, initial_permission="workspace-write", on_user_question=answer) as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        _command(runtime, session, "/plan Prepare a plan and present it for review.")
+        _wait_activity(runtime, session)
+        requests = runtime.model.captured()
+        assert seen == ([_PLAN, _PLAN_REVISED] if first_choice == "Keep planning" else [_PLAN])
+        assert _GUIDANCE in _effective_system(requests[0])
+        if first_choice == "Keep planning":
+            assert _GUIDANCE in _effective_system(requests[1])
+            assert "Include rollback." in str(requests[1]["messages"])
+        if first_choice != "dismiss":
+            assert _GUIDANCE not in _effective_system(requests[-1])
+            assert _tool_names(requests[0]) == _tool_names(requests[-1])
+        else:
+            assert len(requests) == 1, "dismissal started another model request"
+        for request in requests:
+            _assert_request_permission(request, "workspace-write", "ask")
+    events = _read_events(runtime.logs)
+    assert [e["data"]["active"] for e in _of_type(events, "plan/mode")] == ([True] if first_choice == "dismiss" else [True, False])
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 1
+    assert len(_permission_facts(events)) == 3
+    if first_choice == "dismiss":
+        results = _of_type(events, "tool/result")
+        assert len(results) == 1
+        assert "The user dismissed the plan review" in str(results[0]["data"])
+
+
+def test_no_review_provider_keeps_plan_mode_and_reports_tool_failure(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [_review(), _Completion("REVIEW_UNAVAILABLE")], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        _command(runtime, session, "/plan Prepare a plan.")
+        _wait_activity(runtime, session)
+        first, second = runtime.model.captured()
+        assert _GUIDANCE in _effective_system(first) and _GUIDANCE in _effective_system(second)
+        assert "no user-questions answerer accepted the request" in str(second["messages"])
+        _command(runtime, session, "/plan off")
+    events = _read_events(runtime.logs)
+    [result] = _of_type(events, "tool/result")
+    assert result["data"]["message"]["isError"] is True
+    assert result["data"]["error"]["code"] == "NO_PROVIDER"
+    assert len(_permission_facts(events)) == 3
+
+
+def test_plan_state_survives_resume_without_changing_permissions(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        _command(runtime, session, "/plan")
+    with _runtime(tmp_path, [_Completion("RESTORED_PLAN")], initial_permission="workspace-write") as resumed:
+        session = resumed.call(resumed.harness.resume_session, _SESSION_ID)
+        resumed.call(session.query, "Continue planning without tools.")
+        assert isinstance(resumed.call(_receive_result, session), RunResult)
+        [request] = resumed.model.captured()
+        assert _GUIDANCE in _effective_system(request)
+        _assert_request_permission(request, "workspace-write", "ask")
+    events = _read_events(resumed.logs)
+    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:
+    with _runtime(tmp_path, [_Completion("OTHER"), _Completion("PLANNING"), _Completion("RAW_INPUT")],
+                  initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        _command(runtime, session, "/plan")
+        other = runtime.harness.start_session("plan-other")
+        runtime.call(other.query, "Inspect your own task without tools.")
+        assert isinstance(runtime.call(_receive_result, other), RunResult)
+        runtime.call(session.query, "Inspect the plan without tools.")
+        assert isinstance(runtime.call(_receive_result, session), RunResult)
+        result = runtime.call(session.run, "/plan off")
+        assert isinstance(result, RunResult)
+        first, second, third = runtime.model.captured()
+        assert _GUIDANCE not in _effective_system(first)
+        assert _GUIDANCE in _effective_system(second)
+        assert _GUIDANCE in _effective_system(third)
+        assert "/plan off" in _request_texts(third)
+    assert not _of_type(_read_events(runtime.logs, session_id="plan-other"), "plan/mode")
+    assert [e["data"] for e in _of_type(_read_events(runtime.logs), "plan/mode")] == [{"active": True}]

python/sdk/tests/test_queries.py

@@ -141,7 +141,7 @@ def test_query_only_sends_then_receive_returns_command(sdk: DeepSeekHarness) ->
     assert_released(sdk)
 
 
-@pytest.mark.parametrize("text", ["ordinary", "fast-prompt", "/plan", "/permission"])
+@pytest.mark.parametrize("text", ["ordinary", "fast-prompt", "/goal", "/permission"])
 @pytest.mark.parametrize("delivery", ["queue", "steer"])
 def test_prompt_yields_receipt_through_idle_then_run_result(sdk: DeepSeekHarness, text: str, delivery: str) -> None:
     session = sdk.start_session("main")

python/sdk/tests/test_user_questions.py

@@ -102,6 +102,15 @@ def test_missing_callback_does_not_advertise_question_support(runtime: Path, tmp
         assert client.request("stats", {}, response_model=Response).capabilities is None
 
 
+def test_callback_can_dismiss_a_question_without_a_provider_error(runtime: Path, tmp_path: Path) -> None:
+    with HarnessClient(config(lambda _request, _cancelled: {"cancelled": True}),
+                       _launch_args=(sys.executable, str(runtime))) as client:
+        initialize(client, tmp_path)
+        result = client.request("ask", {}, response_model=Response)
+        assert result.answer == {"cancelled": True}
+        assert result.error is None
+
+
 @pytest.mark.parametrize("action", ["cancel", "disconnect", "close"])
 def test_cancellation_stops_cooperative_host_work_and_suppresses_late_answer(runtime: Path, tmp_path: Path, action: str) -> None:
     entered = threading.Event()