feat(sdk): support session permission commands

python/sdk/README.i18n.yaml

@@ -12,8 +12,8 @@
   en: c4c30913fe2642ef
   zh: 0ed42bca37fba60b
 /deepseek-harness-python-sdk/query-and-response:
-  en: b3ad6ea82350b467
-  zh: 85dacc783fd12992
+  en: a3f6c4ee867928a8
+  zh: f5af634f4fcdc547
 /deepseek-harness-python-sdk/client-approvals:
   en: fc69be1beb8abff1
   zh: 91b56781931a6f68

python/sdk/README.md

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

python/sdk/README.zh.md

@@ -61,13 +61,15 @@ 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`，不等待运行时响应。运行时只将 `/compact` 拦截为命令；包括 `/plan` 和其他以斜杠开头的输入在内，其余文本仍作为普通提示词。此输入 API 只接受文本字符串。`delivery="queue"` 是默认值；`delivery="steer"` 将普通提示词交给下一个 step。命令执行时机仍由命令决定。
+调用 `session.query(text)` 发送文本，再迭代 `session.receive_response()` 观察结果。`query()` 发送后即返回 `None`，不等待运行时路由或执行。完整 SDK 接管裸 `/compact` 和带非空 preset 的 `/permission <preset>`，包括绑定到其定义的 Desktop 别名。裸 `/permission` 或 `/权限`、`/plan`、`/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 提供被拦截的 `/compact` 命令。它可能发起辅助模型请求，完成时不要求普通轮次或 idle 通知；空历史可以不生成摘要而成功返回。默认 `sdk-minimal` profile 不提供命令执行能力。`session.run("/compact")` 仍按普通提示词文本处理。底层阻塞式 `HarnessClient.session_command(session_id, line)` 仍可用于严格命令执行，并拒绝 `/compact` 之外的命令名。另见[完整 SDK 压缩示例](examples/README.zh.md#compact-a-full-sdk-session)。
+完整 `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)。
+
+`/permission <preset>` 委托给原生权限 handler。未知 preset 返回 `CommandExecution.result.kind="error"`，不改变权限状态。成功的变更属于该 Session：后续 query 和恢复保留该状态，其他 Session 保留各自状态，新 Session 使用配置的默认值。`DeepSeekHarnessConfig` 仍是启动输入，不是共享的实时状态镜像；命令不重写它或已保存的用户默认值。运行中改变审批策略可能注入通知并触发额外模型 step。另见[权限示例](examples/README.zh.md#change-session-permissions)。
 
 <a id="client-approvals"></a>
 ## 客户端审批

python/sdk/examples/README.i18n.yaml

@@ -9,8 +9,11 @@
   en: 4b79fd6576e2759d
   zh: 6258540acaf4bc37
 /python-sdk-example/compact-a-full-sdk-session:
-  en: 0bada22014bf5e6c
-  zh: 4308ea1a14b48f57
+  en: ad4d67225570ed01
+  zh: 0d59f6dac40129d6
+/python-sdk-example/change-session-permissions:
+  en: c1162e41cd57dec4
+  zh: 96f608fa25e26c77
 /python-sdk-example/add-plugins:
   en: 5e1e30dac13f6c9e
   zh: 0af98ebb75c88f43

python/sdk/examples/README.md

@@ -42,6 +42,20 @@ python python/sdk/examples/compact.py \
 
 The script consumes notifications and prints the command id and outcome separately from the model's responses. A failed turn or command stops the example. See [Query and response](../README.md#query-and-response) for completion and error behavior; `minimal.py` retains its `run()` entry and does not enable command execution.
 
+<a id="change-session-permissions"></a>
+## Change Session permissions
+
+The [permission example](permission.py) starts the full `sdk` profile with a `workspace-write` default, sends `/permission <preset>`, then submits an ordinary query on the same Session. Supply the preset explicitly:
+
+```sh
+python python/sdk/examples/permission.py \
+  --dsh-home /absolute/path/to/permission-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace \
+  --preset read-only
+```
+
+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.
+
 ## Add plugins
 
 Use the runtime wheel's `dsh` command against the same explicit home for persistent profile changes:

python/sdk/examples/README.zh.md

@@ -42,6 +42,20 @@ python python/sdk/examples/compact.py \
 
 脚本消费通知，并将命令 id 和结果与模型回复分开打印。轮次或命令失败时，示例会停止。完成和错误行为见[查询与接收](../README.zh.md#query-and-response)；`minimal.py` 保留其 `run()` 入口，不启用命令执行。
 
+<a id="change-session-permissions"></a>
+## 修改 Session 权限
+
+[权限示例](permission.py)以 `workspace-write` 为默认值启动完整 `sdk` profile，发送 `/permission <preset>`，随后在同一个 Session 中提交普通 query。请显式指定 preset：
+
+```sh
+python python/sdk/examples/permission.py \
+  --dsh-home /absolute/path/to/permission-dsh-home \
+  --workspace /absolute/path/to/disposable-workspace \
+  --preset read-only
+```
+
+命令会修改 Session 的策略，而打印出的启动设置仍为 `workspace-write`。后续 query 使用 Session 的当前策略。裸 `/permission` 是普通文本，未知 preset 返回命令错误。结果和状态语义见[查询与接收](../README.zh.md#query-and-response)。
+
 ## 添加插件
 
 对同一个显式 home 使用运行时 wheel 包提供的 `dsh` 命令，以进行持久 profile 变更：

python/sdk/examples/permission.py

@@ -0,0 +1,58 @@
+#!/usr/bin/env python3
+"""Select a full SDK Session's permission preset and continue without resetting it."""
+
+from __future__ import annotations
+
+import argparse
+import os
+from contextlib import closing
+from pathlib import Path
+
+from deepseek_harness import CommandExecution, DeepSeekHarness, RunResult, Session
+
+
+def receive(session: Session) -> None:
+    """Print a command or activity result and stop on a reported failure."""
+    with closing(session.receive_response()) as responses:
+        for result in responses:
+            if isinstance(result, CommandExecution):
+                print(f"Command {result.command_id}: {result.result.kind}")
+                if result.result.kind == "error":
+                    raise RuntimeError(result.result.text)
+                if result.result.text is not None:
+                    print(result.result.text)
+            elif isinstance(result, RunResult):
+                if result.finish_reason != "completed":
+                    raise RuntimeError(f"Agent turn ended with {result.finish_reason}")
+                print(result.final_response)
+
+
+def main() -> None:
+    """Send an explicit permission command, then query the same Session."""
+    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("--preset", required=True)
+    parser.add_argument("--provider", default="deepseek-official")
+    parser.add_argument("--model", default=os.environ.get("DSH_MODEL", "deepseek-v4-flash"))
+    args = parser.parse_args()
+    if not args.preset.strip():
+        parser.error("--preset must be non-blank")
+
+    with DeepSeekHarness(
+        dsh_home=str(args.dsh_home.resolve()), cwd=str(args.workspace.resolve()),
+        profile="sdk", provider=args.provider, model=args.model,
+        env={"DSH_PERMISSION_MODE": "workspace-write"},
+    ) as harness:
+        session = harness.start_session()
+        session.query("Summarize the current file and approval policy in one sentence. Do not use tools.")
+        receive(session)
+        session.query(f"/permission {args.preset}")
+        receive(session)
+        print(f"Launch setting: {harness.config.env['DSH_PERMISSION_MODE']}")
+        session.query("Summarize the current file and approval policy again. Do not use tools.")
+        receive(session)
+
+
+if __name__ == "__main__":
+    main()

python/sdk/src/deepseek_harness/api.py

@@ -20,6 +20,8 @@ class DeepSeekHarnessConfig:
     The runtime inherits the caller's environment by default, so existing
     DEEPSEEK_API_KEY and DEEPSEEK_BASE_URL settings keep working. Use ``env`` to
     intentionally override or inject variables for a subprocess.
+    These values are startup input, not a live Session state mirror; Session
+    commands do not rewrite this object.
     """
 
     provider: str = "deepseek-official"

python/sdk/tests/test_commands_runtime.py

@@ -1,4 +1,4 @@
-"""Exercise /compact through Python and the real sdk profile with a local model.
+"""Exercise Session commands through Python and the real sdk profile with a local model.
 
 The default carrier is the repository's built dsh CLI. Set
 DSH_COMMANDS_TEST_RUNTIME=installed to run these same cases against the imported
@@ -171,9 +171,10 @@ class _Runtime:
 @contextmanager
 def _runtime(
     tmp_path: Path, completions: list[_Completion], *, capture_patch: Path | None = None,
+    initial_permission: str = "danger-full-access",
 ) -> Iterator[_Runtime]:
     workspace = tmp_path / "workspace"
-    workspace.mkdir()
+    workspace.mkdir(exist_ok=True)
     home = tmp_path / "home"
     logs = tmp_path / "sessions"
     patch = tmp_path / "commands.patch.yml"
@@ -191,7 +192,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,
-            env={"DSH_PERMISSION_MODE": "danger-full-access",
+            env={"DSH_PERMISSION_MODE": initial_permission,
                  "DSH_TELEMETRY_DISABLED": "1", "DSH_TOOLS_MODE": "native"},
         )
         try:
@@ -203,9 +204,14 @@ def _runtime(
             harness.close()
 
 
-def _read_events(root: Path, *, live: bool = False) -> list[dict]:
-    paths = list(root.rglob("session*.jsonl"))
-    assert len(paths) == 1, f"expected one fresh Session log, found {paths}"
+def _read_events(root: Path, *, live: bool = False, session_id: str = _SESSION_ID) -> list[dict]:
+    paths = []
+    for path in root.rglob("session*.jsonl"):
+        with path.open(encoding="utf-8") as log:
+            header = json.loads(log.readline())
+        if header.get("id") == session_id:
+            paths.append(path)
+    assert len(paths) == 1, f"expected one Session log for {session_id}, found {paths}"
     content = paths[0].read_bytes()
     if not live:
         assert content.endswith(b"\n"), "closed Session log has an incomplete final row"
@@ -213,7 +219,7 @@ def _read_events(root: Path, *, live: bool = False) -> list[dict]:
     complete = content[:content.rfind(b"\n") + 1]
     records = [json.loads(line) for line in complete.splitlines()]
     assert records and records[0]["type"] == "session"
-    assert records[0]["id"] == _SESSION_ID
+    assert records[0]["id"] == session_id
     events = records[1:]
     for seq, event in enumerate(events):
         assert isinstance(event, dict)
@@ -225,13 +231,13 @@ def _of_type(events: list[dict], name: str) -> list[dict]:
     return [event for event in events if event["type"] == name]
 
 
-def _command_pair(events: list[dict], execution: CommandExecution) -> None:
+def _command_pair(events: list[dict], execution: CommandExecution, name: str = "compact") -> None:
     runs = [event for event in _of_type(events, "command/run")
             if event["data"]["commandId"] == execution.command_id]
     done = [event for event in _of_type(events, "command/done")
             if event["data"]["commandId"] == execution.command_id]
     assert len(runs) == len(done) == 1
-    assert runs[0]["data"]["name"] == "compact"
+    assert runs[0]["data"]["name"] == name
     assert runs[0]["seq"] < done[0]["seq"]
     assert done[0]["data"]["kind"] == execution.result.kind
     assert done[0]["data"].get("text") == execution.result.text
@@ -463,3 +469,153 @@ def test_unimplemented_slash_remains_plain_text_without_changing_plan_mode(tmp_p
     assert not _of_type(events, "command/run")
     assert not any(event["data"].get("active") is True for event in _of_type(events, "plan/mode"))
     assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 2
+
+
+_PERMISSION_FACTS = {"permission/preset", "sandbox/mode", "approval/policy"}
+_ASK_POLICY = "Approval policy: ask."
+_NEVER_POLICY = "Approval prompts are disabled in this session"
+
+
+def _assert_request_permission(request: dict, sandbox: str, approval: str) -> None:
+    contexts = [text for text in _request_texts(request) if text.startswith("Current runtime context.")]
+    assert contexts, "model request has no runtime-context snapshot"
+    assert f"Current DSH file policy: {sandbox}." in contexts[-1]
+    assert (_ASK_POLICY if approval == "ask" else _NEVER_POLICY) in contexts[-1]
+
+
+def _permission_facts(events: list[dict]) -> list[dict]:
+    return [{"type": event["type"], "data": event["data"]}
+            for event in events if event["type"] in _PERMISSION_FACTS]
+
+
+def _permission_command(runtime: _Runtime, session: Session, preset: str) -> CommandExecution:
+    assert runtime.call(session.query, f"/permission {preset}") is None
+    result = runtime.call(_receive_result, session)
+    assert isinstance(result, CommandExecution)
+    return result
+
+
+def test_permission_changes_only_the_selected_session_and_retains_prompt_semantics(tmp_path: Path) -> None:
+    completions = [_Completion(f"PERMISSION_REPLY_{index}") for index in range(9)]
+    with _runtime(tmp_path, completions, initial_permission="workspace-write") as runtime:
+        main = runtime.harness.start_session(_SESSION_ID)
+        other = runtime.harness.start_session("permission-other")
+        initial_config = dict(runtime.harness.config.env)
+        for session, text in [(main, "Initial main policy; do not use tools."),
+                              (other, "Initial other policy; do not use tools.")]:
+            runtime.call(session.query, text)
+            assert isinstance(runtime.call(_receive_result, session), RunResult)
+
+        readonly = _permission_command(runtime, main, "read-only")
+        invalid = _permission_command(runtime, main, "no-such-preset")
+        repeated = _permission_command(runtime, main, "read-only")
+        assert readonly.result.kind == repeated.result.kind == "success"
+        assert invalid.result.kind == "error"
+        assert "unknown preset" in invalid.result.text
+        assert len(runtime.model.captured()) == 2, "idle permission commands woke the model"
+
+        runtime.call(main.query, "Read-only policy after a plain query; do not use tools.")
+        assert isinstance(runtime.call(_receive_result, main), RunResult)
+        full = _permission_command(runtime, main, "danger-full-access")
+        assert full.result.kind == "success"
+        assert len(runtime.model.captured()) == 3
+        runtime.call(main.query, "Full policy after a plain query; do not use tools.")
+        assert isinstance(runtime.call(_receive_result, main), RunResult)
+        restored = _permission_command(runtime, main, "workspace-write")
+        assert restored.result.kind == "success"
+        runtime.call(main.query, "Switched-back policy; do not use tools.")
+        assert isinstance(runtime.call(_receive_result, main), RunResult)
+
+        for session in [other, runtime.harness.start_session("permission-new")]:
+            runtime.call(session.query, "Report your own policy; do not use tools.")
+            assert isinstance(runtime.call(_receive_result, session), RunResult)
+        ordinary = runtime.call(main.run, "/permission danger-full-access")
+        assert isinstance(ordinary, RunResult)
+        runtime.call(main.query, "/permission")
+        assert isinstance(runtime.call(_receive_result, main), RunResult)
+        assert runtime.harness.config.env == initial_config
+
+        requests = runtime.model.captured()
+        assert len(requests) == 9
+        for index in [0, 1, 4, 5, 6, 7, 8]:
+            _assert_request_permission(requests[index], "workspace-write", "ask")
+        _assert_request_permission(requests[2], "read-only", "ask")
+        _assert_request_permission(requests[3], "danger-full-access", "never")
+        assert 'The approval policy changed from "ask" to "never" (changed by the user).' in _request_texts(requests[3])
+        assert 'The approval policy changed from "never" to "ask" (changed by the user).' in _request_texts(requests[4])
+        for index in [2, 3, 4]:
+            assert requests[index]["system"] == requests[0]["system"]
+            assert requests[index]["tools"] == requests[0]["tools"]
+        assert "/permission danger-full-access" in _request_texts(requests[7])
+        assert "/permission" in _request_texts(requests[8])
+
+    events = _read_events(runtime.logs)
+    executions = [readonly, invalid, repeated, full, restored]
+    assert len(_of_type(events, "command/run")) == len(executions)
+    for execution in executions:
+        _command_pair(events, execution, "permission")
+    for execution in [invalid, repeated]:
+        start = next(event["seq"] for event in _of_type(events, "command/run")
+                     if event["data"]["commandId"] == execution.command_id)
+        done = next(event["seq"] for event in _of_type(events, "command/done")
+                    if event["data"]["commandId"] == execution.command_id)
+        assert not _permission_facts(events[start + 1:done])
+    for session_id in [_SESSION_ID, "permission-other", "permission-new"]:
+        state = {event["type"]: event["data"] for event in _permission_facts(_read_events(runtime.logs, session_id=session_id))}
+        assert state["permission/preset"] == {"preset": "workspace-write"}
+        assert state["sandbox/mode"]["mode"] == "workspace-write"
+        assert state["approval/policy"] == {"policy": "ask"}
+
+
+def test_running_permission_change_keeps_the_native_extra_step_and_notifications(tmp_path: Path) -> None:
+    first = _Completion("PERMISSION_FIRST_DONE", blocked=True)
+    second = _Completion("PERMISSION_SWITCH_SEEN")
+    with _runtime(tmp_path, [first, second], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        runtime.call(session.query, "Wait for the policy change, then finish without tools.")
+        assert first.entered.wait(_WAIT_SECONDS), runtime.model.failures
+        claimed = threading.Event()
+        observing = runtime.workers.submit(_receive_result, session, claimed)
+        assert claimed.wait(_WAIT_SECONDS)
+        execution = _permission_command(runtime, session, "danger-full-access")
+        assert execution.result.kind == "success"
+        assert not first.release.is_set()
+        assert not observing.done()
+        assert len(runtime.model.captured()) == 1
+        first.release.set()
+        result = observing.result(timeout=_WAIT_SECONDS)
+        assert isinstance(result, RunResult)
+        assert result.final_response == second.text
+        requests = runtime.model.captured()
+        assert len(requests) == 2
+        _assert_request_permission(requests[0], "workspace-write", "ask")
+        _assert_request_permission(requests[1], "danger-full-access", "never")
+        assert requests[0]["system"] == requests[1]["system"]
+        assert requests[0]["tools"] == requests[1]["tools"]
+        notice = 'The approval policy changed from "ask" to "never" (changed by the user).'
+        assert notice not in _request_texts(requests[0])
+        assert notice in _request_texts(requests[1])
+        assert runtime.harness.config.env["DSH_PERMISSION_MODE"] == "workspace-write"
+    events = _read_events(runtime.logs)
+    _command_pair(events, execution, "permission")
+    assert [event["data"] for event in _of_type(events, "step/start")] == [
+        {"turn": 1, "step": 1}, {"turn": 1, "step": 2},
+    ]
+    assert len(_of_type(events, "turn/start")) == len(_of_type(events, "turn/end")) == 1
+
+
+def test_resumed_permission_survives_different_launch_defaults(tmp_path: Path) -> None:
+    with _runtime(tmp_path, [], initial_permission="workspace-write") as runtime:
+        session = runtime.harness.start_session(_SESSION_ID)
+        execution = _permission_command(runtime, session, "read-only")
+        assert execution.result.kind == "success"
+        assert runtime.model.captured() == []
+    saved = _permission_facts(_read_events(runtime.logs))
+    with _runtime(tmp_path, [_Completion("RESUMED_PERMISSION")], initial_permission="danger-full-access") as resumed:
+        session = resumed.call(resumed.harness.resume_session, _SESSION_ID)
+        resumed.call(session.query, "Report the restored policy without using tools.")
+        assert isinstance(resumed.call(_receive_result, session), RunResult)
+        [request] = resumed.model.captured()
+        _assert_request_permission(request, "read-only", "ask")
+        assert resumed.harness.config.env["DSH_PERMISSION_MODE"] == "danger-full-access"
+    assert _permission_facts(_read_events(resumed.logs)) == saved

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"])
+@pytest.mark.parametrize("text", ["ordinary", "fast-prompt", "/plan", "/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")