feat: add minimal runtime service registry seam

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, RuntimeOnlyServiceRegistry, RuntimeServiceRegistry
 from .status_projection import StatusProjection
 from .types import to_json_value
 from .workflow_values import Workflow
@@ -105,10 +106,17 @@ 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
+        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 = registry
         self._status_projection = StatusProjection(self)
         self._active_command_claim = threading.local()
         if read_only:
@@ -118,6 +126,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}")

src/hermes_workflows/runtime_services.py

@@ -0,0 +1,73 @@
+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, 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: ...
+
+
+class RuntimeOnlyServiceRegistry:
+    """Nominal marker for process-local runtime service registries."""
+
+    __slots__ = ()
+
+    def __reduce_ex__(self, protocol: SupportsIndex):
+        raise TypeError("runtime service registries are process-local and cannot be pickled")
+
+
+@dataclass(frozen=True)
+class RuntimeServicesV1(RuntimeOnlyServiceRegistry):
+    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(RuntimeOnlyServiceRegistry):
+
+    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")

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
 
@@ -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):
@@ -41,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")
@@ -56,6 +64,59 @@ 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."""
+
+    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")
+
+    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
+        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
+        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
+    finally:
+        active_ids.remove(value_id)
+    return cycle_found
+
+
 def _is_framework_json_value(value: object) -> bool:
     try:
         from .workflow_values import Workflow