fix(relay): close child sessions inside turn ownership

Signed-off-by: Alex Fournier <afournier@nvidia.com>

agent/relay_runtime.py

@@ -635,8 +635,8 @@ class RelaySessionCoordinator:
                 self._reset_turn_context(turn)
                 return
             turn.closed = True
+            lease = turn.lease
             try:
-                lease = turn.lease
                 if isinstance(lease.host, RelayRuntime) and lease.session is not None:
                     self._finish_logical_calls(turn, outcome=outcome)
                     if turn.handle is not None:
@@ -656,8 +656,25 @@ class RelaySessionCoordinator:
                                 "Hermes Relay turn finalization failed", exc_info=True
                             )
             finally:
-                self._unregister_active_turn(turn)
-                self._reset_turn_context(turn)
+                try:
+                    # Delegated agents own one turn. Close their conversation
+                    # while the active-turn guard is still held so a parent
+                    # timeout fallback cannot race this terminal boundary.
+                    if (
+                        lease.parent_session_id
+                        and isinstance(lease.host, RelayRuntime)
+                    ):
+                        lease.host.unregister_subagent({
+                            "child_session_id": lease.session_id
+                        })
+                except Exception:
+                    logger.warning(
+                        "Hermes Relay child conversation finalization failed",
+                        exc_info=True,
+                    )
+                finally:
+                    self._unregister_active_turn(turn)
+                    self._reset_turn_context(turn)
 
     def has_active_turn(self, *, profile_key: str, session_id: str) -> bool:
         """Return whether a turn is still running for one profile/session."""

run_agent.py

@@ -6942,16 +6942,9 @@ class AIAgent:
             finally:
                 try:
                     if relay_lease is not None:
-                        try:
-                            if relay_lease.parent_session_id:
-                                relay_runtime.SESSION_COORDINATOR.finalize_conversation(
-                                    profile_key=relay_lease.profile_key,
-                                    session_id=relay_lease.session_id,
-                                )
-                        finally:
-                            relay_runtime.SESSION_COORDINATOR.release_conversation(
-                                relay_lease
-                            )
+                        relay_runtime.SESSION_COORDINATOR.release_conversation(
+                            relay_lease
+                        )
                 finally:
                     if getattr(self, "_relay_pending_turn_id", None) == relay_turn_id:
                         self._relay_pending_turn_id = None

tests/hermes_cli/test_relay_shared_metrics_runtime.py

@@ -1846,6 +1846,61 @@ def test_coordinator_tracks_active_turns_across_threads(direct_runtime):
     )
 
 
+def test_child_session_closes_before_active_turn_guard_is_released(
+    direct_runtime,
+    monkeypatch,
+):
+    del direct_runtime
+    coordinator = relay_runtime.SESSION_COORDINATOR
+    profile_key = relay_runtime.current_profile_key()
+    parent_lease = coordinator.acquire_conversation(
+        profile_key=profile_key,
+        session_id="parent",
+        platform="cli",
+    )
+    child_lease = coordinator.acquire_conversation(
+        profile_key=profile_key,
+        session_id="child",
+        platform="subagent",
+        parent_session_id="parent",
+    )
+    child_turn = coordinator.begin_turn(
+        child_lease,
+        turn_id="child-turn",
+        task_id="child-task",
+    )
+    runtime = relay_runtime.get_runtime(create=False)
+    assert runtime is not None
+    close_observations = []
+    original_close = runtime.close_session
+
+    def observe_close(event):
+        close_observations.append(
+            coordinator.has_active_turn(
+                profile_key=profile_key,
+                session_id="child",
+            )
+        )
+        original_close(event)
+
+    monkeypatch.setattr(runtime, "close_session", observe_close)
+
+    coordinator.end_turn(child_turn, outcome="success")
+
+    assert close_observations == [True]
+    assert runtime.get_session("child") is None
+    assert not coordinator.has_active_turn(
+        profile_key=profile_key,
+        session_id="child",
+    )
+    coordinator.release_conversation(child_lease)
+    coordinator.release_conversation(parent_lease)
+    coordinator.finalize_conversation(
+        profile_key=profile_key,
+        session_id="parent",
+    )
+
+
 def test_core_runtime_ignores_self_parenting_subagent_event(direct_runtime):
     runtime = relay_runtime.get_runtime()
     assert runtime is not None