diff --git a/CHANGELOG.md b/CHANGELOG.md index b30da6c..7485558 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,15 @@ All notable changes to Ledger are documented here. ## [Unreleased] ### Added +- **Runtime-contract enforcement** (#250). The Agent Trajectory Schema + + Evidence Chain from arXiv:2608.11274: hash-chained trajectory events + (`tool_call`, `tool_result`, `file_read`, `file_write`, `shell_exec`, + `commit`, `screenshot`, `citation_lookup`, `human_approval`, + `model_message`), a deterministic verifier registry separating hard from + soft evidence, evidence-chain construction, a fail-closed evidence-gated + submission contract, and a `ComposedGate` realizing the paper's + compositional gating proposition (preventive monitors + evidential gates + over disjoint requirement sets). See `docs/runtime-contract.md`. - **Evidence levels for receipts** (#235). Receipts now state what they prove: a four-level ladder (`structural` → `attested` → `replay` → `inclusion`) under `verification.evidence`, with stable per-level reason codes. A signed diff --git a/docs/runtime-contract.md b/docs/runtime-contract.md new file mode 100644 index 0000000..8d68980 --- /dev/null +++ b/docs/runtime-contract.md @@ -0,0 +1,239 @@ +# Runtime-contract enforcement + +Ledger issue #250 adds a small, dependency-free runtime contract for agent +trajectories. The design follows *Agent Safety Should Be a Runtime Contract*, +arXiv:2608.11274 (especially its evidence-gated completion and compositional +gating arguments): . + +The contract is intentionally **additive**. It does not change Ledger's +prebind, receipt, HTTP, or OpenAPI schemas. A caller can put the trajectory root +hash in an existing `evidence_hashes` collection when it wants to bind a +submission to the observed trajectory. + +## Trajectory schema + +`ledger_agent.trajectory.Trajectory` represents a trajectory +`τ = (e_1, ..., e_T)`. Every event has the shape below. The serialized envelope +has `schema: "perseus-ledger-trajectory/v1"`, an `events` list, and a +calculated `head_hash`. + +| Field | Type | Meaning | +| --- | --- | --- | +| `kind` | string | One of `tool_call`, `tool_result`, `file_read`, `file_write`, `shell_exec`, `commit`, `screenshot`, `citation_lookup`, `human_approval`, or `model_message`. Unknown kinds are rejected. | +| `timestamp_ms` | integer | Event time in Unix milliseconds. | +| `payload` | object | Structured event data. It is copied on append and must be JSON-serializable. | +| `prev_hash` | 64-character hex string | The preceding event hash. The first event points to `sha256("genesis")`. | +| `hash` | 64-character hex string | `sha256(canonical_json(kind, timestamp_ms, payload, prev_hash))`. | + +Canonical JSON uses sorted keys, compact separators (`(',', ':')`), and UTF-8. +The event's own `hash` is excluded from the covered fields. Consequently, +changing an event makes that event fail verification and makes every unchanged +successor fail its predecessor link. This is the tamper-evident suffix +property needed by the runtime contract in arXiv:2608.11274. + +```python +from ledger_agent.trajectory import Trajectory + +trajectory = Trajectory() +trajectory.append("tool_call", {"name": "pytest"}) +trajectory.append("tool_result", {"exit_code": 0}) +assert trajectory.verify_chain() == (True, "ok") +serialized = trajectory.to_dict() +restored = Trajectory.from_dict(serialized) +assert restored.to_dict() == serialized +``` + +`Trajectory.from_dict` preserves event-level defects so a caller can inspect a +suspect record; `verify_chain()` is the explicit check and is fail-closed. +`Trajectory.head_hash` is the genesis hash for an empty trajectory and the last +stored event hash otherwise. + +## Payload shapes used by the evidence registry + +The evidence verifier registry is deterministic. Each verifier receives +`(event, property, ref_state)` and returns exactly `accept`, `reject`, or +`soft`. + +| Verifier | Required property | Payload / reference state | +| --- | --- | --- | +| `test_run` | `test_suite_passes` | `payload.exit_code == 0` and `ref_state.expected_pass is True`. | +| `citation_lookup` | `citation_real` | `payload.cited_url` (also `url`/`source_url` accepted) is a member of `ref_state.source_urls`. | +| `file_diff` | `diff_present` or `diff_matches` | `payload.diff`/`diff_text`/`hunk`/`patch`; `diff_matches` requires equality with `ref_state.expected_hunk`, while `diff_present` requires a non-empty matching hunk when one is supplied. | +| `log_capture` | `log_contains` | `ref_state.marker` is a non-empty substring of `payload.log_text`. | +| `screenshot` | `screenshot_matches` (or the equivalent screenshot property) | `payload.image_sha256 == ref_state.expected_image_sha256`. | +| `human_approval` | an approval property | `payload.approved_by` is non-empty and equals `ref_state.approval_ref` when that reference is supplied. | +| `shell_exec` | an execution property | `payload.exit_code` is captured as an integer, whether zero or non-zero. Captured failure is evidence of execution, not evidence of success. | +| `commit` | a commit property | `payload.commit_sha == ref_state.expected_commit_sha`. | + +An unknown or missing verifier is `soft`, never `accept`. A verifier marked +non-deterministic by its event or reference state is also `soft`. In particular, +a `model_message` that says `done` is soft evidence and cannot satisfy a +load-bearing requirement. This is the key false-completion distinction in +arXiv:2608.11274: a completion claim is not an observation that the claimed +work occurred. + +## Evidence chains and the submission gate + +A requirement is a small object such as: + +```python +{ + "property": "test_suite_passes", + "verifier": "test_run", + "ref_state": {"expected_pass": True}, +} +``` + +`find_evidence_chain(trajectory, requirements)` searches the observed events. +It returns `(found, chain_events, unmet_requirements)`. Every requirement must +have an event whose registered verifier returns hard `accept`; `reject` and +`soft` do not count. A single event may establish more than one explicitly +requested property, but the returned event list is de-duplicated. + +`evaluate_submission(trajectory, requirements)` is the fail-closed contract: + +```json +{ + "accepted": true, + "decision": "accepted_with_evidence", + "evidence_chain": ["..."], + "unmet_requirements": [] +} +``` + +If the trajectory is invalid or any requirement is unmet, the decision is +`rejected_missing_evidence`; it is never inferred from a terminal model +message. This operationalizes the evidence-gated completion rule discussed in +arXiv:2608.11274. + +## Compositional gating proposition + +`ComposedGate` combines deterministic trajectory monitors with one or more +independent evidence gates. The built-in monitors are: + +* `no_shell_exec_without_prior_human_approval`: every `shell_exec` must be + preceded by an approved `human_approval` event; +* `no_file_write_outside_allowed_paths`: every `file_write` path must be within + a caller-provided `ref_state["allowed_paths"]` root. + +The preferred API is: + +```python +from ledger_agent.trajectory import ComposedGate + +gate = ComposedGate( + monitors=[{"name": "no_shell_exec_without_prior_human_approval"}], + gates=[requirements], +) +report = gate.evaluate(trajectory) +``` + +The gate accepts iff every monitor holds and every evidence gate finds a full +chain. Requirement sets across evidence gates must be disjoint; overlap is +rejected rather than silently allowing interference. + +**Proposition (compositional runtime enforcement, adapted from §4.3 of +arXiv:2608.11274).** Let monitors `h_1, ..., h_n` be deterministic finite +state monitors (DFAs) with disjoint observation alphabets, and let `H_1, ..., +H_m` be evidence gates over disjoint requirement sets. Their parallel +composition accepts exactly the trajectories satisfying + +``` +(h_1 || ... || h_n || H_1 || ... || H_m)(τ) + = (∧ᵢ φᵢ(τ)) ∧ (∧ⱼ ηⱼ(τ)) +``` + +where `φ_i` is the safety language of monitor `h_i` and `η_j` is a complete +hard-evidence chain for gate `H_j`. Disjoint alphabets make the monitors +non-interfering and permit product evaluation in polynomial time in the +trajectory and monitor sizes. If observations are shared, use an +assume-guarantee contract: each monitor states the events it assumes and the +properties it guarantees, and the composition must discharge the shared-event +obligations. General shared-alphabet DFA composition can require an exponential +state product. Sequential evidence-chain scans and disjoint monitor evaluation +remain polynomial; unrestricted general composition has the usual exponential +worst case. This is the runtime-contract composition boundary described in +arXiv:2608.11274. + +## Preventive-face taxonomy + +A runtime contract has four complementary faces, rather than treating every +control as a post-hoc audit: + +1. **Preventive** — block or hold an action before its side effect (for example, + the shell-approval and allowed-path monitors). +2. **Detective** — observe and hash what happened (the trajectory chain, logs, + screenshots, citations, and test results). +3. **Corrective** — stop, roll back, quarantine, or require re-approval after a + monitor violation or missing evidence. +4. **Structural** — make the safe path the natural path through schemas, + least authority, explicit references, and cryptographic binding. + +The five Saltzer-Schroeder principles adapted to this contract are: + +* **Economy of mechanism:** use a small closed event vocabulary, canonical JSON, + and simple deterministic predicates. +* **Fail-safe defaults:** absent, unknown, rejected, or soft evidence denies a + submission; callers must prove acceptance. +* **Complete mediation:** evaluate every submission and every monitored side + effect, not only the first event or the final model message. +* **Open design:** the schema and verifier behavior are inspectable and + reproducible; security does not depend on hiding the implementation. +* **Separation of privilege:** require independent evidence classes and, where + appropriate, an independent human approval instead of treating one claim as + sufficient. + +These faces and principles turn the paper's runtime-contract argument into +operational controls while preserving the distinction between an observed +fact and a model assertion (arXiv:2608.11274). + +## Six evidence classes for false-completion audits + +The paper's false-completion audit motivates collecting multiple evidence +classes. A deployment can require whichever classes are appropriate, but should +not silently substitute a model message for any of them: + +1. **Citation grounding** — the cited URL or external source was looked up and + is in the trusted source set. +2. **Log capture** — the expected marker is present in captured execution output. +3. **Test run** — a concrete test process returned exit code zero against the + expected reference state. +4. **Human approval** — an identified approver authorized the relevant action. +5. **External state** — a commit, file diff, or other independently inspectable + state matches the expected reference. +6. **Screenshot** — a captured visual state is bound by its image digest. + +The registry's `test_run`, `log_capture`, `citation_lookup`, +`human_approval`, `commit`/`file_diff`, and `screenshot` verifiers implement +these classes as hard evidence where their deterministic reference checks +succeed. This layered evidence is the practical antidote to false completion +identified by arXiv:2608.11274. + +## AAR / prebind integration + +`trajectory_root_hash(trajectory)` computes +`sha256(trajectory.head_hash)` as a hexadecimal string. Callers may append this +value to an existing prebind or receipt `evidence_hashes` list, for example: + +```python +from ledger_agent.trajectory import trajectory_root_hash + +root = trajectory_root_hash(trajectory) +# Existing prebind/receipt builder, unchanged: +# evidence_hashes = prior_hashes + [root] +``` + +This binds the caller's existing evidence block to the observed trajectory +without changing any existing function signature or schema. The root helper is +an integration aid, not a replacement for verifying the trajectory itself. + +## Demo + +Run the end-to-end example with the repository's configured interpreter: + +```bash +/opt/data/venv-ledger/bin/python examples/runtime_contract_demo.py +``` + +It prints a rejection for the evidence-less `done` claim, then appends a test +run, captured log marker, and citation lookup and prints an accepted report. diff --git a/examples/runtime_contract_demo.py b/examples/runtime_contract_demo.py new file mode 100644 index 0000000..5cd5114 --- /dev/null +++ b/examples/runtime_contract_demo.py @@ -0,0 +1,63 @@ +"""Minimal runtime-contract evidence-gating demonstration for Ledger #250.""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +# Keep the example runnable directly from a clean checkout, before a wheel is +# installed into the selected interpreter. +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from ledger_agent.trajectory import Trajectory, evaluate_submission + + +REQUIREMENTS = [ + { + "property": "test_suite_passes", + "verifier": "test_run", + "ref_state": {"expected_pass": True}, + }, + { + "property": "log_contains", + "verifier": "log_capture", + "ref_state": {"marker": "ALL TESTS PASSED"}, + }, + { + "property": "citation_real", + "verifier": "citation_lookup", + "ref_state": {"source_urls": ["https://arxiv.org/abs/2608.11274"]}, + }, +] + + +def report(title: str, value: dict) -> None: + print(title) + print(json.dumps(value, indent=2, sort_keys=True)) + + +def main() -> None: + trajectory = Trajectory() + trajectory.append("model_message", {"text": "done"}, timestamp_ms=1) + report("Evidence-less done claim", evaluate_submission(trajectory, REQUIREMENTS)) + + trajectory.append( + "tool_result", + {"command": "pytest tests/ -q", "exit_code": 0}, + timestamp_ms=2, + ) + trajectory.append( + "tool_result", + {"log_text": "pytest: ALL TESTS PASSED"}, + timestamp_ms=3, + ) + trajectory.append( + "citation_lookup", + {"cited_url": "https://arxiv.org/abs/2608.11274"}, + timestamp_ms=4, + ) + report("Complete evidence chain", evaluate_submission(trajectory, REQUIREMENTS)) + + +if __name__ == "__main__": + main() diff --git a/ledger_agent/trajectory.py b/ledger_agent/trajectory.py new file mode 100644 index 0000000..d0fa6e6 --- /dev/null +++ b/ledger_agent/trajectory.py @@ -0,0 +1,925 @@ +"""Runtime-contract trajectory and evidence enforcement (#250). + +This module is deliberately dependency-free. It records an agent trajectory as +an append-only hash chain and provides deterministic, fail-closed evidence +verifiers for submission decisions. The schema and evidence distinction are +based on *Agent Safety Should Be a Runtime Contract*, arXiv:2608.11274. + +The module is additive to Ledger's prebind and receipt schemas: callers can use +:func:`trajectory_root_hash` as another value in an existing ``evidence_hashes`` +collection without changing those schemas. +""" +from __future__ import annotations + +import copy +import hashlib +import json +import os +import time +from collections.abc import Callable, Iterable, Mapping, Sequence +from typing import Any, Optional + + +TRAJECTORY_SCHEMA = "perseus-ledger-trajectory/v1" +"""Versioned schema identifier for a serialized trajectory.""" + +GENESIS_HASH = hashlib.sha256(b"genesis").hexdigest() +"""The hash used as ``h_0`` for every trajectory.""" + +EVENT_KINDS = frozenset( + { + "tool_call", + "tool_result", + "file_read", + "file_write", + "shell_exec", + "commit", + "screenshot", + "citation_lookup", + "human_approval", + "model_message", + } +) +"""The closed event-kind vocabulary from Definition 1.""" + +# Friendly aliases make the schema easy to discover without duplicating the +# canonical constants. +TRAJECTORY_VERSION = TRAJECTORY_SCHEMA +ALLOWED_EVENT_KINDS = EVENT_KINDS +GENESIS = GENESIS_HASH +HARD = "accept" +REJECT = "reject" +SOFT = "soft" + +_EVENT_FIELDS = frozenset({"kind", "timestamp_ms", "payload", "prev_hash", "hash"}) +_HEX_DIGITS = frozenset("0123456789abcdef") + + +def canonical_json(value: Any) -> str: + """Return Ledger's stable canonical JSON representation. + + Hashes intentionally use sorted keys and compact separators. Unicode is + kept as UTF-8 rather than escaped so the representation is unambiguous and + compact while remaining deterministic. + """ + + return json.dumps( + value, + sort_keys=True, + separators=(",", ":"), + ) + + +def _canonical_bytes(value: Any) -> bytes: + return canonical_json(value).encode("utf-8") + + +def _is_sha256(value: Any) -> bool: + return ( + isinstance(value, str) + and len(value) == 64 + and set(value.lower()) <= _HEX_DIGITS + ) + + +def _event_material(event: Mapping[str, Any], prev_hash: str) -> dict[str, Any]: + """Return the fields covered by an event hash, excluding ``hash`` itself.""" + + return { + "kind": event.get("kind"), + "timestamp_ms": event.get("timestamp_ms"), + "payload": event.get("payload"), + "prev_hash": prev_hash, + } + + +def event_hash(event: Mapping[str, Any], prev_hash: Optional[str] = None) -> str: + """Hash an event's canonical fields and its predecessor hash. + + ``prev_hash`` defaults to the event's stored predecessor, which is useful + for independently checking a serialized event. The stored ``hash`` is + never included in the digest. + """ + + predecessor = event.get("prev_hash") if prev_hash is None else prev_hash + if not isinstance(predecessor, str): + raise ValueError("prev_hash must be a string") + return hashlib.sha256(_canonical_bytes(_event_material(event, predecessor))).hexdigest() + + +class Trajectory: + """An append-only sequence of hash-chained agent events. + + Events are plain dictionaries with ``kind``, ``timestamp_ms``, ``payload``, + ``prev_hash``, and ``hash`` fields. The first event links to + :data:`GENESIS_HASH`; the current ``head_hash`` is the last event hash, or + the genesis hash for an empty trajectory. + """ + + def __init__(self, events: Optional[Iterable[Mapping[str, Any]]] = None): + if events is None: + copied: list[dict[str, Any]] = [] + else: + if isinstance(events, (str, bytes)): + raise TypeError("events must be an iterable of event mappings") + copied = [] + for event in events: + if not isinstance(event, Mapping): + raise TypeError("each event must be a mapping") + copied.append(copy.deepcopy(dict(event))) + self.events = copied + self._schema = TRAJECTORY_SCHEMA + self._declared_head_hash: Optional[str] = None + + def __iter__(self): + return iter(self.events) + + def __len__(self) -> int: + return len(self.events) + + @property + def head_hash(self) -> str: + """Return the stored head hash, or the genesis hash when empty.""" + + if not self.events: + return GENESIS_HASH + value = self.events[-1].get("hash") + return value if isinstance(value, str) else "" + + def append( + self, + kind: str, + payload: dict[str, Any], + timestamp_ms: Optional[int] = None, + ) -> dict[str, Any]: + """Append one event and return its plain dictionary representation. + + Unknown kinds, non-dict payloads, invalid timestamps, and payloads that + cannot be represented in canonical JSON are rejected before mutation. + """ + + if not isinstance(kind, str) or kind not in EVENT_KINDS: + raise ValueError(f"unknown event kind: {kind!r}") + if not isinstance(payload, dict): + raise TypeError("event payload must be a dict") + if timestamp_ms is None: + timestamp_ms = time.time_ns() // 1_000_000 + if isinstance(timestamp_ms, bool) or not isinstance(timestamp_ms, int): + raise TypeError("timestamp_ms must be an integer") + + payload_copy = copy.deepcopy(payload) + # Validate serializability before changing the trajectory. + try: + _canonical_bytes(payload_copy) + except (TypeError, ValueError) as exc: + raise TypeError("event payload must be JSON serializable") from exc + + predecessor = self.head_hash + event: dict[str, Any] = { + "kind": kind, + "timestamp_ms": timestamp_ms, + "payload": payload_copy, + "prev_hash": predecessor, + } + event["hash"] = event_hash(event, predecessor) + self.events.append(event) + self._declared_head_hash = event["hash"] + return event + + def verify_chain(self) -> tuple[bool, str]: + """Verify schema, predecessor links, and every event hash. + + The reason is a stable lowercase code so callers can make a deterministic + fail-closed decision without parsing human prose. + """ + + if self._schema != TRAJECTORY_SCHEMA: + return False, "schema_mismatch" + predecessor = GENESIS_HASH + for event in self.events: + if not isinstance(event, Mapping): + return False, "invalid_event" + if set(event) != _EVENT_FIELDS: + return False, "invalid_event" + kind = event.get("kind") + if not isinstance(kind, str) or kind not in EVENT_KINDS: + return False, "unknown_kind" + timestamp_ms = event.get("timestamp_ms") + if isinstance(timestamp_ms, bool) or not isinstance(timestamp_ms, int): + return False, "invalid_timestamp" + if not isinstance(event.get("payload"), dict): + return False, "invalid_payload" + stored_prev = event.get("prev_hash") + stored_hash = event.get("hash") + if not _is_sha256(stored_prev) or not _is_sha256(stored_hash): + return False, "invalid_hash" + if stored_prev != predecessor: + return False, "prev_hash_mismatch" + try: + expected = event_hash(event, predecessor) + except (TypeError, ValueError): + return False, "invalid_event" + if stored_hash != expected: + return False, "hash_mismatch" + predecessor = stored_hash + + if self._declared_head_hash is not None and self._declared_head_hash != predecessor: + return False, "head_hash_mismatch" + return True, "ok" + + def to_dict(self) -> dict[str, Any]: + """Serialize the trajectory to a deterministic plain dictionary.""" + + return { + "schema": TRAJECTORY_SCHEMA, + "events": copy.deepcopy(self.events), + "head_hash": self.head_hash, + } + + @classmethod + def from_dict(cls, value: Mapping[str, Any]) -> "Trajectory": + """Restore a trajectory without hiding later chain verification errors. + + A malformed top-level envelope raises immediately. Event-level + tampering is retained so :meth:`verify_chain` can report it rather than + making it impossible to inspect a suspect trajectory. + """ + + if not isinstance(value, Mapping): + raise TypeError("trajectory must be a mapping") + if value.get("schema") != TRAJECTORY_SCHEMA: + raise ValueError("invalid trajectory schema") + events = value.get("events") + if not isinstance(events, list): + raise ValueError("trajectory events must be a list") + trajectory = cls(events) + declared = value.get("head_hash") + if declared is not None and not isinstance(declared, str): + raise ValueError("trajectory head_hash must be a string") + trajectory._declared_head_hash = declared + return trajectory + + +# ── Deterministic evidence verifiers ──────────────────────────────────────── + + +def _event_payload(event: Mapping[str, Any]) -> Optional[Mapping[str, Any]]: + payload = event.get("payload") if isinstance(event, Mapping) else None + return payload if isinstance(payload, Mapping) else None + + +def _common_verifier_guard( + event: Mapping[str, Any], ref_state: Mapping[str, Any] +) -> Optional[str]: + """Return a non-accepting classification for soft/non-deterministic input.""" + + payload = _event_payload(event) + if payload is None: + return REJECT + kind = event.get("kind") + # A model's assertion is not execution evidence, even if it says "done" or + # happens to contain fields that resemble a tool result. + if kind == "model_message": + return SOFT + if ( + event.get("deterministic") is False + or payload.get("deterministic") is False + or payload.get("non_deterministic") is True + or ref_state.get("deterministic") is False + or ref_state.get("non_deterministic") is True + ): + return SOFT + return None + + +def _test_run(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if property != "test_suite_passes": + return REJECT + payload = _event_payload(event) + exit_code = payload.get("exit_code") if payload is not None else None + if ( + isinstance(exit_code, int) + and not isinstance(exit_code, bool) + and exit_code == 0 + and ref_state.get("expected_pass") is True + ): + return HARD + return REJECT + + +def _citation_lookup(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if property != "citation_real": + return REJECT + payload = _event_payload(event) + if payload is None: + return REJECT + url = payload.get("cited_url") or payload.get("url") or payload.get("source_url") + source_urls = ref_state.get("source_urls") + if isinstance(source_urls, str): + allowed = {source_urls} + elif isinstance(source_urls, Iterable): + allowed = set(source_urls) + else: + allowed = set() + return HARD if isinstance(url, str) and url in allowed else REJECT + + +def _diff_value(payload: Mapping[str, Any]) -> Any: + for key in ("diff", "diff_text", "hunk", "patch"): + if key in payload: + return payload[key] + return None + + +def _file_diff(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if property not in {"diff_present", "diff_matches"}: + return REJECT + payload = _event_payload(event) + expected = ref_state.get("expected_hunk") + actual = _diff_value(payload) if payload is not None else None + if property == "diff_matches": + return HARD if expected is not None and actual == expected else REJECT + if actual in (None, "", [], {}): + return REJECT + if expected is None: + return HARD + if isinstance(actual, str) and isinstance(expected, str): + return HARD if expected in actual else REJECT + return HARD if actual == expected else REJECT + + +def _log_capture(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if property != "log_contains": + return REJECT + payload = _event_payload(event) + marker = ref_state.get("marker") + log_text = payload.get("log_text") if payload is not None else None + return HARD if isinstance(marker, str) and marker and isinstance(log_text, str) and marker in log_text else REJECT + + +def _screenshot(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if property not in {"screenshot", "screenshot_matches", "screenshot_real", "image_matches"}: + return REJECT + payload = _event_payload(event) + expected = ref_state.get("expected_image_sha256") + return HARD if payload is not None and payload.get("image_sha256") == expected and _is_sha256(expected) else REJECT + + +def _human_approval(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if not isinstance(property, str) or not property: + return REJECT + payload = _event_payload(event) + approved_by = payload.get("approved_by") if payload is not None else None + if not isinstance(approved_by, str) or not approved_by.strip(): + return REJECT + approval_ref = ref_state.get("approval_ref") + if approval_ref is not None and approved_by != approval_ref: + return REJECT + return HARD + + +def _shell_exec(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if not isinstance(property, str) or not property: + return REJECT + payload = _event_payload(event) + exit_code = payload.get("exit_code") if payload is not None else None + return HARD if isinstance(exit_code, int) and not isinstance(exit_code, bool) else REJECT + + +def _commit(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + if not isinstance(property, str) or not property: + return REJECT + payload = _event_payload(event) + expected = ref_state.get("expected_commit_sha") + return HARD if payload is not None and expected is not None and payload.get("commit_sha") == expected else REJECT + + +def _guarded( + function: Callable[[Mapping[str, Any], str, Mapping[str, Any]], str] +) -> Callable[[Mapping[str, Any], str, Mapping[str, Any]], str]: + def wrapped(event: Mapping[str, Any], property: str, ref_state: Mapping[str, Any]) -> str: + guard = _common_verifier_guard(event, ref_state) + return guard if guard is not None else function(event, property, ref_state) + + wrapped.__name__ = function.__name__ + wrapped.__doc__ = function.__doc__ + return wrapped + + +VERIFIER_REGISTRY: dict[str, Callable[[Mapping[str, Any], str, Mapping[str, Any]], str]] = { + "test_run": _guarded(_test_run), + "citation_lookup": _guarded(_citation_lookup), + "file_diff": _guarded(_file_diff), + "log_capture": _guarded(_log_capture), + "screenshot": _guarded(_screenshot), + "human_approval": _guarded(_human_approval), + "shell_exec": _guarded(_shell_exec), + "commit": _guarded(_commit), +} +# Public aliases used by integrations that call the registry directly. +EVIDENCE_VERIFIERS = VERIFIER_REGISTRY +VERIFIERS = VERIFIER_REGISTRY + + +def verify_evidence( + event: Mapping[str, Any], + property: Any, + verifier: Any, + ref_state: Optional[Mapping[str, Any]] = None, +) -> str: + """Classify one evidence event as hard ``accept``, ``reject``, or ``soft``. + + ``verifier`` is normally a registry name. A callable with the same + ``(event, property, ref_state)`` signature is also accepted for local + deterministic extensions. Unknown or missing verifiers deliberately return + ``soft`` rather than granting evidence. + """ + + # Be forgiving for callers that naturally write (event, verifier, property, + # ref_state); the explicit names in the public signature remain canonical. + if isinstance(property, str) and property in VERIFIER_REGISTRY and ( + not isinstance(verifier, str) or verifier not in VERIFIER_REGISTRY + ): + property, verifier = verifier, property + + state: Mapping[str, Any] + if isinstance(ref_state, Mapping): + state = ref_state + else: + state = {} + if not isinstance(event, Mapping): + return REJECT + if verifier is None or verifier == "": + return SOFT + if isinstance(verifier, str): + function = VERIFIER_REGISTRY.get(verifier) + if function is None: + return SOFT + elif callable(verifier): + function = verifier + else: + return SOFT + try: + result = function(event, property, state) + except (KeyError, TypeError, ValueError): + return REJECT + return result if isinstance(result, str) and result in {HARD, REJECT, SOFT} else SOFT + + +# Names used by earlier/adjacent evidence integrations. +classify_evidence = verify_evidence +verify_event_evidence = verify_evidence + + +def _trajectory_events(trajectory: Any) -> list[Mapping[str, Any]]: + if isinstance(trajectory, Trajectory): + return list(trajectory.events) + if isinstance(trajectory, Mapping): + events = trajectory.get("events", []) + return list(events) if isinstance(events, list) else [] + if isinstance(trajectory, Iterable) and not isinstance(trajectory, (str, bytes)): + return list(trajectory) + return [] + + +def _requirement_state( + requirement: Mapping[str, Any], global_ref_state: Optional[Mapping[str, Any]] +) -> Mapping[str, Any]: + for key in ("ref_state", "reference_state", "ref"): + value = requirement.get(key) + if isinstance(value, Mapping): + return value + return global_ref_state if isinstance(global_ref_state, Mapping) else {} + + +def find_evidence_chain( + trajectory: Any, + requirements: Sequence[Mapping[str, Any]], + *, + ref_state: Optional[Mapping[str, Any]] = None, +) -> tuple[bool, list[dict[str, Any]], list[dict[str, Any]]]: + """Find an evidence chain satisfying every requested property. + + A requirement is satisfied only by a verifier returning hard ``accept``. + Unknown verifiers and model assertions therefore remain soft and cannot + enter the returned chain. One event may establish more than one explicitly + requested property; the chain result deduplicates such an event. + """ + + events = _trajectory_events(trajectory) + chain: list[dict[str, Any]] = [] + chain_indices: set[int] = set() + unmet: list[dict[str, Any]] = [] + for raw_requirement in requirements or []: + if not isinstance(raw_requirement, Mapping): + unmet.append(copy.deepcopy(raw_requirement)) + continue + requirement = dict(raw_requirement) + property_name = requirement.get("property") + verifier_name = requirement.get("verifier") + state = _requirement_state(requirement, ref_state) + match_index: Optional[int] = None + for index, event in enumerate(events): + if verify_evidence(event, property_name, verifier_name, state) == HARD: + match_index = index + break + if match_index is None: + unmet.append(copy.deepcopy(requirement)) + elif match_index not in chain_indices: + matched = events[match_index] + if isinstance(matched, Mapping): + chain.append(copy.deepcopy(dict(matched))) + chain_indices.add(match_index) + return not unmet, chain, unmet + + +def _trajectory_is_valid(trajectory: Any) -> tuple[bool, str]: + if isinstance(trajectory, Trajectory): + return trajectory.verify_chain() + if isinstance(trajectory, Mapping): + try: + return Trajectory.from_dict(trajectory).verify_chain() + except (TypeError, ValueError): + return False, "invalid_trajectory" + return True, "ok" + + +def evaluate_submission( + trajectory: Any, + requirements: Sequence[Mapping[str, Any]], + *, + ref_state: Optional[Mapping[str, Any]] = None, +) -> dict[str, Any]: + """Apply the fail-closed evidence-gated submission contract.""" + + valid, reason = _trajectory_is_valid(trajectory) + if valid: + found, chain, unmet = find_evidence_chain( + trajectory, requirements, ref_state=ref_state + ) + else: + found, chain, unmet = False, [], [copy.deepcopy(r) for r in (requirements or [])] + unmet.insert( + 0, + { + "property": "trajectory_integrity", + "verifier": "trajectory_chain", + "reason": reason, + }, + ) + accepted = bool(valid and found) + return { + "accepted": accepted, + "decision": "accepted_with_evidence" if accepted else "rejected_missing_evidence", + "evidence_chain": chain if accepted else [], + "unmet_requirements": [] if accepted else unmet, + } + + +# ── Deterministic monitors and compositional gates ─────────────────────────── + + +def _approved_payload(payload: Mapping[str, Any]) -> bool: + approved_by = payload.get("approved_by") + if not isinstance(approved_by, str) or not approved_by.strip(): + return False + return payload.get("approved", True) is not False + + +def monitor_no_shell_exec_without_prior_human_approval( + trajectory: Any, ref_state: Optional[Mapping[str, Any]] = None +) -> tuple[bool, str]: + """Require a preceding approved ``human_approval`` before every shell exec.""" + + approved = False + for event in _trajectory_events(trajectory): + if not isinstance(event, Mapping): + return False, "invalid_event" + if event.get("kind") == "human_approval": + payload = _event_payload(event) + if payload is not None and _approved_payload(payload): + approved = True + elif event.get("kind") == "shell_exec" and not approved: + return False, "shell_exec_without_prior_human_approval" + return True, "ok" + + +def _path_is_within(path: str, allowed: str) -> bool: + try: + candidate = os.path.abspath(os.path.normpath(path)) + root = os.path.abspath(os.path.normpath(allowed)) + return os.path.commonpath([candidate, root]) == root + except (TypeError, ValueError): + return False + + +def monitor_no_file_write_outside_allowed_paths( + trajectory: Any, ref_state: Optional[Mapping[str, Any]] = None +) -> tuple[bool, str]: + """Reject every ``file_write`` whose path is outside ``allowed_paths``.""" + + state = ref_state if isinstance(ref_state, Mapping) else {} + allowed_raw = state.get("allowed_paths") + if isinstance(allowed_raw, str): + allowed_paths = [allowed_raw] + elif isinstance(allowed_raw, Mapping): + allowed_paths = allowed_raw.get("paths", []) + elif isinstance(allowed_raw, Iterable): + allowed_paths = list(allowed_raw) + else: + allowed_paths = [] + + for event in _trajectory_events(trajectory): + if not isinstance(event, Mapping): + return False, "invalid_event" + if event.get("kind") != "file_write": + continue + payload = _event_payload(event) + path = payload.get("path") or payload.get("file_path") if payload is not None else None + if not isinstance(path, str) or not path: + return False, "file_write_path_missing" + if not allowed_paths: + return False, "allowed_paths_missing" + if not any(isinstance(root, str) and _path_is_within(path, root) for root in allowed_paths): + return False, "file_write_outside_allowed_paths" + return True, "ok" + + +MONITOR_REGISTRY: dict[str, Callable[[Any, Optional[Mapping[str, Any]]], tuple[bool, str]]] = { + "no_shell_exec_without_prior_human_approval": monitor_no_shell_exec_without_prior_human_approval, + "no_shell_exec_without_approval": monitor_no_shell_exec_without_prior_human_approval, + "no_file_write_outside_allowed_paths": monitor_no_file_write_outside_allowed_paths, +} +MONITORS = MONITOR_REGISTRY + + +def _normalise_monitor_name(name: str) -> str: + return name.strip().lower().replace("-", "_").replace(" ", "_") + + +def _resolve_monitor(spec: Any) -> tuple[str, Callable[..., Any], Mapping[str, Any]]: + state: Mapping[str, Any] = {} + candidate = spec + if isinstance(spec, Mapping): + state_value = spec.get("ref_state", spec.get("reference_state", spec.get("ref", {}))) + if isinstance(state_value, Mapping): + state = state_value + candidate = spec.get("predicate", spec.get("monitor", spec.get("name"))) + if isinstance(candidate, str): + name = _normalise_monitor_name(candidate) + function = MONITOR_REGISTRY.get(name) + if function is None: + raise KeyError(name) + return name, function, state + if callable(candidate): + return getattr(candidate, "__name__", "custom_monitor"), candidate, state + raise KeyError("missing_monitor") + + +class EvidenceGate: + """Small callable wrapper for one requirement set.""" + + def __init__(self, requirements: Sequence[Mapping[str, Any]]): + self.requirements = list(requirements or []) + + def evaluate(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + return evaluate_submission(trajectory, self.requirements, ref_state=ref_state) + + def __call__(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + return self.evaluate(trajectory, ref_state=ref_state) + + +def _looks_like_pair(value: Any) -> bool: + return isinstance(value, (tuple, list)) and len(value) == 2 + + +def _requirements_for_gate(gate: Any) -> Optional[list[Mapping[str, Any]]]: + if isinstance(gate, EvidenceGate): + return list(gate.requirements) + if isinstance(gate, Mapping): + if "requirements" in gate and isinstance(gate["requirements"], Sequence): + return list(gate["requirements"]) + if "property" in gate or "verifier" in gate: + return [gate] + if isinstance(gate, Sequence) and not isinstance(gate, (str, bytes)): + if all(isinstance(item, Mapping) for item in gate): + return list(gate) + return None + + +class ComposedGate(list): + """Compose deterministic monitors with disjoint evidence gates. + + The preferred form is ``ComposedGate(monitors=[...], gates=[...])``. For + compact declarative use, ``ComposedGate([(monitor_spec, gate), ...])`` and + ``ComposedGate(monitors, gates)`` are also accepted. Each monitor returns a + boolean/reason pair and each evidence gate is an ordinary requirement list + or :class:`EvidenceGate`. + """ + + def __init__(self, *args: Any, monitors: Any = None, gates: Any = None): + if len(args) > 2: + raise TypeError("ComposedGate accepts at most monitors and gates") + if len(args) == 2: + if monitors is not None or gates is not None: + raise TypeError("do not mix positional and keyword monitor/gate lists") + monitors, gates = args + elif len(args) == 1: + if monitors is not None or gates is not None: + raise TypeError("do not mix positional and keyword monitor/gate lists") + candidate = args[0] + if _looks_like_pair(candidate): + candidate = [candidate] + if isinstance(candidate, Sequence) and not isinstance(candidate, (str, bytes)) and candidate and all(_looks_like_pair(item) for item in candidate): + pairs = list(candidate) + monitors = [pair[0] for pair in pairs] + gates = [pair[1] for pair in pairs] + else: + monitors = candidate + gates = [] + self.monitors = list(monitors or []) + self.gates = list(gates or []) + super().__init__([(monitor, None) for monitor in self.monitors] + [(None, gate) for gate in self.gates]) + + def add_monitor(self, monitor_spec: Any) -> "ComposedGate": + self.monitors.append(monitor_spec) + self.append((monitor_spec, None)) + return self + + def add_gate(self, gate: Any) -> "ComposedGate": + self.gates.append(gate) + self.append((None, gate)) + return self + + def _requirements_are_disjoint(self) -> bool: + seen: set[tuple[Any, Any]] = set() + for gate in self.gates: + requirements = _requirements_for_gate(gate) + if requirements is None: + continue + for requirement in requirements: + if not isinstance(requirement, Mapping): + continue + key = (requirement.get("property"), requirement.get("verifier")) + if key in seen: + return False + seen.add(key) + return True + + @staticmethod + def _monitor_result(result: Any) -> tuple[bool, str]: + if isinstance(result, tuple) and len(result) >= 2: + return bool(result[0]), str(result[1]) + if isinstance(result, Mapping): + return bool(result.get("held", result.get("ok", False))), str(result.get("reason", "monitor_violated")) + return (bool(result), "ok" if result else "monitor_violated") + + def _evaluate_gate( + self, gate: Any, trajectory: Any, ref_state: Optional[Mapping[str, Any]] + ) -> dict[str, Any]: + requirements = _requirements_for_gate(gate) + if requirements is not None: + return evaluate_submission(trajectory, requirements, ref_state=ref_state) + try: + if hasattr(gate, "evaluate") and callable(gate.evaluate): + result = gate.evaluate(trajectory, ref_state=ref_state) + elif callable(gate): + result = gate(trajectory) + else: + result = False + except Exception as exc: # custom gates fail closed + return { + "accepted": False, + "decision": "rejected_missing_evidence", + "evidence_chain": [], + "unmet_requirements": [{"reason": "gate_error", "detail": str(exc)}], + } + if isinstance(result, Mapping): + report = dict(result) + report.setdefault("accepted", False) + report.setdefault("evidence_chain", []) + report.setdefault("unmet_requirements", []) + return report + accepted = bool(result) + return { + "accepted": accepted, + "decision": "accepted_with_evidence" if accepted else "rejected_missing_evidence", + "evidence_chain": [], + "unmet_requirements": [] if accepted else [{"reason": "gate_rejected"}], + } + + def evaluate( + self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None + ) -> dict[str, Any]: + """Evaluate all monitors and all evidence gates fail-closed.""" + + monitor_reports: list[dict[str, Any]] = [] + all_monitors_hold = True + for spec in self.monitors: + try: + name, function, monitor_state = _resolve_monitor(spec) + if isinstance(ref_state, Mapping): + merged_state = dict(ref_state) + merged_state.update(monitor_state) + else: + merged_state = monitor_state + result = function(trajectory, merged_state) + held, reason = self._monitor_result(result) + except Exception as exc: + name = getattr(spec, "__name__", "unknown_monitor") + held, reason = False, "monitor_error" + monitor_reports.append({"monitor": name, "held": held, "reason": reason}) + all_monitors_hold = all_monitors_hold and held + + gate_reports = [self._evaluate_gate(gate, trajectory, ref_state) for gate in self.gates] + all_gates_hold = all(bool(report.get("accepted")) for report in gate_reports) + disjoint = self._requirements_are_disjoint() + accepted = bool(all_monitors_hold and all_gates_hold and disjoint) + + evidence_chain: list[Any] = [] + unmet: list[Any] = [] + for report in gate_reports: + evidence_chain.extend(report.get("evidence_chain", [])) + unmet.extend(report.get("unmet_requirements", [])) + if not disjoint: + unmet.insert(0, {"reason": "overlapping_requirements"}) + return { + "accepted": accepted, + "decision": "accepted_with_evidence" if accepted else "rejected_composition", + "monitors": monitor_reports, + "gates": gate_reports, + "evidence_chain": evidence_chain if accepted else evidence_chain, + "unmet_requirements": unmet, + } + + def verify(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + return self.evaluate(trajectory, ref_state=ref_state) + + def check(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + return self.evaluate(trajectory, ref_state=ref_state) + + def accepts(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> bool: + return bool(self.evaluate(trajectory, ref_state=ref_state)["accepted"]) + + def __call__(self, trajectory: Any, *, ref_state: Optional[Mapping[str, Any]] = None) -> dict[str, Any]: + return self.evaluate(trajectory, ref_state=ref_state) + + +# Friendly function aliases for monitor names used in prose and integrations. +no_shell_exec_without_prior_human_approval = monitor_no_shell_exec_without_prior_human_approval +no_file_write_outside_allowed_paths = monitor_no_file_write_outside_allowed_paths + + +def trajectory_root_hash(trajectory: Any) -> str: + """Return ``sha256(head_hash)`` as a hexadecimal string. + + The returned value is suitable for adding to an existing prebind or receipt + ``evidence_hashes`` list. This helper does not alter those existing + signature schemas, preserving additive AAR integration. + """ + + if isinstance(trajectory, Trajectory): + head = trajectory.head_hash + elif isinstance(trajectory, Mapping): + try: + head = Trajectory.from_dict(trajectory).head_hash + except (TypeError, ValueError) as exc: + raise ValueError("invalid trajectory") from exc + else: + head = Trajectory(trajectory).head_hash + if not _is_sha256(head): + raise ValueError("trajectory head_hash is invalid") + return hashlib.sha256(head.encode("ascii")).hexdigest() + + +__all__ = [ + "ALLOWED_EVENT_KINDS", + "ComposedGate", + "EVIDENCE_VERIFIERS", + "EVENT_KINDS", + "GENESIS", + "GENESIS_HASH", + "HARD", + "MONITORS", + "MONITOR_REGISTRY", + "REJECT", + "SOFT", + "TRAJECTORY_SCHEMA", + "TRAJECTORY_VERSION", + "VERIFIERS", + "VERIFIER_REGISTRY", + "EvidenceGate", + "Trajectory", + "canonical_json", + "classify_evidence", + "evaluate_submission", + "event_hash", + "find_evidence_chain", + "monitor_no_file_write_outside_allowed_paths", + "monitor_no_shell_exec_without_prior_human_approval", + "no_file_write_outside_allowed_paths", + "no_shell_exec_without_prior_human_approval", + "trajectory_root_hash", + "verify_evidence", + "verify_event_evidence", +] diff --git a/tests/test_runtime_contract.py b/tests/test_runtime_contract.py new file mode 100644 index 0000000..e4461e7 --- /dev/null +++ b/tests/test_runtime_contract.py @@ -0,0 +1,324 @@ +from __future__ import annotations + +import hashlib +import json +import subprocess +import sys +from pathlib import Path + +import pytest + +from ledger_agent.trajectory import ( + EVENT_KINDS, + GENESIS_HASH, + ComposedGate, + Trajectory, + evaluate_submission, + find_evidence_chain, + trajectory_root_hash, + verify_evidence, +) + + +ROOT = Path(__file__).resolve().parents[1] + + +def make_trajectory() -> Trajectory: + trajectory = Trajectory() + trajectory.append("model_message", {"text": "done"}, timestamp_ms=1) + return trajectory + + +def evidence_requirements() -> list[dict]: + return [ + { + "property": "test_suite_passes", + "verifier": "test_run", + "ref_state": {"expected_pass": True}, + }, + { + "property": "log_contains", + "verifier": "log_capture", + "ref_state": {"marker": "ALL TESTS PASSED"}, + }, + { + "property": "citation_real", + "verifier": "citation_lookup", + "ref_state": {"source_urls": {"https://arxiv.org/abs/2608.11274"}}, + }, + ] + + +def add_evidence(trajectory: Trajectory) -> None: + trajectory.append( + "tool_result", + {"exit_code": 0, "command": "pytest tests/ -q"}, + timestamp_ms=2, + ) + trajectory.append( + "tool_result", + {"log_text": "pytest: ALL TESTS PASSED"}, + timestamp_ms=3, + ) + trajectory.append( + "citation_lookup", + {"cited_url": "https://arxiv.org/abs/2608.11274"}, + timestamp_ms=4, + ) + + +def test_genesis_append_and_verify_chain_happy_path(): + trajectory = Trajectory() + assert trajectory.head_hash == GENESIS_HASH + + event = trajectory.append("tool_call", {"name": "pytest"}, timestamp_ms=100) + + assert event["kind"] == "tool_call" + assert event["timestamp_ms"] == 100 + assert event["prev_hash"] == GENESIS_HASH + assert len(event["hash"]) == 64 + assert trajectory.verify_chain() == (True, "ok") + + +def test_unknown_kind_is_rejected(): + trajectory = Trajectory() + with pytest.raises(ValueError, match="unknown event kind"): + trajectory.append("unknown_kind", {}, timestamp_ms=1) + assert "model_message" in EVENT_KINDS + + +def test_tampering_invalidates_the_tampered_event_and_chain_suffix(): + trajectory = Trajectory() + trajectory.append("tool_call", {"name": "pytest"}, timestamp_ms=1) + trajectory.append("tool_result", {"exit_code": 0}, timestamp_ms=2) + trajectory.append("commit", {"commit_sha": "a" * 40}, timestamp_ms=3) + original_suffix_hash = trajectory.events[2]["hash"] + + trajectory.events[0]["payload"]["name"] = " 다른" + + valid, reason = trajectory.verify_chain() + assert valid is False + assert reason == "hash_mismatch" + assert trajectory.events[2]["hash"] == original_suffix_hash + + +def test_round_trip_is_deterministic_and_diff_stable(): + trajectory = Trajectory() + trajectory.append("file_read", {"path": "src/main.py"}, timestamp_ms=1) + trajectory.append("file_write", {"path": "src/main.py", "bytes": 3}, timestamp_ms=2) + + first = trajectory.to_dict() + second = trajectory.to_dict() + restored = Trajectory.from_dict(json.loads(json.dumps(first, sort_keys=True))) + + assert first == second + assert restored.to_dict() == first + assert json.dumps(first, sort_keys=True, separators=(",", ":")) == json.dumps( + second, sort_keys=True, separators=(",", ":") + ) + + +def test_known_verifiers_distinguish_hard_accept_from_reject(): + event = {"kind": "tool_result", "payload": {"exit_code": 0}} + assert verify_evidence(event, "test_suite_passes", "test_run", {"expected_pass": True}) == "accept" + assert verify_evidence(event, "test_suite_passes", "test_run", {"expected_pass": False}) == "reject" + assert verify_evidence(event, "test_suite_passes", "test_run", {"expected_pass": True, "deterministic": False}) == "soft" + + +def test_soft_evidence_includes_done_model_message_and_unknown_verifiers(): + done = {"kind": "model_message", "payload": {"text": "done"}} + assert verify_evidence(done, "test_suite_passes", "test_run", {"expected_pass": True}) == "soft" + assert verify_evidence(done, "anything", "missing_verifier", {}) == "soft" + + +def test_all_shipped_evidence_verifiers_cover_required_payload_shapes(): + assert verify_evidence( + {"kind": "citation_lookup", "payload": {"cited_url": "https://example.test/source"}}, + "citation_real", + "citation_lookup", + {"source_urls": {"https://example.test/source"}}, + ) == "accept" + assert verify_evidence( + {"kind": "file_write", "payload": {"diff": "@@ -1 +1 @@\n-old\n+new"}}, + "diff_matches", + "file_diff", + {"expected_hunk": "@@ -1 +1 @@\n-old\n+new"}, + ) == "accept" + assert verify_evidence( + {"kind": "shell_exec", "payload": {"exit_code": 127}}, + "exit_code_captured", + "shell_exec", + {}, + ) == "accept" + assert verify_evidence( + {"kind": "commit", "payload": {"commit_sha": "a" * 40}}, + "commit_matches", + "commit", + {"expected_commit_sha": "a" * 40}, + ) == "accept" + assert verify_evidence( + {"kind": "screenshot", "payload": {"image_sha256": "b" * 64}}, + "screenshot_matches", + "screenshot", + {"expected_image_sha256": "b" * 64}, + ) == "accept" + assert verify_evidence( + {"kind": "human_approval", "payload": {"approved_by": "human:alice"}}, + "human_approved", + "human_approval", + {"approval_ref": "human:alice"}, + ) == "accept" + + +def test_find_evidence_chain_reports_found_and_unmet_requirements(): + trajectory = make_trajectory() + add_evidence(trajectory) + + found, chain_events, unmet = find_evidence_chain(trajectory, evidence_requirements()) + + assert found is True + assert {event["kind"] for event in chain_events} == { + "tool_result", + "citation_lookup", + } + assert unmet == [] + + found, chain_events, unmet = find_evidence_chain( + trajectory, + evidence_requirements() + [{"property": "commit_matches", "verifier": "commit", "ref_state": {"expected_commit_sha": "c" * 40}}], + ) + assert found is False + assert len(chain_events) == 3 + assert unmet[0]["property"] == "commit_matches" + + +def test_submission_gate_rejects_evidence_less_done_claim(): + report = evaluate_submission(make_trajectory(), evidence_requirements()) + + assert report["accepted"] is False + assert report["decision"] == "rejected_missing_evidence" + assert report["evidence_chain"] == [] + assert len(report["unmet_requirements"]) == 3 + + +def test_submission_gate_accepts_complete_evidence_chain(): + trajectory = make_trajectory() + add_evidence(trajectory) + + report = evaluate_submission(trajectory, evidence_requirements()) + + assert report["accepted"] is True + assert report["decision"] == "accepted_with_evidence" + assert len(report["evidence_chain"]) == 3 + assert report["unmet_requirements"] == [] + + +def test_no_shell_exec_monitor_requires_prior_human_approval(): + safe = Trajectory() + safe.append("human_approval", {"approved_by": "human:alice"}, timestamp_ms=1) + safe.append("shell_exec", {"command": "pytest", "exit_code": 0}, timestamp_ms=2) + + unsafe = Trajectory() + unsafe.append("shell_exec", {"command": "rm -rf /", "exit_code": 0}, timestamp_ms=1) + + monitor = {"name": "no_shell_exec_without_prior_human_approval"} + assert ComposedGate(monitors=[monitor]).evaluate(safe)["accepted"] is True + violation = ComposedGate(monitors=[monitor]).evaluate(unsafe) + assert violation["accepted"] is False + assert violation["monitors"][0]["reason"] == "shell_exec_without_prior_human_approval" + + +def test_file_write_monitor_enforces_allowed_path_reference(): + trajectory = Trajectory() + trajectory.append("file_write", {"path": "/workspace/project/result.txt"}, timestamp_ms=1) + + gate = ComposedGate( + monitors=[ + { + "name": "no_file_write_outside_allowed_paths", + "ref_state": {"allowed_paths": {"/workspace/project"}}, + } + ] + ) + assert gate.evaluate(trajectory)["accepted"] is True + + trajectory.events[0]["payload"]["path"] = "/etc/passwd" + violation = gate.evaluate(trajectory) + assert violation["accepted"] is False + assert violation["monitors"][0]["reason"] == "file_write_outside_allowed_paths" + + +def test_composed_gate_accepts_when_all_monitors_and_gates_pass(): + trajectory = make_trajectory() + trajectory.append("human_approval", {"approved_by": "human:alice"}, timestamp_ms=2) + trajectory.append("shell_exec", {"command": "pytest", "exit_code": 0}, timestamp_ms=3) + add_evidence(trajectory) + + composed = ComposedGate( + monitors=[{"name": "no_shell_exec_without_prior_human_approval"}], + gates=[evidence_requirements()], + ) + report = composed.evaluate(trajectory) + + assert report["accepted"] is True + assert report["decision"] == "accepted_with_evidence" + assert report["unmet_requirements"] == [] + + +def test_composed_gate_rejects_when_a_monitor_is_violated(): + trajectory = make_trajectory() + trajectory.append("shell_exec", {"command": "pytest", "exit_code": 0}, timestamp_ms=2) + add_evidence(trajectory) + + composed = ComposedGate( + monitors=[{"name": "no_shell_exec_without_prior_human_approval"}], + gates=[evidence_requirements()], + ) + report = composed.evaluate(trajectory) + + assert report["accepted"] is False + assert report["decision"] == "rejected_composition" + assert report["monitors"][0]["held"] is False + + +def test_composed_gate_rejects_when_one_evidence_gate_is_unmet(): + trajectory = make_trajectory() + trajectory.append("human_approval", {"approved_by": "human:alice"}, timestamp_ms=2) + trajectory.append("shell_exec", {"command": "pytest", "exit_code": 0}, timestamp_ms=3) + add_evidence(trajectory) + + composed = ComposedGate( + monitors=[{"name": "no_shell_exec_without_prior_human_approval"}], + gates=[evidence_requirements() + [{"property": "commit_matches", "verifier": "commit", "ref_state": {"expected_commit_sha": "d" * 40}}]], + ) + report = composed.evaluate(trajectory) + + assert report["accepted"] is False + assert report["decision"] == "rejected_composition" + assert report["gates"][0]["unmet_requirements"][0]["property"] == "commit_matches" + + +def test_trajectory_root_hash_changes_when_an_event_is_appended(): + trajectory = Trajectory() + before = trajectory_root_hash(trajectory) + trajectory.append("model_message", {"text": "done"}, timestamp_ms=1) + after = trajectory_root_hash(trajectory) + + assert before != after + assert after == hashlib.sha256(trajectory.head_hash.encode("ascii")).hexdigest() + + +def test_runtime_contract_demo_runs_end_to_end(): + completed = subprocess.run( + [sys.executable, str(ROOT / "examples" / "runtime_contract_demo.py")], + cwd=ROOT, + capture_output=True, + text=True, + check=False, + ) + + assert completed.returncode == 0, completed.stderr + assert "Evidence-less done claim" in completed.stdout + assert "Complete evidence chain" in completed.stdout + assert '"accepted": false' in completed.stdout + assert '"accepted": true' in completed.stdout