From 580faa6d6bde19014dd7621fa9b9b935f08cb13e Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 16:53:28 -0700
Subject: [PATCH 01/10] feat: add runtime service registry seam

---
 src/hermes_workflows/engine.py           |   6 ++
 src/hermes_workflows/runtime_services.py |  62 ++++++++++++
 tests/test_runtime_services_contract.py  | 114 +++++++++++++++++++++++
 3 files changed, 182 insertions(+)
 create mode 100644 src/hermes_workflows/runtime_services.py
 create mode 100644 tests/test_runtime_services_contract.py

diff --git a/src/hermes_workflows/engine.py b/src/hermes_workflows/engine.py
index 56c3f91..7505619 100644
--- a/src/hermes_workflows/engine.py
+++ b/src/hermes_workflows/engine.py
@@ -18,6 +18,7 @@
 from .approvals import ApprovalDecision, ApprovalDecisionInput, ApprovalReceipt, ApprovalView, OperatorResponseReceipt
 from .domain import CommandType, WorkflowStatus, decode_command_row, decode_event_row, make_command, make_event
 from .input_parsing import coerce_workflow_input
+from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry
 from .status_projection import StatusProjection
 from .types import to_json_value
 from .workflow_values import Workflow
@@ -105,10 +106,12 @@ def __init__(
         *,
         agent_runner: Optional[Callable[[Dict[str, Any]], Any]] = None,
         read_only: bool = False,
+        runtime_services: RuntimeServiceRegistry | None = None,
     ):
         self.db_path = Path(db_path)
         self.agent_runner = agent_runner
         self.read_only = read_only
+        self.runtime_services = runtime_services if runtime_services is not None else EmptyRuntimeServicesV1()
         self._status_projection = StatusProjection(self)
         self._active_command_claim = threading.local()
         if read_only:
@@ -118,6 +121,9 @@ def __init__(
             self.db_path.parent.mkdir(parents=True, exist_ok=True)
             self._init_db()
 
+    def resolve_runtime_service(self, service_id: str, contract_version: int) -> object | None:
+        return self.runtime_services.resolve(service_id, contract_version)
+
     def _ensure_writable(self, operation: str) -> None:
         if self.read_only:
             raise RuntimeError(f"WorkflowEngine is read-only; cannot {operation}")
diff --git a/src/hermes_workflows/runtime_services.py b/src/hermes_workflows/runtime_services.py
new file mode 100644
index 0000000..9f9a669
--- /dev/null
+++ b/src/hermes_workflows/runtime_services.py
@@ -0,0 +1,62 @@
+from __future__ import annotations
+
+import re
+from collections.abc import Mapping
+from dataclasses import dataclass, field
+from types import MappingProxyType
+from typing import Protocol, SupportsIndex
+
+
+_SERVICE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
+
+
+class RuntimeServiceRegistry(Protocol):
+    def resolve(self, service_id: str, contract_version: int) -> object | None: ...
+
+
+@dataclass(frozen=True)
+class RuntimeServicesV1:
+    schema_version: int = 1
+    services: Mapping[str, object] = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        if type(self.schema_version) is not int or self.schema_version != 1:
+            raise ValueError("schema_version must equal 1")
+        if not isinstance(self.services, Mapping):
+            raise TypeError("services must be a mapping")
+
+        validated: dict[str, object] = {}
+        for service_id, service in self.services.items():
+            _validate_service_id(service_id)
+            if service_id in validated:
+                raise ValueError(f"duplicate service_id: {service_id}")
+            validated[service_id] = service
+        object.__setattr__(self, "services", MappingProxyType(validated))
+
+    def resolve(self, service_id: str, contract_version: int) -> object | None:
+        _validate_resolution(service_id, contract_version)
+        return self.services.get(service_id)
+
+    def __reduce_ex__(self, protocol: SupportsIndex):
+        raise TypeError("runtime service registries are process-local and cannot be pickled")
+
+
+@dataclass(frozen=True)
+class EmptyRuntimeServicesV1:
+    def resolve(self, service_id: str, contract_version: int) -> object | None:
+        _validate_resolution(service_id, contract_version)
+        return None
+
+    def __reduce_ex__(self, protocol: SupportsIndex):
+        raise TypeError("runtime service registries are process-local and cannot be pickled")
+
+
+def _validate_service_id(service_id: object) -> None:
+    if not isinstance(service_id, str) or _SERVICE_ID_PATTERN.fullmatch(service_id) is None:
+        raise ValueError("service_id must match ^[a-z][a-z0-9_.-]{0,63}$")
+
+
+def _validate_resolution(service_id: object, contract_version: object) -> None:
+    _validate_service_id(service_id)
+    if type(contract_version) is not int or contract_version < 1:
+        raise ValueError("contract_version must be an integer >= 1")
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
new file mode 100644
index 0000000..4be1abf
--- /dev/null
+++ b/tests/test_runtime_services_contract.py
@@ -0,0 +1,114 @@
+from __future__ import annotations
+
+import json
+import pickle
+from collections.abc import Iterator, Mapping
+
+import pytest
+
+from hermes_workflows import WorkflowEngine, workflow
+from hermes_workflows.runtime_services import EmptyRuntimeServicesV1, RuntimeServicesV1
+
+
+@workflow
+async def runtime_service_contract_workflow(inputs):
+    return {"value": inputs["value"]}
+
+
+class DuplicateItemsMapping(Mapping[str, object]):
+    """Deliberately broken Mapping used to prove duplicate input rejection."""
+
+    def __getitem__(self, key: str) -> object:
+        if key == "test.recording":
+            return object()
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[str]:
+        yield "test.recording"
+
+    def __len__(self) -> int:
+        return 1
+
+    def items(self):  # type: ignore[override]
+        recording = object()
+        return (("test.recording", recording), ("test.recording", recording))
+
+
+def test_runtime_services_resolve_recording_object_by_identity():
+    recording = object()
+    services = RuntimeServicesV1(services={"test.recording": recording})
+
+    assert services.resolve("test.recording", 1) is recording
+    assert services.resolve("test.missing", 1) is None
+
+
+def test_empty_runtime_services_preserve_default_engine_path(tmp_path):
+    engine = WorkflowEngine(tmp_path / "workflow.sqlite")
+
+    assert isinstance(engine.runtime_services, EmptyRuntimeServicesV1)
+    assert engine.resolve_runtime_service("test.missing", 1) is None
+    result = engine.run_until_idle(
+        runtime_service_contract_workflow,
+        {"value": "unchanged"},
+        workflow_id="wf_runtime_service_default",
+    )
+    assert result.status == "completed"
+    assert result.result == {"value": "unchanged"}
+
+
+@pytest.mark.parametrize("service_id", ["", "Uppercase", "1leading", "has space", "a" * 65, 1, None])
+def test_runtime_services_reject_malformed_service_ids(service_id):
+    with pytest.raises(ValueError, match="service_id"):
+        RuntimeServicesV1(services={service_id: object()})
+
+    with pytest.raises(ValueError, match="service_id"):
+        EmptyRuntimeServicesV1().resolve(service_id, 1)
+
+
+@pytest.mark.parametrize("contract_version", [0, -1, True, 1.0, "1", None])
+def test_runtime_services_reject_malformed_contract_versions(contract_version):
+    services = RuntimeServicesV1(services={})
+
+    with pytest.raises(ValueError, match="contract_version"):
+        services.resolve("test.recording", contract_version)
+
+
+@pytest.mark.parametrize("schema_version", [0, 2, True, "1", None])
+def test_runtime_services_require_schema_version_one(schema_version):
+    with pytest.raises(ValueError, match="schema_version"):
+        RuntimeServicesV1(schema_version=schema_version, services={})
+
+
+def test_runtime_services_reject_duplicate_ids():
+    with pytest.raises(ValueError, match="duplicate service_id"):
+        RuntimeServicesV1(services=DuplicateItemsMapping())
+
+
+def test_runtime_services_are_process_local_and_nonserializable():
+    services = RuntimeServicesV1(services={"test.recording": object()})
+    empty = EmptyRuntimeServicesV1()
+
+    for registry in (services, empty):
+        with pytest.raises(TypeError):
+            json.dumps(registry)
+        with pytest.raises(TypeError, match="process-local"):
+            pickle.dumps(registry)
+
+
+def test_engine_stores_one_registry_without_persisting_it(tmp_path):
+    db_path = tmp_path / "workflow.sqlite"
+    recording = object()
+    services = RuntimeServicesV1(services={"test.recording": recording})
+    engine = WorkflowEngine(db_path, runtime_services=services)
+
+    assert engine.runtime_services is services
+    assert engine.resolve_runtime_service("test.recording", 1) is recording
+    result = engine.run_until_idle(
+        runtime_service_contract_workflow,
+        {"value": "recording stays process-local"},
+        workflow_id="wf_runtime_service_injected",
+    )
+    assert result.status == "completed"
+
+    assert "test.recording" not in db_path.read_bytes().decode("utf-8", errors="ignore")
+    assert all("test.recording" not in json.dumps(event) for event in engine.events("wf_runtime_service_injected"))

From 618fac6481b7c01ca348d625dcbcd85e692a5061 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 17:20:53 -0700
Subject: [PATCH 02/10] fix(runtime): reject service registry persistence

---
 src/hermes_workflows/engine.py          | 28 +++++++++++++++++++++++--
 tests/test_runtime_services_contract.py |  7 ++++++-
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/src/hermes_workflows/engine.py b/src/hermes_workflows/engine.py
index 7505619..b36fe28 100644
--- a/src/hermes_workflows/engine.py
+++ b/src/hermes_workflows/engine.py
@@ -9,8 +9,9 @@
 import sqlite3
 import threading
 import time
+from collections.abc import Mapping, Sequence
 from contextlib import contextmanager
-from dataclasses import dataclass
+from dataclasses import dataclass, fields, is_dataclass
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union, get_type_hints
 from urllib.parse import quote
@@ -18,7 +19,7 @@
 from .approvals import ApprovalDecision, ApprovalDecisionInput, ApprovalReceipt, ApprovalView, OperatorResponseReceipt
 from .domain import CommandType, WorkflowStatus, decode_command_row, decode_event_row, make_command, make_event
 from .input_parsing import coerce_workflow_input
-from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry
+from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry, RuntimeServicesV1
 from .status_projection import StatusProjection
 from .types import to_json_value
 from .workflow_values import Workflow
@@ -3644,9 +3645,32 @@ def _validate_step_request_fingerprint(step_key: str, stored_payload: Any, curre
 
 
 def _to_jsonable(value: Any) -> Any:
+    _reject_runtime_service_registries(value)
     return to_json_value(value)
 
 
+def _reject_runtime_service_registries(value: Any, seen: set[int] | None = None) -> None:
+    if isinstance(value, (RuntimeServicesV1, EmptyRuntimeServicesV1)):
+        raise TypeError("runtime service registries are process-local and cannot be serialized")
+
+    seen = seen if seen is not None else set()
+    value_id = id(value)
+    if value_id in seen:
+        return
+    seen.add(value_id)
+
+    if is_dataclass(value) and not isinstance(value, type):
+        for field in fields(value):
+            _reject_runtime_service_registries(getattr(value, field.name), seen)
+    elif isinstance(value, Mapping):
+        for key, item in value.items():
+            _reject_runtime_service_registries(key, seen)
+            _reject_runtime_service_registries(item, seen)
+    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        for item in value:
+            _reject_runtime_service_registries(item, seen)
+
+
 def _from_jsonable(value: Any) -> Any:
     if isinstance(value, dict):
         if value.get("__hermes_type__") == "Workflow":
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index 4be1abf..ea1a05e 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -7,6 +7,7 @@
 import pytest
 
 from hermes_workflows import WorkflowEngine, workflow
+from hermes_workflows.engine import JsonCodec
 from hermes_workflows.runtime_services import EmptyRuntimeServicesV1, RuntimeServicesV1
 
 
@@ -85,7 +86,7 @@ def test_runtime_services_reject_duplicate_ids():
 
 
 def test_runtime_services_are_process_local_and_nonserializable():
-    services = RuntimeServicesV1(services={"test.recording": object()})
+    services = RuntimeServicesV1(services={"test.recording": "must not leak"})
     empty = EmptyRuntimeServicesV1()
 
     for registry in (services, empty):
@@ -93,6 +94,10 @@ def test_runtime_services_are_process_local_and_nonserializable():
             json.dumps(registry)
         with pytest.raises(TypeError, match="process-local"):
             pickle.dumps(registry)
+        with pytest.raises(TypeError, match="process-local"):
+            JsonCodec.dumps(registry)
+        with pytest.raises(TypeError, match="process-local"):
+            JsonCodec.dumps({"nested": [registry]})
 
 
 def test_engine_stores_one_registry_without_persisting_it(tmp_path):

From 52da4d977387c28630b070b5754ee4f8e5d64b12 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 17:46:58 -0700
Subject: [PATCH 03/10] Block runtime service helper serialization

---
 src/hermes_workflows/runtime_services.py | 13 +++++++++++++
 tests/test_runtime_services_contract.py  | 20 ++++++++++++++++++++
 2 files changed, 33 insertions(+)

diff --git a/src/hermes_workflows/runtime_services.py b/src/hermes_workflows/runtime_services.py
index 9f9a669..42f03f4 100644
--- a/src/hermes_workflows/runtime_services.py
+++ b/src/hermes_workflows/runtime_services.py
@@ -16,9 +16,15 @@ def resolve(self, service_id: str, contract_version: int) -> object | None: ...
 
 @dataclass(frozen=True)
 class RuntimeServicesV1:
+    _serialization_guard: object = field(init=False, repr=False, compare=False)
     schema_version: int = 1
     services: Mapping[str, object] = field(default_factory=dict)
 
+    def __getattribute__(self, name: str) -> object:
+        if name == "_serialization_guard":
+            raise TypeError("runtime service registries are process-local and cannot be serialized")
+        return super().__getattribute__(name)
+
     def __post_init__(self) -> None:
         if type(self.schema_version) is not int or self.schema_version != 1:
             raise ValueError("schema_version must equal 1")
@@ -43,6 +49,13 @@ def __reduce_ex__(self, protocol: SupportsIndex):
 
 @dataclass(frozen=True)
 class EmptyRuntimeServicesV1:
+    _serialization_guard: object = field(init=False, repr=False, compare=False)
+
+    def __getattribute__(self, name: str) -> object:
+        if name == "_serialization_guard":
+            raise TypeError("runtime service registries are process-local and cannot be serialized")
+        return super().__getattribute__(name)
+
     def resolve(self, service_id: str, contract_version: int) -> object | None:
         _validate_resolution(service_id, contract_version)
         return None
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index ea1a05e..ca0ac36 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -7,8 +7,11 @@
 import pytest
 
 from hermes_workflows import WorkflowEngine, workflow
+from hermes_workflows.artifacts import JsonArtifact
 from hermes_workflows.engine import JsonCodec
 from hermes_workflows.runtime_services import EmptyRuntimeServicesV1, RuntimeServicesV1
+from hermes_workflows.status_projection import JsonCodec as StatusProjectionJsonCodec
+from hermes_workflows.types import to_json_value
 
 
 @workflow
@@ -100,6 +103,23 @@ def test_runtime_services_are_process_local_and_nonserializable():
             JsonCodec.dumps({"nested": [registry]})
 
 
+@pytest.mark.parametrize(
+    "registry",
+    [
+        RuntimeServicesV1(services={"test.recording": "must not leak"}),
+        EmptyRuntimeServicesV1(),
+    ],
+)
+def test_runtime_services_reject_framework_persistence_helpers(registry):
+    for serialize in (
+        to_json_value,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("runtime services", value),
+    ):
+        with pytest.raises(TypeError, match="process-local"):
+            serialize(registry)
+
+
 def test_engine_stores_one_registry_without_persisting_it(tmp_path):
     db_path = tmp_path / "workflow.sqlite"
     recording = object()

From 0e86026e2688466056d3fdcc7d3249ed9f685b75 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 18:31:17 -0700
Subject: [PATCH 04/10] Reject arbitrary runtime registry persistence

---
 src/hermes_workflows/engine.py           |  9 +++--
 src/hermes_workflows/runtime_services.py | 44 ++++++++++++++++++++++--
 tests/test_runtime_services_contract.py  | 41 ++++++++++++++++++++++
 3 files changed, 89 insertions(+), 5 deletions(-)

diff --git a/src/hermes_workflows/engine.py b/src/hermes_workflows/engine.py
index b36fe28..ef9170b 100644
--- a/src/hermes_workflows/engine.py
+++ b/src/hermes_workflows/engine.py
@@ -19,7 +19,7 @@
 from .approvals import ApprovalDecision, ApprovalDecisionInput, ApprovalReceipt, ApprovalView, OperatorResponseReceipt
 from .domain import CommandType, WorkflowStatus, decode_command_row, decode_event_row, make_command, make_event
 from .input_parsing import coerce_workflow_input
-from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry, RuntimeServicesV1
+from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry, _make_registry_process_local
 from .status_projection import StatusProjection
 from .types import to_json_value
 from .workflow_values import Workflow
@@ -112,7 +112,10 @@ def __init__(
         self.db_path = Path(db_path)
         self.agent_runner = agent_runner
         self.read_only = read_only
-        self.runtime_services = runtime_services if runtime_services is not None else EmptyRuntimeServicesV1()
+        registry = runtime_services if runtime_services is not None else EmptyRuntimeServicesV1()
+        if not isinstance(registry, RuntimeServiceRegistry):
+            raise TypeError("runtime_services must implement RuntimeServiceRegistry")
+        self.runtime_services = _make_registry_process_local(registry)
         self._status_projection = StatusProjection(self)
         self._active_command_claim = threading.local()
         if read_only:
@@ -3650,7 +3653,7 @@ def _to_jsonable(value: Any) -> Any:
 
 
 def _reject_runtime_service_registries(value: Any, seen: set[int] | None = None) -> None:
-    if isinstance(value, (RuntimeServicesV1, EmptyRuntimeServicesV1)):
+    if isinstance(value, RuntimeServiceRegistry):
         raise TypeError("runtime service registries are process-local and cannot be serialized")
 
     seen = seen if seen is not None else set()
diff --git a/src/hermes_workflows/runtime_services.py b/src/hermes_workflows/runtime_services.py
index 42f03f4..347213a 100644
--- a/src/hermes_workflows/runtime_services.py
+++ b/src/hermes_workflows/runtime_services.py
@@ -2,18 +2,58 @@
 
 import re
 from collections.abc import Mapping
-from dataclasses import dataclass, field
+from dataclasses import dataclass, field, is_dataclass, make_dataclass
 from types import MappingProxyType
-from typing import Protocol, SupportsIndex
+from typing import Any, Callable, Protocol, SupportsIndex, cast, runtime_checkable
 
 
 _SERVICE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
 
 
+@runtime_checkable
 class RuntimeServiceRegistry(Protocol):
     def resolve(self, service_id: str, contract_version: int) -> object | None: ...
 
 
+def _make_registry_process_local(registry: RuntimeServiceRegistry) -> RuntimeServiceRegistry:
+    if isinstance(registry, (RuntimeServicesV1, EmptyRuntimeServicesV1)):
+        return registry
+
+    registry_type = cast(type[Any], type(registry))
+    original_getattribute = cast(Callable[[object, str], object], registry_type.__getattribute__)
+
+    def guarded_getattribute(self: object, name: str) -> object:
+        if name == "_serialization_guard":
+            raise TypeError("runtime service registries are process-local and cannot be serialized")
+        return original_getattribute(self, name)
+
+    def reject_pickle(self: object, protocol: SupportsIndex):
+        raise TypeError("runtime service registries are process-local and cannot be pickled")
+
+    namespace = {
+        "__module__": registry_type.__module__,
+        "__getattribute__": guarded_getattribute,
+        "__reduce_ex__": reject_pickle,
+    }
+    if is_dataclass(registry) and not isinstance(registry, type):
+        dataclass_params = registry_type.__dataclass_params__
+        guarded_type = make_dataclass(
+            f"_ProcessLocal{registry_type.__name__}",
+            [("_serialization_guard", object, field(init=False, repr=False, compare=False))],
+            bases=(registry_type,),
+            namespace=namespace,
+            frozen=dataclass_params.frozen,
+        )
+    else:
+        guarded_type = type(f"_ProcessLocal{registry_type.__name__}", (registry_type,), namespace)
+
+    try:
+        object.__setattr__(registry, "__class__", guarded_type)
+    except TypeError as exc:
+        raise TypeError("runtime service registry cannot be made process-local") from exc
+    return registry
+
+
 @dataclass(frozen=True)
 class RuntimeServicesV1:
     _serialization_guard: object = field(init=False, repr=False, compare=False)
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index ca0ac36..a1e9138 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -3,6 +3,7 @@
 import json
 import pickle
 from collections.abc import Iterator, Mapping
+from dataclasses import dataclass
 
 import pytest
 
@@ -38,6 +39,14 @@ def items(self):  # type: ignore[override]
         return (("test.recording", recording), ("test.recording", recording))
 
 
+@dataclass(frozen=True)
+class IntegrationOwnedRuntimeServices:
+    marker: str
+
+    def resolve(self, service_id: str, contract_version: int) -> object | None:
+        return None
+
+
 def test_runtime_services_resolve_recording_object_by_identity():
     recording = object()
     services = RuntimeServicesV1(services={"test.recording": recording})
@@ -137,3 +146,35 @@ def test_engine_stores_one_registry_without_persisting_it(tmp_path):
 
     assert "test.recording" not in db_path.read_bytes().decode("utf-8", errors="ignore")
     assert all("test.recording" not in json.dumps(event) for event in engine.events("wf_runtime_service_injected"))
+
+
+def test_engine_rejects_arbitrary_protocol_registry_persistence(tmp_path):
+    db_path = tmp_path / "workflow.sqlite"
+    marker = "integration-registry-must-not-persist"
+    registry = IntegrationOwnedRuntimeServices(marker=marker)
+    engine = WorkflowEngine(db_path, runtime_services=registry)
+
+    assert engine.runtime_services is registry
+    assert engine.resolve_runtime_service("integration.any", 1) is None
+    with pytest.raises(TypeError):
+        json.dumps(registry)
+    for serialize in (
+        pickle.dumps,
+        to_json_value,
+        JsonCodec.dumps,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("runtime services", value),
+    ):
+        with pytest.raises(TypeError, match="process-local"):
+            serialize(registry)
+
+    with pytest.raises(TypeError, match="process-local"):
+        engine.start(
+            runtime_service_contract_workflow,
+            {"value": {"registry": registry}},
+            workflow_id="wf_arbitrary_runtime_service_registry",
+        )
+
+    assert marker not in db_path.read_bytes().decode("utf-8", errors="ignore")
+    with pytest.raises(KeyError, match="unknown workflow_id"):
+        engine.events("wf_arbitrary_runtime_service_registry")

From adbf605c201edd3839864357b7c51ddfd17aac48 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 20:32:55 -0700
Subject: [PATCH 05/10] fix(runtime): require explicit process-local registry
 marker

---
 src/hermes_workflows/engine.py           | 32 ++-----------
 src/hermes_workflows/runtime_services.py | 58 ++++--------------------
 src/hermes_workflows/types.py            |  4 ++
 tests/test_runtime_services_contract.py  | 46 +++++++++++++++----
 4 files changed, 53 insertions(+), 87 deletions(-)

diff --git a/src/hermes_workflows/engine.py b/src/hermes_workflows/engine.py
index ef9170b..dc0d0ca 100644
--- a/src/hermes_workflows/engine.py
+++ b/src/hermes_workflows/engine.py
@@ -9,9 +9,8 @@
 import sqlite3
 import threading
 import time
-from collections.abc import Mapping, Sequence
 from contextlib import contextmanager
-from dataclasses import dataclass, fields, is_dataclass
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any, Callable, Dict, List, Optional, Tuple, Union, get_type_hints
 from urllib.parse import quote
@@ -19,7 +18,7 @@
 from .approvals import ApprovalDecision, ApprovalDecisionInput, ApprovalReceipt, ApprovalView, OperatorResponseReceipt
 from .domain import CommandType, WorkflowStatus, decode_command_row, decode_event_row, make_command, make_event
 from .input_parsing import coerce_workflow_input
-from .runtime_services import EmptyRuntimeServicesV1, RuntimeServiceRegistry, _make_registry_process_local
+from .runtime_services import EmptyRuntimeServicesV1, RuntimeOnlyServiceRegistry, RuntimeServiceRegistry
 from .status_projection import StatusProjection
 from .types import to_json_value
 from .workflow_values import Workflow
@@ -113,9 +112,11 @@ def __init__(
         self.agent_runner = agent_runner
         self.read_only = read_only
         registry = runtime_services if runtime_services is not None else EmptyRuntimeServicesV1()
+        if not isinstance(registry, RuntimeOnlyServiceRegistry):
+            raise TypeError("runtime_services must implement the runtime-only marker contract")
         if not isinstance(registry, RuntimeServiceRegistry):
             raise TypeError("runtime_services must implement RuntimeServiceRegistry")
-        self.runtime_services = _make_registry_process_local(registry)
+        self.runtime_services = registry
         self._status_projection = StatusProjection(self)
         self._active_command_claim = threading.local()
         if read_only:
@@ -3648,32 +3649,9 @@ def _validate_step_request_fingerprint(step_key: str, stored_payload: Any, curre
 
 
 def _to_jsonable(value: Any) -> Any:
-    _reject_runtime_service_registries(value)
     return to_json_value(value)
 
 
-def _reject_runtime_service_registries(value: Any, seen: set[int] | None = None) -> None:
-    if isinstance(value, RuntimeServiceRegistry):
-        raise TypeError("runtime service registries are process-local and cannot be serialized")
-
-    seen = seen if seen is not None else set()
-    value_id = id(value)
-    if value_id in seen:
-        return
-    seen.add(value_id)
-
-    if is_dataclass(value) and not isinstance(value, type):
-        for field in fields(value):
-            _reject_runtime_service_registries(getattr(value, field.name), seen)
-    elif isinstance(value, Mapping):
-        for key, item in value.items():
-            _reject_runtime_service_registries(key, seen)
-            _reject_runtime_service_registries(item, seen)
-    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
-        for item in value:
-            _reject_runtime_service_registries(item, seen)
-
-
 def _from_jsonable(value: Any) -> Any:
     if isinstance(value, dict):
         if value.get("__hermes_type__") == "Workflow":
diff --git a/src/hermes_workflows/runtime_services.py b/src/hermes_workflows/runtime_services.py
index 347213a..bc22d9f 100644
--- a/src/hermes_workflows/runtime_services.py
+++ b/src/hermes_workflows/runtime_services.py
@@ -2,9 +2,9 @@
 
 import re
 from collections.abc import Mapping
-from dataclasses import dataclass, field, is_dataclass, make_dataclass
+from dataclasses import dataclass, field
 from types import MappingProxyType
-from typing import Any, Callable, Protocol, SupportsIndex, cast, runtime_checkable
+from typing import Protocol, SupportsIndex, runtime_checkable
 
 
 _SERVICE_ID_PATTERN = re.compile(r"^[a-z][a-z0-9_.-]{0,63}$")
@@ -15,56 +15,20 @@ class RuntimeServiceRegistry(Protocol):
     def resolve(self, service_id: str, contract_version: int) -> object | None: ...
 
 
-def _make_registry_process_local(registry: RuntimeServiceRegistry) -> RuntimeServiceRegistry:
-    if isinstance(registry, (RuntimeServicesV1, EmptyRuntimeServicesV1)):
-        return registry
+class RuntimeOnlyServiceRegistry:
+    """Nominal marker for process-local runtime service registries."""
 
-    registry_type = cast(type[Any], type(registry))
-    original_getattribute = cast(Callable[[object, str], object], registry_type.__getattribute__)
+    __slots__ = ()
 
-    def guarded_getattribute(self: object, name: str) -> object:
-        if name == "_serialization_guard":
-            raise TypeError("runtime service registries are process-local and cannot be serialized")
-        return original_getattribute(self, name)
-
-    def reject_pickle(self: object, protocol: SupportsIndex):
+    def __reduce_ex__(self, protocol: SupportsIndex):
         raise TypeError("runtime service registries are process-local and cannot be pickled")
 
-    namespace = {
-        "__module__": registry_type.__module__,
-        "__getattribute__": guarded_getattribute,
-        "__reduce_ex__": reject_pickle,
-    }
-    if is_dataclass(registry) and not isinstance(registry, type):
-        dataclass_params = registry_type.__dataclass_params__
-        guarded_type = make_dataclass(
-            f"_ProcessLocal{registry_type.__name__}",
-            [("_serialization_guard", object, field(init=False, repr=False, compare=False))],
-            bases=(registry_type,),
-            namespace=namespace,
-            frozen=dataclass_params.frozen,
-        )
-    else:
-        guarded_type = type(f"_ProcessLocal{registry_type.__name__}", (registry_type,), namespace)
-
-    try:
-        object.__setattr__(registry, "__class__", guarded_type)
-    except TypeError as exc:
-        raise TypeError("runtime service registry cannot be made process-local") from exc
-    return registry
-
 
 @dataclass(frozen=True)
-class RuntimeServicesV1:
-    _serialization_guard: object = field(init=False, repr=False, compare=False)
+class RuntimeServicesV1(RuntimeOnlyServiceRegistry):
     schema_version: int = 1
     services: Mapping[str, object] = field(default_factory=dict)
 
-    def __getattribute__(self, name: str) -> object:
-        if name == "_serialization_guard":
-            raise TypeError("runtime service registries are process-local and cannot be serialized")
-        return super().__getattribute__(name)
-
     def __post_init__(self) -> None:
         if type(self.schema_version) is not int or self.schema_version != 1:
             raise ValueError("schema_version must equal 1")
@@ -88,13 +52,7 @@ def __reduce_ex__(self, protocol: SupportsIndex):
 
 
 @dataclass(frozen=True)
-class EmptyRuntimeServicesV1:
-    _serialization_guard: object = field(init=False, repr=False, compare=False)
-
-    def __getattribute__(self, name: str) -> object:
-        if name == "_serialization_guard":
-            raise TypeError("runtime service registries are process-local and cannot be serialized")
-        return super().__getattribute__(name)
+class EmptyRuntimeServicesV1(RuntimeOnlyServiceRegistry):
 
     def resolve(self, service_id: str, contract_version: int) -> object | None:
         _validate_resolution(service_id, contract_version)
diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index 14afce5..bf19f48 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -32,6 +32,10 @@ def to_json_value(value: object) -> JsonValue:
     at typed input boundaries; this helper only guarantees JSON shape.
     """
 
+    from .runtime_services import RuntimeOnlyServiceRegistry
+
+    if isinstance(value, RuntimeOnlyServiceRegistry):
+        raise TypeError("runtime service registries are process-local and cannot be serialized")
     if value is None or isinstance(value, (str, int, float, bool)):
         return value
     if _is_framework_json_value(value):
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index a1e9138..1d95884 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -10,7 +10,11 @@
 from hermes_workflows import WorkflowEngine, workflow
 from hermes_workflows.artifacts import JsonArtifact
 from hermes_workflows.engine import JsonCodec
-from hermes_workflows.runtime_services import EmptyRuntimeServicesV1, RuntimeServicesV1
+from hermes_workflows.runtime_services import (
+    EmptyRuntimeServicesV1,
+    RuntimeOnlyServiceRegistry,
+    RuntimeServicesV1,
+)
 from hermes_workflows.status_projection import JsonCodec as StatusProjectionJsonCodec
 from hermes_workflows.types import to_json_value
 
@@ -40,13 +44,18 @@ def items(self):  # type: ignore[override]
 
 
 @dataclass(frozen=True)
-class IntegrationOwnedRuntimeServices:
+class IntegrationOwnedRuntimeServices(RuntimeOnlyServiceRegistry):
     marker: str
 
     def resolve(self, service_id: str, contract_version: int) -> object | None:
         return None
 
 
+class UnmarkedStructuralRuntimeServices:
+    def resolve(self, service_id: str, contract_version: int) -> object | None:
+        return None
+
+
 def test_runtime_services_resolve_recording_object_by_identity():
     recording = object()
     services = RuntimeServicesV1(services={"test.recording": recording})
@@ -148,33 +157,50 @@ def test_engine_stores_one_registry_without_persisting_it(tmp_path):
     assert all("test.recording" not in json.dumps(event) for event in engine.events("wf_runtime_service_injected"))
 
 
-def test_engine_rejects_arbitrary_protocol_registry_persistence(tmp_path):
+def test_engine_preserves_marked_registry_identity_without_mutation_or_wrapping(tmp_path):
     db_path = tmp_path / "workflow.sqlite"
     marker = "integration-registry-must-not-persist"
     registry = IntegrationOwnedRuntimeServices(marker=marker)
+    original_type = type(registry)
+    original_state = dict(registry.__dict__)
     engine = WorkflowEngine(db_path, runtime_services=registry)
 
     assert engine.runtime_services is registry
+    assert type(registry) is original_type
+    assert registry.__dict__ == original_state
     assert engine.resolve_runtime_service("integration.any", 1) is None
-    with pytest.raises(TypeError):
-        json.dumps(registry)
+
+
+def test_engine_rejects_unmarked_structural_registry(tmp_path):
+    registry = UnmarkedStructuralRuntimeServices()
+
+    with pytest.raises(TypeError, match="runtime-only marker"):
+        WorkflowEngine(tmp_path / "workflow.sqlite", runtime_services=registry)
+
+
+def test_marked_registry_rejected_by_framework_serializers_and_persistence(tmp_path):
+    db_path = tmp_path / "workflow.sqlite"
+    marker = "integration-registry-must-not-persist"
+    registry = IntegrationOwnedRuntimeServices(marker=marker)
+    engine = WorkflowEngine(db_path, runtime_services=registry)
+    nested = {"outer": [{"registry": registry}]}
+
     for serialize in (
-        pickle.dumps,
         to_json_value,
         JsonCodec.dumps,
         StatusProjectionJsonCodec.dumps,
         lambda value: JsonArtifact("runtime services", value),
     ):
         with pytest.raises(TypeError, match="process-local"):
-            serialize(registry)
+            serialize(nested)
 
     with pytest.raises(TypeError, match="process-local"):
         engine.start(
             runtime_service_contract_workflow,
-            {"value": {"registry": registry}},
-            workflow_id="wf_arbitrary_runtime_service_registry",
+            {"value": nested},
+            workflow_id="wf_marked_runtime_service_registry",
         )
 
     assert marker not in db_path.read_bytes().decode("utf-8", errors="ignore")
     with pytest.raises(KeyError, match="unknown workflow_id"):
-        engine.events("wf_arbitrary_runtime_service_registry")
+        engine.events("wf_marked_runtime_service_registry")

From c60c1e8dd41d9b47b345b6cd8d43bf36a9e55e12 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 21:08:35 -0700
Subject: [PATCH 06/10] fix(runtime): reject registries in mapping keys

---
 src/hermes_workflows/types.py           | 25 +++++++++++++-
 tests/test_runtime_services_contract.py | 45 +++++++++++++++++++++++++
 2 files changed, 69 insertions(+), 1 deletion(-)

diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index bf19f48..e669654 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -45,7 +45,11 @@ def to_json_value(value: object) -> JsonValue:
     if is_dataclass(value) and not isinstance(value, type):
         return {field.name: to_json_value(getattr(value, field.name)) for field in fields(value)}
     if isinstance(value, Mapping):
-        return {str(key): to_json_value(item) for key, item in value.items()}
+        normalized = {}
+        for key, item in value.items():
+            _reject_runtime_service_registry(key)
+            normalized[str(key)] = to_json_value(item)
+        return normalized
     if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
         return [to_json_value(item) for item in value]
     raise TypeError(f"value of type {type(value).__name__} is not JSON-serializable")
@@ -60,6 +64,25 @@ def to_json_object(value: object) -> JsonObject:
     return normalized
 
 
+def _reject_runtime_service_registry(value: object) -> None:
+    """Reject process-local registries nested in values about to become mapping keys."""
+
+    from .runtime_services import RuntimeOnlyServiceRegistry
+
+    if isinstance(value, RuntimeOnlyServiceRegistry):
+        raise TypeError("runtime service registries are process-local and cannot be serialized")
+    if is_dataclass(value) and not isinstance(value, type):
+        for field in fields(value):
+            _reject_runtime_service_registry(getattr(value, field.name))
+    elif isinstance(value, Mapping):
+        for key, item in value.items():
+            _reject_runtime_service_registry(key)
+            _reject_runtime_service_registry(item)
+    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
+        for item in value:
+            _reject_runtime_service_registry(item)
+
+
 def _is_framework_json_value(value: object) -> bool:
     try:
         from .workflow_values import Workflow
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index 1d95884..0afd452 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -204,3 +204,48 @@ def test_marked_registry_rejected_by_framework_serializers_and_persistence(tmp_p
     assert marker not in db_path.read_bytes().decode("utf-8", errors="ignore")
     with pytest.raises(KeyError, match="unknown workflow_id"):
         engine.events("wf_marked_runtime_service_registry")
+
+
+@pytest.mark.parametrize(
+    ("key_factory", "workflow_id"),
+    [
+        pytest.param(lambda registry: registry, "wf_registry_direct_mapping_key", id="direct-key"),
+        pytest.param(lambda registry: ("nested", registry), "wf_registry_nested_mapping_key", id="nested-key"),
+    ],
+)
+def test_marked_registry_rejected_as_mapping_key_by_serializers_and_persistence(
+    tmp_path,
+    key_factory,
+    workflow_id,
+):
+    db_path = tmp_path / "workflow.sqlite"
+    secret = "mapping-key-registry-secret-must-not-persist"
+    registry = IntegrationOwnedRuntimeServices(marker=secret)
+    payload = {key_factory(registry): "safe value"}
+
+    for serialize in (
+        to_json_value,
+        JsonCodec.dumps,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("runtime services", value),
+    ):
+        with pytest.raises(TypeError, match="process-local"):
+            serialize(payload)
+
+    engine = WorkflowEngine(db_path, runtime_services=registry)
+    with pytest.raises(TypeError, match="process-local"):
+        engine.start(
+            runtime_service_contract_workflow,
+            {"value": payload},
+            workflow_id=workflow_id,
+        )
+
+    assert secret not in db_path.read_bytes().decode("utf-8", errors="ignore")
+    with pytest.raises(KeyError, match="unknown workflow_id"):
+        engine.events(workflow_id)
+
+
+def test_safe_mapping_keys_preserve_string_conversion_compatibility():
+    payload = {7: "integer", ("safe", 1): "tuple"}
+
+    assert to_json_value(payload) == {"7": "integer", "('safe', 1)": "tuple"}

From be6b7326f358b8c6f6c01211c1a78e05f047ef60 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 21:32:24 -0700
Subject: [PATCH 07/10] Reject runtime registries in set-like mapping keys

---
 src/hermes_workflows/types.py           |  5 ++++-
 tests/test_runtime_services_contract.py | 10 ++++++++++
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index e669654..062dbcc 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from collections.abc import Mapping, Sequence
+from collections.abc import Mapping, Sequence, Set
 from dataclasses import fields, is_dataclass
 from typing import TYPE_CHECKING, Any, Protocol, Union, cast
 
@@ -81,6 +81,9 @@ def _reject_runtime_service_registry(value: object) -> None:
     elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
         for item in value:
             _reject_runtime_service_registry(item)
+    elif isinstance(value, Set):
+        for item in value:
+            _reject_runtime_service_registry(item)
 
 
 def _is_framework_json_value(value: object) -> bool:
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index 0afd452..e0482cb 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -211,6 +211,16 @@ def test_marked_registry_rejected_by_framework_serializers_and_persistence(tmp_p
     [
         pytest.param(lambda registry: registry, "wf_registry_direct_mapping_key", id="direct-key"),
         pytest.param(lambda registry: ("nested", registry), "wf_registry_nested_mapping_key", id="nested-key"),
+        pytest.param(
+            lambda registry: frozenset({registry}),
+            "wf_registry_direct_frozenset_mapping_key",
+            id="direct-frozenset-key",
+        ),
+        pytest.param(
+            lambda registry: ("nested", frozenset({registry})),
+            "wf_registry_nested_frozenset_mapping_key",
+            id="nested-frozenset-key",
+        ),
     ],
 )
 def test_marked_registry_rejected_as_mapping_key_by_serializers_and_persistence(

From d9f175063c789ce6f56bccc43a3ce677c6c33f0a Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:12:49 -0700
Subject: [PATCH 08/10] fix: reject cyclic runtime registry mapping keys

---
 src/hermes_workflows/types.py           |  52 +++++++---
 tests/test_runtime_services_contract.py | 122 +++++++++++++++++++++++-
 2 files changed, 160 insertions(+), 14 deletions(-)

diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index 062dbcc..68a6812 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -67,23 +67,49 @@ def to_json_object(value: object) -> JsonObject:
 def _reject_runtime_service_registry(value: object) -> None:
     """Reject process-local registries nested in values about to become mapping keys."""
 
+    if _mapping_key_contains_registry_or_cycle(value, active_ids=set()):
+        raise TypeError("cyclic mapping keys are not JSON-serializable")
+
+
+def _mapping_key_contains_registry_or_cycle(value: object, *, active_ids: set[int]) -> bool:
+    """Reject registries eagerly and report cycles after inspecting sibling values."""
+
     from .runtime_services import RuntimeOnlyServiceRegistry
 
     if isinstance(value, RuntimeOnlyServiceRegistry):
         raise TypeError("runtime service registries are process-local and cannot be serialized")
-    if is_dataclass(value) and not isinstance(value, type):
-        for field in fields(value):
-            _reject_runtime_service_registry(getattr(value, field.name))
-    elif isinstance(value, Mapping):
-        for key, item in value.items():
-            _reject_runtime_service_registry(key)
-            _reject_runtime_service_registry(item)
-    elif isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)):
-        for item in value:
-            _reject_runtime_service_registry(item)
-    elif isinstance(value, Set):
-        for item in value:
-            _reject_runtime_service_registry(item)
+
+    is_dataclass_instance = is_dataclass(value) and not isinstance(value, type)
+    is_mapping = isinstance(value, Mapping)
+    is_sequence = isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray))
+    is_set = isinstance(value, Set)
+    if not (is_dataclass_instance or is_mapping or is_sequence or is_set):
+        return False
+
+    value_id = id(value)
+    if value_id in active_ids:
+        return True
+
+    active_ids.add(value_id)
+    cycle_found = False
+    try:
+        if is_dataclass_instance:
+            for field in fields(cast(Any, value)):
+                if _mapping_key_contains_registry_or_cycle(getattr(value, field.name), active_ids=active_ids):
+                    cycle_found = True
+        elif is_mapping:
+            for key, item in cast(Mapping[object, object], value).items():
+                if _mapping_key_contains_registry_or_cycle(key, active_ids=active_ids):
+                    cycle_found = True
+                if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
+                    cycle_found = True
+        else:
+            for item in cast(Sequence[object] | Set[object], value):
+                if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
+                    cycle_found = True
+    finally:
+        active_ids.remove(value_id)
+    return cycle_found
 
 
 def _is_framework_json_value(value: object) -> bool:
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index e0482cb..4542b84 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -2,7 +2,7 @@
 
 import json
 import pickle
-from collections.abc import Iterator, Mapping
+from collections.abc import Iterator, Mapping, Sequence, Set
 from dataclasses import dataclass
 
 import pytest
@@ -56,6 +56,69 @@ def resolve(self, service_id: str, contract_version: int) -> object | None:
         return None
 
 
+class HashableCycleSequence(Sequence[object]):
+    def __init__(self, registry: RuntimeOnlyServiceRegistry | None = None):
+        self._items = (self, registry) if registry is not None else (self,)
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __getitem__(self, index):
+        return self._items[index]
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+
+class HashableCycleMapping(Mapping[object, object]):
+    def __init__(self, registry: RuntimeOnlyServiceRegistry):
+        self._items = (("self", self), ("registry", registry))
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __getitem__(self, key: object) -> object:
+        for candidate, value in self._items:
+            if candidate == key:
+                return value
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[object]:
+        return (key for key, _ in self._items)
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+    def items(self):  # type: ignore[override]
+        return self._items
+
+
+class HashableCycleSet(Set[object]):
+    def __init__(self, registry: RuntimeOnlyServiceRegistry):
+        self._items = (self, registry)
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __contains__(self, item: object) -> bool:
+        return any(candidate is item for candidate in self._items)
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(self._items)
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+
+@dataclass(eq=False)
+class HashableCycleDataclass:
+    self_reference: object | None = None
+    registry: RuntimeOnlyServiceRegistry | None = None
+
+    def __hash__(self) -> int:
+        return id(self)
+
+
 def test_runtime_services_resolve_recording_object_by_identity():
     recording = object()
     services = RuntimeServicesV1(services={"test.recording": recording})
@@ -259,3 +322,60 @@ def test_safe_mapping_keys_preserve_string_conversion_compatibility():
     payload = {7: "integer", ("safe", 1): "tuple"}
 
     assert to_json_value(payload) == {"7": "integer", "('safe', 1)": "tuple"}
+
+
+def _cyclic_key_with_registry(shape: str, registry: RuntimeOnlyServiceRegistry) -> object:
+    if shape == "sequence":
+        return HashableCycleSequence(registry)
+    if shape == "mapping":
+        return HashableCycleMapping(registry)
+    if shape == "set":
+        return HashableCycleSet(registry)
+    if shape == "dataclass":
+        value = HashableCycleDataclass(registry=registry)
+        value.self_reference = value
+        return value
+    raise AssertionError(f"unknown shape: {shape}")
+
+
+@pytest.mark.parametrize("shape", ["sequence", "mapping", "set", "dataclass"])
+def test_cyclic_mapping_keys_do_not_mask_marked_registry_rejection(tmp_path, shape):
+    db_path = tmp_path / "workflow.sqlite"
+    secret = f"cyclic-{shape}-registry-secret-must-not-persist"
+    registry = IntegrationOwnedRuntimeServices(marker=secret)
+    payload = {_cyclic_key_with_registry(shape, registry): "safe value"}
+
+    for serialize in (
+        to_json_value,
+        JsonCodec.dumps,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("runtime services", value),
+    ):
+        with pytest.raises(TypeError, match="process-local"):
+            serialize(payload)
+
+    workflow_id = f"wf_cyclic_{shape}_registry_mapping_key"
+    engine = WorkflowEngine(db_path, runtime_services=registry)
+    with pytest.raises(TypeError, match="process-local"):
+        engine.start(
+            runtime_service_contract_workflow,
+            {"value": payload},
+            workflow_id=workflow_id,
+        )
+
+    assert secret not in db_path.read_bytes().decode("utf-8", errors="ignore")
+    with pytest.raises(KeyError, match="unknown workflow_id"):
+        engine.events(workflow_id)
+
+
+def test_safe_cyclic_non_string_mapping_key_fails_closed_deterministically():
+    payload = {HashableCycleSequence(): "safe value"}
+
+    for serialize in (
+        to_json_value,
+        JsonCodec.dumps,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("cyclic key", value),
+    ):
+        with pytest.raises(TypeError, match="cyclic mapping keys are not JSON-serializable"):
+            serialize(payload)

From cfd1a644054b45c808f5d4d3779ded5cd6e9c212 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:15:23 -0700
Subject: [PATCH 09/10] fix: keep cycle traversal Python 3.9 compatible

---
 src/hermes_workflows/types.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index 68a6812..b4a2021 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -104,7 +104,7 @@ def _mapping_key_contains_registry_or_cycle(value: object, *, active_ids: set[in
                 if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
                     cycle_found = True
         else:
-            for item in cast(Sequence[object] | Set[object], value):
+            for item in cast(Any, value):
                 if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
                     cycle_found = True
     finally:

From 4cc4549fce268d109b1fe2d0698a9438e7930d34 Mon Sep 17 00:00:00 2001
From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com>
Date: Fri, 10 Jul 2026 22:41:08 -0700
Subject: [PATCH 10/10] fix(runtime): inspect overlapping registry key shapes

---
 src/hermes_workflows/types.py           |   9 +-
 tests/test_runtime_services_contract.py | 163 ++++++++++++++++++++++++
 2 files changed, 170 insertions(+), 2 deletions(-)

diff --git a/src/hermes_workflows/types.py b/src/hermes_workflows/types.py
index b4a2021..e945315 100644
--- a/src/hermes_workflows/types.py
+++ b/src/hermes_workflows/types.py
@@ -97,13 +97,18 @@ def _mapping_key_contains_registry_or_cycle(value: object, *, active_ids: set[in
             for field in fields(cast(Any, value)):
                 if _mapping_key_contains_registry_or_cycle(getattr(value, field.name), active_ids=active_ids):
                     cycle_found = True
-        elif is_mapping:
+        if is_mapping:
             for key, item in cast(Mapping[object, object], value).items():
                 if _mapping_key_contains_registry_or_cycle(key, active_ids=active_ids):
                     cycle_found = True
                 if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
                     cycle_found = True
-        else:
+        if is_sequence:
+            sequence = cast(Sequence[object], value)
+            for index in range(len(sequence)):
+                if _mapping_key_contains_registry_or_cycle(sequence[index], active_ids=active_ids):
+                    cycle_found = True
+        if is_set:
             for item in cast(Any, value):
                 if _mapping_key_contains_registry_or_cycle(item, active_ids=active_ids):
                     cycle_found = True
diff --git a/tests/test_runtime_services_contract.py b/tests/test_runtime_services_contract.py
index 4542b84..1dcfb24 100644
--- a/tests/test_runtime_services_contract.py
+++ b/tests/test_runtime_services_contract.py
@@ -4,6 +4,7 @@
 import pickle
 from collections.abc import Iterator, Mapping, Sequence, Set
 from dataclasses import dataclass
+from typing import overload
 
 import pytest
 
@@ -119,6 +120,119 @@ def __hash__(self) -> int:
         return id(self)
 
 
+@dataclass(eq=False)
+class HashableDataclassMapping(Mapping[object, object]):
+    label: str
+
+    def __post_init__(self) -> None:
+        self._items: tuple[tuple[object, object], ...] = ()
+
+    def hide(self, *items: tuple[object, object]) -> None:
+        self._items = items
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __str__(self) -> str:
+        for _, value in self._items:
+            if isinstance(value, IntegrationOwnedRuntimeServices):
+                return value.marker
+        return self.label
+
+    def __getitem__(self, key: object) -> object:
+        for candidate, value in self._items:
+            if candidate == key:
+                return value
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[object]:
+        return (key for key, _ in self._items)
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+    def items(self):  # type: ignore[override]
+        return self._items
+
+
+@dataclass(eq=False)
+class HashableDataclassSequence(Sequence[object]):
+    label: str
+
+    def __post_init__(self) -> None:
+        self._items: tuple[object, ...] = ()
+
+    def hide(self, *items: object) -> None:
+        self._items = items
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    @overload
+    def __getitem__(self, index: int) -> object: ...
+
+    @overload
+    def __getitem__(self, index: slice) -> Sequence[object]: ...
+
+    def __getitem__(self, index: int | slice) -> object | Sequence[object]:
+        return self._items[index]
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+
+@dataclass(eq=False)
+class HashableDataclassSet(Set[object]):
+    label: str
+
+    def __post_init__(self) -> None:
+        self._items: tuple[object, ...] = ()
+
+    def hide(self, *items: object) -> None:
+        self._items = items
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __contains__(self, item: object) -> bool:
+        return any(candidate is item for candidate in self._items)
+
+    def __iter__(self) -> Iterator[object]:
+        return iter(self._items)
+
+    def __len__(self) -> int:
+        return len(self._items)
+
+
+class HashableMappingSequence(Mapping[object, object], Sequence[object]):
+    def __init__(self, registry: IntegrationOwnedRuntimeServices):
+        self._mapping_items = (("safe", "value"), ("also-safe", "value"))
+        self._sequence_items = (self, registry)
+
+    def __hash__(self) -> int:
+        return id(self)
+
+    def __str__(self) -> str:
+        return self._sequence_items[1].marker
+
+    def __getitem__(self, key):  # type: ignore[override]
+        if isinstance(key, int):
+            return self._sequence_items[key]
+        for candidate, value in self._mapping_items:
+            if candidate == key:
+                return value
+        raise KeyError(key)
+
+    def __iter__(self) -> Iterator[object]:
+        return (key for key, _ in self._mapping_items)
+
+    def __len__(self) -> int:
+        return len(self._mapping_items)
+
+    def items(self):  # type: ignore[override]
+        return self._mapping_items
+
+
 def test_runtime_services_resolve_recording_object_by_identity():
     recording = object()
     services = RuntimeServicesV1(services={"test.recording": recording})
@@ -324,6 +438,55 @@ def test_safe_mapping_keys_preserve_string_conversion_compatibility():
     assert to_json_value(payload) == {"7": "integer", "('safe', 1)": "tuple"}
 
 
+def _overlapping_shape_key(shape: str, registry: RuntimeOnlyServiceRegistry) -> object:
+    if shape == "mapping":
+        value = HashableDataclassMapping("safe dataclass field")
+        value.hide(("self", value), ("registry", registry))
+        return value
+    if shape == "sequence":
+        value = HashableDataclassSequence("safe dataclass field")
+        value.hide(value, registry)
+        return value
+    if shape == "set":
+        value = HashableDataclassSet("safe dataclass field")
+        value.hide(value, registry)
+        return value
+    if shape == "mapping-sequence":
+        assert isinstance(registry, IntegrationOwnedRuntimeServices)
+        return HashableMappingSequence(registry)
+    raise AssertionError(f"unknown shape: {shape}")
+
+
+@pytest.mark.parametrize("shape", ["mapping", "sequence", "set", "mapping-sequence"])
+def test_overlapping_dataclass_container_keys_traverse_every_applicable_shape(tmp_path, shape):
+    db_path = tmp_path / "workflow.sqlite"
+    secret = f"overlapping-{shape}-registry-secret-must-not-persist"
+    registry = IntegrationOwnedRuntimeServices(marker=secret)
+    payload = {_overlapping_shape_key(shape, registry): "safe value"}
+
+    for serialize in (
+        to_json_value,
+        JsonCodec.dumps,
+        StatusProjectionJsonCodec.dumps,
+        lambda value: JsonArtifact("overlapping key", value),
+    ):
+        with pytest.raises(TypeError, match="process-local"):
+            serialize(payload)
+
+    workflow_id = f"wf_overlapping_{shape}_registry_mapping_key"
+    engine = WorkflowEngine(db_path, runtime_services=registry)
+    with pytest.raises(TypeError, match="process-local"):
+        engine.start(
+            runtime_service_contract_workflow,
+            {"value": payload},
+            workflow_id=workflow_id,
+        )
+
+    assert secret not in db_path.read_bytes().decode("utf-8", errors="ignore")
+    with pytest.raises(KeyError, match="unknown workflow_id"):
+        engine.events(workflow_id)
+
+
 def _cyclic_key_with_registry(shape: str, registry: RuntimeOnlyServiceRegistry) -> object:
     if shape == "sequence":
         return HashableCycleSequence(registry)