Merge remote-tracking branch 'origin/master' into claude/unified-environment-credentials-c8841a

# Conflicts:
#	examples/headless-agent/tests/headless.snapshot.ts
#	examples/headless-agent/tests/snapshots/missing-credential/stream-json.expected.jsonl
#	packages/llm/llm-deepseek/tests/adapter.spec.ts

python/sdk/README.i18n.yaml

@@ -2,5 +2,5 @@
 # side as of the last confirmed-consistent state. Both languages carry equal authority;
 # after editing either side, bring the other along and re-record with:
 #   pnpm run verify-translation-pairing --write python/sdk/README.md
-README.md: bb3420f1a1bd461facbd0eb1312a255df1da4412
-README.zh.md: 2cb48ac9aaae115bbcce5943449bca9e42f2592c
+README.md: 2d545688c58a2f1b755e647d7cda9555249e41c4
+README.zh.md: 653256584f06c0e63b725953f7f27bc91fc198cc

python/sdk/README.md

@@ -35,7 +35,9 @@ with DeepSeekHarness(
 
 `provider` selects a provider route registered by the chosen Cordis composition; `model` is the model id resolved by that adapter. `max_tokens` is an optional positive per-request output-token cap for the root agent and its in-process descendants; omission leaves the provider default in control. Compaction summaries keep the separate limit configured by their compaction plugin. The bundled default composition registers `deepseek-official`. A custom composition can mount `llm-pi-ai`, configure provider-specific credentials/endpoints there, and select any provider/model present in pi-ai's installed catalog.
 
-`HarnessClient` retains discovered subagent ancestry for the lifetime of the runtime process. During each `Session.run()`, `TurnResult.notifications` and `on_notification` receive the root session and all known descendant notifications in wire order, including nested subagent lifecycle and session events. `TurnResult.events` remains the root session's complete event stream, and `TurnResult.final_response` is the text content from its last `assistant/message`; descendant messages therefore cannot replace the root response.
+`Session.run()` owns an activity interval from its prompt's durable inbox receipt through the next whole-agent idle and returns `RunResult(session_id, final_response, events, notifications, session_root)`. The result has no prompt-level status or turn reason: `final_response` is the last committed root-session assistant text in the interval, not an output causally assigned to the prompt. Steering, injected context, and other queued work may contribute before idle.
+
+`HarnessClient` retains discovered subagent ancestry for the lifetime of the runtime process. During each `Session.run()`, `RunResult.notifications` and `on_notification` receive the root session and all known descendant notifications in wire order, including nested subagent lifecycle and session events. `RunResult.events` contains root-session events only, so descendant messages cannot replace the root response. The low-level `session_prompt()` returns the queued `MessageId` immediately; callers that bypass `Session.run()` own any later activity boundary themselves.
 
 The same behavior can be selected for the runtime subprocess with `DSH_CORDIS_CONFIG`. The injection lives in `HarnessClient.start()`, so the low-level client's default launch gets it too: when the launch resolves to the bundled runtime and neither `cordis` nor a non-empty `DSH_CORDIS_CONFIG` is set (the runtime treats an empty value as absent, and so does the injection check), the bundled default configuration is used; an explicit `runtime_bin`, `bridge_bin`, or `launch_args_override` disables the injection entirely. See the [sdk-runtime README](../sdk-runtime/README.md) for the runtime carriers (production exe vs dev-only node closure) and how to obtain them.
 

python/sdk/README.zh.md

@@ -31,7 +31,9 @@ with DeepSeekHarness(
 
 `provider` 用于选择当前 Cordis 组合已注册的提供方路由；`model` 是该适配器解析的模型 ID。`max_tokens` 是可选的正整数，用于限制根 agent（智能体）及其进程内后代每次请求的输出 token；省略时由提供方默认值控制。压缩摘要继续使用压缩插件单独配置的上限。内置默认组合注册 `deepseek-official`。自定义组合可以挂载 `llm-pi-ai`，在其中配置各提供方的凭据与端点，再选择 pi-ai 已安装目录中的任意提供方/模型组合。
 
-`HarnessClient` 会在运行时进程的生命周期内保留已发现的 subagent（子 agent）祖先关系。每次执行 `Session.run()` 时，`TurnResult.notifications` 与 `on_notification` 会按协议传输顺序收到根会话及所有已知后代的通知，其中包括嵌套 subagent 的生命周期事件与会话事件。`TurnResult.events` 仍只保存根会话的完整事件流，`TurnResult.final_response` 则取该会话最后一个 `assistant/message` 的文本内容，因此后代消息不会覆盖根会话回复。
+`Session.run()` 拥有一个从提示词的持久 inbox 回执开始、到整个 agent 下一次进入 idle 为止的活动区间，并返回 `RunResult(session_id, final_response, events, notifications, session_root)`。结果不携带提示词级状态或轮次原因：`final_response` 是该区间内根会话最后提交的助手文本，并非因果上归属于该提示词的输出。steering（中途引导）、注入的上下文和其他排队工作都可能在 idle 前参与其中。
+
+`HarnessClient` 会在运行时进程的生命周期内保留已发现的 subagent（子 agent）祖先关系。每次执行 `Session.run()` 时，`RunResult.notifications` 与 `on_notification` 会按协议传输顺序收到根会话及所有已知后代的通知，其中包括嵌套 subagent 的生命周期事件与会话事件。`RunResult.events` 只包含根会话事件，因此后代消息不会覆盖根会话回复。底层 `session_prompt()` 会立即返回已排队消息的 `MessageId`；绕过 `Session.run()` 的调用方必须自行负责后续的活动边界。
 
 同样的行为也可以通过 `DSH_CORDIS_CONFIG` 为运行时子进程选定。注入逻辑位于 `HarnessClient.start()`，因此底层客户端的默认启动也具有此行为：当启动解析到内置运行时，且 `cordis` 与非空的 `DSH_CORDIS_CONFIG` 均未设置时（运行时把空值视为缺省，注入检查与之一致），使用内置的默认配置；显式给出 `runtime_bin`、`bridge_bin` 或 `launch_args_override` 则完全禁用注入。运行时载体（生产用 exe 与仅限开发的 `node` 闭包）及其获取方式见 [sdk-runtime README](../sdk-runtime/README.md)。
 

python/sdk/src/deepseek_harness/__init__.py

@@ -1,4 +1,4 @@
-from .api import DeepSeekHarness, DeepSeekHarnessConfig, Session, TurnResult
+from .api import DeepSeekHarness, DeepSeekHarnessConfig, RunResult, Session
 from .client import HarnessClient, HarnessConfig
 from .models import IncomingRequest, InitializeResponse, JsonObject, Notification, ServerInfo
 
@@ -6,7 +6,7 @@ __all__ = [
     "DeepSeekHarness",
     "DeepSeekHarnessConfig",
     "Session",
-    "TurnResult",
+    "RunResult",
     "HarnessClient",
     "HarnessConfig",
     "IncomingRequest",

python/sdk/src/deepseek_harness/api.py

@@ -35,9 +35,8 @@ class DeepSeekHarnessConfig:
 
 
 @dataclass(slots=True)
-class TurnResult:
+class RunResult:
     session_id: str
-    status: str
     final_response: str
     events: list[JsonObject]
     notifications: list[Notification]
@@ -119,7 +118,7 @@ class DeepSeekHarness:
         *,
         session_id: str | None = None,
         on_notification: Callable[[Notification], None] | None = None,
-    ) -> TurnResult:
+    ) -> RunResult:
         return self.start_session(session_id).run(input, on_notification=on_notification)
 
 
@@ -133,15 +132,12 @@ class Session:
         input: str | list[JsonObject],
         *,
         on_notification: Callable[[Notification], None] | None = None,
-    ) -> TurnResult:
+    ) -> RunResult:
         content_blocks = normalize_input(input)
         notifications: list[Notification] = []
         events: list[JsonObject] = []
-        status = "error"
-        finished = False
 
         def collect(notification: Notification) -> None:
-            nonlocal finished, status
             notifications.append(notification)
             if on_notification is not None:
                 on_notification(notification)
@@ -152,25 +148,31 @@ class Session:
                 event = notification.payload.get("event")
                 if isinstance(event, dict):
                     events.append(event)
-            if notification.method == "session.finished" and notification.payload.get("sessionId") == self.id:
-                status = str(notification.payload.get("status") or "ok")
-                finished = True
 
         with self.harness.client.subscribe_session_notifications(self.id) as subscription:
-            self.harness.client.session_prompt(
+            message_id = self.harness.client.session_prompt(
                 self.id,
                 content_blocks,
-                on_notification=collect,
                 notification_subscription=subscription,
             )
 
-            while not finished:
+            received = False
+            while True:
                 notification = subscription.next()
+                if not received:
+                    if not _is_inbox_receipt(notification, self.id, message_id):
+                        continue
+                    received = True
                 collect(notification)
+                if (
+                    notification.method == "session.status"
+                    and notification.payload.get("sessionId") == self.id
+                    and notification.payload.get("status") == "idle"
+                ):
+                    break
 
-        return TurnResult(
+        return RunResult(
             session_id=self.id,
-            status=status,
             final_response=final_response(events),
             events=events,
             notifications=notifications,
@@ -178,6 +180,19 @@ class Session:
         )
 
 
+def _is_inbox_receipt(notification: Notification, session_id: str, message_id: str) -> bool:
+    if notification.method != "session.event" or notification.payload.get("sessionId") != session_id:
+        return False
+    event = notification.payload.get("event")
+    if not isinstance(event, dict) or event.get("type") != "agent/inbox/spliced":
+        return False
+    data = event.get("data")
+    inserted = data.get("inserted") if isinstance(data, dict) else None
+    return isinstance(inserted, list) and any(
+        isinstance(message, dict) and message.get("id") == message_id for message in inserted
+    )
+
+
 def normalize_input(input: str | list[JsonObject]) -> list[JsonObject]:
     if isinstance(input, str):
         return [{"type": "text", "text": input}]

python/sdk/src/deepseek_harness/client.py

@@ -10,7 +10,7 @@ import uuid
 from collections import deque
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Callable, Literal, TypeAlias, TypeVar
+from typing import Callable, TypeAlias, TypeVar
 
 from pydantic import BaseModel
 
@@ -142,9 +142,9 @@ class HarnessClient:
         *,
         on_notification: Callable[[Notification], None] | None = None,
         notification_subscription: "NotificationSubscription | None" = None,
-    ) -> None:
+    ) -> str:
         payload: JsonObject = {"sessionId": session_id, "contentBlocks": content_blocks}
-        self.request(
+        response = self.request(
             "session/prompt",
             payload,
             response_model=_SessionPromptResponse,
@@ -152,6 +152,7 @@ class HarnessClient:
             notification_filter=self._notification_belongs_to_session_tree(session_id),
             notification_subscription=notification_subscription,
         )
+        return response.messageId
 
     def request(
         self,
@@ -536,7 +537,7 @@ class NotificationSubscription:
 
 
 class _SessionPromptResponse(BaseModel):
-    accepted: Literal[True]
+    messageId: str
 
 
 class _ShutdownResponse(BaseModel):

python/sdk/tests/manual_sdk_agent_smoke.py

@@ -73,9 +73,7 @@ def run_smoke(repo_root: Path, keep_sessions: bool) -> None:
                 "Please reply with a short confirmation and do not call tools.",
                 session_id="sdk-smoke-main",
             )
-        print(f"turn_status={result.status}")
         print(f"final_response={result.final_response}")
-        assert result.status == "ok", result
         assert "configured HTTP model endpoint" in result.final_response
         assert len(MockCompletionHandler.requests) == 1
         request = MockCompletionHandler.requests[0]

python/sdk/tests/test_client.py

@@ -39,6 +39,9 @@ for line in sys.stdin:
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
     elif method == "session/prompt":
         params = msg.get("params") or {}
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({
             "jsonrpc": "2.0",
             "method": "session.event",
@@ -57,10 +60,9 @@ for line in sys.stdin:
         }), flush=True)
         print(json.dumps({
             "jsonrpc": "2.0",
-            "method": "session.finished",
-            "params": {"sessionId": params["sessionId"], "status": "ok"},
+            "method": "session.status",
+            "params": {"sessionId": params["sessionId"], "status": "idle"},
         }), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -83,9 +85,8 @@ for line in sys.stdin:
     ) as harness:
         result = harness.run("say hello", session_id="main")
 
-    assert result.status == "ok"
     assert result.final_response == "hello from runtime"
-    assert result.events[0]["type"] == "assistant/message"
+    assert result.events[-1]["type"] == "assistant/message"
     dumped_env = json.loads(env_dump.read_text())
     assert dumped_env["DEEPSEEK_API_KEY"] == "env-key"
     assert dumped_env["DEEPSEEK_BASE_URL"] == "http://127.0.0.1:4321"
@@ -113,9 +114,11 @@ for line in sys.stdin:
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
     elif method == "session/prompt":
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "main", "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "main", "childSessionId": "child"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": "main", "status": "ok"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -133,8 +136,7 @@ for line in sys.stdin:
             on_notification=lambda notification: seen.append(notification.method),
         )
 
-    assert result.status == "ok"
-    assert seen == ["subagent.started", "session.finished"]
+    assert seen == ["session.event", "session.status", "subagent.started", "session.status"]
 
 
 def test_relative_cwd_is_absolute_in_process_environment_and_wire(
@@ -189,10 +191,12 @@ for line in sys.stdin:
     if method == "initialize":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
     elif method == "session/prompt":
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "main", "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "main", "childSessionId": "child"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.finished", "params": {"parentSessionId": "main", "childSessionId": "child", "status": "ok", "stopReason": "completed"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": "main", "status": "ok"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "main", "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -205,11 +209,12 @@ for line in sys.stdin:
     ) as harness:
         result = harness.run("spawn a helper", session_id="main")
 
-    assert result.status == "ok"
     assert [notification.method for notification in result.notifications] == [
+        "session.event",
+        "session.status",
         "subagent.started",
         "subagent.finished",
-        "session.finished",
+        "session.status",
     ]
 
 
@@ -229,6 +234,9 @@ for line in sys.stdin:
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
     elif method == "session/prompt":
         root = (msg.get("params") or {})["sessionId"]
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": root, "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": root, "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": root, "childSessionId": "child"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "child", "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "child response"}]}}}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.started", "params": {"parentSessionId": "child", "childSessionId": "grandchild"}}), flush=True)
@@ -236,8 +244,7 @@ for line in sys.stdin:
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.finished", "params": {"parentSessionId": "child", "childSessionId": "grandchild", "status": "ok"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "subagent.finished", "params": {"parentSessionId": root, "childSessionId": "child", "status": "ok"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": root, "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "root response"}]}}}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": root, "status": "ok"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": root, "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -256,10 +263,11 @@ for line in sys.stdin:
         )
         assert harness.client._notifications.qsize() == 0
 
-    assert result.status == "ok"
     assert result.final_response == "root response"
-    assert [event["data"]["content"][0]["text"] for event in result.events] == ["root response"]
+    assert [event["data"]["content"][0]["text"] for event in result.events if event["type"] == "assistant/message"] == ["root response"]
     assert [notification.method for notification in result.notifications] == [
+        "session.event",
+        "session.status",
         "subagent.started",
         "session.event",
         "subagent.started",
@@ -267,7 +275,7 @@ for line in sys.stdin:
         "subagent.finished",
         "subagent.finished",
         "session.event",
-        "session.finished",
+        "session.status",
     ]
     assert seen == [notification.method for notification in result.notifications]
 
@@ -287,10 +295,12 @@ for line in sys.stdin:
     elif method == "session/prompt":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "other", "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "wrong session"}]}}}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": "other", "status": "ok"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": "other", "status": "idle"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "right session"}]}}}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": params["sessionId"], "status": "ok"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -303,9 +313,8 @@ for line in sys.stdin:
     ) as harness:
         result = harness.run("stay in your lane", session_id="main")
 
-    assert result.status == "ok"
     assert result.final_response == "right session"
-    assert [notification.payload.get("sessionId") for notification in result.notifications] == ["main", "main"]
+    assert [notification.payload.get("sessionId") for notification in result.notifications] == ["main"] * 4
 
 
 def test_high_level_session_run_does_not_accumulate_global_notifications(tmp_path: Path) -> None:
@@ -322,9 +331,11 @@ for line in sys.stdin:
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
     elif method == "session/prompt":
         params = msg.get("params") or {}
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": "message-1"}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
         print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": params["sessionId"], "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "ok"}]}}}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": params["sessionId"], "status": "ok"}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": params["sessionId"], "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -333,11 +344,10 @@ for line in sys.stdin:
 
     with DeepSeekHarness(launch_args_override=(sys.executable, str(script)), cwd=str(tmp_path)) as harness:
         result = harness.run("one turn", session_id="main")
-        assert result.status == "ok"
         assert harness.client._notifications.qsize() == 0
 
 
-def test_session_run_waits_for_late_finished_without_replaying_stale_notifications(tmp_path: Path) -> None:
+def test_session_run_waits_for_late_idle_without_replaying_stale_notifications(tmp_path: Path) -> None:
     script = tmp_path / "fake_runtime.py"
     script.write_text(
         """
@@ -355,15 +365,17 @@ for line in sys.stdin:
         turn += 1
         params = msg.get("params") or {}
         session_id = params["sessionId"]
+        message_id = f"message-{turn}"
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": session_id, "event": {"type": "agent/inbox/spliced", "data": {"target": "next-turn", "start": 0, "inserted": [{"id": message_id}]}}}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": session_id, "status": "running"}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": message_id}}), flush=True)
         if turn == 1:
             print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": session_id, "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "first"}]}}}}), flush=True)
-            print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": session_id, "status": "ok"}}), flush=True)
-            print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+            print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": session_id, "status": "idle"}}), flush=True)
         else:
-            print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
             time.sleep(0.05)
             print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": session_id, "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "second"}]}}}}), flush=True)
-            print(json.dumps({"jsonrpc": "2.0", "method": "session.finished", "params": {"sessionId": session_id, "status": "ok"}}), flush=True)
+            print(json.dumps({"jsonrpc": "2.0", "method": "session.status", "params": {"sessionId": session_id, "status": "idle"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -376,7 +388,7 @@ for line in sys.stdin:
 
     assert first.final_response == "first"
     assert second.final_response == "second"
-    assert [notification.payload.get("sessionId") for notification in second.notifications] == ["main", "main"]
+    assert [notification.payload.get("sessionId") for notification in second.notifications] == ["main"] * 4
 
 
 def test_client_starts_subprocess_sends_requests_and_routes_notifications(tmp_path: Path) -> None:
@@ -394,7 +406,7 @@ for line in sys.stdin:
     elif method == "session/prompt":
         params = msg.get("params") or {}
         print(json.dumps({"jsonrpc": "2.0", "method": "llm/request", "params": {"requestId": "req-1", "sessionId": params["sessionId"], "model": "dsagent", "messages": []}}), flush=True)
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break
@@ -531,7 +543,7 @@ for line in sys.stdin:
     elif method in {"emit-first", "emit-second"}:
         print(json.dumps({"jsonrpc": "2.0", "method": "tick", "params": {"source": method}}), flush=True)
     elif method == "session/prompt":
-        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"accepted": True}}), flush=True)
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"messageId": "message-1"}}), flush=True)
     elif method == "shutdown":
         print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
         break