diff --git a/adapters/common/src/nemo_fabric_adapters/common/utils.py b/adapters/common/src/nemo_fabric_adapters/common/utils.py
index 96ae9c72..e44d1cc6 100644
--- a/adapters/common/src/nemo_fabric_adapters/common/utils.py
+++ b/adapters/common/src/nemo_fabric_adapters/common/utils.py
@@ -219,9 +219,11 @@ def merge_unique(*values: Any) -> list[str]:
                 merged.append(item)
     return merged
 
+
 def without_none(mapping: dict[str, Any]) -> dict[str, Any]:
     return {key: value for key, value in mapping.items() if value is not None}
 
+
 def dump_yaml(value: dict[str, Any]) -> str:
     try:
         import yaml
@@ -409,7 +411,7 @@ def write_relay_configs(
             relay_config_path.write_text(tomli_w.dumps(relay_config), encoding="utf-8")
 
         if plugin_config is not None:
-            if observability_version != 2:
+            if observability_version not in {2, 3}:
                 raise ValueError(
                     f"unsupported NeMo Relay observability config version {observability_version}"
                 )
diff --git a/adapters/hermes/pyproject.toml b/adapters/hermes/pyproject.toml
index a308eded..5d2f215e 100644
--- a/adapters/hermes/pyproject.toml
+++ b/adapters/hermes/pyproject.toml
@@ -34,11 +34,11 @@ harness = [
   "hermes-agent[mcp]>=0.19.0; python_version < '3.14'",
 ]
 relay = [
-  "nemo-relay>=0.6.0,<0.7",
+  "nemo-relay>=0.6.0,<0.8",
 ]
 full = [
   "hermes-agent[mcp]>=0.19.0; python_version < '3.14'",
-  "nemo-relay>=0.6.0,<0.7",
+  "nemo-relay>=0.6.0,<0.8",
 ]
 
 [project.urls]
diff --git a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py
index 29d69b7a..75eb9951 100755
--- a/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py
+++ b/adapters/hermes/src/nemo_fabric_adapters/hermes/adapter.py
@@ -11,11 +11,13 @@
 from __future__ import annotations
 
 import asyncio
+import copy
 import inspect
 import json
 import logging
 import os
 from contextlib import redirect_stdout
+from importlib.metadata import version as distribution_version
 from io import StringIO
 from pathlib import Path
 from typing import Any
@@ -35,6 +37,28 @@
     "openai": "OPENAI_API_KEY",
     "openrouter": "OPENROUTER_API_KEY",
 }
+# Hermes 0.16+ discovers Relay from this TOML path and falls back to direct
+# ATIF/ATOF only when TOML initialization fails. Clear only those enable flags.
+HERMES_RELAY_ENV_NAMES = (
+    "HERMES_NEMO_RELAY_PLUGINS_TOML",
+    "HERMES_NEMO_RELAY_ATIF_ENABLED",
+    "HERMES_NEMO_RELAY_ATOF_ENABLED",
+)
+
+
+def finalize_hermes_relay_session(session_id: str) -> None:
+    """Finalize one Relay session through the installed Hermes lifecycle API."""
+    try:
+        from hermes_cli.lifecycle import finalize_session
+    except ModuleNotFoundError as error:
+        if error.name != "hermes_cli.lifecycle":
+            raise
+        # Hermes 0.19 exposes the same finalization boundary as a plugin hook.
+        from hermes_cli.plugins import invoke_hook
+
+        invoke_hook("on_session_finalize", session_id=session_id, platform="fabric")
+    else:
+        finalize_session(session_id=session_id, platform="fabric")
 
 
 def _api_key_env(model_config: dict[str, Any]) -> str:
@@ -50,26 +74,6 @@ def _api_key_env(model_config: dict[str, Any]) -> str:
     return default
 
 
-def _fabric_stream_sink_enabled(config: dict[str, Any] | None) -> bool:
-    if config is None:
-        return False
-    for component in config.get("components") or []:
-        if not isinstance(component, dict) or component.get("kind") != "observability":
-            continue
-        component_config = component.get("config")
-        if not isinstance(component_config, dict):
-            continue
-        atof = component_config.get("atof")
-        if not isinstance(atof, dict):
-            continue
-        if any(
-            isinstance(sink, dict) and sink.get("name") == "nemo-fabric-stream"
-            for sink in atof.get("sinks") or []
-        ):
-            return True
-    return False
-
-
 def validate_hermes_telemetry_provider(payload: dict[str, Any]) -> None:
     providers = common_utils.telemetry_providers(payload)
     if any(provider != "relay" for provider in providers):
@@ -153,6 +157,71 @@ def write_hermes_config(
     return config_path, config
 
 
+def write_hermes_relay_plugin_config(
+    payload: dict[str, Any],
+) -> tuple[Path, dict[str, Any]]:
+    """Stage Fabric's resolved Relay config for Hermes' bundled integration."""
+
+    plugin_config = common_utils.load_relay_plugin_config(payload)
+    hermes_plugin_config = copy.deepcopy(plugin_config)
+    relay_version = distribution_version("nemo-relay")
+    try:
+        relay_major, relay_minor = (
+            int(part) for part in relay_version.split(".", maxsplit=2)[:2]
+        )
+    except ValueError as error:
+        raise RuntimeError(
+            f"unsupported NeMo Relay version {relay_version!r}"
+        ) from error
+    observability_version = 3 if (relay_major, relay_minor) >= (0, 7) else 2
+    for component in hermes_plugin_config.get("components", []):
+        if component.get("kind") != "observability":
+            continue
+        observability = component.get("config")
+        if not isinstance(observability, dict):
+            continue
+
+        if observability_version == 3 and observability.get("version") != 3:
+            # Relay 0.7 combines Fabric's legacy OTLP and OpenInference exporter
+            # settings into typed OpenTelemetry endpoints in its v3 schema.
+            endpoints = []
+            for config_name, endpoint_type in (
+                ("opentelemetry", "full"),
+                ("openinference", "openinference"),
+            ):
+                exporter = observability.pop(config_name, None)
+                if not isinstance(exporter, dict) or not exporter.get("enabled"):
+                    continue
+                endpoint = {
+                    key: value
+                    for key, value in exporter.items()
+                    if key != "enabled" and value is not None
+                }
+                endpoint["type"] = endpoint_type
+                endpoints.append(endpoint)
+            if endpoints:
+                observability["opentelemetry"] = {
+                    "enabled": True,
+                    "endpoints": endpoints,
+                }
+            observability["version"] = 3
+
+        # Fabric finalizes Hermes' Relay session after every invocation. Each
+        # finalization reinitializes Relay for the next turn, so a file sink
+        # cannot overwrite the runtime-scoped artifact it created previously.
+        for sink in (observability.get("atof") or {}).get("sinks") or []:
+            if isinstance(sink, dict) and sink.get("type") == "file":
+                if sink.get("mode") == "overwrite":
+                    sink["mode"] = "append"
+    _, plugin_config_path = common_utils.write_relay_configs(
+        plugin_config=hermes_plugin_config,
+        observability_version=observability_version,
+    )
+    if plugin_config_path is None:
+        raise RuntimeError("Hermes Relay plugin configuration was not generated")
+    return plugin_config_path, plugin_config
+
+
 def hermes_mcp_server_config(server: dict[str, Any]) -> dict[str, Any]:
     transport = str(server.get("transport") or "").strip().lower()
     raw_target = server.get("url")
@@ -231,13 +300,9 @@ def __init__(self) -> None:
         self._conversation_history: list[dict[str, Any]] | None = None
         self._session_db: Any = None
         self._agent: Any = None
-        self._invoke_hook: Any = None
         self._relay_plugin_config: dict[str, Any] | None = None
-        self._relay_context: Any = None
-        self._relay_context_entered = False
-        self._relay_session_pending = False
-        self._relay_finalize_hook_invoked = False
-        self._relay_model_name = "unknown"
+        self._relay_plugin_config_path: Path | None = None
+        self._active_invoke_task: asyncio.Task[tuple[dict[str, Any], str]] | None = None
 
     async def start(self, payload: dict[str, Any]) -> None:
         if self._started:
@@ -247,8 +312,6 @@ async def start(self, payload: dict[str, Any]) -> None:
             )
 
         try:
-            self._relay_session_pending = False
-            self._relay_finalize_hook_invoked = False
             validate_hermes_telemetry_provider(payload)
             self._settings = common_utils.settings_payload(payload)
             self._model_config = common_utils.selected_model_config(payload)
@@ -270,14 +333,15 @@ async def start(self, payload: dict[str, Any]) -> None:
 
             relay_enabled = common_utils.relay_enabled(payload)
             if relay_enabled:
-                self._relay_plugin_config = common_utils.load_relay_plugin_config(
-                    payload
+                (
+                    self._relay_plugin_config_path,
+                    self._relay_plugin_config,
+                ) = write_hermes_relay_plugin_config(payload)
+                for name in HERMES_RELAY_ENV_NAMES:
+                    os.environ.pop(name, None)
+                os.environ["HERMES_NEMO_RELAY_PLUGINS_TOML"] = str(
+                    self._relay_plugin_config_path
                 )
-                from nemo_relay import plugin
-
-                self._relay_context = plugin.plugin(self._relay_plugin_config)
-                await self._relay_context.__aenter__()
-                self._relay_context_entered = True
 
             self._hermes_config_path, self._hermes_config = write_hermes_config(
                 payload,
@@ -289,11 +353,9 @@ async def start(self, payload: dict[str, Any]) -> None:
             if not api_key:
                 raise RuntimeError(f"{api_key_env} is required for Hermes mode")
             self._base_url = common_utils.get_base_url(self._model_config)
-            self._relay_model_name = common_utils.relay_model_name(payload)
 
             from hermes_cli.config import load_config
             from hermes_cli.plugins import discover_plugins
-            from hermes_cli.plugins import invoke_hook
             from hermes_state import SessionDB
             from run_agent import AIAgent
 
@@ -303,10 +365,11 @@ async def start(self, payload: dict[str, Any]) -> None:
                 # Hermes 0.12+ no longer discovers MCP tools as an import side effect
                 # (#16856). Fabric is a Hermes host: discover after config.yaml exists
                 # and before AIAgent resolves mcp-* toolsets.
-                # discover_mcp_tools uses a blocking 120s wait, wrapping it in 
+                # discover_mcp_tools uses a blocking 120s wait, wrapping it in
                 # asyncio.to_thread to avoid blocking the loop.
                 if self._hermes_config.get("mcp_servers"):
                     from tools.mcp_tool import discover_mcp_tools
+
                     await asyncio.to_thread(discover_mcp_tools)
 
                 self._enabled_toolsets = resolve_hermes_toolsets(
@@ -348,7 +411,6 @@ async def start(self, payload: dict[str, Any]) -> None:
                         session_db=self._session_db,
                     )
                 )
-            self._invoke_hook = invoke_hook
             self._start_payload = payload
             self._started = True
         except BaseException:
@@ -378,40 +440,51 @@ async def invoke(self, invocation: dict[str, Any]) -> dict[str, Any]:
         if not isinstance(user_message, str):
             user_message = json.dumps(user_message, sort_keys=True)
 
-        def invoke_turn() -> tuple[dict[str, Any], str]:
-            return _invoke_hermes_turn(
-                agent=self._agent,
-                system_prompt=common_utils.system_instruction(start_payload),
-                user_message=user_message,
-                conversation_history=self._conversation_history,
-            )
-
-        self._relay_session_pending = self._relay_plugin_config is not None
-        self._relay_finalize_hook_invoked = False
-        if _fabric_stream_sink_enabled(self._relay_plugin_config):
-            from nemo_relay import ScopeType, scope
-
-            with scope.scope(
-                "nemo-fabric-invocation",
-                ScopeType.Agent,
-                metadata={
-                    "nemo_fabric_request_id": request.get("request_id"),
-                },
-            ):
-                try:
-                    result, adapter_stdout = invoke_turn()
-                finally:
-                    # The Hermes plugin pushes its session below this correlation
-                    # scope, so finalize that session before popping the parent.
-                    self._finalize_relay_session()
-        else:
+        def run_hermes_turn() -> tuple[dict[str, Any], str]:
             try:
-                result, adapter_stdout = invoke_turn()
+                return _invoke_hermes_turn(
+                    agent=self._agent,
+                    system_prompt=common_utils.system_instruction(start_payload),
+                    user_message=user_message,
+                    conversation_history=self._conversation_history,
+                    task_id=(
+                        request["request_id"]
+                        if isinstance(request.get("request_id"), str)
+                        else None
+                    ),
+                )
             finally:
-                # Hermes' Relay plugin materializes ATIF when its session-finalize
-                # hook runs. Finalize the telemetry session for each Fabric
-                # invocation while retaining the native AIAgent and SessionDB.
-                self._finalize_relay_session()
+                if self._relay_plugin_config is not None:
+                    # Hermes writes TOML-configured ATIF at its session-finalization
+                    # boundary for every supported Relay version. Fabric defines
+                    # each invoke as an artifact-complete boundary, so finalize
+                    # through Hermes' lifecycle instead of reaching into Relay
+                    # directly.
+                    finalize_hermes_relay_session(str(self._agent.session_id))
+
+        # Hermes' upstream Relay integration drives async Relay hooks from its
+        # synchronous agent loop. Run that loop outside this lifecycle server's
+        # event-loop thread so Hermes can own its Relay event loop.
+        if self._active_invoke_task is not None:
+            raise lifecycle.LifecycleError(
+                "hermes_invocation_in_progress",
+                "Hermes runtime already has an active invocation",
+            )
+        invoke_task = asyncio.create_task(asyncio.to_thread(run_hermes_turn))
+        self._active_invoke_task = invoke_task
+
+        def clear_active_invoke_task(
+            completed_task: asyncio.Task[tuple[dict[str, Any], str]],
+        ) -> None:
+            if self._active_invoke_task is completed_task:
+                self._active_invoke_task = None
+
+        invoke_task.add_done_callback(clear_active_invoke_task)
+        try:
+            result, adapter_stdout = await asyncio.shield(invoke_task)
+        finally:
+            if invoke_task.done() and self._active_invoke_task is invoke_task:
+                self._active_invoke_task = None
         messages = result.get("messages") or []
         if isinstance(messages, list):
             self._conversation_history = messages
@@ -439,51 +512,30 @@ def invoke_turn() -> tuple[dict[str, Any], str]:
             output["relay_runtime"] = {
                 "enabled": True,
                 "config_path": os.environ.get("FABRIC_RELAY_CONFIG_PATH"),
-                "emitter": "hermes.observability/nemo_relay",
+                "plugin_config_path": str(self._relay_plugin_config_path),
+                "emitter": "hermes-agent/nemo-relay",
             }
             output["relay_artifacts"] = common_utils.collect_relay_artifacts(
                 self._relay_plugin_config
             )
         return output
 
-    def _finalize_relay_session(self) -> None:
-        if (
-            self._relay_plugin_config is None
-            or self._agent is None
-            or self._invoke_hook is None
-            or not self._relay_session_pending
-        ):
-            return
-        if not self._relay_finalize_hook_invoked:
-            self._invoke_hook(
-                "on_session_finalize",
-                session_id=getattr(self._agent, "session_id", ""),
-                model=getattr(self._agent, "model", None) or self._relay_model_name,
-                platform=getattr(self._agent, "platform", None) or "fabric",
-            )
-            self._relay_finalize_hook_invoked = True
-        # Relay subscriber callbacks are queued. The long-lived plugin context
-        # does not flush them until runtime shutdown, but invocation results
-        # must include artifacts produced by this turn.
-        from nemo_relay import subscribers
-
-        subscribers.flush()
-        self._relay_session_pending = False
-        self._relay_finalize_hook_invoked = False
-
     async def stop(self) -> None:
-        agent = self._agent
-        session_db = self._session_db
-        relay_context = self._relay_context
-        relay_context_entered = self._relay_context_entered
-        relay_plugin_config = self._relay_plugin_config
-        had_mcp_servers = bool(self._hermes_config.get("mcp_servers"))
+        active_invoke_task = self._active_invoke_task
         errors: list[BaseException] = []
-        if relay_plugin_config is not None and agent is not None:
+        if active_invoke_task is not None:
             try:
-                self._finalize_relay_session()
+                await asyncio.shield(active_invoke_task)
             except BaseException as error:
                 errors.append(error)
+            finally:
+                if self._active_invoke_task is active_invoke_task:
+                    self._active_invoke_task = None
+
+        agent = self._agent
+        session_db = self._session_db
+        had_mcp_servers = bool(self._hermes_config.get("mcp_servers"))
+        had_relay_plugin = self._relay_plugin_config_path is not None
         self._agent = None
         self._session_db = None
         self._start_payload = None
@@ -496,15 +548,14 @@ async def stop(self) -> None:
         self._hermes_config = {}
         self._enabled_toolsets = None
         self._conversation_history = None
-        self._relay_context = None
-        self._relay_context_entered = False
-        self._relay_session_pending = False
-        self._relay_finalize_hook_invoked = False
-        self._invoke_hook = None
         self._relay_plugin_config = None
-        self._relay_model_name = "unknown"
+        self._relay_plugin_config_path = None
         self._started = False
 
+        if had_relay_plugin:
+            for name in HERMES_RELAY_ENV_NAMES:
+                os.environ.pop(name, None)
+
         if had_mcp_servers:
             try:
                 from tools.mcp_tool import shutdown_mcp_servers
@@ -524,12 +575,6 @@ async def stop(self) -> None:
                 session_db.close()
             except BaseException as error:
                 errors.append(error)
-        if relay_context is not None and relay_context_entered:
-            try:
-                await relay_context.__aexit__(None, None, None)
-            except BaseException as error:
-                errors.append(error)
-
         if errors:
             for error in errors:
                 if isinstance(error, asyncio.CancelledError):
@@ -550,6 +595,7 @@ def _invoke_hermes_turn(
     system_prompt: str | None,
     user_message: str,
     conversation_history: list[dict[str, Any]] | None,
+    task_id: str | None,
 ) -> tuple[dict[str, Any], str]:
     hermes_stdout = StringIO()
     with redirect_stdout(hermes_stdout):
@@ -557,6 +603,7 @@ def _invoke_hermes_turn(
             agent.run_conversation,
             system_message=system_prompt,
             conversation_history=conversation_history,
+            task_id=task_id,
             sync_honcho=False,
             dont_review=True,
         )
diff --git a/adapters/hermes/uv.lock b/adapters/hermes/uv.lock
index 2a501733..383582c0 100644
--- a/adapters/hermes/uv.lock
+++ b/adapters/hermes/uv.lock
@@ -647,24 +647,24 @@ requires-dist = [
     { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" },
     { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" },
     { name = "nemo-fabric-adapters-common", editable = "../common" },
-    { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.7" },
-    { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.7" },
+    { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.8" },
+    { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.8" },
 ]
 provides-extras = ["harness", "relay", "full"]
 
 [[package]]
 name = "nemo-relay"
-version = "0.6.0"
+version = "0.7.2"
 source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/16/db/44d7258ee620c5cce6dc588983fcb11be3ee03ae71089a65686c14d9ee02/nemo_relay-0.6.0.tar.gz", hash = "sha256:f3d3088019609bc953357b5598a47481dc3e7dc8f11ecf27002ede251f37eb7b", size = 1071046, upload-time = "2026-08-03T14:55:49.702Z" }
+sdist = { url = "https://files.pythonhosted.org/packages/58/81/a7a545ac3a2f8c670d261c89df599aa8fbf49d8be45fd1f52efb36b489eb/nemo_relay-0.7.2.tar.gz", hash = "sha256:828d9f6c7d7e4e42276bb7192bd44202c761e0c76fa4943d84e051b5a99028e5", size = 1295616, upload-time = "2026-08-08T01:54:00.953Z" }
 wheels = [
-    { url = "https://files.pythonhosted.org/packages/25/65/d320016505457cc30971f575e8dadffb923b7cfc780ab8bb25a4ce9d305c/nemo_relay-0.6.0-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:ad5dae6febf6532d7b113abc2a404679c8feffc499df3034b93d9a078185d2bb", size = 9917779, upload-time = "2026-07-22T20:07:48.961Z" },
-    { url = "https://files.pythonhosted.org/packages/ae/c0/f33250e71c4206da1b339072893f9a1e39295fe1aceb9a2fef4b8620a0f2/nemo_relay-0.6.0-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:c0cd9570f64c6956fe3bfb82af1cdb3ee70cb50b51098cdb0de831c3f9b4e904", size = 8888375, upload-time = "2026-07-22T20:07:51.049Z" },
-    { url = "https://files.pythonhosted.org/packages/a3/f4/d1dfaed022da0f6f14765a122867f976a69cc520fe1faaf99757f5719d1f/nemo_relay-0.6.0-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:849daa9e45158ac581e54506e0fcc7a24f557d1ed06dbdc074f5de7a00393cbc", size = 9336372, upload-time = "2026-07-22T20:07:53.224Z" },
-    { url = "https://files.pythonhosted.org/packages/60/9e/f8b80509eef5e05702b940a3d1e2f60c962548d87dec2d712fd3804e6cd4/nemo_relay-0.6.0-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:8c80e534b76bb0455cfc222aaf5c10fa064c088b3c0d5e137eb3c97db46dbc47", size = 10578834, upload-time = "2026-07-30T15:44:44.726Z" },
-    { url = "https://files.pythonhosted.org/packages/f0/a0/84ee49d45a1a874457f2f9260d30fd573af30c9be360d9d97b8bb7835ad9/nemo_relay-0.6.0-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:c8bc4a792a2f8c35ddef1b900cc43be2b2bbbfcd5e7cf65aa0829c04d25eb77f", size = 10849651, upload-time = "2026-07-30T15:44:40.48Z" },
-    { url = "https://files.pythonhosted.org/packages/3a/b0/908d77f75b9054e1e403da78f7d9430249a45939070d4298abbb41a04b5b/nemo_relay-0.6.0-cp311-abi3-win_amd64.whl", hash = "sha256:bfbbedfd130fa95c9b8c04643c30910df850e0ed3500beeab82b00ca2d94e7ea", size = 9613425, upload-time = "2026-07-22T20:07:55.175Z" },
-    { url = "https://files.pythonhosted.org/packages/cd/71/c438b9d746303ff7f270d99f13b250bf947cdac3e52de2a83fd132cbca0b/nemo_relay-0.6.0-cp311-abi3-win_arm64.whl", hash = "sha256:82fe132943399d89e6ec34dc28df0be7bbe41b84f6698c545928b8816b6010f6", size = 9034810, upload-time = "2026-07-22T20:07:57.467Z" },
+    { url = "https://files.pythonhosted.org/packages/5a/cd/f50440257f01bc5ab3d668331c90e06cf4edcc84dc7dc582d322ad05b622/nemo_relay-0.7.2-cp311-abi3-macosx_11_0_arm64.whl", hash = "sha256:e7c7977f0903793cc34c5542bf2b2e44d107def8a5ae9f1b28f06dd61ddec4ed", size = 9246341, upload-time = "2026-08-08T01:53:19.832Z" },
+    { url = "https://files.pythonhosted.org/packages/ed/9f/4041446dd134218799a34b5b5fad3a62d3e1d0a6c322ba2ca4b896ba1393/nemo_relay-0.7.2-cp311-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:b4ae77c1f3d58eabda264e82ffaca54548df80caede7dd6af8cbd8f72b4a82ed", size = 8454070, upload-time = "2026-08-08T01:53:22.524Z" },
+    { url = "https://files.pythonhosted.org/packages/11/83/90230c2e9fae1aee39f768d4a9ef57e9f2716bcaed1a5923cce8b526c66b/nemo_relay-0.7.2-cp311-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0ce7103aec546766649c182619d16aa6ad07439e4d0ebd16d95c5004afb3e56a", size = 8954377, upload-time = "2026-08-08T01:53:25.267Z" },
+    { url = "https://files.pythonhosted.org/packages/71/e7/463fa461d0801146fec6a00cbc02e8961b30089d65ba170f9dfa9e6e3dcd/nemo_relay-0.7.2-cp311-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:2e7d0c2629ade7313aaed71d2272dca96a2fafad0248d0f87cf40a7720b252a0", size = 10322132, upload-time = "2026-08-08T01:53:27.991Z" },
+    { url = "https://files.pythonhosted.org/packages/32/8c/e20ec9c52bd1edd953157aaf24d0d9f9ab8afcbf108fc2356f398e252da8/nemo_relay-0.7.2-cp311-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:b841c92395d7686c7f233036008294b9d362af1ec5123ab0babbfd11cbb04054", size = 10704141, upload-time = "2026-08-08T01:53:30.453Z" },
+    { url = "https://files.pythonhosted.org/packages/5a/c1/92a73961ea759b433b1f897b225662d499123cb962b48dc8ece19f610a09/nemo_relay-0.7.2-cp311-abi3-win_amd64.whl", hash = "sha256:0cdcc5e09d6d62d5c1d385dc62c9233eb714a25f36a09da81e5b9731e3c67903", size = 8803938, upload-time = "2026-08-08T01:53:33.437Z" },
+    { url = "https://files.pythonhosted.org/packages/9d/ec/2de114dab437431173988b9b11f46e8d377e12d57e1b4903258f3e03c2df/nemo_relay-0.7.2-cp311-abi3-win_arm64.whl", hash = "sha256:ca5f66e617311f836a10d96f120f3f32a99b4267d65048453b31951de3419a9d", size = 8438997, upload-time = "2026-08-08T01:53:36.12Z" },
 ]
 
 [[package]]
diff --git a/python/src/nemo_fabric/streaming.py b/python/src/nemo_fabric/streaming.py
index 0999d18f..5de3d397 100644
--- a/python/src/nemo_fabric/streaming.py
+++ b/python/src/nemo_fabric/streaming.py
@@ -254,6 +254,7 @@ def __init__(
         self._accepting = False
         self._request_id: str | None = None
         self._turn_index: int | None = None
+        self._upstream_hermes_turn_id: str | None = None
         self._turn_root_uuid: str | None = None
         self._turn_scope_uuids: set[str] = set()
         self._saw_atof_data = False
@@ -307,6 +308,7 @@ def begin_stream(
             self._queue.get_nowait()
         self._request_id = request_id
         self._turn_index = turn_index
+        self._upstream_hermes_turn_id = None
         self._turn_root_uuid = None
         self._turn_scope_uuids.clear()
         self._saw_atof_data = False
@@ -321,6 +323,7 @@ def end_stream(self) -> None:
         self._accepting = False
         self._request_id = None
         self._turn_index = None
+        self._upstream_hermes_turn_id = None
         self._turn_root_uuid = None
         self._turn_scope_uuids.clear()
 
@@ -521,9 +524,28 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool:
         uuid = record.get("uuid")
         if not isinstance(uuid, str):
             return False
+        metadata = record.get("metadata")
+
+        # Hermes copies the task ID passed by Fabric into its Relay turn markers.
+        if (
+            self._upstream_hermes_turn_id is not None
+            and isinstance(metadata, dict)
+            and metadata.get("turn_id") == self._upstream_hermes_turn_id
+        ):
+            if (
+                record.get("kind") == "scope"
+                and record.get("scope_category") == "start"
+            ):
+                self._turn_scope_uuids.add(uuid)
+            return True
+
         if self._turn_root_uuid is None:
             if not self._matches_turn_root(record):
                 return False
+            if isinstance(metadata, dict) and record.get("kind") == "mark":
+                turn_id = metadata.get("turn_id")
+                if isinstance(turn_id, str):
+                    self._upstream_hermes_turn_id = turn_id
             self._turn_root_uuid = uuid
             self._turn_scope_uuids.add(uuid)
             self._matched_turn_root = True
@@ -542,11 +564,20 @@ def _belongs_to_active_turn(self, record: dict[str, Any]) -> bool:
         return True
 
     def _matches_turn_root(self, record: dict[str, Any]) -> bool:
-        if record.get("kind") != "scope" or record.get("scope_category") != "start":
-            return False
         metadata = record.get("metadata")
         if not isinstance(metadata, dict):
             return False
+        if (
+            record.get("kind") == "mark"
+            and record.get("name") == "hermes.turn.start"
+            and metadata.get("platform") == "fabric"
+            and self._request_id is not None
+            and metadata.get("task_id") == self._request_id
+            and isinstance(metadata.get("turn_id"), str)
+        ):
+            return True
+        if record.get("kind") != "scope" or record.get("scope_category") != "start":
+            return False
         if (
             self._request_id is not None
             and metadata.get("nemo_fabric_request_id") == self._request_id
diff --git a/tests/adapters/test_adapter_package_metadata.py b/tests/adapters/test_adapter_package_metadata.py
index 43e46089..e7a7f44d 100644
--- a/tests/adapters/test_adapter_package_metadata.py
+++ b/tests/adapters/test_adapter_package_metadata.py
@@ -50,10 +50,8 @@ def load_pyproject(path: str) -> dict:
             f"nemo-fabric-adapters-hermes[harness] == {PACKAGE_VERSION}; "
             "python_version < '3.14'"
         ),
-        "harness": [
-            "hermes-agent[mcp]>=0.19.0; python_version < '3.14'"
-        ],
-        "relay": ["nemo-relay>=0.6.0,<0.7"],
+        "harness": ["hermes-agent[mcp]>=0.19.0; python_version < '3.14'"],
+        "relay": ["nemo-relay>=0.6.0,<0.8"],
     },
 }
 
diff --git a/tests/adapters/test_hermes_adapter.py b/tests/adapters/test_hermes_adapter.py
index e32ccacb..e8b340e3 100644
--- a/tests/adapters/test_hermes_adapter.py
+++ b/tests/adapters/test_hermes_adapter.py
@@ -5,13 +5,16 @@
 
 from __future__ import annotations
 
+import asyncio
 import importlib.util
 import inspect
 import json
 import os
 import sys
+import threading
+import tomllib
 from pathlib import Path
-from types import ModuleType, SimpleNamespace
+from types import ModuleType
 from unittest.mock import MagicMock
 
 import pytest
@@ -58,107 +61,192 @@ def test_validate_hermes_telemetry_provider_rejects_mixed_native_and_relay():
         adapter.validate_hermes_telemetry_provider(payload)
 
 
-def test_finalize_relay_session_flushes_before_artifact_collection(monkeypatch):
-    calls: list[str] = []
-    invoke_hook = MagicMock(side_effect=lambda *args, **kwargs: calls.append("hook"))
-    runtime = adapter.HermesRuntime()
-    runtime._relay_plugin_config = {"components": []}
-    runtime._agent = SimpleNamespace(
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
+def test_write_hermes_relay_plugin_config_uses_upstream_toml(
+    monkeypatch,
+    tmp_path: Path,
+):
+    monkeypatch.setattr(adapter, "distribution_version", lambda _name: "0.6.0")
+    relay_config_path = tmp_path / "relay.json"
+    relay_config_path.write_text(
+        json.dumps(
+            {
+                "relay": {
+                    "config": {
+                        "atof": {
+                            "enabled": True,
+                            "sinks": [{"type": "file"}],
+                        },
+                        "atif": {"enabled": True},
+                        "opentelemetry": {
+                            "enabled": True,
+                            "endpoint": "https://otel.example/v1/traces",
+                            "service_name": "fabric",
+                        },
+                    }
+                }
+            }
+        ),
+        encoding="utf-8",
     )
-    runtime._invoke_hook = invoke_hook
-    runtime._relay_session_pending = True
-
-    from nemo_relay import subscribers
-
-    monkeypatch.setattr(subscribers, "flush", lambda: calls.append("flush"))
+    monkeypatch.setenv("FABRIC_RELAY_CONFIG_PATH", str(relay_config_path))
+    payload = {
+        "agent_name": "hermes-test-agent",
+        "base_dir": str(tmp_path),
+        "config": {
+            "models": {"default": {"provider": "nvidia", "model": "nvidia/test-model"}}
+        },
+        "runtime_context": {"runtime_id": "runtime-hermes-relay"},
+    }
 
-    runtime._finalize_relay_session()
+    plugin_config_path, plugin_config = adapter.write_hermes_relay_plugin_config(
+        payload
+    )
 
-    assert calls == ["hook", "flush"]
-    invoke_hook.assert_called_once_with(
-        "on_session_finalize",
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
+    assert plugin_config_path == tmp_path / "relay-config" / "plugins.toml"
+    with plugin_config_path.open("rb") as stream:
+        staged_plugin_config = tomllib.load(stream)
+    staged_observability = staged_plugin_config["components"][0]["config"]
+    assert staged_observability["version"] == 2
+    assert staged_observability["atif"]["enabled"] is True
+    assert staged_observability["atof"]["sinks"][0]["mode"] == "append"
+    assert staged_observability["opentelemetry"] == {
+        "enabled": True,
+        "endpoint": "https://otel.example/v1/traces",
+        "service_name": "fabric",
+    }
+    assert plugin_config["components"][0]["config"]["atof"]["sinks"][0][
+        "output_directory"
+    ] == str(tmp_path / "artifacts" / "relay" / "runtime-hermes-relay")
+    assert (
+        plugin_config["components"][0]["config"]["atof"]["sinks"][0]["mode"]
+        == "overwrite"
     )
 
 
-async def test_stop_does_not_refinalize_completed_relay_turn(monkeypatch):
-    invoke_hook = MagicMock()
-    runtime = adapter.HermesRuntime()
-    runtime._started = True
-    runtime._relay_plugin_config = {"components": []}
-    agent = MagicMock(
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
+def test_write_hermes_relay_plugin_config_migrates_otlp_exporters_to_relay_v3(
+    monkeypatch,
+    tmp_path: Path,
+):
+    monkeypatch.setattr(adapter, "distribution_version", lambda _name: "0.7.2")
+    relay_config_path = tmp_path / "relay.json"
+    relay_config_path.write_text(
+        json.dumps(
+            {
+                "relay": {
+                    "config": {
+                        "opentelemetry": {
+                            "enabled": True,
+                            "endpoint": "https://otel.example/v1/traces",
+                            "service_name": "fabric",
+                        },
+                        "openinference": {
+                            "enabled": True,
+                            "endpoint": "https://openinference.example/v1/traces",
+                            "service_name": "fabric",
+                        },
+                    }
+                }
+            }
+        ),
+        encoding="utf-8",
     )
-    session_db = MagicMock()
-    runtime._agent = agent
-    runtime._session_db = session_db
-    runtime._invoke_hook = invoke_hook
-    runtime._relay_session_pending = True
+    monkeypatch.setenv("FABRIC_RELAY_CONFIG_PATH", str(relay_config_path))
+    payload = {
+        "agent_name": "hermes-test-agent",
+        "base_dir": str(tmp_path),
+        "config": {
+            "models": {"default": {"provider": "nvidia", "model": "nvidia/test-model"}}
+        },
+        "runtime_context": {"runtime_id": "runtime-hermes-relay"},
+    }
 
-    from nemo_relay import subscribers
+    plugin_config_path, _ = adapter.write_hermes_relay_plugin_config(payload)
 
-    flush = MagicMock()
-    monkeypatch.setattr(subscribers, "flush", flush)
+    with plugin_config_path.open("rb") as stream:
+        staged_observability = tomllib.load(stream)["components"][0]["config"]
+    assert staged_observability["version"] == 3
+    assert staged_observability["opentelemetry"] == {
+        "enabled": True,
+        "endpoints": [
+            {
+                "type": "full",
+                "endpoint": "https://otel.example/v1/traces",
+                "service_name": "fabric",
+            },
+            {
+                "type": "openinference",
+                "endpoint": "https://openinference.example/v1/traces",
+                "service_name": "fabric",
+            },
+        ],
+    }
 
-    runtime._finalize_relay_session()
-    await runtime.stop()
 
-    invoke_hook.assert_called_once_with(
-        "on_session_finalize",
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
-    )
-    flush.assert_called_once_with()
-    agent.close.assert_called_once_with()
-    session_db.close.assert_called_once_with()
-    assert runtime._started is False
-    assert runtime._relay_session_pending is False
-    assert runtime._relay_finalize_hook_invoked is False
+def test_finalize_hermes_relay_session_uses_legacy_plugin_hook(monkeypatch):
+    hermes_cli = ModuleType("hermes_cli")
+    hermes_plugins = ModuleType("hermes_cli.plugins")
+    mock_invoke_hook = MagicMock()
+    hermes_plugins.invoke_hook = mock_invoke_hook  # type: ignore[attr-defined]
 
+    monkeypatch.setitem(sys.modules, "hermes_cli", hermes_cli)
+    monkeypatch.setitem(sys.modules, "hermes_cli.plugins", hermes_plugins)
+    monkeypatch.delitem(sys.modules, "hermes_cli.lifecycle", raising=False)
 
-async def test_stop_retries_failed_relay_flush_without_refinalizing(monkeypatch):
-    invoke_hook = MagicMock()
-    runtime = adapter.HermesRuntime()
-    runtime._started = True
-    runtime._relay_plugin_config = {"components": []}
-    runtime._agent = MagicMock(
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
+    adapter.finalize_hermes_relay_session("session-legacy")
+
+    mock_invoke_hook.assert_called_once_with(
+        "on_session_finalize", session_id="session-legacy", platform="fabric"
     )
-    runtime._session_db = MagicMock()
-    runtime._invoke_hook = invoke_hook
-    runtime._relay_session_pending = True
 
-    from nemo_relay import subscribers
 
-    flush = MagicMock(side_effect=[RuntimeError("flush failed"), None])
-    monkeypatch.setattr(subscribers, "flush", flush)
+async def test_runtime_start_stages_upstream_relay_plugin_configuration(
+    monkeypatch,
+    tmp_path: Path,
+):
+    plugin_config_path = tmp_path / "relay-config" / "plugins.toml"
 
-    with pytest.raises(RuntimeError, match="flush failed"):
-        runtime._finalize_relay_session()
+    monkeypatch.setattr(
+        adapter,
+        "write_hermes_relay_plugin_config",
+        lambda _payload: (plugin_config_path, {"version": 1}),
+    )
 
-    assert runtime._relay_session_pending is True
-    assert runtime._relay_finalize_hook_invoked is True
-    await runtime.stop()
+    def stop_after_staging(
+        _payload: dict[str, object],
+        _hermes_home: Path,
+        *,
+        relay_enabled: bool,
+    ) -> tuple[Path, dict[str, object]]:
+        assert relay_enabled is True
+        assert os.environ["HERMES_NEMO_RELAY_PLUGINS_TOML"] == str(plugin_config_path)
+        assert all(
+            name not in os.environ
+            for name in adapter.HERMES_RELAY_ENV_NAMES
+            if name != "HERMES_NEMO_RELAY_PLUGINS_TOML"
+        )
+        raise RuntimeError("stop after Relay plugin staging")
+
+    monkeypatch.setattr(adapter, "write_hermes_config", stop_after_staging)
+    for name in adapter.HERMES_RELAY_ENV_NAMES:
+        monkeypatch.setenv(name, "before")
+    payload = {
+        "base_dir": str(tmp_path),
+        "config": {
+            "harness": {"settings": {}},
+            "models": {"default": {"provider": "nvidia", "model": "test-model"}},
+        },
+        "runtime_context": {
+            "runtime_id": "runtime-relay-plugin",
+            "environment": {"workspace": str(tmp_path)},
+            "artifacts": {"root": str(tmp_path / "artifacts")},
+        },
+        "telemetry_plan": {"providers": ["relay"], "relay_enabled": True},
+    }
 
-    invoke_hook.assert_called_once_with(
-        "on_session_finalize",
-        session_id="runtime-1",
-        model="test-model",
-        platform="fabric",
-    )
-    assert flush.call_count == 2
-    assert runtime._relay_session_pending is False
-    assert runtime._relay_finalize_hook_invoked is False
+    with pytest.raises(RuntimeError, match="stop after Relay plugin staging"):
+        await adapter.HermesRuntime().start(payload)
+
+    assert all(name not in os.environ for name in adapter.HERMES_RELAY_ENV_NAMES)
 
 
 def test_build_hermes_config_maps_fabric_config_to_hermes_config():
@@ -706,9 +794,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
         "base_dir": str(tmp_path),
         "config": {
             "harness": {"settings": {}},
-            "instructions": {
-                "system": {"content": "system", "mode": "replace"}
-            },
+            "instructions": {"system": {"content": "system", "mode": "replace"}},
             "runtime": {"max_turns": None},
             "tools": {"enabled": []},
             "models": {
@@ -726,6 +812,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
         },
         "request": {
             "input": "hello",
+            "request_id": "request-1",
             "context": {"history": [{"role": "user", "content": "stale"}]},
         },
         "capability_plan": {"native": {}},
@@ -743,6 +830,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
     )
     payload["runtime_context"]["invocation_id"] = "invocation-2"
     payload["request"]["input"] = "continue"
+    payload["request"]["request_id"] = "request-2"
     second = await runtime.invoke(
         {
             "runtime_context": payload["runtime_context"],
@@ -777,11 +865,13 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
     assert first_call.kwargs == {
         "system_message": "system",
         "conversation_history": None,
+        "task_id": "request-1",
     }
     assert second_call.args == ("continue",)
     assert second_call.kwargs == {
         "system_message": "system",
         "conversation_history": first_messages,
+        "task_id": "request-2",
     }
     mock_ai_agent.close.assert_called_once_with()
     mock_session_db.close.assert_called_once_with()
@@ -789,6 +879,7 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
     assert runtime._session_db is None
     assert runtime._start_payload is None
     assert runtime._conversation_history is None
+    assert runtime._relay_plugin_config_path is None
     assert first["response"] == "first response"
     assert second["response"] == "second response"
     assert "session_id" not in second
@@ -802,6 +893,112 @@ async def test_persistent_runtime_reuses_hermes_agent_session_and_history(
     )
 
 
+async def test_runtime_stop_waits_for_cancelled_invoke_worker(monkeypatch):
+    worker_started = threading.Event()
+    worker_release = threading.Event()
+    mock_agent = MagicMock()
+    mock_session_db = MagicMock()
+
+    def run_turn(**_kwargs):
+        worker_started.set()
+        assert worker_release.wait(timeout=1)
+        return (
+            {
+                "response": "completed after cancellation",
+                "completed": True,
+                "failed": False,
+                "messages": [],
+            },
+            "",
+        )
+
+    monkeypatch.setattr(adapter, "_invoke_hermes_turn", run_turn)
+    runtime = adapter.HermesRuntime()
+    runtime._started = True
+    runtime._runtime_id = "runtime-cancelled-invoke"
+    runtime._start_payload = {"config": {"instructions": {}}}
+    runtime._agent = mock_agent
+    runtime._session_db = mock_session_db
+
+    invoke_task = asyncio.create_task(
+        runtime.invoke(
+            {
+                "runtime_context": {"runtime_id": runtime._runtime_id},
+                "request": {"input": "wait"},
+            }
+        )
+    )
+    assert await asyncio.to_thread(worker_started.wait, 1)
+
+    invoke_task.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await invoke_task
+
+    stop_task = asyncio.create_task(runtime.stop())
+    await asyncio.sleep(0)
+    mock_agent.close.assert_not_called()
+    mock_session_db.close.assert_not_called()
+
+    worker_release.set()
+    await stop_task
+
+    mock_agent.close.assert_called_once_with()
+    mock_session_db.close.assert_called_once_with()
+
+
+async def test_runtime_allows_invoke_after_cancelled_worker_finishes(monkeypatch):
+    worker_started = threading.Event()
+    worker_release = threading.Event()
+    mock_agent = MagicMock()
+    mock_session_db = MagicMock()
+    calls = 0
+
+    def run_turn(**_kwargs):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            worker_started.set()
+            assert worker_release.wait(timeout=1)
+        return (
+            {
+                "response": f"turn-{calls}",
+                "completed": True,
+                "failed": False,
+                "messages": [],
+            },
+            "",
+        )
+
+    monkeypatch.setattr(adapter, "_invoke_hermes_turn", run_turn)
+    runtime = adapter.HermesRuntime()
+    runtime._started = True
+    runtime._runtime_id = "runtime-cancelled-invoke"
+    runtime._start_payload = {"config": {"instructions": {}}}
+    runtime._agent = mock_agent
+    runtime._session_db = mock_session_db
+    invocation = {
+        "runtime_context": {"runtime_id": runtime._runtime_id},
+        "request": {"input": "wait"},
+    }
+
+    cancelled_invoke = asyncio.create_task(runtime.invoke(invocation))
+    assert await asyncio.to_thread(worker_started.wait, 1)
+    active_invoke_task = runtime._active_invoke_task
+    assert active_invoke_task is not None
+
+    cancelled_invoke.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await cancelled_invoke
+
+    worker_release.set()
+    await asyncio.wait_for(asyncio.shield(active_invoke_task), timeout=1)
+
+    result = await runtime.invoke(invocation)
+
+    assert result["response"] == "turn-2"
+    await runtime.stop()
+
+
 def test_main_serves_persistent_runtime(monkeypatch):
     serve = MagicMock()
     monkeypatch.setattr(adapter.lifecycle, "serve", serve)
diff --git a/tests/adapters/test_hermes_streaming.py b/tests/adapters/test_hermes_streaming.py
index f9e5081f..4aa0c672 100644
--- a/tests/adapters/test_hermes_streaming.py
+++ b/tests/adapters/test_hermes_streaming.py
@@ -4,7 +4,6 @@
 """Dependency-free tests for Hermes Relay streaming integration."""
 
 import sys
-from contextlib import contextmanager
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -19,42 +18,12 @@
 from nemo_fabric_adapters.hermes import adapter
 
 
-@pytest.mark.parametrize(
-    ("relay_plugin_config", "expected_metadata"),
-    [
-        (
-            {
-                "components": [
-                    {
-                        "kind": "observability",
-                        "config": {
-                            "atof": {
-                                "enabled": True,
-                                "sinks": [
-                                    {
-                                        "type": "stream",
-                                        "name": "nemo-fabric-stream",
-                                        "url": "http://127.0.0.1:1234/atof",
-                                    }
-                                ],
-                            }
-                        },
-                    }
-                ]
-            },
-            [{"nemo_fabric_request_id": "request-1"}],
-        ),
-        ({"components": []}, []),
-    ],
-    ids=["streaming", "non-streaming"],
-)
-async def test_relay_invocation_scope_carries_fabric_request_id(
+async def test_relay_invocation_passes_fabric_request_id_to_hermes(
     monkeypatch,
     tmp_path: Path,
-    relay_plugin_config: dict[str, object],
-    expected_metadata: list[object],
 ):
     events: list[str] = []
+    task_ids: list[object] = []
     runtime = adapter.HermesRuntime()
     runtime._started = True
     runtime._start_payload = {}
@@ -64,14 +33,14 @@ async def test_relay_invocation_scope_carries_fabric_request_id(
         model="test-model",
         platform="fabric",
     )
-    runtime._invoke_hook = lambda *_args, **_kwargs: events.append("finalize")
-    runtime._relay_plugin_config = relay_plugin_config
+    runtime._relay_plugin_config = {"components": []}
     runtime._hermes_home = tmp_path
     runtime._hermes_config_path = tmp_path / "config.yaml"
     runtime._enabled_toolsets = []
 
     def invoke_turn(**_kwargs: object):
         events.append("turn")
+        task_ids.append(_kwargs["task_id"])
         return (
             {
                 "response": "done",
@@ -83,28 +52,17 @@ def invoke_turn(**_kwargs: object):
         )
 
     monkeypatch.setattr(adapter, "_invoke_hermes_turn", invoke_turn)
+    monkeypatch.setattr(
+        adapter,
+        "finalize_hermes_relay_session",
+        lambda _session_id: events.append("finalize"),
+    )
     monkeypatch.setattr(
         adapter.common_utils,
         "collect_relay_artifacts",
         lambda _config: [],
     )
 
-    from nemo_relay import scope, subscribers
-
-    captured_metadata: list[object] = []
-
-    @contextmanager
-    def capture_scope(*_args: object, **kwargs: object):
-        captured_metadata.append(kwargs["metadata"])
-        events.append("scope-enter")
-        try:
-            yield
-        finally:
-            events.append("scope-exit")
-
-    monkeypatch.setattr(scope, "scope", capture_scope)
-    monkeypatch.setattr(subscribers, "flush", lambda: None)
-
     await runtime.invoke(
         {
             "runtime_context": {"runtime_id": "runtime-1"},
@@ -112,8 +70,5 @@ def capture_scope(*_args: object, **kwargs: object):
         }
     )
 
-    assert captured_metadata == expected_metadata
-    if expected_metadata:
-        assert events == ["scope-enter", "turn", "finalize", "scope-exit"]
-    else:
-        assert events == ["turn", "finalize"]
+    assert task_ids == ["request-1"]
+    assert events == ["turn", "finalize"]
diff --git a/tests/e2e/test_hermes_e2e.py b/tests/e2e/test_hermes_e2e.py
index b7c43d1e..6a630bde 100644
--- a/tests/e2e/test_hermes_e2e.py
+++ b/tests/e2e/test_hermes_e2e.py
@@ -7,9 +7,11 @@
 import os
 import sys
 import warnings
+from importlib.metadata import version as distribution_version
 from pathlib import Path
 from types import ModuleType
 
+from packaging.version import Version
 import pytest
 import requests
 import yaml
@@ -58,6 +60,11 @@ async def test_hermes_persistent_host_reuses_native_session(
     assert first["metadata"]["adapter_runner"] == "persistent_local_host", results
     assert first["metadata"]["host_pid"] == second["metadata"]["host_pid"], results
     assert "user_count=2" in second["output"]["response"], results
+    for turn in (first, second):
+        assert {artifact["kind"] for artifact in turn["output"]["relay_artifacts"]} >= {
+            "atof",
+            "atif",
+        }, turn.to_mapping()
 
 
 @pytest.mark.usefixtures("mock_nvidia_api_key", "nemo_relay")
@@ -106,14 +113,19 @@ async def test_mcp_stdio_transport(
 ):
     os.environ["ADAPTER_PYTHON"] = sys.executable
     tool_name = "mcp__mcp_server_time__get_current_time"
+    tool_arguments = {"timezone": "America/Los_Angeles"}
+    if Version(distribution_version("hermes-agent")) < Version("0.20"):
+        # The released 0.19 integration accepts the configured MCP tool directly.
+        tool_call = {"name": tool_name, "arguments": tool_arguments}
+    else:
+        # Newer Hermes versions dispatch MCP schemas through their native bridge.
+        tool_call = {
+            "name": "tool_call",
+            "arguments": {"name": tool_name, "arguments": tool_arguments},
+        }
     scenario_response = requests.post(
         f"{api_server}/_scenario",
-        json={
-            "tool_call": {
-                "name": tool_name,
-                "arguments": {"timezone": "America/Los_Angeles"},
-            }
-        },
+        json={"tool_call": tool_call},
         timeout=5,
     )
     scenario_response.raise_for_status()
@@ -177,12 +189,15 @@ async def test_mcp_stdio_transport(
             "args": ["-m", "mcp_server_time"],
             "env": {"MCP_TIME_TEST": "enabled"},
         }
+        assert "mcp_server_time" in result["output"]["enabled_toolsets"]
         assert {record["scope_category"] for record in tool_records} == {"start", "end"}
         tool_end = next(
             record for record in tool_records if record["scope_category"] == "end"
         )
         assert tool_end["category"] == "tool"
-        assert tool_end["metadata"]["status"] == "ok"
+        assert tool_end["metadata"].get("otel.status_code") == "OK" or (
+            tool_end["metadata"].get("status") == "ok"
+        )
         assert "America/Los_Angeles" in tool_end["data"]
 
 
@@ -243,7 +258,7 @@ async def test_artifacts(self):
         assert output["base_url"] == f"{self.api_server}/v1"
         assert output["error"] is None
         assert output["relay_runtime"]["enabled"] is True
-        assert output["relay_runtime"]["emitter"] == "hermes.observability/nemo_relay"
+        assert output["relay_runtime"]["emitter"] == "hermes-agent/nemo-relay"
         assert output["failed"] is False
 
         assert "echo user_count=" in output["response"]
@@ -257,7 +272,10 @@ async def test_artifacts(self):
 
         hermes_config = yaml.safe_load(hermes_config_path.read_text(encoding="utf-8"))
         assert hermes_config["model"]["provider"] == "nvidia"
-        assert hermes_config["model"]["default"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"
+        assert (
+            hermes_config["model"]["default"]
+            == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"
+        )
         assert hermes_config["model"]["base_url"] == f"{self.api_server}/v1"
         assert hermes_config["plugins"]["enabled"] == ["observability/nemo_relay"]
         assert output["hermes_native_config"]["plugins"] == ["observability/nemo_relay"]
@@ -315,16 +333,42 @@ async def test_atof_artifacts(self):
         actual_atof_fields = set().union(*(record.keys() for record in atof_records))
         assert actual_atof_fields.issuperset(expected_atof_fields)
 
-        assert len(atof_records) == 7
+        record_kinds = {
+            (record["name"], record.get("scope_category")) for record in atof_records
+        }
+        assert record_kinds.issuperset({("nvidia", "start"), ("nvidia", "end")})
 
-        assert all(
-            record["metadata"]["model"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"
-            and record["metadata"]["platform"] == self.atof_platform
+        session_scopes = [
+            record
             for record in atof_records
-        )
+            if record["name"] == "hermes.session"
+            or str(record["name"]).startswith("hermes-session-")
+        ]
+        assert {record.get("scope_category") for record in session_scopes} >= {
+            "start",
+            "end",
+        }
+
+        current_turn_scopes = [
+            record
+            for record in atof_records
+            if record["name"] == "hermes.turn"
+            and record.get("scope_category") in {"start", "end"}
+        ]
+        upstream_turn_marks = [
+            record
+            for record in atof_records
+            if record["name"] in {"hermes.turn.start", "hermes.turn.end"}
+        ]
+        turn_marks = current_turn_scopes or upstream_turn_marks
+        assert turn_marks
 
-        assert atof_records[-2]["name"] == "hermes.session.end"
-        assert atof_records[-1]["scope_category"] == "end"
+        fabric_scopes = [*session_scopes, *turn_marks]
+        assert all(
+            record["metadata"].get("hermes.execution_surface") == "fabric"
+            or record["metadata"].get("platform") == self.atof_platform
+            for record in fabric_scopes
+        )
 
     async def test_atif_artifacts(self):
         kinds = {artifact["kind"] for artifact in self.relay_artifacts}
@@ -354,7 +398,12 @@ async def test_atif_artifacts(self):
 
         last_step = steps[-1]
         assert last_step["source"] == "agent"
-        assert last_step["message"] == self.output["response"]
-        assert last_step["model_name"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"
+        # The upstream exporter derives this field from the provider's final
+        # wire response. The mock streaming response has no final text field,
+        # while Fabric's normalized response is assembled from its deltas.
+        assert last_step["message"] in {"", self.output["response"]}
+        assert (
+            last_step["model_name"] == "nvidia/nemotron-3-nano-omni-30b-a3b-reasoning"
+        )
         assert last_step["extra"]["invocation"]["framework"] == "nemo_relay"
         assert last_step["extra"]["invocation"]["status"] == "completed"
diff --git a/tests/e2e/test_hermes_runtime.py b/tests/e2e/test_hermes_runtime.py
index 4f0e57a0..791e88ee 100644
--- a/tests/e2e/test_hermes_runtime.py
+++ b/tests/e2e/test_hermes_runtime.py
@@ -31,8 +31,6 @@ async def test_hermes_runtime():
 
 async def test_hermes_runtime_with_relay():
     _require_hermes_integration()
-    if importlib.util.find_spec("nemo_relay") is None:
-        pytest.fail("the nemo-relay Python package is required")
     await _run(relay=True)
 
 
@@ -103,8 +101,7 @@ async def _run(*, relay: bool) -> None:
             for result in (r1, r2):
                 assert result.telemetry[0].provider == "relay", result.to_mapping()
                 assert {
-                    artifact["kind"]
-                    for artifact in result["output"]["relay_artifacts"]
+                    artifact["kind"] for artifact in result["output"]["relay_artifacts"]
                 } >= {"atof", "atif"}, result.to_mapping()
 
     assert runtime.status is RuntimeStatus.STOPPED, runtime.status
diff --git a/tests/python/test_streaming.py b/tests/python/test_streaming.py
index 31736cec..9adf8236 100644
--- a/tests/python/test_streaming.py
+++ b/tests/python/test_streaming.py
@@ -928,6 +928,96 @@ async def test_listener_correlates_records_to_active_turn(
     await listener.close()
 
 
+async def test_listener_correlates_upstream_hermes_turn_records():
+    listener = await _AtofStreamListener(maxsize=4).start()
+    listener.begin_stream(request_id="request-2", turn_index=2)
+    turn_id = "upstream-turn"
+    current = [
+        {
+            "kind": "mark",
+            "name": "hermes.turn.start",
+            "uuid": "turn-start",
+            "metadata": {
+                "platform": "fabric",
+                "task_id": "request-2",
+                "turn_id": turn_id,
+            },
+        },
+        {
+            "kind": "scope",
+            "scope_category": "start",
+            "uuid": "llm",
+            "parent_uuid": "session",
+            "metadata": {"turn_id": turn_id},
+        },
+        {"kind": "mark", "uuid": "llm-child", "parent_uuid": "llm"},
+        {
+            "kind": "mark",
+            "name": "hermes.turn.end",
+            "uuid": "turn-end",
+            "metadata": {
+                "platform": "fabric",
+                "task_id": "request-2",
+                "turn_id": turn_id,
+            },
+        },
+    ]
+
+    await _post_chunked(
+        listener.url,
+        [
+            {
+                "kind": "scope",
+                "scope_category": "start",
+                "uuid": "previous",
+                "metadata": {"nemo_fabric_request_id": "request-1"},
+            },
+            *current,
+        ],
+    )
+
+    assert [await listener.records.get() for _ in current] == current
+    assert listener.records.empty()
+    listener.end_stream()
+    await listener.close()
+
+
+async def test_listener_rejects_late_upstream_hermes_turn_marker():
+    listener = await _AtofStreamListener().start()
+    previous = {
+        "kind": "mark",
+        "name": "hermes.turn.start",
+        "uuid": "previous-turn",
+        "metadata": {
+            "platform": "fabric",
+            "task_id": "request-1",
+            "turn_id": "previous-turn",
+        },
+    }
+    listener.begin_stream(request_id="request-1")
+    await _post_chunked(listener.url, [previous])
+    assert await listener.records.get() == previous
+    listener.end_stream()
+
+    current = {
+        "kind": "mark",
+        "name": "hermes.turn.start",
+        "uuid": "current-turn",
+        "metadata": {
+            "platform": "fabric",
+            "task_id": "request-2",
+            "turn_id": "current-turn",
+        },
+    }
+    listener.begin_stream(request_id="request-2")
+    await _post_chunked(listener.url, [previous, current])
+
+    assert await listener.records.get() == current
+    assert listener.records.empty()
+    listener.end_stream()
+    await listener.close()
+
+
 async def test_listener_applies_byte_budget_backpressure():
     record = {"uuid": "record", "payload": "x" * 16}
     record_size = len(json.dumps(record).encode())
diff --git a/uv.lock b/uv.lock
index 3def94c4..6f530c90 100644
--- a/uv.lock
+++ b/uv.lock
@@ -2359,8 +2359,8 @@ requires-dist = [
     { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'full'", specifier = ">=0.19.0" },
     { name = "hermes-agent", extras = ["mcp"], marker = "python_full_version < '3.14' and extra == 'harness'", specifier = ">=0.19.0" },
     { name = "nemo-fabric-adapters-common", editable = "adapters/common" },
-    { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.7" },
-    { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.7" },
+    { name = "nemo-relay", marker = "extra == 'full'", specifier = ">=0.6.0,<0.8" },
+    { name = "nemo-relay", marker = "extra == 'relay'", specifier = ">=0.6.0,<0.8" },
 ]
 provides-extras = ["harness", "relay", "full"]