fix(python-sdk): preserve reused session ancestry

.agents/notes/implemented/bug-fix/2026-07-24-recursive-python-sdk-session-notifications.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
-2026-07-24-recursive-python-sdk-session-notifications.md: c608cfa584e568602d599c7e18ffc36fabbd0186
-2026-07-24-recursive-python-sdk-session-notifications.zh.md: 397669cc13a5c77dabce55209be3cd2f0c7a82ea
+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

@@ -10,7 +10,7 @@ The Python SDK filtered turn notifications by comparing each payload directly wi
 
 ## Decision
 
-`HarnessClient` records every valid `subagent.started` and `subagent.finished` child-to-parent edge before dispatching the notification. Session subscriptions classify each payload session id, parent id, and child id by walking that client-lifetime ancestry graph to their 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.
+`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.
 
@@ -22,6 +22,8 @@ The Python SDK filtered turn notifications by comparing each payload directly wi
 
 **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 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, and ancestry reuse across subscriptions.
+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

@@ -10,7 +10,7 @@ Python SDK 过去通过将每条通知的 payload 与根会话 ID 直接比较
 
 ## 决策
 
-`HarnessClient` 会在分发通知前，记录每条有效 `subagent.started` 和 `subagent.finished` 所包含的 child-to-parent（子到父）关系。会话订阅会沿客户端生命周期内保存的祖先关系图回溯每个 payload 中的 session ID、parent ID 与 child ID，判断它们是否属于请求的根会话。该关系图会跨连续订阅保留，因此某个后代即使跨过一次 `Session.run()`，在后续轮次中发出通知时仍能正确归属；客户端启动新的运行时进程时会重置关系图。
+`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` 或参与最终回复重建。因此调用方能够观察后代事件，同时子会话回复不会覆盖根会话回复。
 
@@ -22,6 +22,8 @@ Python SDK 过去通过将每条通知的 payload 与根会话 ID 直接比较
 
 **只订阅后代生命周期通知。** 这可以修复关系与结束状态的上报，但后代会话事件仍会堆积在全局队列中，回调看到的会话树也不完整。
 
+**在 JSON-RPC 线路上公开并索引每个 subagent run ID。** 当客户端必须关联同一 child 的两个并发结果时，精确 run 身份很有价值；但会话树路由已经拥有权威 start 关系和每条终止通知中不可变的 parent。没有必要为这一归属决策扩展协议。
+
 ## 后果
 
-高层消费者会按线上的原始顺序收到嵌套生命周期与会话通知，同时根轮次结果保持原有回复语义。客户端会为每个已观察到的子会话保留一条父关系，直到运行时重启；祖先回溯能够安全处理环，无关会话通知仍可从全局队列获取。无密钥 Python 测试覆盖两层派生、根回复隔离、会话树通知不堆积，以及跨订阅复用祖先关系。
+高层消费者会按线上的原始顺序收到嵌套生命周期与会话通知，同时根轮次结果保持原有回复语义。客户端会为每个已观察到的子会话保留一条当前父关系，直到运行时重启；祖先回溯能够安全处理环，无关会话通知仍可从全局队列获取。无密钥 Python 测试覆盖两层派生、根回复隔离、会话树通知不堆积、跨订阅复用祖先关系，以及旧 run 乱序结束的复用 child ID。

python/sdk/src/deepseek_harness/client.py

@@ -445,7 +445,7 @@ class HarnessClient:
             self._notification_subscribers.pop(subscription_id, None)
 
     def _record_session_relationship_locked(self, notification: Notification) -> None:
-        if notification.method not in {"subagent.started", "subagent.finished"}:
+        if notification.method != "subagent.started":
             return
         parent_id = notification.payload.get("parentSessionId")
         child_id = notification.payload.get("childSessionId")
@@ -461,15 +461,18 @@ class HarnessClient:
     def _notification_belongs_to_session_tree(self, session_id: str) -> NotificationFilter:
         def belongs(notification: Notification) -> bool:
             payload = notification.payload
-            related_ids = (
-                payload.get("sessionId"),
-                payload.get("parentSessionId"),
-                payload.get("childSessionId"),
-            )
-            return any(
+            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)
-                for related_id in related_ids
             )
 
         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:
@@ -442,6 +442,65 @@ def test_session_subscription_keeps_descendant_relationships_across_subscription
     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(