From 29c291ae640f9c6ec3b74149b57076699b6afbe6 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Fri, 10 Jul 2026 23:49:55 -0700 Subject: [PATCH 01/11] feat(effects): add stable operation identity and durable receipts --- docs/contracts/effects.md | 37 ++ src/hermes_workflows/effects.py | 573 ++++++++++++++++++ .../contracts/test_effect_adapter_contract.py | 163 +++++ tests/fixtures/effect_contract_v1.json | 28 + tests/probes/kill_after_effect.py | 208 +++++++ tests/test_effect_idempotency.py | 257 ++++++++ 6 files changed, 1266 insertions(+) create mode 100644 docs/contracts/effects.md create mode 100644 src/hermes_workflows/effects.py create mode 100644 tests/contracts/test_effect_adapter_contract.py create mode 100644 tests/fixtures/effect_contract_v1.json create mode 100644 tests/probes/kill_after_effect.py create mode 100644 tests/test_effect_idempotency.py diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md new file mode 100644 index 0000000..f6684b8 --- /dev/null +++ b/docs/contracts/effects.md @@ -0,0 +1,37 @@ +# Durable effects contract v1 + +Hermes Workflows identifies a logical external operation independently of an execution attempt. The operation ID is the SHA-256 of canonical JSON containing the workflow instance ID, logical effect key, adapter ID, and canonical input hash. Attempt numbers, claim tokens, worker IDs, and clocks are deliberately excluded. + +## Delivery semantics + +This contract is **at least once**, not exactly once. A worker may start more than once. An idempotent adapter must accept the stable operation ID, expose `lookup_receipt(operation_id)`, and ensure repeated `perform(operation_id, input)` calls converge on one externally observable operation. The framework's durable receipt proves what the adapter reported; it cannot prove universal exactly-once delivery. + +Policies are explicit: + +- `pure`: no externally observable mutation. +- `idempotent`: repeats with the same operation ID converge on the same external result. +- `unsafe`: execution is refused unless a caller explicitly authorizes it; a pending intent can be claimed once, but an expired uncertain claim cannot be reclaimed automatically and must go to human reconciliation. +- `unclassified`: refused. Legacy effects are unclassified until an owner classifies and adapts them. + +No policy is inferred from a function name, tool name, or implementation. + +## Durable records and fencing + +`effect_intents` stores canonical input, input hash, policy, state, attempt count, and the active claim token. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. + +`effect_receipts` stores the adapter receipt, its SHA-256, sensitivity flag, completion time, and the claim token that fenced the write. Intent completion and receipt insertion are one SQLite transaction. + +The public receipt descriptor contains only operation ID, adapter receipt ID, receipt hash, presence, sensitivity, and completion time. It never includes the raw payload. Sensitive payloads remain available only through the durable effect store's privileged lookup path. + +## Adapter lookup + +Adapters are obtained through the FND-RT process-local service registry at service ID `effects.adapters`, contract version `1`. The resolver returns an adapter whose `adapter_id` exactly matches the request and which implements: + +- `lookup_receipt(operation_id)` +- `perform(operation_id, canonical_input)` + +The coordinator always queries the adapter receipt before performing the operation. This closes the recovery window where the external system accepted the operation but the local receipt transaction did not commit, provided the external adapter can look up the stable operation ID. + +## Crash windows + +The probe covers the four approved windows: before the adapter call; during the adapter call; after external success before adapter-receipt commit; and after durable receipt commit before linked-command completion. Each scenario runs at least two worker attempts, produces one external effect, and ends with one completed local receipt. This is evidence for the supplied file-backed adapter, not a general exactly-once guarantee. diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py new file mode 100644 index 0000000..bf16eaa --- /dev/null +++ b/src/hermes_workflows/effects.py @@ -0,0 +1,573 @@ +from __future__ import annotations + +import hashlib +import json +import math +import re +import secrets +import sqlite3 +import time +from collections.abc import Mapping +from contextlib import contextmanager +from dataclasses import dataclass +from enum import Enum +from pathlib import Path +from types import MappingProxyType +from typing import Any, Iterator, Protocol, Union, cast, runtime_checkable + + +EFFECT_ADAPTER_SERVICE_ID = "effects.adapters" +EFFECT_ADAPTER_CONTRACT_VERSION = 1 +_ID_PATTERN = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:/-]{0,255}$") +_STATE_VALUES = frozenset({"pending", "claimed", "completed", "failed"}) + + +class EffectPolicy(str, Enum): + PURE = "pure" + IDEMPOTENT = "idempotent" + UNSAFE = "unsafe" + UNCLASSIFIED = "unclassified" + + +JsonScalar = Union[None, bool, int, float, str] +JsonValue = Union[JsonScalar, list["JsonValue"], dict[str, "JsonValue"]] + + +@dataclass(frozen=True) +class OperationIdentity: + operation_id: str + workflow_id: str + effect_key: str + adapter_id: str + input_hash: str + + +@dataclass(frozen=True) +class EffectClaim: + operation_id: str + token: str + attempt: int + expires_at: float + + +@dataclass(frozen=True) +class EffectReceipt: + operation_id: str + adapter_receipt_id: str + payload: Mapping[str, Any] + receipt_hash: str + sensitive: bool + completed_at: float + claim_token: str + + def descriptor(self) -> dict[str, Any]: + """Return a bounded projection which never includes the receipt payload.""" + return { + "operation_id": self.operation_id, + "adapter_receipt_id": self.adapter_receipt_id, + "receipt_hash": self.receipt_hash, + "receipt_present": True, + "sensitive": self.sensitive, + "completed_at": self.completed_at, + } + + +@dataclass(frozen=True) +class EffectRecord: + identity: OperationIdentity + policy: EffectPolicy + state: str + attempts: int + claim_token: str | None + claim_expires_at: float | None + receipt: EffectReceipt | None + error: Mapping[str, Any] | None + + +@runtime_checkable +class EffectAdapter(Protocol): + adapter_id: str + + def lookup_receipt(self, operation_id: str) -> Mapping[str, Any] | None: ... + + def perform(self, operation_id: str, input_value: JsonValue) -> Mapping[str, Any]: ... + + +@runtime_checkable +class EffectAdapterResolver(Protocol): + def resolve_adapter(self, adapter_id: str) -> EffectAdapter | None: ... + + +def canonical_json(value: Any) -> str: + normalized = _normalize_json(value) + return json.dumps( + normalized, + ensure_ascii=False, + allow_nan=False, + separators=(",", ":"), + sort_keys=True, + ) + + +def operation_identity( + *, + workflow_id: str, + effect_key: str, + adapter_id: str, + input_value: Any, + attempt: int | None = None, +) -> OperationIdentity: + """Build an attempt-independent identity from logical effect coordinates and input.""" + del attempt + _validate_id(workflow_id, "workflow_id") + _validate_id(effect_key, "effect_key") + _validate_id(adapter_id, "adapter_id") + input_json = canonical_json(input_value) + input_hash = _sha256(input_json) + operation_document = canonical_json( + { + "schema_version": 1, + "workflow_id": workflow_id, + "effect_key": effect_key, + "adapter_id": adapter_id, + "input_hash": input_hash, + } + ) + return OperationIdentity( + operation_id=f"op_{_sha256(operation_document)}", + workflow_id=workflow_id, + effect_key=effect_key, + adapter_id=adapter_id, + input_hash=input_hash, + ) + + +def resolve_effect_adapter(runtime_services: Any, adapter_id: str) -> EffectAdapter: + """Resolve the issue-owned adapter service through the FND-RT generic seam.""" + _validate_id(adapter_id, "adapter_id") + service = runtime_services.resolve(EFFECT_ADAPTER_SERVICE_ID, EFFECT_ADAPTER_CONTRACT_VERSION) + if service is None: + raise LookupError("effect adapter resolver service is unavailable") + resolve_adapter = getattr(service, "resolve_adapter", None) + if not callable(resolve_adapter): + raise TypeError("effect adapter resolver does not implement contract version 1") + adapter = resolve_adapter(adapter_id) + if adapter is None: + raise LookupError(f"effect adapter is unavailable: {adapter_id}") + if getattr(adapter, "adapter_id", None) != adapter_id: + raise ValueError("resolved effect adapter identity mismatch") + if not callable(getattr(adapter, "lookup_receipt", None)) or not callable( + getattr(adapter, "perform", None) + ): + raise TypeError("effect adapter does not implement contract version 1") + return cast(EffectAdapter, adapter) + + +class SQLiteEffectStore: + """Durable intent, fencing claim, and receipt storage for effect adapters.""" + + def __init__(self, path: str | Path): + self.path = Path(path) + self.path.parent.mkdir(parents=True, exist_ok=True) + self._initialize() + + def ensure_intent( + self, + identity: OperationIdentity, + policy: EffectPolicy | str, + input_value: Any, + *, + now: float | None = None, + ) -> EffectRecord: + policy_value = _coerce_policy(policy) + input_json = canonical_json(input_value) + if _sha256(input_json) != identity.input_hash: + raise ValueError("input conflicts with operation identity") + timestamp = _timestamp(now) + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + """ + INSERT OR IGNORE INTO effect_intents ( + operation_id, workflow_id, effect_key, adapter_id, policy, + input_hash, input_json, state, attempts, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?) + """, + ( + identity.operation_id, + identity.workflow_id, + identity.effect_key, + identity.adapter_id, + policy_value.value, + identity.input_hash, + input_json, + timestamp, + timestamp, + ), + ) + row = self._intent_row(conn, identity.operation_id) + expected = ( + identity.workflow_id, + identity.effect_key, + identity.adapter_id, + policy_value.value, + identity.input_hash, + input_json, + ) + actual = tuple(row[key] for key in expected_intent_columns()) + if actual != expected: + raise ValueError("operation identity conflicts with durable intent") + conn.commit() + return self.get(identity.operation_id) + + def claim( + self, + operation_id: str, + *, + ttl_seconds: float = 30.0, + now: float | None = None, + token: str | None = None, + ) -> EffectClaim: + _validate_operation_id(operation_id) + if not isinstance(ttl_seconds, (int, float)) or isinstance(ttl_seconds, bool): + raise TypeError("ttl_seconds must be numeric") + if not math.isfinite(float(ttl_seconds)) or ttl_seconds <= 0: + raise ValueError("ttl_seconds must be finite and greater than zero") + timestamp = _timestamp(now) + claim_token = token or secrets.token_urlsafe(24) + if not isinstance(claim_token, str) or not claim_token: + raise ValueError("claim token must be a nonempty string") + expires_at = timestamp + float(ttl_seconds) + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.execute( + """ + UPDATE effect_intents + SET state = 'claimed', claim_token = ?, claim_expires_at = ?, + attempts = attempts + 1, updated_at = ?, error_json = NULL + WHERE operation_id = ? + AND ( + (state = 'pending' AND policy IN ('pure', 'idempotent', 'unsafe')) + OR ( + state = 'claimed' AND claim_expires_at <= ? + AND policy IN ('pure', 'idempotent') + ) + ) + """, + (claim_token, expires_at, timestamp, operation_id, timestamp), + ) + if cursor.rowcount != 1: + if conn.execute( + "SELECT 1 FROM effect_intents WHERE operation_id = ?", (operation_id,) + ).fetchone() is None: + raise KeyError(f"unknown operation_id: {operation_id}") + raise RuntimeError("effect intent is not claimable") + attempt = int( + conn.execute( + "SELECT attempts FROM effect_intents WHERE operation_id = ?", (operation_id,) + ).fetchone()[0] + ) + conn.commit() + return EffectClaim(operation_id, claim_token, attempt, expires_at) + + def complete( + self, + operation_id: str, + claim_token: str, + receipt_payload: Mapping[str, Any], + *, + adapter_receipt_id: str | None = None, + sensitive: bool = False, + now: float | None = None, + ) -> EffectRecord: + _validate_operation_id(operation_id) + payload_json = canonical_json(dict(receipt_payload)) + timestamp = _timestamp(now) + receipt_hash = _sha256(payload_json) + receipt_id = adapter_receipt_id or receipt_hash + if not isinstance(receipt_id, str) or not receipt_id: + raise ValueError("adapter_receipt_id must be a nonempty string") + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.execute( + """ + UPDATE effect_intents + SET state = 'completed', updated_at = ? + WHERE operation_id = ? AND state = 'claimed' AND claim_token = ? + AND claim_expires_at > ? + """, + (timestamp, operation_id, claim_token, timestamp), + ) + if cursor.rowcount != 1: + raise RuntimeError("stale claim token cannot complete effect") + conn.execute( + """ + INSERT INTO effect_receipts ( + operation_id, adapter_receipt_id, receipt_json, receipt_hash, + sensitive, completed_at, claim_token + ) VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + operation_id, + receipt_id, + payload_json, + receipt_hash, + int(bool(sensitive)), + timestamp, + claim_token, + ), + ) + conn.commit() + return self.get(operation_id) + + def fail( + self, + operation_id: str, + claim_token: str, + error: Mapping[str, Any], + *, + now: float | None = None, + ) -> EffectRecord: + _validate_operation_id(operation_id) + error_json = canonical_json(dict(error)) + timestamp = _timestamp(now) + with self._connect() as conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.execute( + """ + UPDATE effect_intents + SET state = 'failed', error_json = ?, updated_at = ? + WHERE operation_id = ? AND state = 'claimed' AND claim_token = ? + AND claim_expires_at > ? + """, + (error_json, timestamp, operation_id, claim_token, timestamp), + ) + if cursor.rowcount != 1: + raise RuntimeError("stale claim token cannot fail effect") + conn.commit() + return self.get(operation_id) + + def get(self, operation_id: str) -> EffectRecord: + _validate_operation_id(operation_id) + with self._connect() as conn: + row = conn.execute( + "SELECT * FROM effect_intents WHERE operation_id = ?", (operation_id,) + ).fetchone() + if row is None: + raise KeyError(f"unknown operation_id: {operation_id}") + receipt_row = conn.execute( + "SELECT * FROM effect_receipts WHERE operation_id = ?", (operation_id,) + ).fetchone() + return _record_from_rows(row, receipt_row) + + def lookup_receipt(self, operation_id: str) -> EffectReceipt | None: + return self.get(operation_id).receipt + + def _initialize(self) -> None: + with self._connect() as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS effect_intents ( + operation_id TEXT PRIMARY KEY, + workflow_id TEXT NOT NULL, + effect_key TEXT NOT NULL, + adapter_id TEXT NOT NULL, + policy TEXT NOT NULL CHECK (policy IN ('pure','idempotent','unsafe','unclassified')), + input_hash TEXT NOT NULL, + input_json TEXT NOT NULL, + state TEXT NOT NULL CHECK (state IN ('pending','claimed','completed','failed')), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + claim_token TEXT, + claim_expires_at REAL, + error_json TEXT, + created_at REAL NOT NULL, + updated_at REAL NOT NULL, + CHECK ( + (state = 'pending' AND claim_token IS NULL AND claim_expires_at IS NULL) + OR (state IN ('claimed','completed','failed') AND claim_token IS NOT NULL) + ) + ); + CREATE TABLE IF NOT EXISTS effect_receipts ( + operation_id TEXT PRIMARY KEY REFERENCES effect_intents(operation_id), + adapter_receipt_id TEXT NOT NULL, + receipt_json TEXT NOT NULL, + receipt_hash TEXT NOT NULL, + sensitive INTEGER NOT NULL CHECK (sensitive IN (0,1)), + completed_at REAL NOT NULL, + claim_token TEXT NOT NULL + ); + CREATE INDEX IF NOT EXISTS effect_intents_claimable + ON effect_intents(state, claim_expires_at); + """ + ) + + @contextmanager + def _connect(self) -> Iterator[sqlite3.Connection]: + conn = sqlite3.connect(str(self.path), timeout=30.0, isolation_level=None) + try: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys = ON") + conn.execute("PRAGMA busy_timeout = 30000") + yield conn + finally: + conn.close() + + @staticmethod + def _intent_row(conn: sqlite3.Connection, operation_id: str) -> sqlite3.Row: + row = conn.execute( + "SELECT * FROM effect_intents WHERE operation_id = ?", (operation_id,) + ).fetchone() + if row is None: + raise KeyError(f"unknown operation_id: {operation_id}") + return row + + +class EffectCoordinator: + """Explicit one-attempt coordinator; callers own retry policy and scheduling.""" + + def __init__(self, store: SQLiteEffectStore): + self.store = store + + def prepare( + self, + *, + workflow_id: str, + effect_key: str, + adapter_id: str, + input_value: Any, + policy: EffectPolicy | str, + allow_unsafe: bool = False, + ) -> EffectRecord: + policy_value = _coerce_policy(policy) + if policy_value is EffectPolicy.UNCLASSIFIED: + raise ValueError("unclassified or legacy effects refuse execution") + if policy_value is EffectPolicy.UNSAFE and not allow_unsafe: + raise ValueError("unsafe effects require explicit authorization") + identity = operation_identity( + workflow_id=workflow_id, + effect_key=effect_key, + adapter_id=adapter_id, + input_value=input_value, + ) + return self.store.ensure_intent(identity, policy_value, input_value) + + def execute_claimed( + self, + record: EffectRecord, + claim: EffectClaim, + adapter: EffectAdapter, + input_value: Any, + *, + sensitive_receipt: bool = False, + ) -> EffectRecord: + if record.identity.adapter_id != adapter.adapter_id: + raise ValueError("effect adapter identity mismatch") + if _sha256(canonical_json(input_value)) != record.identity.input_hash: + raise ValueError("effect input conflicts with durable intent") + existing = adapter.lookup_receipt(record.identity.operation_id) + payload = existing if existing is not None else adapter.perform(record.identity.operation_id, input_value) + if not isinstance(payload, Mapping): + raise TypeError("effect adapter receipt must be a mapping") + adapter_receipt_id = payload.get("adapter_receipt_id") + return self.store.complete( + record.identity.operation_id, + claim.token, + payload, + adapter_receipt_id=str(adapter_receipt_id) if adapter_receipt_id is not None else None, + sensitive=sensitive_receipt, + ) + + +def expected_intent_columns() -> tuple[str, ...]: + return ("workflow_id", "effect_key", "adapter_id", "policy", "input_hash", "input_json") + + +def _record_from_rows(row: sqlite3.Row, receipt_row: sqlite3.Row | None) -> EffectRecord: + state = str(row["state"]) + if state not in _STATE_VALUES: + raise ValueError(f"invalid durable effect state: {state}") + identity = OperationIdentity( + operation_id=str(row["operation_id"]), + workflow_id=str(row["workflow_id"]), + effect_key=str(row["effect_key"]), + adapter_id=str(row["adapter_id"]), + input_hash=str(row["input_hash"]), + ) + receipt = None + if receipt_row is not None: + payload = json.loads(str(receipt_row["receipt_json"])) + receipt = EffectReceipt( + operation_id=identity.operation_id, + adapter_receipt_id=str(receipt_row["adapter_receipt_id"]), + payload=MappingProxyType(payload), + receipt_hash=str(receipt_row["receipt_hash"]), + sensitive=bool(receipt_row["sensitive"]), + completed_at=float(receipt_row["completed_at"]), + claim_token=str(receipt_row["claim_token"]), + ) + error_value = json.loads(str(row["error_json"])) if row["error_json"] is not None else None + return EffectRecord( + identity=identity, + policy=EffectPolicy(str(row["policy"])), + state=state, + attempts=int(row["attempts"]), + claim_token=str(row["claim_token"]) if row["claim_token"] is not None else None, + claim_expires_at=( + float(row["claim_expires_at"]) if row["claim_expires_at"] is not None else None + ), + receipt=receipt, + error=MappingProxyType(error_value) if error_value is not None else None, + ) + + +def _normalize_json(value: Any) -> JsonValue: + if value is None or isinstance(value, (str, bool)): + return value + if isinstance(value, int) and not isinstance(value, bool): + return value + if isinstance(value, float): + if not math.isfinite(value): + raise ValueError("JSON numbers must be finite") + return value + if isinstance(value, list): + return [_normalize_json(item) for item in value] + if isinstance(value, tuple): + return [_normalize_json(item) for item in value] + if isinstance(value, Mapping): + result: dict[str, JsonValue] = {} + for key, item in value.items(): + if not isinstance(key, str): + raise TypeError("JSON object keys must be strings") + result[key] = _normalize_json(item) + return result + raise TypeError(f"value is not canonical JSON: {type(value).__name__}") + + +def _coerce_policy(value: EffectPolicy | str) -> EffectPolicy: + try: + return value if isinstance(value, EffectPolicy) else EffectPolicy(value) + except (TypeError, ValueError) as exc: + raise ValueError("effect policy must be pure, idempotent, unsafe, or unclassified") from exc + + +def _validate_id(value: Any, label: str) -> None: + if not isinstance(value, str) or _ID_PATTERN.fullmatch(value) is None: + raise ValueError(f"{label} must be a nonempty canonical identifier") + + +def _validate_operation_id(value: Any) -> None: + if not isinstance(value, str) or re.fullmatch(r"op_[0-9a-f]{64}", value) is None: + raise ValueError("operation_id must be an op_ prefixed SHA-256") + + +def _timestamp(value: float | None) -> float: + timestamp = time.time() if value is None else value + if not isinstance(timestamp, (int, float)) or isinstance(timestamp, bool): + raise TypeError("timestamp must be numeric") + if not math.isfinite(float(timestamp)): + raise ValueError("timestamp must be finite") + return float(timestamp) + + +def _sha256(value: str) -> str: + return hashlib.sha256(value.encode("utf-8")).hexdigest() diff --git a/tests/contracts/test_effect_adapter_contract.py b/tests/contracts/test_effect_adapter_contract.py new file mode 100644 index 0000000..a71d878 --- /dev/null +++ b/tests/contracts/test_effect_adapter_contract.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +import json +import sqlite3 +import subprocess +import sys +from pathlib import Path + +from hermes_workflows.effects import ( + EFFECT_ADAPTER_CONTRACT_VERSION, + EFFECT_ADAPTER_SERVICE_ID, + EffectCoordinator, + EffectPolicy, + SQLiteEffectStore, + operation_identity, + resolve_effect_adapter, +) +from hermes_workflows.runtime_services import RuntimeServicesV1 + + +class FileAdapter: + adapter_id = "contract.file.v1" + + def __init__(self, path: Path): + self.path = path + with sqlite3.connect(path) as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS external_effects ( + operation_id TEXT PRIMARY KEY, + input_hash TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS adapter_receipts ( + operation_id TEXT PRIMARY KEY, + adapter_receipt_id TEXT NOT NULL + ); + """ + ) + + def lookup_receipt(self, operation_id: str): + with sqlite3.connect(self.path) as conn: + row = conn.execute( + "SELECT adapter_receipt_id FROM adapter_receipts WHERE operation_id = ?", + (operation_id,), + ).fetchone() + if row is None: + return None + return {"adapter_receipt_id": row[0], "operation_id": operation_id} + + def perform(self, operation_id: str, input_value): + input_hash = operation_identity( + workflow_id="wf-contract-001", + effect_key="publish-report", + adapter_id=self.adapter_id, + input_value=input_value, + ).input_hash + receipt_id = f"file-{operation_id[-16:]}" + with sqlite3.connect(self.path) as conn: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + "INSERT OR IGNORE INTO external_effects(operation_id, input_hash) VALUES (?, ?)", + (operation_id, input_hash), + ) + conn.execute( + "INSERT OR IGNORE INTO adapter_receipts(operation_id, adapter_receipt_id) VALUES (?, ?)", + (operation_id, receipt_id), + ) + conn.commit() + return {"adapter_receipt_id": receipt_id, "operation_id": operation_id} + + +class Resolver: + def __init__(self, adapter): + self.adapter = adapter + + def resolve_adapter(self, adapter_id: str): + return self.adapter if adapter_id == self.adapter.adapter_id else None + + +def _run_adapter_worker(db_path: Path, operation_id: str, input_json: str) -> dict: + command = [ + sys.executable, + str(Path(__file__).resolve()), + "--adapter-worker", + str(db_path), + operation_id, + input_json, + ] + completed = subprocess.run(command, check=True, capture_output=True, text=True) + return json.loads(completed.stdout) + + +def test_file_backed_adapter_subprocess_is_idempotent_and_lookupable(tmp_path): + fixture = json.loads( + (Path(__file__).parents[1] / "fixtures" / "effect_contract_v1.json").read_text() + ) + vector = fixture["identity_vectors"][0] + identity = operation_identity( + workflow_id=vector["workflow_id"], + effect_key=vector["effect_key"], + adapter_id=vector["adapter_id"], + input_value=vector["input"], + ) + assert identity.operation_id == vector["operation_id"] + assert identity.input_hash == vector["input_hash"] + adapter_db = tmp_path / "adapter.sqlite" + input_json = json.dumps(vector["input"], sort_keys=True, separators=(",", ":")) + + first = _run_adapter_worker(adapter_db, identity.operation_id, input_json) + second = _run_adapter_worker(adapter_db, identity.operation_id, input_json) + + assert first["adapter_receipt_id"] == second["adapter_receipt_id"] + assert first["external_effect_count"] == second["external_effect_count"] == 1 + assert FileAdapter(adapter_db).lookup_receipt(identity.operation_id) == { + "adapter_receipt_id": first["adapter_receipt_id"], + "operation_id": identity.operation_id, + } + + +def test_adapter_resolves_only_through_generic_runtime_service_lookup(tmp_path): + adapter = FileAdapter(tmp_path / "adapter.sqlite") + resolver = Resolver(adapter) + services = RuntimeServicesV1(services={EFFECT_ADAPTER_SERVICE_ID: resolver}) + + assert EFFECT_ADAPTER_CONTRACT_VERSION == 1 + assert resolve_effect_adapter(services, adapter.adapter_id) is adapter + + +def test_coordinator_queries_receipt_before_perform(tmp_path): + adapter = FileAdapter(tmp_path / "adapter.sqlite") + coordinator = EffectCoordinator(SQLiteEffectStore(tmp_path / "effects.sqlite")) + input_value = {"report_id": "r-7", "sections": ["summary", "risks"]} + record = coordinator.prepare( + workflow_id="wf-contract-001", + effect_key="publish-report", + adapter_id=adapter.adapter_id, + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + adapter.perform(record.identity.operation_id, input_value) + claim = coordinator.store.claim(record.identity.operation_id) + completed = coordinator.execute_claimed(record, claim, adapter, input_value) + + assert completed.state == "completed" + with sqlite3.connect(adapter.path) as conn: + assert conn.execute("SELECT COUNT(*) FROM external_effects").fetchone()[0] == 1 + + +def _adapter_worker(args: list[str]) -> int: + db_path = Path(args[0]) + operation_id = args[1] + input_value = json.loads(args[2]) + receipt = FileAdapter(db_path).perform(operation_id, input_value) + with sqlite3.connect(db_path) as conn: + count = conn.execute("SELECT COUNT(*) FROM external_effects").fetchone()[0] + print(json.dumps({**receipt, "external_effect_count": count}, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + if len(sys.argv) >= 2 and sys.argv[1] == "--adapter-worker": + raise SystemExit(_adapter_worker(sys.argv[2:])) + raise SystemExit("this module is an adapter worker only when invoked with --adapter-worker") diff --git a/tests/fixtures/effect_contract_v1.json b/tests/fixtures/effect_contract_v1.json new file mode 100644 index 0000000..139b071 --- /dev/null +++ b/tests/fixtures/effect_contract_v1.json @@ -0,0 +1,28 @@ +{ + "schema_version": 1, + "identity_vectors": [ + { + "workflow_id": "wf-contract-001", + "effect_key": "publish-report", + "adapter_id": "contract.file.v1", + "input": {"report_id": "r-7", "sections": ["summary", "risks"]}, + "input_hash": "0443d565a2434c98e2d2034c8bdc5621cc77e23864d75088f7b6e29e3d8ed928", + "operation_id": "op_779cfb267aaf4485e131af3c56d0366053f28814563e018beefab5b9627eaf95" + } + ], + "policies": ["pure", "idempotent", "unsafe", "unclassified"], + "claim_states": ["pending", "claimed", "completed", "failed"], + "adapter_contract": { + "service_id": "effects.adapters", + "contract_version": 1, + "lookup_before_perform": true, + "receipt_projection_excludes_payload": true, + "delivery_semantics": "at-least-once" + }, + "crash_windows": [ + "before_adapter_call", + "during_adapter_call", + "after_effect_before_adapter_receipt", + "after_receipt_commit_before_command_completion" + ] +} diff --git a/tests/probes/kill_after_effect.py b/tests/probes/kill_after_effect.py new file mode 100644 index 0000000..1dd8844 --- /dev/null +++ b/tests/probes/kill_after_effect.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import argparse +import hashlib +import json +import os +import sqlite3 +import subprocess +import sys +import tempfile +import time +from pathlib import Path + + +REPO_ROOT = Path(__file__).resolve().parents[2] +SRC_ROOT = REPO_ROOT / "src" +if str(SRC_ROOT) not in sys.path: + sys.path.insert(0, str(SRC_ROOT)) + +from hermes_workflows.effects import ( # noqa: E402 + EffectCoordinator, + EffectPolicy, + SQLiteEffectStore, + operation_identity, +) + + +INPUT_VALUE = {"message": "crash-safe"} +WORKFLOW_ID = "wf-crash-probe" +EFFECT_KEY = "emit" +ADAPTER_ID = "probe.file.v1" +CRASH_WINDOWS = ( + "before_adapter_call", + "during_adapter_call", + "after_effect_before_adapter_receipt", + "after_receipt_commit_before_command_completion", +) + + +class CrashProbeAdapter: + adapter_id = ADAPTER_ID + + def __init__(self, path: Path, crash_window: str | None): + self.path = path + self.crash_window = crash_window + with sqlite3.connect(path) as conn: + conn.executescript( + """ + CREATE TABLE IF NOT EXISTS external_effects ( + operation_id TEXT PRIMARY KEY, + input_hash TEXT NOT NULL + ); + CREATE TABLE IF NOT EXISTS adapter_receipts ( + operation_id TEXT PRIMARY KEY, + adapter_receipt_id TEXT NOT NULL + ); + """ + ) + + def lookup_receipt(self, operation_id: str): + with sqlite3.connect(self.path) as conn: + external = conn.execute( + "SELECT input_hash FROM external_effects WHERE operation_id = ?", (operation_id,) + ).fetchone() + if external is None: + return None + receipt_id = f"probe-{operation_id[-16:]}" + conn.execute( + "INSERT OR IGNORE INTO adapter_receipts(operation_id, adapter_receipt_id) VALUES (?, ?)", + (operation_id, receipt_id), + ) + conn.commit() + return {"adapter_receipt_id": receipt_id, "operation_id": operation_id} + + def perform(self, operation_id: str, input_value): + if self.crash_window == "during_adapter_call": + os._exit(86) + input_hash = hashlib.sha256( + json.dumps(input_value, sort_keys=True, separators=(",", ":")).encode() + ).hexdigest() + receipt_id = f"probe-{operation_id[-16:]}" + with sqlite3.connect(self.path) as conn: + conn.execute("BEGIN IMMEDIATE") + conn.execute( + "INSERT OR IGNORE INTO external_effects(operation_id, input_hash) VALUES (?, ?)", + (operation_id, input_hash), + ) + conn.commit() + if self.crash_window == "after_effect_before_adapter_receipt": + os._exit(86) + with sqlite3.connect(self.path) as conn: + conn.execute( + "INSERT OR IGNORE INTO adapter_receipts(operation_id, adapter_receipt_id) VALUES (?, ?)", + (operation_id, receipt_id), + ) + conn.commit() + return {"adapter_receipt_id": receipt_id, "operation_id": operation_id} + + +def worker(effect_db: Path, adapter_db: Path, crash_window: str | None) -> int: + store = SQLiteEffectStore(effect_db) + identity = operation_identity( + workflow_id=WORKFLOW_ID, + effect_key=EFFECT_KEY, + adapter_id=ADAPTER_ID, + input_value=INPUT_VALUE, + ) + record = store.get(identity.operation_id) + if record.state == "completed": + return 0 + claim = store.claim(identity.operation_id, ttl_seconds=0.05) + if crash_window == "before_adapter_call": + os._exit(86) + completed = EffectCoordinator(store).execute_claimed( + record, + claim, + CrashProbeAdapter(adapter_db, crash_window), + INPUT_VALUE, + sensitive_receipt=True, + ) + if completed.state != "completed": + raise RuntimeError("effect did not complete") + if crash_window == "after_receipt_commit_before_command_completion": + os._exit(86) + return 0 + + +def run_scenario(root: Path, crash_window: str) -> dict[str, object]: + scenario = root / crash_window + scenario.mkdir(parents=True) + effect_db = scenario / "effects.sqlite" + adapter_db = scenario / "adapter.sqlite" + store = SQLiteEffectStore(effect_db) + identity = operation_identity( + workflow_id=WORKFLOW_ID, + effect_key=EFFECT_KEY, + adapter_id=ADAPTER_ID, + input_value=INPUT_VALUE, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, INPUT_VALUE) + base = [ + sys.executable, + str(Path(__file__).resolve()), + "--worker", + str(effect_db), + str(adapter_db), + ] + first = subprocess.run([*base, "--crash-window", crash_window], check=False) + if first.returncode != 86: + raise RuntimeError(f"first worker did not crash at {crash_window}: {first.returncode}") + time.sleep(0.08) + second = subprocess.run(base, check=False) + if second.returncode != 0: + raise RuntimeError(f"recovery worker failed at {crash_window}: {second.returncode}") + + record = store.get(identity.operation_id) + with sqlite3.connect(adapter_db) as conn: + external_count = conn.execute("SELECT COUNT(*) FROM external_effects").fetchone()[0] + adapter_receipt_count = conn.execute("SELECT COUNT(*) FROM adapter_receipts").fetchone()[0] + with sqlite3.connect(effect_db) as conn: + local_receipt_count = conn.execute("SELECT COUNT(*) FROM effect_receipts").fetchone()[0] + fencing_row = conn.execute( + "SELECT state, attempts, claim_token FROM effect_intents WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone() + result = { + "window": crash_window, + "operation_id": identity.operation_id, + "input_hash": identity.input_hash, + "attempts": 2, + "claim_attempts": record.attempts, + "external_effect_count": external_count, + "adapter_receipt_count": adapter_receipt_count, + "completed_receipt_count": local_receipt_count, + "state": record.state, + "fencing_row": [ + fencing_row[0], + fencing_row[1], + hashlib.sha256(fencing_row[2].encode()).hexdigest(), + ], + } + if external_count != 1 or adapter_receipt_count != 1 or local_receipt_count != 1: + raise RuntimeError(f"duplicate or missing effect receipt: {result}") + if record.state != "completed" or result["attempts"] < 2: + raise RuntimeError(f"recovery did not converge: {result}") + return result + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--worker", action="store_true") + parser.add_argument("effect_db", nargs="?") + parser.add_argument("adapter_db", nargs="?") + parser.add_argument("--crash-window", choices=CRASH_WINDOWS) + args = parser.parse_args(argv) + if args.worker: + if args.effect_db is None or args.adapter_db is None: + parser.error("worker requires effect_db and adapter_db") + return worker(Path(args.effect_db), Path(args.adapter_db), args.crash_window) + + with tempfile.TemporaryDirectory(prefix="hw01-crash-") as temporary: + results = [run_scenario(Path(temporary), window) for window in CRASH_WINDOWS] + print(json.dumps({"schema_version": 1, "scenarios": results}, indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py new file mode 100644 index 0000000..6fafa4e --- /dev/null +++ b/tests/test_effect_idempotency.py @@ -0,0 +1,257 @@ +from __future__ import annotations + +import json +import sqlite3 +import threading +from concurrent.futures import ThreadPoolExecutor + +import pytest + + +def test_stable_operation_id_is_attempt_independent_and_input_sensitive(): + from hermes_workflows.effects import operation_identity + + first = operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"b": [2, 3], "a": 1}, + attempt=1, + ) + replay = operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"a": 1, "b": [2, 3]}, + attempt=99, + ) + changed = operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"a": 2, "b": [2, 3]}, + attempt=1, + ) + + assert first.operation_id == replay.operation_id + assert first.input_hash == replay.input_hash + assert first.operation_id != changed.operation_id + assert first.input_hash != changed.input_hash + assert first.operation_id.startswith("op_") + assert len(first.input_hash) == 64 + + +def test_noncanonical_or_secret_bearing_inputs_are_rejected(): + from hermes_workflows.effects import operation_identity + + with pytest.raises(TypeError, match="JSON"): + operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"not-json": object()}, + ) + with pytest.raises(ValueError, match="finite"): + operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"bad": float("nan")}, + ) + + +def test_sqlite_intent_claim_completion_and_stale_token_fencing(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-cas", + effect_key="send", + adapter_id="test.file.v1", + input_value={"message": "hello"}, + ) + intent = store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"message": "hello"}) + assert intent.state == "pending" + + claim_a = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0, token="token-a") + assert claim_a.attempt == 1 + with pytest.raises(RuntimeError, match="not claimable"): + store.claim(identity.operation_id, now=10.5, ttl_seconds=1.0, token="token-b") + + claim_b = store.claim(identity.operation_id, now=11.1, ttl_seconds=5.0, token="token-b") + assert claim_b.attempt == 2 + with pytest.raises(RuntimeError, match="stale claim token"): + store.complete(identity.operation_id, "token-a", {"provider_id": "wrong"}, now=12.0) + + completed = store.complete( + identity.operation_id, + "token-b", + {"provider_id": "receipt-1", "access_token": "must-not-project"}, + sensitive=True, + now=12.0, + ) + assert completed.state == "completed" + assert completed.receipt is not None + assert completed.receipt.payload["provider_id"] == "receipt-1" + assert completed.receipt.descriptor()["sensitive"] is True + assert "access_token" not in json.dumps(completed.receipt.descriptor()) + + with sqlite3.connect(tmp_path / "effects.sqlite") as conn: + row = conn.execute( + "SELECT state, attempts, claim_token FROM effect_intents WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone() + receipt_row = conn.execute( + "SELECT claim_token, receipt_hash FROM effect_receipts WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone() + assert row == ("completed", 2, "token-b") + assert receipt_row is not None + assert receipt_row[0] == "token-b" + assert len(receipt_row[1]) == 64 + + +def test_failure_is_fenced_and_terminal(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-fail", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) + claim = store.claim(identity.operation_id, token="winner") + with pytest.raises(RuntimeError, match="stale claim token"): + store.fail(identity.operation_id, "loser", {"kind": "wrong-owner"}) + failed = store.fail(identity.operation_id, claim.token, {"kind": "adapter-error"}) + assert failed.state == "failed" + assert failed.error == {"kind": "adapter-error"} + with pytest.raises(RuntimeError, match="not claimable"): + store.claim(identity.operation_id) + + +def test_expired_claim_cannot_complete_without_reclaim(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-expired", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) + claim = store.claim(identity.operation_id, token="expired", now=10.0, ttl_seconds=1.0) + + with pytest.raises(RuntimeError, match="stale claim token"): + store.complete(identity.operation_id, claim.token, {"receipt": 1}, now=11.0) + + replacement = store.claim(identity.operation_id, token="current", now=11.0) + completed = store.complete( + identity.operation_id, replacement.token, {"receipt": 1}, now=12.0 + ) + assert completed.state == "completed" + + +def test_concurrent_claim_race_has_one_winner(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-race", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) + barrier = threading.Barrier(2) + + def compete(token): + barrier.wait() + try: + return store.claim(identity.operation_id, token=token, now=10.0) + except RuntimeError: + return None + + with ThreadPoolExecutor(max_workers=2) as pool: + claims = list(pool.map(compete, ("racer-a", "racer-b"))) + + winners = [claim for claim in claims if claim is not None] + assert len(winners) == 1 + record = store.get(identity.operation_id) + assert record.state == "claimed" + assert record.attempts == 1 + assert record.claim_token == winners[0].token + + +def test_intent_identity_conflicts_fail_closed(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-conflict", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}) + with pytest.raises(ValueError, match="conflicts with durable intent"): + store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) + + +def test_unclassified_and_legacy_effects_refuse_execution(tmp_path): + from hermes_workflows.effects import ( + EffectCoordinator, + EffectPolicy, + SQLiteEffectStore, + operation_identity, + ) + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + with pytest.raises(ValueError, match="unclassified"): + coordinator.prepare( + workflow_id="wf-legacy", + effect_key="legacy-call", + adapter_id="legacy.adapter.v0", + input_value={"value": 1}, + policy=EffectPolicy.UNCLASSIFIED, + ) + with pytest.raises(ValueError, match="unsafe"): + coordinator.prepare( + workflow_id="wf-unsafe", + effect_key="charge-card", + adapter_id="payments.v1", + input_value={"amount": 100}, + policy=EffectPolicy.UNSAFE, + ) + + legacy_identity = operation_identity( + workflow_id="wf-legacy-row", + effect_key="old-step", + adapter_id="legacy.adapter.v0", + input_value={"value": 1}, + ) + store.ensure_intent(legacy_identity, EffectPolicy.UNCLASSIFIED, {"value": 1}) + with pytest.raises(RuntimeError, match="not claimable"): + store.claim(legacy_identity.operation_id) + + +def test_unsafe_effect_can_be_claimed_once_but_not_reclaimed_automatically(tmp_path): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + record = EffectCoordinator(store).prepare( + workflow_id="wf-unsafe-once", + effect_key="charge-card", + adapter_id="payments.v1", + input_value={"amount": 100}, + policy=EffectPolicy.UNSAFE, + allow_unsafe=True, + ) + store.claim(record.identity.operation_id, token="only-attempt", now=10.0, ttl_seconds=1.0) + + with pytest.raises(RuntimeError, match="not claimable"): + store.claim(record.identity.operation_id, token="forbidden-retry", now=11.0) From 782786b2f2b8120bdf40fdde5be02caed0a59079 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 00:51:23 -0700 Subject: [PATCH 02/11] fix(effects): enforce operation identity integrity --- docs/contracts/effects.md | 4 +-- src/hermes_workflows/effects.py | 30 ++++++++++++++-- tests/test_effect_idempotency.py | 62 ++++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+), 4 deletions(-) diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md index f6684b8..c953de5 100644 --- a/docs/contracts/effects.md +++ b/docs/contracts/effects.md @@ -1,6 +1,6 @@ # Durable effects contract v1 -Hermes Workflows identifies a logical external operation independently of an execution attempt. The operation ID is the SHA-256 of canonical JSON containing the workflow instance ID, logical effect key, adapter ID, and canonical input hash. Attempt numbers, claim tokens, worker IDs, and clocks are deliberately excluded. +Hermes Workflows identifies a logical external operation independently of an execution attempt. The operation ID is the SHA-256 of canonical JSON containing the workflow instance ID, logical effect key, adapter ID, and canonical input hash. Attempt numbers, claim tokens, worker IDs, and clocks are deliberately excluded. JSON object keys must be unique; mapping item streams with repeated keys are rejected rather than silently collapsed during normalization. ## Delivery semantics @@ -17,7 +17,7 @@ No policy is inferred from a function name, tool name, or implementation. ## Durable records and fencing -`effect_intents` stores canonical input, input hash, policy, state, attempt count, and the active claim token. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. +`effect_intents` stores canonical input, input hash, policy, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. `effect_receipts` stores the adapter receipt, its SHA-256, sensitivity flag, completion time, and the claim token that fenced the write. Intent completion and receipt insertion are one SQLite transaction. diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index bf16eaa..4daba2c 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -119,10 +119,21 @@ def operation_identity( ) -> OperationIdentity: """Build an attempt-independent identity from logical effect coordinates and input.""" del attempt + input_json = canonical_json(input_value) + return _operation_identity_from_input_json( + workflow_id=workflow_id, + effect_key=effect_key, + adapter_id=adapter_id, + input_json=input_json, + ) + + +def _operation_identity_from_input_json( + *, workflow_id: str, effect_key: str, adapter_id: str, input_json: str +) -> OperationIdentity: _validate_id(workflow_id, "workflow_id") _validate_id(effect_key, "effect_key") _validate_id(adapter_id, "adapter_id") - input_json = canonical_json(input_value) input_hash = _sha256(input_json) operation_document = canonical_json( { @@ -181,8 +192,21 @@ def ensure_intent( ) -> EffectRecord: policy_value = _coerce_policy(policy) input_json = canonical_json(input_value) - if _sha256(input_json) != identity.input_hash: + expected_identity = _operation_identity_from_input_json( + workflow_id=identity.workflow_id, + effect_key=identity.effect_key, + adapter_id=identity.adapter_id, + input_json=input_json, + ) + if identity.input_hash != expected_identity.input_hash: raise ValueError("input conflicts with operation identity") + if ( + identity.operation_id != expected_identity.operation_id + or identity.workflow_id != expected_identity.workflow_id + or identity.effect_key != expected_identity.effect_key + or identity.adapter_id != expected_identity.adapter_id + ): + raise ValueError("operation identity mismatch") timestamp = _timestamp(now) with self._connect() as conn: conn.execute("BEGIN IMMEDIATE") @@ -538,6 +562,8 @@ def _normalize_json(value: Any) -> JsonValue: for key, item in value.items(): if not isinstance(key, str): raise TypeError("JSON object keys must be strings") + if key in result: + raise ValueError(f"duplicate JSON object key: {key}") result[key] = _normalize_json(item) return result raise TypeError(f"value is not canonical JSON: {type(value).__name__}") diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index 6fafa4e..f3b3bb8 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -3,11 +3,32 @@ import json import sqlite3 import threading +from collections.abc import Iterator, Mapping from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from typing import Any import pytest +class _RepeatedItemsMapping(Mapping[str, Any]): + """Adversarial Mapping whose item stream is not a JSON object.""" + + def __getitem__(self, key: str) -> Any: + if key == "value": + return 2 + raise KeyError(key) + + def __iter__(self) -> Iterator[str]: + yield "value" + + def __len__(self) -> int: + return 1 + + def items(self) -> Any: + return (("value", 1), ("value", 2)) + + def test_stable_operation_id_is_attempt_independent_and_input_sensitive(): from hermes_workflows.effects import operation_identity @@ -60,6 +81,27 @@ def test_noncanonical_or_secret_bearing_inputs_are_rejected(): ) +def test_custom_mapping_with_repeated_items_cannot_collapse_into_json_object(): + from hermes_workflows.effects import operation_identity + + ordinary = operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value={"value": 2}, + ) + + with pytest.raises(ValueError, match="duplicate JSON object key"): + operation_identity( + workflow_id="wf-123", + effect_key="publish-report", + adapter_id="test.file.v1", + input_value=_RepeatedItemsMapping(), + ) + + assert ordinary.operation_id.startswith("op_") + + def test_sqlite_intent_claim_completion_and_stale_token_fencing(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity @@ -201,6 +243,26 @@ def test_intent_identity_conflicts_fail_closed(tmp_path): store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) +def test_intent_rejects_forged_hash_shaped_operation_id_before_persistence(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-forged", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + forged = replace(identity, operation_id="op_" + "0" * 64) + + with pytest.raises(ValueError, match="operation identity mismatch"): + store.ensure_intent(forged, EffectPolicy.IDEMPOTENT, {"value": 1}) + + with sqlite3.connect(tmp_path / "effects.sqlite") as conn: + count = conn.execute("SELECT COUNT(*) FROM effect_intents").fetchone()[0] + assert count == 0 + + def test_unclassified_and_legacy_effects_refuse_execution(tmp_path): from hermes_workflows.effects import ( EffectCoordinator, From fc0096772f95dd57fdd2bd75e81302288d5924a5 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 01:14:31 -0700 Subject: [PATCH 03/11] fix(effects): reject conflicting receipt operation identity --- docs/contracts/effects.md | 2 +- src/hermes_workflows/effects.py | 5 +++ .../contracts/test_effect_adapter_contract.py | 33 +++++++++++++++ tests/test_effect_idempotency.py | 41 +++++++++++++++++++ 4 files changed, 80 insertions(+), 1 deletion(-) diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md index c953de5..238cb73 100644 --- a/docs/contracts/effects.md +++ b/docs/contracts/effects.md @@ -30,7 +30,7 @@ Adapters are obtained through the FND-RT process-local service registry at servi - `lookup_receipt(operation_id)` - `perform(operation_id, canonical_input)` -The coordinator always queries the adapter receipt before performing the operation. This closes the recovery window where the external system accepted the operation but the local receipt transaction did not commit, provided the external adapter can look up the stable operation ID. +The coordinator always queries the adapter receipt before performing the operation. This closes the recovery window where the external system accepted the operation but the local receipt transaction did not commit, provided the external adapter can look up the stable operation ID. A receipt may omit `operation_id`; if it includes one, it must exactly match the requested operation or the coordinator rejects it before durable completion. ## Crash windows diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index 4daba2c..1923007 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -492,6 +492,11 @@ def execute_claimed( payload = existing if existing is not None else adapter.perform(record.identity.operation_id, input_value) if not isinstance(payload, Mapping): raise TypeError("effect adapter receipt must be a mapping") + if ( + "operation_id" in payload + and payload["operation_id"] != record.identity.operation_id + ): + raise ValueError("effect adapter receipt operation_id mismatch") adapter_receipt_id = payload.get("adapter_receipt_id") return self.store.complete( record.identity.operation_id, diff --git a/tests/contracts/test_effect_adapter_contract.py b/tests/contracts/test_effect_adapter_contract.py index a71d878..441906d 100644 --- a/tests/contracts/test_effect_adapter_contract.py +++ b/tests/contracts/test_effect_adapter_contract.py @@ -6,6 +6,8 @@ import sys from pathlib import Path +import pytest + from hermes_workflows.effects import ( EFFECT_ADAPTER_CONTRACT_VERSION, EFFECT_ADAPTER_SERVICE_ID, @@ -146,6 +148,37 @@ def test_coordinator_queries_receipt_before_perform(tmp_path): assert conn.execute("SELECT COUNT(*) FROM external_effects").fetchone()[0] == 1 +@pytest.mark.parametrize("source", ["lookup", "perform"]) +def test_adapter_receipt_operation_id_must_match_requested_operation(tmp_path, source): + class ConflictingFileAdapter(FileAdapter): + def lookup_receipt(self, operation_id: str): + if source == "lookup": + return {"adapter_receipt_id": "wrong", "operation_id": "op_" + "0" * 64} + return None + + def perform(self, operation_id: str, input_value): + return {"adapter_receipt_id": "wrong", "operation_id": "op_" + "0" * 64} + + adapter = ConflictingFileAdapter(tmp_path / "adapter.sqlite") + coordinator = EffectCoordinator(SQLiteEffectStore(tmp_path / "effects.sqlite")) + input_value = {"report_id": "r-conflict"} + record = coordinator.prepare( + workflow_id="wf-contract-001", + effect_key="publish-report", + adapter_id=adapter.adapter_id, + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + claim = coordinator.store.claim(record.identity.operation_id) + + with pytest.raises(ValueError, match="receipt operation_id mismatch"): + coordinator.execute_claimed(record, claim, adapter, input_value) + + rejected = coordinator.store.get(record.identity.operation_id) + assert rejected.state == "claimed" + assert rejected.receipt is None + + def _adapter_worker(args: list[str]) -> int: db_path = Path(args[0]) operation_id = args[1] diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index f3b3bb8..c49eee3 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -29,6 +29,21 @@ def items(self) -> Any: return (("value", 1), ("value", 2)) +class _ConflictingReceiptAdapter: + adapter_id = "test.file.v1" + + def __init__(self, source: str): + self.source = source + + def lookup_receipt(self, operation_id: str): + if self.source == "lookup": + return {"operation_id": "op_" + "0" * 64, "adapter_receipt_id": "wrong"} + return None + + def perform(self, operation_id: str, input_value: Any): + return {"operation_id": "op_" + "0" * 64, "adapter_receipt_id": "wrong"} + + def test_stable_operation_id_is_attempt_independent_and_input_sensitive(): from hermes_workflows.effects import operation_identity @@ -197,6 +212,32 @@ def test_expired_claim_cannot_complete_without_reclaim(tmp_path): assert completed.state == "completed" +@pytest.mark.parametrize("source", ["lookup", "perform"]) +def test_coordinator_rejects_conflicting_adapter_receipt_operation_id(tmp_path, source): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + input_value = {"value": 1} + record = coordinator.prepare( + workflow_id="wf-receipt-conflict", + effect_key="write", + adapter_id="test.file.v1", + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + claim = store.claim(record.identity.operation_id, token=f"{source}-claim") + + with pytest.raises(ValueError, match="receipt operation_id mismatch"): + coordinator.execute_claimed( + record, claim, _ConflictingReceiptAdapter(source), input_value + ) + + rejected = store.get(record.identity.operation_id) + assert rejected.state == "claimed" + assert rejected.receipt is None + + def test_concurrent_claim_race_has_one_winner(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity From e7f25e28c770beb39a89728338cd9dd0f7288133 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 01:40:24 -0700 Subject: [PATCH 04/11] fix(effects): reject malformed terminal mappings --- src/hermes_workflows/effects.py | 4 +-- tests/test_effect_idempotency.py | 54 ++++++++++++++++++++++++++++++++ 2 files changed, 56 insertions(+), 2 deletions(-) diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index 1923007..ba05c54 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -305,7 +305,7 @@ def complete( now: float | None = None, ) -> EffectRecord: _validate_operation_id(operation_id) - payload_json = canonical_json(dict(receipt_payload)) + payload_json = canonical_json(receipt_payload) timestamp = _timestamp(now) receipt_hash = _sha256(payload_json) receipt_id = adapter_receipt_id or receipt_hash @@ -353,7 +353,7 @@ def fail( now: float | None = None, ) -> EffectRecord: _validate_operation_id(operation_id) - error_json = canonical_json(dict(error)) + error_json = canonical_json(error) timestamp = _timestamp(now) with self._connect() as conn: conn.execute("BEGIN IMMEDIATE") diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index c49eee3..6cccced 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -189,6 +189,60 @@ def test_failure_is_fenced_and_terminal(tmp_path): store.claim(identity.operation_id) +def test_completion_rejects_repeated_mapping_items_without_terminal_state(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-receipt-shape", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}) + claim = store.claim(identity.operation_id, token="receipt-owner") + + with pytest.raises(ValueError, match="duplicate JSON object key"): + store.complete(identity.operation_id, claim.token, _RepeatedItemsMapping()) + + rejected = store.get(identity.operation_id) + assert rejected.state == "claimed" + assert rejected.receipt is None + with sqlite3.connect(tmp_path / "effects.sqlite") as conn: + receipt_count = conn.execute( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone()[0] + assert receipt_count == 0 + + +def test_failure_rejects_repeated_mapping_items_without_terminal_state(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-error-shape", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) + claim = store.claim(identity.operation_id, token="error-owner") + + with pytest.raises(ValueError, match="duplicate JSON object key"): + store.fail(identity.operation_id, claim.token, _RepeatedItemsMapping()) + + rejected = store.get(identity.operation_id) + assert rejected.state == "claimed" + assert rejected.error is None + with sqlite3.connect(tmp_path / "effects.sqlite") as conn: + state, error_json = conn.execute( + "SELECT state, error_json FROM effect_intents WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone() + assert (state, error_json) == ("claimed", None) + + def test_expired_claim_cannot_complete_without_reclaim(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity From 4e65e72bbc9d4e3efb748d3647cc6add74b6a31a Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 02:21:05 -0700 Subject: [PATCH 05/11] fix(effects): bind claims to operation identity --- src/hermes_workflows/effects.py | 2 ++ tests/test_effect_idempotency.py | 40 ++++++++++++++++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index ba05c54..3d703d3 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -484,6 +484,8 @@ def execute_claimed( *, sensitive_receipt: bool = False, ) -> EffectRecord: + if claim.operation_id != record.identity.operation_id: + raise ValueError("effect claim operation_id mismatch") if record.identity.adapter_id != adapter.adapter_id: raise ValueError("effect adapter identity mismatch") if _sha256(canonical_json(input_value)) != record.identity.input_hash: diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index 6cccced..2ca7dc4 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -292,6 +292,46 @@ def test_coordinator_rejects_conflicting_adapter_receipt_operation_id(tmp_path, assert rejected.receipt is None +def test_coordinator_rejects_claim_for_another_operation_before_adapter_call(tmp_path): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + class RecordingAdapter: + adapter_id = "test.file.v1" + + def __init__(self): + self.calls = [] + + def lookup_receipt(self, operation_id: str): + self.calls.append(("lookup", operation_id)) + return None + + def perform(self, operation_id: str, input_value: Any): + self.calls.append(("perform", operation_id)) + return {"operation_id": operation_id, "adapter_receipt_id": "unexpected"} + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + input_value = {"value": 1} + record = coordinator.prepare( + workflow_id="wf-claim-conflict", + effect_key="write", + adapter_id="test.file.v1", + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + valid_claim = store.claim(record.identity.operation_id, token="valid-token") + conflicting_claim = replace(valid_claim, operation_id="op_" + "0" * 64) + adapter = RecordingAdapter() + + with pytest.raises(ValueError, match="claim operation_id mismatch"): + coordinator.execute_claimed(record, conflicting_claim, adapter, input_value) + + assert adapter.calls == [] + rejected = store.get(record.identity.operation_id) + assert rejected.state == "claimed" + assert rejected.receipt is None + + def test_concurrent_claim_race_has_one_winner(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity From 79d0e5677b09859be70632c6003c96fcddbb671f Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 02:49:00 -0700 Subject: [PATCH 06/11] fix(effects): fence adapter calls by durable claim --- src/hermes_workflows/effects.py | 42 +++++++++++++++++++++------ tests/test_effect_idempotency.py | 50 ++++++++++++++++++++++++++++++++ 2 files changed, 83 insertions(+), 9 deletions(-) diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index 3d703d3..b866a8e 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -387,6 +387,29 @@ def get(self, operation_id: str) -> EffectRecord: def lookup_receipt(self, operation_id: str) -> EffectReceipt | None: return self.get(operation_id).receipt + def require_active_claim( + self, + operation_id: str, + claim_token: str, + *, + now: float | None = None, + ) -> EffectRecord: + """Return the durable intent only when the token currently owns its live claim.""" + _validate_operation_id(operation_id) + timestamp = _timestamp(now) + with self._connect() as conn: + row = conn.execute( + """ + SELECT * FROM effect_intents + WHERE operation_id = ? AND state = 'claimed' AND claim_token = ? + AND claim_expires_at > ? + """, + (operation_id, claim_token, timestamp), + ).fetchone() + if row is None: + raise RuntimeError("stale claim token cannot execute effect") + return _record_from_rows(row, None) + def _initialize(self) -> None: with self._connect() as conn: conn.executescript( @@ -486,22 +509,23 @@ def execute_claimed( ) -> EffectRecord: if claim.operation_id != record.identity.operation_id: raise ValueError("effect claim operation_id mismatch") - if record.identity.adapter_id != adapter.adapter_id: + durable_record = self.store.require_active_claim(claim.operation_id, claim.token) + if durable_record.identity != record.identity: + raise ValueError("effect record conflicts with durable intent") + if durable_record.identity.adapter_id != adapter.adapter_id: raise ValueError("effect adapter identity mismatch") - if _sha256(canonical_json(input_value)) != record.identity.input_hash: + if _sha256(canonical_json(input_value)) != durable_record.identity.input_hash: raise ValueError("effect input conflicts with durable intent") - existing = adapter.lookup_receipt(record.identity.operation_id) - payload = existing if existing is not None else adapter.perform(record.identity.operation_id, input_value) + operation_id = durable_record.identity.operation_id + existing = adapter.lookup_receipt(operation_id) + payload = existing if existing is not None else adapter.perform(operation_id, input_value) if not isinstance(payload, Mapping): raise TypeError("effect adapter receipt must be a mapping") - if ( - "operation_id" in payload - and payload["operation_id"] != record.identity.operation_id - ): + if "operation_id" in payload and payload["operation_id"] != operation_id: raise ValueError("effect adapter receipt operation_id mismatch") adapter_receipt_id = payload.get("adapter_receipt_id") return self.store.complete( - record.identity.operation_id, + operation_id, claim.token, payload, adapter_receipt_id=str(adapter_receipt_id) if adapter_receipt_id is not None else None, diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index 2ca7dc4..a1e3384 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -332,6 +332,56 @@ def perform(self, operation_id: str, input_value: Any): assert rejected.receipt is None +@pytest.mark.parametrize("claim_state", ["forged", "stale", "expired"]) +def test_coordinator_rejects_noncurrent_claim_before_adapter_call(tmp_path, claim_state): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + class RecordingAdapter: + adapter_id = "test.file.v1" + + def __init__(self): + self.calls = [] + + def lookup_receipt(self, operation_id: str): + self.calls.append(("lookup", operation_id)) + return None + + def perform(self, operation_id: str, input_value: Any): + self.calls.append(("perform", operation_id)) + return {"operation_id": operation_id, "adapter_receipt_id": "unexpected"} + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + input_value = {"value": 1} + record = coordinator.prepare( + workflow_id=f"wf-{claim_state}-claim", + effect_key="write", + adapter_id="test.file.v1", + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + claim_options = ( + {"now": 10.0, "ttl_seconds": 1.0} if claim_state in {"stale", "expired"} else {} + ) + claim = store.claim(record.identity.operation_id, token="original-token", **claim_options) + if claim_state == "forged": + rejected_claim = replace(claim, token="forged-token", expires_at=10_000.0) + elif claim_state == "stale": + store.claim(record.identity.operation_id, token="current-token", now=11.0) + rejected_claim = replace(claim, expires_at=10_000.0) + else: + rejected_claim = claim + adapter = RecordingAdapter() + + with pytest.raises(RuntimeError, match="stale claim token"): + coordinator.execute_claimed(record, rejected_claim, adapter, input_value) + + assert adapter.calls == [] + rejected = store.get(record.identity.operation_id) + assert rejected.state == "claimed" + assert rejected.receipt is None + + def test_concurrent_claim_race_has_one_winner(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity From 1f91be7c3742a539cd16aac810018e7783c7161c Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 03:12:59 -0700 Subject: [PATCH 07/11] fix(effects): revalidate claim before perform --- src/hermes_workflows/effects.py | 8 +++++- tests/test_effect_idempotency.py | 46 ++++++++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 1 deletion(-) diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index b866a8e..84de148 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -518,7 +518,13 @@ def execute_claimed( raise ValueError("effect input conflicts with durable intent") operation_id = durable_record.identity.operation_id existing = adapter.lookup_receipt(operation_id) - payload = existing if existing is not None else adapter.perform(operation_id, input_value) + if existing is not None: + payload = existing + else: + current_record = self.store.require_active_claim(operation_id, claim.token) + if current_record.identity != durable_record.identity: + raise ValueError("effect record conflicts with durable intent") + payload = adapter.perform(operation_id, input_value) if not isinstance(payload, Mapping): raise TypeError("effect adapter receipt must be a mapping") if "operation_id" in payload and payload["operation_id"] != operation_id: diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index a1e3384..fb5c1fc 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -382,6 +382,52 @@ def perform(self, operation_id: str, input_value: Any): assert rejected.receipt is None +def test_coordinator_revalidates_claim_after_receipt_lookup_before_perform(tmp_path): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + input_value = {"value": 1} + record = coordinator.prepare( + workflow_id="wf-lookup-claim-loss", + effect_key="write", + adapter_id="test.file.v1", + input_value=input_value, + policy=EffectPolicy.IDEMPOTENT, + ) + claim = store.claim(record.identity.operation_id, token="original-token") + + class ClaimReplacingAdapter: + adapter_id = "test.file.v1" + + def __init__(self): + self.calls = [] + + def lookup_receipt(self, operation_id: str): + self.calls.append(("lookup", operation_id)) + with sqlite3.connect(store.path) as conn: + conn.execute( + "UPDATE effect_intents SET claim_token = ? WHERE operation_id = ?", + ("replacement-token", operation_id), + ) + return None + + def perform(self, operation_id: str, input_value: Any): + self.calls.append(("perform", operation_id)) + return {"operation_id": operation_id, "adapter_receipt_id": "unexpected"} + + adapter = ClaimReplacingAdapter() + + with pytest.raises(RuntimeError, match="stale claim token"): + coordinator.execute_claimed(record, claim, adapter, input_value) + + assert adapter.calls == [("lookup", record.identity.operation_id)] + rejected = store.get(record.identity.operation_id) + assert rejected.state == "claimed" + assert rejected.claim_token == "replacement-token" + assert rejected.receipt is None + + def test_concurrent_claim_race_has_one_winner(tmp_path): from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity From a62fbd8c80a4d3bfc09933a3fcaa7eecb57292e7 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 03:39:21 -0700 Subject: [PATCH 08/11] fix(effects): enforce unsafe authorization at execution --- docs/contracts/effects.md | 4 +- src/hermes_workflows/effects.py | 29 +++++- .../contracts/test_effect_adapter_contract.py | 1 + tests/fixtures/effect_contract_v1.json | 1 + tests/test_effect_idempotency.py | 90 +++++++++++++++++++ 5 files changed, 119 insertions(+), 6 deletions(-) diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md index 238cb73..45fa041 100644 --- a/docs/contracts/effects.md +++ b/docs/contracts/effects.md @@ -10,14 +10,14 @@ Policies are explicit: - `pure`: no externally observable mutation. - `idempotent`: repeats with the same operation ID converge on the same external result. -- `unsafe`: execution is refused unless a caller explicitly authorizes it; a pending intent can be claimed once, but an expired uncertain claim cannot be reclaimed automatically and must go to human reconciliation. +- `unsafe`: execution is refused unless a caller explicitly authorizes it. That authorization is persisted with the intent and rechecked by the coordinator immediately before any adapter lookup or external `perform` call. A pending intent can be claimed once, but an expired uncertain claim cannot be reclaimed automatically and must go to human reconciliation. - `unclassified`: refused. Legacy effects are unclassified until an owner classifies and adapts them. No policy is inferred from a function name, tool name, or implementation. ## Durable records and fencing -`effect_intents` stores canonical input, input hash, policy, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. +`effect_intents` stores canonical input, input hash, policy, explicit unsafe-authorization evidence, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. Directly persisted unsafe intents without authorization remain inventoriable and claimable once, but the execution boundary refuses them before adapter lookup, external mutation, or durable completion. `effect_receipts` stores the adapter receipt, its SHA-256, sensitivity flag, completion time, and the claim token that fenced the write. Intent completion and receipt insertion are one SQLite transaction. diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index 84de148..7889fde 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -76,6 +76,7 @@ def descriptor(self) -> dict[str, Any]: class EffectRecord: identity: OperationIdentity policy: EffectPolicy + unsafe_authorized: bool state: str attempts: int claim_token: str | None @@ -188,9 +189,13 @@ def ensure_intent( policy: EffectPolicy | str, input_value: Any, *, + allow_unsafe: bool = False, now: float | None = None, ) -> EffectRecord: policy_value = _coerce_policy(policy) + if not isinstance(allow_unsafe, bool): + raise TypeError("allow_unsafe must be a boolean") + unsafe_authorized = policy_value is EffectPolicy.UNSAFE and allow_unsafe input_json = canonical_json(input_value) expected_identity = _operation_identity_from_input_json( workflow_id=identity.workflow_id, @@ -213,9 +218,9 @@ def ensure_intent( conn.execute( """ INSERT OR IGNORE INTO effect_intents ( - operation_id, workflow_id, effect_key, adapter_id, policy, + operation_id, workflow_id, effect_key, adapter_id, policy, unsafe_authorized, input_hash, input_json, state, attempts, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', 0, ?, ?) """, ( identity.operation_id, @@ -223,6 +228,7 @@ def ensure_intent( identity.effect_key, identity.adapter_id, policy_value.value, + int(unsafe_authorized), identity.input_hash, input_json, timestamp, @@ -235,6 +241,7 @@ def ensure_intent( identity.effect_key, identity.adapter_id, policy_value.value, + int(unsafe_authorized), identity.input_hash, input_json, ) @@ -420,6 +427,7 @@ def _initialize(self) -> None: effect_key TEXT NOT NULL, adapter_id TEXT NOT NULL, policy TEXT NOT NULL CHECK (policy IN ('pure','idempotent','unsafe','unclassified')), + unsafe_authorized INTEGER NOT NULL DEFAULT 0 CHECK (unsafe_authorized IN (0,1)), input_hash TEXT NOT NULL, input_json TEXT NOT NULL, state TEXT NOT NULL CHECK (state IN ('pending','claimed','completed','failed')), @@ -496,7 +504,9 @@ def prepare( adapter_id=adapter_id, input_value=input_value, ) - return self.store.ensure_intent(identity, policy_value, input_value) + return self.store.ensure_intent( + identity, policy_value, input_value, allow_unsafe=allow_unsafe + ) def execute_claimed( self, @@ -512,6 +522,8 @@ def execute_claimed( durable_record = self.store.require_active_claim(claim.operation_id, claim.token) if durable_record.identity != record.identity: raise ValueError("effect record conflicts with durable intent") + if durable_record.policy is EffectPolicy.UNSAFE and not durable_record.unsafe_authorized: + raise ValueError("unsafe effects require explicit authorization") if durable_record.identity.adapter_id != adapter.adapter_id: raise ValueError("effect adapter identity mismatch") if _sha256(canonical_json(input_value)) != durable_record.identity.input_hash: @@ -540,7 +552,15 @@ def execute_claimed( def expected_intent_columns() -> tuple[str, ...]: - return ("workflow_id", "effect_key", "adapter_id", "policy", "input_hash", "input_json") + return ( + "workflow_id", + "effect_key", + "adapter_id", + "policy", + "unsafe_authorized", + "input_hash", + "input_json", + ) def _record_from_rows(row: sqlite3.Row, receipt_row: sqlite3.Row | None) -> EffectRecord: @@ -570,6 +590,7 @@ def _record_from_rows(row: sqlite3.Row, receipt_row: sqlite3.Row | None) -> Effe return EffectRecord( identity=identity, policy=EffectPolicy(str(row["policy"])), + unsafe_authorized=bool(row["unsafe_authorized"]), state=state, attempts=int(row["attempts"]), claim_token=str(row["claim_token"]) if row["claim_token"] is not None else None, diff --git a/tests/contracts/test_effect_adapter_contract.py b/tests/contracts/test_effect_adapter_contract.py index 441906d..d2b388f 100644 --- a/tests/contracts/test_effect_adapter_contract.py +++ b/tests/contracts/test_effect_adapter_contract.py @@ -96,6 +96,7 @@ def test_file_backed_adapter_subprocess_is_idempotent_and_lookupable(tmp_path): fixture = json.loads( (Path(__file__).parents[1] / "fixtures" / "effect_contract_v1.json").read_text() ) + assert fixture["adapter_contract"]["unsafe_requires_durable_authorization"] is True vector = fixture["identity_vectors"][0] identity = operation_identity( workflow_id=vector["workflow_id"], diff --git a/tests/fixtures/effect_contract_v1.json b/tests/fixtures/effect_contract_v1.json index 139b071..56a7ccb 100644 --- a/tests/fixtures/effect_contract_v1.json +++ b/tests/fixtures/effect_contract_v1.json @@ -16,6 +16,7 @@ "service_id": "effects.adapters", "contract_version": 1, "lookup_before_perform": true, + "unsafe_requires_durable_authorization": true, "receipt_projection_excludes_payload": true, "delivery_semantics": "at-least-once" }, diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index fb5c1fc..7be87b7 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -548,3 +548,93 @@ def test_unsafe_effect_can_be_claimed_once_but_not_reclaimed_automatically(tmp_p with pytest.raises(RuntimeError, match="not claimable"): store.claim(record.identity.operation_id, token="forbidden-retry", now=11.0) + + +def test_direct_unsafe_intent_without_authorization_refuses_execution(tmp_path): + from hermes_workflows.effects import ( + EffectCoordinator, + EffectPolicy, + SQLiteEffectStore, + operation_identity, + ) + + class RecordingAdapter: + adapter_id = "payments.v1" + + def __init__(self): + self.calls = [] + + def lookup_receipt(self, operation_id: str): + self.calls.append(("lookup", operation_id)) + return None + + def perform(self, operation_id: str, input_value: Any): + self.calls.append(("perform", operation_id)) + return {"operation_id": operation_id, "adapter_receipt_id": "charge-1"} + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + input_value = {"amount": 100} + identity = operation_identity( + workflow_id="wf-direct-unsafe", + effect_key="charge-card", + adapter_id="payments.v1", + input_value=input_value, + ) + record = store.ensure_intent(identity, EffectPolicy.UNSAFE, input_value) + claim = store.claim(identity.operation_id, token="unauthorized") + adapter = RecordingAdapter() + + with pytest.raises(ValueError, match="explicit authorization"): + EffectCoordinator(store).execute_claimed(record, claim, adapter, input_value) + + assert adapter.calls == [] + refused = store.get(identity.operation_id) + assert refused.state == "claimed" + assert refused.receipt is None + + +def test_authorized_unsafe_intent_executes_with_durable_authorization(tmp_path): + from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore + + class RecordingAdapter: + adapter_id = "payments.v1" + + def __init__(self): + self.calls = [] + + def lookup_receipt(self, operation_id: str): + self.calls.append(("lookup", operation_id)) + return None + + def perform(self, operation_id: str, input_value: Any): + self.calls.append(("perform", operation_id)) + return {"operation_id": operation_id, "adapter_receipt_id": "charge-1"} + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + coordinator = EffectCoordinator(store) + input_value = {"amount": 100} + record = coordinator.prepare( + workflow_id="wf-authorized-unsafe", + effect_key="charge-card", + adapter_id="payments.v1", + input_value=input_value, + policy=EffectPolicy.UNSAFE, + allow_unsafe=True, + ) + claim = store.claim(record.identity.operation_id, token="authorized") + adapter = RecordingAdapter() + + completed = coordinator.execute_claimed(record, claim, adapter, input_value) + + assert completed.state == "completed" + assert completed.receipt is not None + assert adapter.calls == [ + ("lookup", record.identity.operation_id), + ("perform", record.identity.operation_id), + ] + with sqlite3.connect(store.path) as conn: + authorization = conn.execute( + "SELECT unsafe_authorized FROM effect_intents WHERE operation_id = ?", + (record.identity.operation_id,), + ).fetchone()[0] + assert authorization == 1 From 145803a1a1b3c11909130040d66552ba7a8afa02 Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:04:24 -0700 Subject: [PATCH 09/11] fix(effects): fence unsafe completion authorization --- src/hermes_workflows/effects.py | 12 ++++++++++++ tests/test_effect_idempotency.py | 32 ++++++++++++++++++++++++++++++++ 2 files changed, 44 insertions(+) diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index 7889fde..bfed81a 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -326,10 +326,22 @@ def complete( SET state = 'completed', updated_at = ? WHERE operation_id = ? AND state = 'claimed' AND claim_token = ? AND claim_expires_at > ? + AND (policy != 'unsafe' OR unsafe_authorized = 1) """, (timestamp, operation_id, claim_token, timestamp), ) if cursor.rowcount != 1: + unauthorized = conn.execute( + """ + SELECT 1 FROM effect_intents + WHERE operation_id = ? AND state = 'claimed' AND claim_token = ? + AND claim_expires_at > ? AND policy = 'unsafe' + AND unsafe_authorized = 0 + """, + (operation_id, claim_token, timestamp), + ).fetchone() + if unauthorized is not None: + raise ValueError("unsafe effects require explicit authorization") raise RuntimeError("stale claim token cannot complete effect") conn.execute( """ diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index 7be87b7..9a62121 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -593,6 +593,38 @@ def perform(self, operation_id: str, input_value: Any): assert refused.receipt is None +def test_direct_unsafe_intent_without_authorization_refuses_completion(tmp_path): + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + input_value = {"amount": 100} + identity = operation_identity( + workflow_id="wf-direct-unsafe-completion", + effect_key="charge-card", + adapter_id="payments.v1", + input_value=input_value, + ) + store.ensure_intent(identity, EffectPolicy.UNSAFE, input_value) + claim = store.claim(identity.operation_id, token="unauthorized") + + with pytest.raises(ValueError, match="explicit authorization"): + store.complete( + identity.operation_id, + claim.token, + {"operation_id": identity.operation_id, "adapter_receipt_id": "charge-1"}, + ) + + refused = store.get(identity.operation_id) + assert refused.state == "claimed" + assert refused.receipt is None + with sqlite3.connect(store.path) as conn: + receipt_count = conn.execute( + "SELECT COUNT(*) FROM effect_receipts WHERE operation_id = ?", + (identity.operation_id,), + ).fetchone()[0] + assert receipt_count == 0 + + def test_authorized_unsafe_intent_executes_with_durable_authorization(tmp_path): from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore From 78d925502120b2cf628a544596e4f4f5f598877d Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 04:58:33 -0700 Subject: [PATCH 10/11] fix(effects): generate fresh ownership per claim --- docs/contracts/effects.md | 2 +- src/hermes_workflows/effects.py | 22 +++++---- tests/test_effect_idempotency.py | 79 ++++++++++++++++++++++---------- 3 files changed, 70 insertions(+), 33 deletions(-) diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md index 45fa041..3f4ab31 100644 --- a/docs/contracts/effects.md +++ b/docs/contracts/effects.md @@ -17,7 +17,7 @@ No policy is inferred from a function name, tool name, or implementation. ## Durable records and fencing -`effect_intents` stores canonical input, input hash, policy, explicit unsafe-authorization evidence, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. Directly persisted unsafe intents without authorization remain inventoriable and claimable once, but the execution boundary refuses them before adapter lookup, external mutation, or durable completion. +`effect_intents` stores canonical input, input hash, policy, explicit unsafe-authorization evidence, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. Every successful claim receives a fresh, unpredictable token generated by the store; callers cannot choose ownership identity, and a reclaim refuses to reuse the preceding generation's token. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. Directly persisted unsafe intents without authorization remain inventoriable and claimable once, but the execution boundary refuses them before adapter lookup, external mutation, or durable completion. `effect_receipts` stores the adapter receipt, its SHA-256, sensitivity flag, completion time, and the claim token that fenced the write. Intent completion and receipt insertion are one SQLite transaction. diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index bfed81a..ce6fed4 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -257,7 +257,6 @@ def claim( *, ttl_seconds: float = 30.0, now: float | None = None, - token: str | None = None, ) -> EffectClaim: _validate_operation_id(operation_id) if not isinstance(ttl_seconds, (int, float)) or isinstance(ttl_seconds, bool): @@ -265,12 +264,23 @@ def claim( if not math.isfinite(float(ttl_seconds)) or ttl_seconds <= 0: raise ValueError("ttl_seconds must be finite and greater than zero") timestamp = _timestamp(now) - claim_token = token or secrets.token_urlsafe(24) - if not isinstance(claim_token, str) or not claim_token: - raise ValueError("claim token must be a nonempty string") expires_at = timestamp + float(ttl_seconds) with self._connect() as conn: conn.execute("BEGIN IMMEDIATE") + row = conn.execute( + "SELECT claim_token FROM effect_intents WHERE operation_id = ?", (operation_id,) + ).fetchone() + if row is None: + raise KeyError(f"unknown operation_id: {operation_id}") + active_token = row[0] + claim_token = "" + for _ in range(8): + candidate = secrets.token_urlsafe(24) + if isinstance(candidate, str) and candidate and candidate != active_token: + claim_token = candidate + break + if not claim_token: + raise RuntimeError("could not generate fresh claim ownership token") cursor = conn.execute( """ UPDATE effect_intents @@ -288,10 +298,6 @@ def claim( (claim_token, expires_at, timestamp, operation_id, timestamp), ) if cursor.rowcount != 1: - if conn.execute( - "SELECT 1 FROM effect_intents WHERE operation_id = ?", (operation_id,) - ).fetchone() is None: - raise KeyError(f"unknown operation_id: {operation_id}") raise RuntimeError("effect intent is not claimable") attempt = int( conn.execute( diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index 9a62121..f3641cc 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -130,19 +130,19 @@ def test_sqlite_intent_claim_completion_and_stale_token_fencing(tmp_path): intent = store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"message": "hello"}) assert intent.state == "pending" - claim_a = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0, token="token-a") + claim_a = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0) assert claim_a.attempt == 1 with pytest.raises(RuntimeError, match="not claimable"): - store.claim(identity.operation_id, now=10.5, ttl_seconds=1.0, token="token-b") + store.claim(identity.operation_id, now=10.5, ttl_seconds=1.0) - claim_b = store.claim(identity.operation_id, now=11.1, ttl_seconds=5.0, token="token-b") + claim_b = store.claim(identity.operation_id, now=11.1, ttl_seconds=5.0) assert claim_b.attempt == 2 with pytest.raises(RuntimeError, match="stale claim token"): - store.complete(identity.operation_id, "token-a", {"provider_id": "wrong"}, now=12.0) + store.complete(identity.operation_id, claim_a.token, {"provider_id": "wrong"}, now=12.0) completed = store.complete( identity.operation_id, - "token-b", + claim_b.token, {"provider_id": "receipt-1", "access_token": "must-not-project"}, sensitive=True, now=12.0, @@ -162,9 +162,9 @@ def test_sqlite_intent_claim_completion_and_stale_token_fencing(tmp_path): "SELECT claim_token, receipt_hash FROM effect_receipts WHERE operation_id = ?", (identity.operation_id,), ).fetchone() - assert row == ("completed", 2, "token-b") + assert row == ("completed", 2, claim_b.token) assert receipt_row is not None - assert receipt_row[0] == "token-b" + assert receipt_row[0] == claim_b.token assert len(receipt_row[1]) == 64 @@ -179,7 +179,7 @@ def test_failure_is_fenced_and_terminal(tmp_path): input_value={"value": 1}, ) store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) - claim = store.claim(identity.operation_id, token="winner") + claim = store.claim(identity.operation_id) with pytest.raises(RuntimeError, match="stale claim token"): store.fail(identity.operation_id, "loser", {"kind": "wrong-owner"}) failed = store.fail(identity.operation_id, claim.token, {"kind": "adapter-error"}) @@ -200,7 +200,7 @@ def test_completion_rejects_repeated_mapping_items_without_terminal_state(tmp_pa input_value={"value": 1}, ) store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}) - claim = store.claim(identity.operation_id, token="receipt-owner") + claim = store.claim(identity.operation_id) with pytest.raises(ValueError, match="duplicate JSON object key"): store.complete(identity.operation_id, claim.token, _RepeatedItemsMapping()) @@ -227,7 +227,7 @@ def test_failure_rejects_repeated_mapping_items_without_terminal_state(tmp_path) input_value={"value": 1}, ) store.ensure_intent(identity, EffectPolicy.PURE, {"value": 1}) - claim = store.claim(identity.operation_id, token="error-owner") + claim = store.claim(identity.operation_id) with pytest.raises(ValueError, match="duplicate JSON object key"): store.fail(identity.operation_id, claim.token, _RepeatedItemsMapping()) @@ -254,18 +254,49 @@ def test_expired_claim_cannot_complete_without_reclaim(tmp_path): input_value={"value": 1}, ) store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) - claim = store.claim(identity.operation_id, token="expired", now=10.0, ttl_seconds=1.0) + claim = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0) with pytest.raises(RuntimeError, match="stale claim token"): store.complete(identity.operation_id, claim.token, {"receipt": 1}, now=11.0) - replacement = store.claim(identity.operation_id, token="current", now=11.0) + replacement = store.claim(identity.operation_id, now=11.0) completed = store.complete( identity.operation_id, replacement.token, {"receipt": 1}, now=12.0 ) assert completed.state == "completed" +def test_reclaim_gets_fresh_ownership_when_token_generator_repeats(tmp_path, monkeypatch): + from hermes_workflows import effects + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + generated_tokens = iter(("repeated-token", "repeated-token", "fresh-token")) + monkeypatch.setattr(effects.secrets, "token_urlsafe", lambda _size: next(generated_tokens)) + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-repeated-claim-token", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) + + stale_claim = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0) + current_claim = store.claim(identity.operation_id, now=11.0, ttl_seconds=5.0) + + assert stale_claim.token == "repeated-token" + assert current_claim.token == "fresh-token" + with pytest.raises(RuntimeError, match="stale claim token"): + store.complete(identity.operation_id, stale_claim.token, {"receipt": "stale"}, now=12.0) + + completed = store.complete( + identity.operation_id, current_claim.token, {"receipt": "current"}, now=12.0 + ) + assert completed.receipt is not None + assert completed.receipt.payload == {"receipt": "current"} + + @pytest.mark.parametrize("source", ["lookup", "perform"]) def test_coordinator_rejects_conflicting_adapter_receipt_operation_id(tmp_path, source): from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore @@ -280,7 +311,7 @@ def test_coordinator_rejects_conflicting_adapter_receipt_operation_id(tmp_path, input_value=input_value, policy=EffectPolicy.IDEMPOTENT, ) - claim = store.claim(record.identity.operation_id, token=f"{source}-claim") + claim = store.claim(record.identity.operation_id) with pytest.raises(ValueError, match="receipt operation_id mismatch"): coordinator.execute_claimed( @@ -319,7 +350,7 @@ def perform(self, operation_id: str, input_value: Any): input_value=input_value, policy=EffectPolicy.IDEMPOTENT, ) - valid_claim = store.claim(record.identity.operation_id, token="valid-token") + valid_claim = store.claim(record.identity.operation_id) conflicting_claim = replace(valid_claim, operation_id="op_" + "0" * 64) adapter = RecordingAdapter() @@ -363,11 +394,11 @@ def perform(self, operation_id: str, input_value: Any): claim_options = ( {"now": 10.0, "ttl_seconds": 1.0} if claim_state in {"stale", "expired"} else {} ) - claim = store.claim(record.identity.operation_id, token="original-token", **claim_options) + claim = store.claim(record.identity.operation_id, **claim_options) if claim_state == "forged": rejected_claim = replace(claim, token="forged-token", expires_at=10_000.0) elif claim_state == "stale": - store.claim(record.identity.operation_id, token="current-token", now=11.0) + store.claim(record.identity.operation_id, now=11.0) rejected_claim = replace(claim, expires_at=10_000.0) else: rejected_claim = claim @@ -395,7 +426,7 @@ def test_coordinator_revalidates_claim_after_receipt_lookup_before_perform(tmp_p input_value=input_value, policy=EffectPolicy.IDEMPOTENT, ) - claim = store.claim(record.identity.operation_id, token="original-token") + claim = store.claim(record.identity.operation_id) class ClaimReplacingAdapter: adapter_id = "test.file.v1" @@ -441,10 +472,10 @@ def test_concurrent_claim_race_has_one_winner(tmp_path): store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) barrier = threading.Barrier(2) - def compete(token): + def compete(_racer): barrier.wait() try: - return store.claim(identity.operation_id, token=token, now=10.0) + return store.claim(identity.operation_id, now=10.0) except RuntimeError: return None @@ -544,10 +575,10 @@ def test_unsafe_effect_can_be_claimed_once_but_not_reclaimed_automatically(tmp_p policy=EffectPolicy.UNSAFE, allow_unsafe=True, ) - store.claim(record.identity.operation_id, token="only-attempt", now=10.0, ttl_seconds=1.0) + store.claim(record.identity.operation_id, now=10.0, ttl_seconds=1.0) with pytest.raises(RuntimeError, match="not claimable"): - store.claim(record.identity.operation_id, token="forbidden-retry", now=11.0) + store.claim(record.identity.operation_id, now=11.0) def test_direct_unsafe_intent_without_authorization_refuses_execution(tmp_path): @@ -581,7 +612,7 @@ def perform(self, operation_id: str, input_value: Any): input_value=input_value, ) record = store.ensure_intent(identity, EffectPolicy.UNSAFE, input_value) - claim = store.claim(identity.operation_id, token="unauthorized") + claim = store.claim(identity.operation_id) adapter = RecordingAdapter() with pytest.raises(ValueError, match="explicit authorization"): @@ -605,7 +636,7 @@ def test_direct_unsafe_intent_without_authorization_refuses_completion(tmp_path) input_value=input_value, ) store.ensure_intent(identity, EffectPolicy.UNSAFE, input_value) - claim = store.claim(identity.operation_id, token="unauthorized") + claim = store.claim(identity.operation_id) with pytest.raises(ValueError, match="explicit authorization"): store.complete( @@ -653,7 +684,7 @@ def perform(self, operation_id: str, input_value: Any): policy=EffectPolicy.UNSAFE, allow_unsafe=True, ) - claim = store.claim(record.identity.operation_id, token="authorized") + claim = store.claim(record.identity.operation_id) adapter = RecordingAdapter() completed = coordinator.execute_claimed(record, claim, adapter, input_value) From 0935b8f9632c2e514f16f5455710d6cdad2715ce Mon Sep 17 00:00:00 2001 From: Skylar Payne <4830598+skylarbpayne@users.noreply.github.com> Date: Sat, 11 Jul 2026 17:10:41 -0700 Subject: [PATCH 11/11] fix(effects): prevent ABA claim token reuse --- docs/contracts/effects.md | 2 +- src/hermes_workflows/effects.py | 24 ++++++++++++++++++--- tests/test_effect_idempotency.py | 37 ++++++++++++++++++++++++++++++++ 3 files changed, 59 insertions(+), 4 deletions(-) diff --git a/docs/contracts/effects.md b/docs/contracts/effects.md index 3f4ab31..11e5ff5 100644 --- a/docs/contracts/effects.md +++ b/docs/contracts/effects.md @@ -17,7 +17,7 @@ No policy is inferred from a function name, tool name, or implementation. ## Durable records and fencing -`effect_intents` stores canonical input, input hash, policy, explicit unsafe-authorization evidence, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. Every successful claim receives a fresh, unpredictable token generated by the store; callers cannot choose ownership identity, and a reclaim refuses to reuse the preceding generation's token. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. Directly persisted unsafe intents without authorization remain inventoriable and claimable once, but the execution boundary refuses them before adapter lookup, external mutation, or durable completion. +`effect_intents` stores canonical input, input hash, policy, explicit unsafe-authorization evidence, state, attempt count, and the active claim token. Before insertion, the store recomputes the complete operation identity from workflow ID, effect key, adapter ID, and the canonicalized input and rejects any mismatch. State moves from `pending` to `claimed`, then to `completed` or `failed`. Every successful claim receives a fresh, unpredictable token generated by the store; callers cannot choose ownership identity, and the durable `effect_claim_tokens` ledger prevents reuse of any token previously issued by that store, including across expiry and reclaim. An expired claim may be replaced. Completion and failure use compare-and-swap on the active opaque claim token; a stale worker cannot write a receipt or terminal state. Directly persisted unsafe intents without authorization remain inventoriable and claimable once, but the execution boundary refuses them before adapter lookup, external mutation, or durable completion. `effect_receipts` stores the adapter receipt, its SHA-256, sensitivity flag, completion time, and the claim token that fenced the write. Intent completion and receipt insertion are one SQLite transaction. diff --git a/src/hermes_workflows/effects.py b/src/hermes_workflows/effects.py index ce6fed4..a99ceed 100644 --- a/src/hermes_workflows/effects.py +++ b/src/hermes_workflows/effects.py @@ -268,15 +268,24 @@ def claim( with self._connect() as conn: conn.execute("BEGIN IMMEDIATE") row = conn.execute( - "SELECT claim_token FROM effect_intents WHERE operation_id = ?", (operation_id,) + "SELECT 1 FROM effect_intents WHERE operation_id = ?", (operation_id,) ).fetchone() if row is None: raise KeyError(f"unknown operation_id: {operation_id}") - active_token = row[0] claim_token = "" for _ in range(8): candidate = secrets.token_urlsafe(24) - if isinstance(candidate, str) and candidate and candidate != active_token: + if not isinstance(candidate, str) or not candidate: + continue + issued = conn.execute( + """ + INSERT OR IGNORE INTO effect_claim_tokens ( + claim_token, operation_id, issued_at + ) VALUES (?, ?, ?) + """, + (candidate, operation_id, timestamp), + ) + if issued.rowcount == 1: claim_token = candidate break if not claim_token: @@ -469,6 +478,15 @@ def _initialize(self) -> None: completed_at REAL NOT NULL, claim_token TEXT NOT NULL ); + CREATE TABLE IF NOT EXISTS effect_claim_tokens ( + claim_token TEXT PRIMARY KEY, + operation_id TEXT NOT NULL REFERENCES effect_intents(operation_id), + issued_at REAL NOT NULL + ); + INSERT OR IGNORE INTO effect_claim_tokens (claim_token, operation_id, issued_at) + SELECT claim_token, operation_id, updated_at + FROM effect_intents + WHERE claim_token IS NOT NULL; CREATE INDEX IF NOT EXISTS effect_intents_claimable ON effect_intents(state, claim_expires_at); """ diff --git a/tests/test_effect_idempotency.py b/tests/test_effect_idempotency.py index f3641cc..6fa0373 100644 --- a/tests/test_effect_idempotency.py +++ b/tests/test_effect_idempotency.py @@ -297,6 +297,43 @@ def test_reclaim_gets_fresh_ownership_when_token_generator_repeats(tmp_path, mon assert completed.receipt.payload == {"receipt": "current"} +def test_reclaim_never_reuses_earlier_claim_token_after_aba_input(tmp_path, monkeypatch): + from hermes_workflows import effects + from hermes_workflows.effects import EffectPolicy, SQLiteEffectStore, operation_identity + + generated_tokens = iter(("token-a", "token-b", "token-a", "token-c")) + monkeypatch.setattr(effects.secrets, "token_urlsafe", lambda _size: next(generated_tokens)) + + store = SQLiteEffectStore(tmp_path / "effects.sqlite") + identity = operation_identity( + workflow_id="wf-aba-claim-token", + effect_key="write", + adapter_id="test.file.v1", + input_value={"value": 1}, + ) + store.ensure_intent(identity, EffectPolicy.IDEMPOTENT, {"value": 1}, now=10.0) + + claim_a = store.claim(identity.operation_id, now=10.0, ttl_seconds=1.0) + claim_b = store.claim(identity.operation_id, now=11.0, ttl_seconds=1.0) + claim_c = store.claim(identity.operation_id, now=12.0, ttl_seconds=5.0) + + assert (claim_a.token, claim_b.token, claim_c.token) == ("token-a", "token-b", "token-c") + with sqlite3.connect(tmp_path / "effects.sqlite") as conn: + issued_tokens = conn.execute( + "SELECT claim_token FROM effect_claim_tokens ORDER BY issued_at" + ).fetchall() + assert issued_tokens == [("token-a",), ("token-b",), ("token-c",)] + with pytest.raises(RuntimeError, match="stale claim token"): + store.complete(identity.operation_id, claim_a.token, {"receipt": "stale-a"}, now=13.0) + + completed = store.complete( + identity.operation_id, claim_c.token, {"receipt": "current-c"}, now=13.0 + ) + assert completed.attempts == 3 + assert completed.receipt is not None + assert completed.receipt.payload == {"receipt": "current-c"} + + @pytest.mark.parametrize("source", ["lookup", "perform"]) def test_coordinator_rejects_conflicting_adapter_receipt_operation_id(tmp_path, source): from hermes_workflows.effects import EffectCoordinator, EffectPolicy, SQLiteEffectStore