fix(relay): gate skipped turn metrics

Signed-off-by: Bryan Bednarski <bbednarski@nvidia.com>

agent/relay_runtime.py

@@ -811,6 +811,12 @@ def current_turn() -> RelayTurnContext | None:
     return _CURRENT_TURN.get()
 
 
+def relay_instrumentation_enabled() -> bool:
+    """Return whether this inherited turn may create Relay instrumentation."""
+    turn = current_turn()
+    return turn is None or turn.relay_enabled
+
+
 def active_turn(session_id: str | None = None) -> RelayTurnContext | None:
     """Return a live turn only when it belongs to the active profile/session."""
     turn = current_turn()

hermes_cli/observability/relay_shared_metrics.py

@@ -671,6 +671,8 @@ def observe_lifecycle(hook_name: str, **kwargs: Any) -> None:
     """Project one Hermes lifecycle event into the core Relay integration."""
     if not handles_hook(hook_name):
         return
+    if not relay_runtime.relay_instrumentation_enabled():
+        return
     runtime = _get_runtime()
     if runtime is None:
         return

tests/hermes_cli/test_relay_shared_metrics_runtime.py

@@ -835,6 +835,43 @@ def test_concurrent_turn_skips_relay_before_scope_stack_can_interleave(
     assert len(turn_closes) == 1
 
 
+def test_concurrent_turn_skips_shared_metrics_scope_creation(direct_runtime):
+    coordinator = relay_runtime.SESSION_COORDINATOR
+    profile_key = relay_runtime.current_profile_key()
+    lease = coordinator.acquire_conversation(
+        profile_key=profile_key,
+        session_id="shared-session",
+        platform="cli",
+    )
+    first = coordinator.begin_turn(lease, turn_id="first", task_id="first-task")
+    second = coordinator.begin_turn(lease, turn_id="second", task_id="second-task")
+
+    relay_shared_metrics.observe_lifecycle(
+        "pre_llm_call",
+        session_id="shared-session",
+        task_id="second-task",
+        platform="cli",
+    )
+    relay_shared_metrics.observe_lifecycle(
+        "pre_api_request",
+        session_id="shared-session",
+        task_id="second-task",
+        api_request_id="second-request",
+        platform="cli",
+    )
+
+    assert second.relay_enabled is False
+    assert not [
+        event
+        for event in direct_runtime.events
+        if event[0] == "scope.push" and event[1] == relay_shared_metrics.TASK_SCOPE
+    ]
+
+    coordinator.end_turn(first, outcome="success")
+    coordinator.end_turn(second, outcome="success")
+    coordinator.release_conversation(lease)
+
+
 
 
 
@@ -1016,4 +1053,3 @@ def test_failed_flush_keeps_daily_export_open_for_later_task(
 
 
 
-