Merge pull request #584 from deepseek-harness/worktree/fix-python-sdk-nested-subagent-notifications

fix(python-sdk): track recursive subagent notifications

.agents/notes/implemented/bug-fix/2026-07-24-recursive-python-sdk-session-notifications.i18n.yaml

@@ -0,0 +1,6 @@
+# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each
+# 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
+2026-07-24-recursive-python-sdk-session-notifications.md: c90213659391b565acd043a1be64e225f8babd31
+2026-07-24-recursive-python-sdk-session-notifications.zh.md: 214a5ef924dcc9da3a97aab6385837acd2b364d9

.agents/notes/implemented/bug-fix/2026-07-24-recursive-python-sdk-session-notifications.md

@@ -0,0 +1,29 @@
+# Agent Note: Recursive Python SDK session notifications
+
+Status: implemented
+
+English | [中文](2026-07-24-recursive-python-sdk-session-notifications.zh.md)
+
+## Problem
+
+The Python SDK filtered turn notifications by comparing each payload directly with the root session id. This admitted a direct child's lifecycle because its parent id named the root, but rejected a grandchild's lifecycle and every descendant `session.event`. The JSON-RPC server still emitted those notifications, so they accumulated on the low-level global queue while high-level consumers lost nested trajectory relationships and completion states.
+
+## Decision
+
+`HarnessClient` records every valid `subagent.started` child-to-parent edge before dispatching the notification. A later `subagent.finished` routes by its immutable parent id but never rewrites current ancestry, so an older run that settles after its child id has been reused cannot displace the replacement session. Other session notifications resolve their session id by walking that client-lifetime ancestry graph to the requested root. The graph survives successive subscriptions so a descendant that outlives one `Session.run()` remains attributable when it emits during a later turn, and it resets when the client starts a new runtime process.
+
+`Session.run()` delivers the complete discovered session-tree notification stream through `TurnResult.notifications` and `on_notification`. Only `session.event` notifications whose `sessionId` equals the requested root enter `TurnResult.events` or final-response reconstruction. Descendant events are therefore observable without allowing a child response to replace the root response.
+
+## Alternatives considered
+
+**Add a root session id to every JSON-RPC notification.** The server already provides exact immediate-parent edges, and duplicating transitive ancestry on the wire would make every producer responsible for client subscription state.
+
+**Limit subagents to one level.** A deployment can set `maxDepth: 1`, but changing the SDK to depend on that policy would silently misreport valid recursive compositions.
+
+**Subscribe only to descendant lifecycle notifications.** This would repair relation and completion reporting, but descendant session events would continue accumulating on the global queue and callbacks would expose an incomplete tree.
+
+**Expose and index every subagent run id on the JSON-RPC wire.** Exact run identity is useful when a client must correlate two concurrent outcomes for the same child, but session-tree routing already has the authoritative start edge and each terminal notification's immutable parent. Expanding the protocol is unnecessary for this ownership decision.
+
+## Consequences
+
+High-level consumers receive nested lifecycle and session notifications in wire order while root turn results preserve their prior response semantics. The client retains one current parent entry per observed child until the runtime restarts; ancestry lookup is cycle-safe, and unrelated session notifications remain available through the global queue. Keyless Python tests cover two-level delegation, root-response isolation, absence of tree-notification queue buildup, ancestry reuse across subscriptions, and reused child ids whose older runs settle out of order.

.agents/notes/implemented/bug-fix/2026-07-24-recursive-python-sdk-session-notifications.zh.md

@@ -0,0 +1,29 @@
+# Agent Note: Python SDK 递归会话通知
+
+Status: implemented
+
+[English](2026-07-24-recursive-python-sdk-session-notifications.md) | 中文
+
+## 问题
+
+Python SDK 过去通过将每条通知的 payload 与根会话 ID 直接比较来过滤轮次通知。直接子 agent 的生命周期通知因 parent ID 指向根会话而能够通过，但孙级生命周期通知与所有后代 `session.event` 都会被拒绝。JSON-RPC 服务器仍会发出这些通知，因此它们会堆积在底层全局队列中，而高层消费者会丢失嵌套轨迹的关系与结束状态。
+
+## 决策
+
+`HarnessClient` 会在分发通知前，记录每条有效 `subagent.started` 所包含的 child-to-parent（子到父）关系。后续的 `subagent.finished` 会依据自身不可变的 parent ID 路由，但不会改写当前祖先关系，因此旧 run 即使在其 child ID 已被复用后才结束，也无法覆盖替代它的新会话。其他会话通知会沿客户端生命周期内保存的祖先关系图回溯自身 session ID，判断它们是否属于请求的根会话。该关系图会跨连续订阅保留，因此某个后代即使跨过一次 `Session.run()`，在后续轮次中发出通知时仍能正确归属；客户端启动新的运行时进程时会重置关系图。
+
+`Session.run()` 通过 `TurnResult.notifications` 与 `on_notification` 提供已发现会话树的完整通知流。只有 `sessionId` 等于请求根会话的 `session.event` 才会进入 `TurnResult.events` 或参与最终回复重建。因此调用方能够观察后代事件，同时子会话回复不会覆盖根会话回复。
+
+## 考虑过的替代方案
+
+**在每条 JSON-RPC 通知中加入根会话 ID。** 服务器已经提供精确的直接父子关系；在线路协议中重复传递祖先关系，会迫使每个生产者承担客户端订阅状态的职责。
+
+**把 subagent 限制为一层。** 部署可以设置 `maxDepth: 1`，但让 SDK 依赖该策略，会对合法的递归组合产生静默误报。
+
+**只订阅后代生命周期通知。** 这可以修复关系与结束状态的上报，但后代会话事件仍会堆积在全局队列中，回调看到的会话树也不完整。
+
+**在 JSON-RPC 线路上公开并索引每个 subagent run ID。** 当客户端必须关联同一 child 的两个并发结果时，精确 run 身份很有价值；但会话树路由已经拥有权威 start 关系和每条终止通知中不可变的 parent。没有必要为这一归属决策扩展协议。
+
+## 后果
+
+高层消费者会按线上的原始顺序收到嵌套生命周期与会话通知，同时根轮次结果保持原有回复语义。客户端会为每个已观察到的子会话保留一条当前父关系，直到运行时重启；祖先回溯能够安全处理环，无关会话通知仍可从全局队列获取。无密钥 Python 测试覆盖两层派生、根回复隔离、会话树通知不堆积、跨订阅复用祖先关系，以及旧 run 乱序结束的复用 child ID。

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
-README.md: 23d15d617b3d295a6cc2d8d20c6d03abc226834b
-README.zh.md: 4f6aef13833af937babc2e5a92bfd14c12170534
+README.md: bfa31a712acd6fccf1458a0a80fc2ff80dfe114e
+README.zh.md: 11ebcdd133b2fd839b73f50ef2be2e531e8bbc2a

python/sdk/README.md

@@ -34,9 +34,7 @@ with DeepSeekHarness(
 
 `provider` selects a provider route registered by the chosen Cordis composition; `model` is the model id resolved by that adapter. The bundled default composition registers `deepseek`. 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.
 
-`TurnResult.final_response` is the text content from the last
-`assistant/message` event in the turn. Use `TurnResult.events` for the complete
-event stream, including intermediate assistant messages and tool activity.
+`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.
 
 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

@@ -30,7 +30,7 @@ with DeepSeekHarness(
 
 `provider` 用于选择当前 Cordis 组合已注册的提供方路由；`model` 是该适配器解析的模型 ID。内置默认组合注册 `deepseek`。自定义组合可以挂载 `llm-pi-ai`，在其中配置各提供方的凭据与端点，再选择 pi-ai 已安装目录中的任意提供方/模型组合。
 
-`TurnResult.final_response` 是本轮次最后一个 `assistant/message` 事件的文本内容。完整的事件流（包括中间的助手消息与工具活动）用 `TurnResult.events` 获取。
+`HarnessClient` 会在运行时进程的生命周期内保留已发现的 subagent（子 agent）祖先关系。每次执行 `Session.run()` 时，`TurnResult.notifications` 与 `on_notification` 会按线上的原始顺序收到根会话及所有已知后代的通知，其中包括嵌套 subagent 的生命周期与会话事件。`TurnResult.events` 仍只保存根会话的完整事件流，`TurnResult.final_response` 则取该会话最后一个 `assistant/message` 的文本内容，因此后代消息不会覆盖根会话回复。
 
 同样的行为也可以通过 `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/api.py

@@ -143,7 +143,10 @@ class Session:
             notifications.append(notification)
             if on_notification is not None:
                 on_notification(notification)
-            if notification.method == "session.event":
+            if (
+                notification.method == "session.event"
+                and notification.payload.get("sessionId") == self.id
+            ):
                 event = notification.payload.get("event")
                 if isinstance(event, dict):
                     events.append(event)

python/sdk/src/deepseek_harness/client.py

@@ -47,6 +47,7 @@ class HarnessClient:
         self._notification_subscribers: dict[
             str, tuple[queue.Queue[Notification | BaseException], NotificationFilter | None]
         ] = {}
+        self._session_parents: dict[str, str] = {}
         self._requests: queue.Queue[IncomingRequest | BaseException] = queue.Queue()
         self._stderr_lines: deque[str] = deque(maxlen=400)
         self._reader_thread: threading.Thread | None = None
@@ -62,6 +63,8 @@ class HarnessClient:
     def start(self) -> None:
         if self._proc is not None:
             return
+        with self._lock:
+            self._session_parents.clear()
         args = list(self.config.launch_args_override or self._default_launch_args())
         env = os.environ.copy()
         if self.config.env:
@@ -143,7 +146,7 @@ class HarnessClient:
             payload,
             response_model=_SessionPromptResponse,
             on_notification=on_notification,
-            notification_filter=_notification_belongs_to_session(session_id),
+            notification_filter=self._notification_belongs_to_session_tree(session_id),
             notification_subscription=notification_subscription,
         )
 
@@ -193,7 +196,8 @@ class HarnessClient:
         return NotificationSubscription(self, subscription_id, notifications)
 
     def subscribe_session_notifications(self, session_id: str) -> "NotificationSubscription":
-        return self.subscribe_notifications(_notification_belongs_to_session(session_id))
+        """Subscribe to a session and descendants discovered from subagent lifecycle edges."""
+        return self.subscribe_notifications(self._notification_belongs_to_session_tree(session_id))
 
     def next_request(self) -> IncomingRequest:
         item = self._requests.get()
@@ -352,6 +356,7 @@ class HarnessClient:
             params = message.get("params")
             notification = Notification(method=method, payload=params if isinstance(params, dict) else {})
             with self._lock:
+                self._record_session_relationship_locked(notification)
                 subscribers = list(self._notification_subscribers.items())
             delivered = False
             for subscription_id, (subscriber, predicate) in subscribers:
@@ -439,6 +444,52 @@ class HarnessClient:
         with self._lock:
             self._notification_subscribers.pop(subscription_id, None)
 
+    def _record_session_relationship_locked(self, notification: Notification) -> None:
+        if notification.method != "subagent.started":
+            return
+        parent_id = notification.payload.get("parentSessionId")
+        child_id = notification.payload.get("childSessionId")
+        if (
+            isinstance(parent_id, str)
+            and parent_id
+            and isinstance(child_id, str)
+            and child_id
+            and parent_id != child_id
+        ):
+            self._session_parents[child_id] = parent_id
+
+    def _notification_belongs_to_session_tree(self, session_id: str) -> NotificationFilter:
+        def belongs(notification: Notification) -> bool:
+            payload = notification.payload
+            if notification.method in {"subagent.started", "subagent.finished"}:
+                parent_id = payload.get("parentSessionId")
+                if (
+                    isinstance(parent_id, str)
+                    and self._session_is_descendant_of(parent_id, session_id)
+                ):
+                    return True
+                return payload.get("childSessionId") == session_id
+            related_id = payload.get("sessionId")
+            return (
+                isinstance(related_id, str)
+                and self._session_is_descendant_of(related_id, session_id)
+            )
+
+        return belongs
+
+    def _session_is_descendant_of(self, session_id: str, root_session_id: str) -> bool:
+        current = session_id
+        visited: set[str] = set()
+        while current not in visited:
+            if current == root_session_id:
+                return True
+            visited.add(current)
+            parent = self._session_parents.get(current)
+            if parent is None:
+                return False
+            current = parent
+        return False
+
 
 class NotificationSubscription:
     def __init__(
@@ -491,15 +542,3 @@ class _ShutdownResponse(BaseModel):
 
 def _int_or_none(value: object) -> int | None:
     return value if isinstance(value, int) else None
-
-
-def _notification_belongs_to_session(session_id: str) -> NotificationFilter:
-    def belongs(notification: Notification) -> bool:
-        payload = notification.payload
-        return (
-            payload.get("sessionId") == session_id
-            or payload.get("parentSessionId") == session_id
-            or payload.get("childSessionId") == session_id
-        )
-
-    return belongs

python/sdk/tests/test_client.py

@@ -9,7 +9,7 @@ from pathlib import Path
 
 import pytest
 
-from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig
+from deepseek_harness import DeepSeekHarness, HarnessClient, HarnessConfig, Notification
 
 
 def test_high_level_sdk_runs_turn_and_collects_final_response(tmp_path: Path) -> None:
@@ -198,6 +198,65 @@ for line in sys.stdin:
     ]
 
 
+def test_session_run_collects_nested_subagent_tree_without_polluting_root_events(
+    tmp_path: Path,
+) -> None:
+    script = tmp_path / "fake_runtime.py"
+    script.write_text(
+        """
+import json
+import sys
+
+for line in sys.stdin:
+    msg = json.loads(line)
+    method = msg.get("method")
+    if method == "initialize":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {"serverInfo": {"name": "fake-runtime"}}}), flush=True)
+    elif method == "session/prompt":
+        root = (msg.get("params") or {})["sessionId"]
+        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)
+        print(json.dumps({"jsonrpc": "2.0", "method": "session.event", "params": {"sessionId": "grandchild", "event": {"type": "assistant/message", "data": {"content": [{"type": "text", "text": "grandchild response"}]}}}}), flush=True)
+        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)
+    elif method == "shutdown":
+        print(json.dumps({"jsonrpc": "2.0", "id": msg["id"], "result": {}}), flush=True)
+        break
+""".strip()
+    )
+
+    seen: list[str] = []
+    with DeepSeekHarness(
+        launch_args_override=(sys.executable, str(script)),
+        cwd=str(tmp_path),
+    ) as harness:
+        result = harness.run(
+            "delegate recursively",
+            session_id="main",
+            on_notification=lambda notification: seen.append(notification.method),
+        )
+        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 [notification.method for notification in result.notifications] == [
+        "subagent.started",
+        "session.event",
+        "subagent.started",
+        "session.event",
+        "subagent.finished",
+        "subagent.finished",
+        "session.event",
+        "session.finished",
+    ]
+    assert seen == [notification.method for notification in result.notifications]
+
+
 def test_session_run_ignores_notifications_for_other_sessions(tmp_path: Path) -> None:
     script = tmp_path / "fake_runtime.py"
     script.write_text(
@@ -356,6 +415,92 @@ def test_client_keeps_unmatched_notifications_available_globally_while_subscribe
         assert notification.payload["sessionId"] == "other"
 
 
+def test_session_subscription_keeps_descendant_relationships_across_subscriptions() -> None:
+    client = HarnessClient()
+    with client.subscribe_session_notifications("main") as first:
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "subagent.started",
+            "params": {"parentSessionId": "main", "childSessionId": "child"},
+        })
+        assert first.next().payload["childSessionId"] == "child"
+
+    with client.subscribe_session_notifications("main") as second:
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "subagent.started",
+            "params": {"parentSessionId": "child", "childSessionId": "grandchild"},
+        })
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "session.event",
+            "params": {"sessionId": "grandchild", "event": {"type": "assistant/message"}},
+        })
+        assert second.next().payload["childSessionId"] == "grandchild"
+        assert second.next().payload["sessionId"] == "grandchild"
+
+    assert client._notifications.qsize() == 0
+
+
+def test_session_subscription_preserves_reused_child_ancestry_after_late_finish() -> None:
+    client = HarnessClient()
+    old_seen: list[Notification] = []
+    new_seen: list[Notification] = []
+    with (
+        client.subscribe_session_notifications("old-parent") as old_subscription,
+        client.subscribe_session_notifications("new-parent") as new_subscription,
+    ):
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "subagent.started",
+            "params": {"parentSessionId": "old-parent", "childSessionId": "reused-child"},
+        })
+        old_subscription.drain(old_seen.append)
+        new_subscription.drain(new_seen.append)
+        assert [notification.method for notification in old_seen] == ["subagent.started"]
+        assert new_seen == []
+
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "subagent.started",
+            "params": {"parentSessionId": "new-parent", "childSessionId": "reused-child"},
+        })
+        old_subscription.drain(old_seen.append)
+        new_subscription.drain(new_seen.append)
+        assert [notification.method for notification in new_seen] == ["subagent.started"]
+
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "subagent.finished",
+            "params": {"parentSessionId": "old-parent", "childSessionId": "reused-child"},
+        })
+        old_subscription.drain(old_seen.append)
+        new_subscription.drain(new_seen.append)
+        assert [notification.method for notification in old_seen] == [
+            "subagent.started",
+            "subagent.finished",
+        ]
+        assert [notification.method for notification in new_seen] == ["subagent.started"]
+
+        client._handle_message({
+            "jsonrpc": "2.0",
+            "method": "session.event",
+            "params": {"sessionId": "reused-child", "event": {"type": "assistant/message"}},
+        })
+        old_subscription.drain(old_seen.append)
+        new_subscription.drain(new_seen.append)
+
+    assert [notification.method for notification in old_seen] == [
+        "subagent.started",
+        "subagent.finished",
+    ]
+    assert [notification.method for notification in new_seen] == [
+        "subagent.started",
+        "session.event",
+    ]
+    assert client._notifications.qsize() == 0
+
+
 def test_client_contains_notification_filter_failure_to_its_subscription(tmp_path: Path) -> None:
     script = tmp_path / "fake_bridge.py"
     script.write_text(