From 030b2dd7ae2e85662354d23ac17de04f53391320 Mon Sep 17 00:00:00 2001 From: Colin Son Date: Fri, 11 Sep 2026 13:46:28 -0500 Subject: [PATCH 1/3] feat: Phase F3 declared snapshot/restore capability with measured-fork branching and honest refusal - BranchingSupport ladder (unspecified < declared < prefix_replay < measured_fork; not_supported is a refusal state, never a requestable mechanism) on WorldCapabilities, riding the same positional task-vs-adapter cross-check as every other gate via gates(). - measure_branch_support: two independent forked rollouts of one seed grade the ladder; measured_fork only from byte-equal canonical digests, prefix_replay only within an explicit finite tolerance, otherwise not_supported with the first difference. - branch_from: every arm re-runs the shared prefix on a fresh instance; a prefix that fails to replay is refused with zero candidate comparisons - a seed alone must not be asserted to restore a full mutable state. BranchEvidence keeps claimed separate from support so a loud declaration cannot overwrite a weak measurement. - require_branch_support: fail-closed gate; a bare declaration may drive arms but can never certify equality; lumen-gym declares prefix_replay (the one mechanism the pinned stack proves), no built-in claims measured_fork. --- src/or_audit/eval/__init__.py | 14 + src/or_audit/eval/branching.py | 589 +++++++++++++++++++++++++++++++++ src/or_audit/eval/worlds.py | 90 ++++- tests/test_branching.py | 495 +++++++++++++++++++++++++++ 4 files changed, 1186 insertions(+), 2 deletions(-) create mode 100644 src/or_audit/eval/branching.py create mode 100644 tests/test_branching.py diff --git a/src/or_audit/eval/__init__.py b/src/or_audit/eval/__init__.py index c9b6093..99b67b3 100644 --- a/src/or_audit/eval/__init__.py +++ b/src/or_audit/eval/__init__.py @@ -15,6 +15,12 @@ ) from or_audit.eval.agent import AgentPackage from or_audit.eval.bind import assert_bind +from or_audit.eval.branching import ( + BranchEvidence, + branch_from, + measure_branch_support, + require_branch_support, +) from or_audit.eval.cartesian import CartesianManifest, replay_cartesian, run_cartesian_job from or_audit.eval.contracts import ( CapabilitySpec, @@ -98,9 +104,11 @@ from or_audit.eval.vector import TrialVector, project from or_audit.eval.worlds import ( WORLD_KIND_ENTRY_POINT_GROUP, + BranchingSupport, DeterminismClass, WorldCapabilities, WorldKindSpec, + branching_at_least, determinism_at_least, list_world_kinds, require_world_kind, @@ -117,6 +125,8 @@ "AttestationLevel", "BaseModalityAdapter", "BaseSimulationBridge", + "BranchEvidence", + "BranchingSupport", "CapabilitySpec", "CartesianManifest", "DatasetSpec", @@ -172,6 +182,8 @@ "assemble_job_result", "assert_bind", "bootstrap_mean_ci", + "branch_from", + "branching_at_least", "builtin_random_agent", "clear_adapter_registry", "clear_simulation_registry", @@ -193,6 +205,7 @@ "make_isaac_bridge", "make_sofa_bridge", "make_warp_bridge", + "measure_branch_support", "preprocess_observation", "project", "reconstitute_trial_vector", @@ -202,6 +215,7 @@ "replay_cartesian", "replay_job", "require_adapter", + "require_branch_support", "require_simulation_engine", "require_world_kind", "reset_default_simulation_engines", diff --git a/src/or_audit/eval/branching.py b/src/or_audit/eval/branching.py new file mode 100644 index 0000000..5d41898 --- /dev/null +++ b/src/or_audit/eval/branching.py @@ -0,0 +1,589 @@ +"""Branching support for counterfactual evaluation: probe, gate, and refusal. + +Phase F3. A counterfactual interface asks what would have happened had the +policy acted differently at some observed step. Answering it honestly needs a +world that can *fork*: replay the shared prefix exactly, then diverge the +candidates from a common state. The kernel ships no verified +``snapshot``/``restore`` primitive — the simulation bridges' ``snapshot()`` +methods are read-only reports of live state, not restorable checkpoints — so +this module measures the one mechanism the pinned stack actually demonstrates +(deterministic prefix replay, see ``tests/test_lumen_branch.py``) and refuses, +rather than assumes, everything else. + +Three rules hold it together, and they are the rules conformance applies to +determinism: + +* **recorded, never assumed.** :attr:`BranchingSupport.MEASURED_FORK` is + earned only when two independent forked rollouts of one seed produce + byte-identical canonical digests; +* **a seed is not a state.** :func:`branch_from` re-runs the shared prefix on + every arm and verifies each arm's prefix segment digests identically to the + reference replay before comparing anything. A prefix that fails to replay + means the world cannot fork at all, and the whole branch set is reported as + a refusal (:attr:`BranchingSupport.NOT_SUPPORTED`) with no candidate + comparisons — not a quiet side-by-side of traces from divergent states; +* **a declaration is not a measurement.** :func:`require_branch_support` fails + closed when the mechanism a caller needs is not backed by the world's + declared support, including when a bare ``declared`` claim is all there is + and the caller needs equality-certifying evidence. +""" + +from __future__ import annotations + +import contextlib +import math +from collections.abc import Callable, Sequence +from typing import Any, Self + +from pydantic import BaseModel, ConfigDict, model_validator + +from or_audit.audit.canonical import digest as canonical_digest +from or_audit.errors import TaskContractError +from or_audit.eval.gym_world import GymEnv, jsonable +from or_audit.eval.worlds import BranchingSupport, WorldCapabilities, branching_at_least + +#: Minimum support a caller needs to *drive* side-by-side arms from a shared +#: prefix: at least an authoritative declaration that prefix replay works. +DRIVING_MECHANISM = BranchingSupport.DECLARED +#: Support a caller needs to *certify* branch equivalence (e.g. publishing a +#: counterfactual claim): only a measured, byte-equal fork pair qualifies. +FORK_EQUALITY_MECHANISM = BranchingSupport.MEASURED_FORK +#: Mechanisms a caller may demand of a world. ``not_supported``/``unspecified`` +#: are states a world reports, never things one can require of it. +REQUESTABLE_MECHANISMS = frozenset( + { + BranchingSupport.DECLARED, + BranchingSupport.PREFIX_REPLAY, + BranchingSupport.MEASURED_FORK, + } +) + +#: A zero-argument factory for one fresh environment instance. Deliberately +#: not :data:`or_audit.eval.gym_world.GymFactory` (which takes a task): a +#: probe must not be able to mutate shared world state between arms, so each +#: arm is built by calling this afresh. +BranchEnvFactory = Callable[[], GymEnv] + + +class _Frozen(BaseModel): + model_config = ConfigDict(frozen=True, extra="forbid") + + +class BranchEvidence(_Frozen): + """What a prefix-replay probe actually observed. Mirrors DeterminismEvidence. + + ``support`` is what the probe *earned*, kept separate from the strongest + level either the task or its adapter *claimed*, so a claim the probe does + not back stays visible as ``claimed > support`` in the evidence instead of + being silently overwritten by it. + """ + + #: Level the probe earned from the rollouts it ran. + support: BranchingSupport + #: Strongest level the task or its installed adapter claims; ``unspecified`` + #: means nothing was claimed and there is nothing to violate. + claimed: BranchingSupport = BranchingSupport.UNSPECIFIED + tolerance: float = 0.0 + #: Every replay of the shared prefix produced equal canonical digests. + prefix_equal: bool + #: The two independent forked rollouts of one seed produced byte-identical + #: canonical digests. The fork pair *is* the twin replay pair. + identical_fork_digests: bool + #: True when the candidate arms were distinguishable at all: at least two + #: arms digested differently. ``False`` on distinct plans means the + #: intervention changed nothing and the comparison measured nothing. + candidate_diverged: bool + #: Largest absolute float delta between any two repeated rollouts. + max_float_delta: float = 0.0 + #: First difference tolerance does not excuse; ``""`` when none. + first_difference: str = "" + + @model_validator(mode="after") + def _tolerance_is_a_finite_measurement(self) -> Self: + """``inf``/``nan`` tolerance would certify two unrelated traces as equal. + + ``delta <= inf`` holds for every pair of floats, so an infinite + tolerance is a comparison that observed nothing while still reporting a + passing measurement. Neither value is a JSON number, so the evidence + would serialize as non-standard ``Infinity``/``NaN``. + """ + if not math.isfinite(self.tolerance) or self.tolerance < 0: + raise TaskContractError( + f"branching tolerance must be a finite non-negative float, got {self.tolerance!r}" + ) + return self + + @model_validator(mode="after") + def _earned_fork_needs_a_real_fork_pair(self) -> Self: + """``measured_fork`` is a claim about byte equality of two rollouts. + + Reporting it without the matching digest evidence would let an + unmeasured world inherit the strongest level in the schema — the exact + failure mode the determinism ladder exists to prevent. + """ + if self.support is BranchingSupport.MEASURED_FORK and not ( + self.prefix_equal and self.identical_fork_digests + ): + raise TaskContractError( + "cannot report 'measured_fork' without prefix_equal=True and " + "identical_fork_digests=True: the level is earned from two " + "byte-identical forked rollouts or not at all" + ) + return self + + +def _claimed_support( + capabilities: WorldCapabilities | None, + declared: BranchingSupport | None, +) -> BranchingSupport: + """Strongest standing claim, from the task declaration and the adapter alike. + + Both are claims a published artifact carries, and the probe must be able + to contradict either, so it is the *strongest* claim that gets compared + against the measurement — the ``_declared_determinism`` rule from + conformance. + """ + strongest = BranchingSupport.UNSPECIFIED + for claim in (capabilities.branching if capabilities is not None else None, declared): + if claim is not None and not branching_at_least(strongest, claim): + strongest = claim + return strongest + + +def _normalize_action(action: Any) -> Any: + """Reduce an action to a hashable, digest-stable form. + + numpy arrays are accepted by gym ``step`` but hash by identity and digest + through ``repr``, so two element-equal arrays would compare unequal. They + are canonicalized through ``tolist`` here so arm plans and digests see + content, not object identity. + """ + tolist = getattr(action, "tolist", None) + if callable(tolist): + return tolist() + return action + + +def _float_delta(left: Any, right: Any) -> float: + """Largest absolute float delta between two canonically-matched values. + + Walks mappings and sequences pairwise; non-numeric leaves contribute + nothing here (a structural difference is caught by the digest comparison, + which is stricter than any tolerance can be). Bools are skipped: a flipped + ``terminated`` flag is already a digest-level difference. + """ + if isinstance(left, bool) or isinstance(right, bool): + return 0.0 + if isinstance(left, float | int) and isinstance(right, float | int): + return abs(float(left) - float(right)) + if isinstance(left, dict) and isinstance(right, dict): + worst = 0.0 + for key in left.keys() & right.keys(): + worst = max(worst, _float_delta(left[key], right[key])) + return worst + if isinstance(left, (list, tuple)) and isinstance(right, (list, tuple)): + worst = 0.0 + for a, b in zip(left, right, strict=False): + worst = max(worst, _float_delta(a, b)) + return worst + return 0.0 + + +def _trace_delta(left: Sequence[dict[str, Any]], right: Sequence[dict[str, Any]]) -> float: + """Max float delta across two aligned rollout traces, step by step.""" + worst = 0.0 + for a, b in zip(left, right, strict=False): + worst = max(worst, _float_delta(a, b)) + return worst + + +def _rollout( + env: GymEnv, + *, + seed: int, + options: dict[str, Any] | None, + actions: Sequence[Any], + max_steps: int, +) -> list[dict[str, Any]]: + """Drive ``env`` with ``actions`` and return the per-step trace. + + Deliberately does not route through :func:`or_audit.eval.gym_world.run_gym_episode`: + that driver injects harness perturbations and stamps policy-side + bookkeeping meant for *scoring*. A probe must observe the engine's own + bytes — anything sanitized before the digest is measured can hide the very + nondeterminism being tested, and an injected perturbation would compare + perturbation schedules instead of worlds. The reset record is part of the + trace, because hidden state chosen at reset is exactly what "a seed + restores the state" would smuggle past a prefix-only comparison. + Termination honours ``terminated``/``truncated`` the way the runner does. + """ + _, reset_info = env.reset(seed=seed, options=options) + trace: list[dict[str, Any]] = [{"reset_info": jsonable(reset_info), "seed": seed}] + for index, action in enumerate(actions): + if index >= max_steps: + break + _, reward, terminated, truncated, info = env.step(action) + trace.append( + { + "action": jsonable(_normalize_action(action)), + "reward": jsonable(reward), + "terminated": bool(terminated), + "truncated": bool(truncated), + "info": jsonable(info) if isinstance(info, dict) else jsonable({"info": info}), + } + ) + if terminated or truncated: + break + return trace + + +def _fresh_rollout( + env_factory: BranchEnvFactory, + *, + seed: int, + options: dict[str, Any] | None, + actions: Sequence[Any], + max_steps: int, + label: str, +) -> list[dict[str, Any]]: + """One rollout on a brand-new environment instance. + + Every arm gets its own instance: an engine with hidden mutable state that + shared an instance across arms would let an earlier arm's stepping leak + into a later arm's trace, which digests as *support for* branching that + was never actually provided. + """ + try: + env = env_factory() + except BaseException as exc: + raise TaskContractError( + f"branching probe could not construct the {label} environment: {exc!r}" + ) from exc + try: + return _rollout(env, seed=seed, options=options, actions=actions, max_steps=max_steps) + except TaskContractError: + raise + except BaseException as exc: + raise TaskContractError( + f"branching probe could not complete the {label} rollout: {exc!r}" + ) from exc + finally: + close = getattr(env, "close", None) + if callable(close): + with contextlib.suppress(BaseException): + close() + + +def _first_difference(first: Sequence[dict[str, Any]], second: Sequence[dict[str, Any]]) -> str: + """Canonical summary of the first step where two traces part company.""" + for index, (a, b) in enumerate(zip(first, second, strict=False)): + if canonical_digest(a) != canonical_digest(b): + if "action" in a: + return f"step {index} action {a['action']!r}" + return f"reset record of step {index}" + if len(first) != len(second): + return f"traces end at different steps ({len(first)} vs {len(second)})" + return "traces differ" + + +def _compare_fork_pair( + first: Sequence[dict[str, Any]], + second: Sequence[dict[str, Any]], + *, + claimed: BranchingSupport, + tolerance: float, +) -> BranchEvidence: + """Grade one twin pair of rollouts into evidence; shared by both probes.""" + byte_equal = canonical_digest(list(first)) == canonical_digest(list(second)) + delta = _trace_delta(first, second) + if byte_equal: + return BranchEvidence( + support=BranchingSupport.MEASURED_FORK, + claimed=claimed, + tolerance=tolerance, + prefix_equal=True, + identical_fork_digests=True, + candidate_diverged=False, + max_float_delta=0.0, + ) + if delta <= tolerance and len(first) == len(second): + # Byte-different but every float within tolerance: faithful enough to + # drive arms, not enough to certify equality, so it stops one rung + # below the top of the ladder. + return BranchEvidence( + support=BranchingSupport.PREFIX_REPLAY, + claimed=claimed, + tolerance=tolerance, + prefix_equal=True, + identical_fork_digests=False, + candidate_diverged=False, + max_float_delta=delta, + first_difference=f"twin digests differ within tolerance {tolerance:g}", + ) + return BranchEvidence( + support=BranchingSupport.NOT_SUPPORTED, + claimed=claimed, + tolerance=tolerance, + prefix_equal=False, + identical_fork_digests=False, + candidate_diverged=False, + max_float_delta=delta, + first_difference=( + f"a seed alone does not restore this world's state: twin rollouts diverged " + f"at {_first_difference(first, second)} beyond tolerance {tolerance:g}" + ), + ) + + +def require_branch_support( + capabilities: WorldCapabilities | None, mechanism: BranchingSupport +) -> None: + """Fail closed unless the world's declared support backs ``mechanism``. + + This is the gate every counterfactual-branching caller passes through + before forking anything. It refuses, in order: a caller demanding a + mechanism that is not a mechanism at all (``not_supported``/``unspecified`` + are states a world reports, never things one can require of one); a + missing capability source; and any support weaker than what is asked. So + a lone :attr:`BranchingSupport.DECLARED` claim may drive arms + (:data:`DRIVING_MECHANISM`) but can never certify equality + (:data:`FORK_EQUALITY_MECHANISM`), which needs the measured rung. + """ + if mechanism not in REQUESTABLE_MECHANISMS: + raise TaskContractError( + f"{mechanism.value!r} is not a branching mechanism a caller can require: " + f"ask for one of {', '.join(sorted(item.value for item in REQUESTABLE_MECHANISMS))}" + ) + if capabilities is None: + raise TaskContractError( + f"branching mechanism {mechanism.value!r} needs capability declarations, but the " + "world has none: install the world adapter or declare " + "[environment.capabilities].branching in the task package" + ) + support = capabilities.branching + if support is BranchingSupport.DECLARED and mechanism is FORK_EQUALITY_MECHANISM: + raise TaskContractError( + f"{mechanism.value!r} branching needs measured evidence, but the world " + f"only declares {support.value!r}: an author's assertion may drive arms, " + "yet it can never certify that two branches were equal. Run " + "measure_branch_support against this adapter and record the result, or " + "serve the comparison without an equality claim" + ) + if not branching_at_least(support, mechanism): + raise TaskContractError( + f"world cannot back {mechanism.value!r} branching: it declares " + f"{support.value!r}, which is weaker. Refusing to compare branches on a world " + "that may not reproduce the shared prefix — any such comparison would be " + "fabricated evidence" + ) + + +def measure_branch_support( + env_factory: BranchEnvFactory, + *, + seed: int, + prefix_actions: Sequence[Any], + tolerance: float = 0.0, + options: dict[str, Any] | None = None, + max_steps: int = 10_000, + capabilities: WorldCapabilities | None = None, + declared: BranchingSupport | None = None, +) -> BranchEvidence: + """Measure whether one world can fork a rollout at a shared prefix. + + Runs the same ``prefix_actions`` from the same ``seed`` twice, each on a + fresh instance, and grades what it saw: byte-identical twins earn + :attr:`BranchingSupport.MEASURED_FORK`; agreement only inside + ``tolerance`` earns :attr:`BranchingSupport.PREFIX_REPLAY` (faithful enough + to drive arms, not to certify equality); anything else reports + :attr:`BranchingSupport.NOT_SUPPORTED` with the first difference. + ``claimed`` carries the strongest standing claim so ``claimed > support`` + is visible to callers. Deterministic engines earn the top rung; nobody is + granted it. + """ + if not math.isfinite(tolerance) or tolerance < 0: + raise TaskContractError(f"tolerance must be a finite non-negative float, got {tolerance!r}") + if not callable(env_factory): + raise TaskContractError("env_factory must be callable (a zero-argument factory)") + prefix = [_normalize_action(action) for action in prefix_actions] + first = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=prefix, + max_steps=max_steps, + label="fork", + ) + second = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=prefix, + max_steps=max_steps, + label="twin", + ) + return _compare_fork_pair( + first, + second, + claimed=_claimed_support(capabilities, declared), + tolerance=tolerance, + ) + + +def branch_from( + env_factory: BranchEnvFactory, + *, + seed: int, + prefix_actions: Sequence[Any], + candidates: Sequence[Sequence[Any]], + tolerance: float = 0.0, + options: dict[str, Any] | None = None, + max_steps: int = 10_000, + capabilities: WorldCapabilities | None = None, + declared: BranchingSupport | None = None, +) -> dict[str, Any]: + """Drive counterfactual candidates from one shared prefix, or refuse. + + For each candidate an arm re-runs the shared prefix on a fresh instance and + continues with the candidate's own actions, and every arm is rolled twice + so its own reproducibility is measured, not assumed. Before any candidate + outcome is trusted, each arm's prefix segment must digest identically to + the reference replay of that prefix — **a seed alone does not restore a + full mutable state**, so an arm whose prefix fails to replay was branched + from a different world state than it claims. When that happens the whole + branch set is refused with :attr:`BranchingSupport.NOT_SUPPORTED` + evidence and *no* candidate comparisons are reported, rather than quietly + comparing traces from divergent states. + + Returns a plain, JSON-safe dict:: + + {"supported": bool, "refused": str, "evidence": BranchEvidence, + "arms": [{"index", "plan", "digest", "steps", "replayed_equal", + "prefix_matches_reference", "max_float_delta"}]} + + ``candidate_diverged`` on the evidence is the honesty check on the whole + experiment: distinct candidate plans that all digest equal means the + intervention changed nothing and the comparison measured nothing. + """ + if not callable(env_factory): + raise TaskContractError("env_factory must be callable (a zero-argument factory)") + if not candidates: + raise TaskContractError("branch_from needs at least one candidate action sequence") + if not math.isfinite(tolerance) or tolerance < 0: + raise TaskContractError(f"tolerance must be a finite non-negative float, got {tolerance!r}") + prefix = [_normalize_action(action) for action in prefix_actions] + claimed = _claimed_support(capabilities, declared) + + # The fork pair: two independent instances replaying the same prefix from + # the same seed. Their equality is what every arm's branch point is graded + # against, so it is established before a single candidate runs. + reference = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=prefix, + max_steps=max_steps, + label="reference", + ) + twin = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=prefix, + max_steps=max_steps, + label="reference-twin", + ) + fork = _compare_fork_pair(reference, twin, claimed=claimed, tolerance=tolerance) + if not fork.prefix_equal: + return { + "supported": False, + "refused": ( + "the shared prefix did not replay identically on two independent instances: " + "a seed alone does not restore this world's state, so counterfactual " + "branches from it would compare trajectories that already diverged before " + "the branch point" + ), + "evidence": fork, + "arms": [], + } + + reference_digest = canonical_digest(reference) + prefix_len = len(reference) + arms: list[dict[str, Any]] = [] + digests: list[str] = [] + worst = fork.max_float_delta + for index, candidate in enumerate(candidates): + plan = prefix + [_normalize_action(action) for action in candidate] + label = f"candidate-{index}" + first_pass = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=plan, + max_steps=max_steps, + label=f"{label} (a)", + ) + second_pass = _fresh_rollout( + env_factory, + seed=seed, + options=options, + actions=plan, + max_steps=max_steps, + label=f"{label} (b)", + ) + arm_prefix = first_pass[:prefix_len] + if canonical_digest(list(arm_prefix)) != reference_digest: + delta = max(worst, _trace_delta(arm_prefix, reference)) + return { + "supported": False, + "refused": ( + f"candidate {index} replayed the shared prefix differently from the " + "reference rollout, so its branch point is not the state the reference " + "arm had; refusing the comparison instead of reporting a fabricated " + "counterfactual" + ), + "evidence": BranchEvidence( + support=BranchingSupport.NOT_SUPPORTED, + claimed=claimed, + tolerance=tolerance, + prefix_equal=False, + identical_fork_digests=fork.identical_fork_digests, + candidate_diverged=False, + max_float_delta=delta, + first_difference=( + f"candidate {index} diverged from the reference prefix before the " + "branch point" + ), + ), + "arms": [], + } + replayed_equal = canonical_digest(first_pass) == canonical_digest(second_pass) + replay_delta = _trace_delta(first_pass, second_pass) + worst = max(worst, replay_delta) + digest = canonical_digest(first_pass) + digests.append(digest) + arms.append( + { + "index": index, + "plan": jsonable(plan), + "digest": digest, + "steps": len(first_pass), + "replayed_equal": replayed_equal, + "prefix_matches_reference": True, + "max_float_delta": replay_delta, + } + ) + + everything_exact = fork.identical_fork_digests and all(arm["replayed_equal"] for arm in arms) + evidence = BranchEvidence( + support=( + BranchingSupport.MEASURED_FORK if everything_exact else BranchingSupport.PREFIX_REPLAY + ), + claimed=claimed, + tolerance=tolerance, + prefix_equal=True, + identical_fork_digests=everything_exact, + candidate_diverged=len(set(digests)) > 1, + max_float_delta=worst, + ) + return {"supported": True, "refused": "", "evidence": evidence, "arms": arms} diff --git a/src/or_audit/eval/worlds.py b/src/or_audit/eval/worlds.py index f17bd2a..621de3b 100644 --- a/src/or_audit/eval/worlds.py +++ b/src/or_audit/eval/worlds.py @@ -27,9 +27,9 @@ from collections.abc import Callable from enum import StrEnum from pathlib import Path -from typing import Annotated, Any +from typing import Annotated, Any, Self -from pydantic import BaseModel, ConfigDict, StringConstraints, field_validator +from pydantic import BaseModel, ConfigDict, StringConstraints, field_validator, model_validator from or_audit.errors import TaskContractError from or_audit.eval.enums import WorldKind @@ -77,6 +77,49 @@ def determinism_at_least(measured: DeterminismClass, declared: DeterminismClass) return _DETERMINISM_STRENGTH[measured] >= _DETERMINISM_STRENGTH[declared] +class BranchingSupport(StrEnum): + """How a branch at a shared prefix is backed. Ordered by strength. + + :data:`_BRANCHING_STRENGTH` is the order. The kernel has never verified a + ``snapshot/restore`` round-trip on any adapter, so no level here is named + for one: the only claimed mechanisms are one the pinned stack proves + (:attr:`PREFIX_REPLAY`, see ``tests/test_lumen_branch.py``) and one a + measurement can earn (:attr:`MEASURED_FORK`, see + :func:`or_audit.eval.branching.measure_branch_support`). + """ + + UNSPECIFIED = "unspecified" + #: The adapter *author* asserts the fork is exact. A declaration, not a + #: measurement: it earns no branch-equivalence claim on its own. + DECLARED = "declared" + #: Replaying the same seed and the same action prefix reproduces the + #: prefix — the mechanism the pinned Lumen stack demonstrates. + PREFIX_REPLAY = "prefix_replay" + #: Two forks of one seed measured byte-identical. The only level that may + #: certify branch equivalence, and only from attached evidence. + MEASURED_FORK = "measured_fork" + #: The world cannot replay a prefix at all; counterfactual branches on it + #: must be refused, not silently compared. + NOT_SUPPORTED = "not_supported" + + +#: Rank for each :class:`BranchingSupport`; absent means unsupported. Higher +#: is stronger, so a measurement can never be talked into a claim a louder +#: declaration made — the :data:`_DETERMINISM_STRENGTH` rule again. +_BRANCHING_STRENGTH: dict[BranchingSupport, int] = { + BranchingSupport.NOT_SUPPORTED: -1, + BranchingSupport.UNSPECIFIED: 0, + BranchingSupport.DECLARED: 1, + BranchingSupport.PREFIX_REPLAY: 2, + BranchingSupport.MEASURED_FORK: 3, +} + + +def branching_at_least(actual: BranchingSupport, required: BranchingSupport) -> bool: + """Whether ``actual`` branching support is at least as strong as ``required``.""" + return _BRANCHING_STRENGTH[actual] >= _BRANCHING_STRENGTH[required] + + class _Frozen(BaseModel): model_config = ConfigDict(frozen=True, extra="forbid") @@ -105,6 +148,38 @@ class WorldCapabilities(_Frozen): #: ``WorldSpec.metrics_only`` and verified per env by the conformance #: suite's gate-state availability check (§2.2). determinism_class: DeterminismClass = DeterminismClass.UNMEASURED + #: How a counterfactual branch at a shared prefix is backed for this world + #: kind. ``unspecified`` until an adapter declares a mechanism or a + #: measurement earns one, because the kernel has never verified a + #: ``snapshot/restore`` round-trip on any adapter it ships: the bridges' + #: ``snapshot()`` methods are read-only reports of live state, not + #: restorable checkpoints. A world whose forkability is unproven must stay + #: ``unspecified`` — the :attr:`DeterminismClass.UNMEASURED` rule again. + branching: BranchingSupport = BranchingSupport.UNSPECIFIED + + @model_validator(mode="after") + def _branching_needs_a_stepped_world(self) -> Self: + """Every branching mechanism this kernel knows replays a live episode. + + ``prefix_replay`` re-steps a shared action prefix and ``measured_fork`` + compares two such replays, so both are meaningless on a world the + runner cannot step. A dataset-backed world (frame source, recorded + counterfactual) claiming either would be a contradiction that silently + upgrades its own eligibility, so it is refused at construction. + """ + if self.branching is not BranchingSupport.UNSPECIFIED and not self.closed_loop: + raise TaskContractError( + f"branching support {self.branching.value!r} requires closed_loop=True: " + "forking a trajectory needs a world the runner can step, and a " + "dataset-backed world has no state to branch from" + ) + if self.branching is BranchingSupport.MEASURED_FORK and not self.physics: + raise TaskContractError( + "branching support 'measured_fork' requires physics=True: a forked " + "rollout that carries no dynamics has no fork to measure, so the " + "measurement could not certify anything" + ) + return self def gates(self) -> tuple[bool, ...]: """The eligibility flags a declaration must not overstate.""" @@ -115,6 +190,11 @@ def gates(self) -> tuple[bool, ...]: self.requires_gym_id, self.requires_world_pin, self.requires_contract, + # Branching rides the same positional cross-check as every other + # gate: a task may not declare a fork mechanism the installed + # adapter withholds, and the strength itself is compared + # separately by ``require_branch_support``. + self.branching is not BranchingSupport.UNSPECIFIED, ) @@ -204,6 +284,12 @@ def adapter_identity(factory: Callable[..., Any]) -> tuple[str, str]: closed_loop=True, requires_gym_id=True, requires_world_pin=True, + # The only fork mechanism the pinned stack actually demonstrates: + # identical seed plus identical action prefix reproduces identical + # trajectories (tests/test_lumen_branch.py). ``measured_fork`` stays + # unclaimed here — that rung is earned per adapter at run time by + # ``measure_branch_support``, never declared from a table. + branching=BranchingSupport.PREFIX_REPLAY, ), WorldKind.LUMEN_REPLAY: WorldCapabilities(physics=True, closed_loop=True), WorldKind.GYM: WorldCapabilities( diff --git a/tests/test_branching.py b/tests/test_branching.py new file mode 100644 index 0000000..82ce150 --- /dev/null +++ b/tests/test_branching.py @@ -0,0 +1,495 @@ +"""Phase F3: declared branching capability, measured-fork proof, honest refusal. + +Every environment here is a deterministic scripted fake — no external stack — +so the probes' verdicts are reproducible on CI. The suite pins the four +contract rules: + +* ``measure_branch_support`` earns ``measured_fork`` only from byte-equal twin + rollouts, ``prefix_replay`` only from within-tolerance agreement, and + ``not_supported`` whenever the twin pair diverges; +* ``branch_from`` verifies every arm's prefix segment against the reference + replay before comparing anything, and a world that cannot replay its own + prefix is refused outright with **no candidate comparisons reported** — + the refusal, not the comparison, is the deliverable; +* ``require_branch_support`` fails closed: a bare ``declared`` claim may drive + arms but can never certify equality, and ``not_supported``/``unspecified`` + are states a world reports, not mechanisms a caller may require; +* ``WorldCapabilities.branching`` rides the same task-vs-adapter cross-check + as every other gate, so a task cannot grant itself a fork mechanism its + installed adapter withholds. +""" + +from __future__ import annotations + +import json +import math +from typing import Any, ClassVar, cast + +import pytest + +from or_audit.errors import TaskContractError +from or_audit.eval import worlds as _worlds +from or_audit.eval.branching import ( + DRIVING_MECHANISM, + FORK_EQUALITY_MECHANISM, + BranchEvidence, + branch_from, + measure_branch_support, + require_branch_support, +) +from or_audit.eval.enums import WorldKind +from or_audit.eval.worlds import ( + BUILTIN_WORLD_CAPABILITIES, + BranchingSupport, + WorldCapabilities, + WorldKindSpec, + branching_at_least, + register_world_kind, + resolve_world_capabilities, +) + + +def _score(history: list[tuple[float, ...]]) -> float: + total = 0.0 + for i, action in enumerate(history): + for k, value in enumerate(action): + total += (i + 1) * (k + 1) * value + return total + + +class DeterministicEnv: + """Info is a pure function of (seed, action history): replayable exactly.""" + + def __init__(self) -> None: + self.history: list[tuple[float, ...]] = [] + + def reset(self, *, seed: int | None = None, options: dict[str, Any] | None = None): + self.seed = seed + self.history = [] + return {"x": float(seed or 0)}, {"origin": seed} + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + value = _score(self.history) + return {"x": value}, value * 0.1, False, False, {"h": value} + + def close(self) -> None: + pass + + +class WobblyEnv(DeterministicEnv): + """Reset draws hidden state from a global counter: a seed does NOT restore it.""" + + counter: ClassVar[list[int]] = [0] + + def reset(self, *, seed: int | None = None, options: dict[str, Any] | None = None): + WobblyEnv.counter[0] += 1 + self.bias = float(WobblyEnv.counter[0]) + self.history = [] + return {"x": self.bias}, {"origin": self.bias} + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + value = self.bias + _score(self.history) + return {"x": value}, value * 0.1, False, False, {"h": value} + + +class JitterEnv(DeterministicEnv): + """Twins differ by 1e-9 only: within any declared tolerance, never byte-equal.""" + + parity: ClassVar[list[int]] = [0] + + def reset(self, *, seed: int | None = None, options: dict[str, Any] | None = None): + JitterEnv.parity[0] += 1 + self.bias = 1e-9 * (JitterEnv.parity[0] % 2) + self.history = [] + return {"x": 0.0}, {"origin": 0} + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + value = self.bias + _score(self.history) + return {"x": value}, value * 0.1, False, False, {"h": value} + + +class ThirdInstanceDriftEnv(DeterministicEnv): + """Instances 1-2 replay the prefix; the third adds a hidden bias during it. + + A scripted way to make the *candidate* arm's prefix disagree with the + reference pair while the reference pair itself agrees — which is exactly + the situation the per-arm prefix check exists to catch. + """ + + constructed: ClassVar[list[int]] = [0] + + def __init__(self) -> None: + ThirdInstanceDriftEnv.constructed[0] += 1 + self.bias = 1.0 if ThirdInstanceDriftEnv.constructed[0] >= 3 else 0.0 + self.history = [] + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + value = self.bias + _score(self.history) + return {"x": value}, value * 0.1, False, False, {"h": value} + + +PREFIX: list[list[float]] = [[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]] +CAND_A: list[list[float]] = [[0.9, 0.9]] +CAND_B: list[list[float]] = [[0.1, 0.1]] + + +# -------------------------------------------------------------------------- +# measure_branch_support: the ladder is earned, never granted +# -------------------------------------------------------------------------- + + +def test_byte_identical_twins_earn_measured_fork() -> None: + evidence = measure_branch_support(DeterministicEnv, seed=7, prefix_actions=PREFIX) + assert evidence.support is BranchingSupport.MEASURED_FORK + assert evidence.prefix_equal + assert evidence.identical_fork_digests + assert evidence.max_float_delta == 0.0 + assert evidence.first_difference == "" + + +def test_hidden_state_at_reset_defeats_the_fork() -> None: + """A seed that does not restore hidden state must not certify a fork. + + The wobbly env's info differs from the first record onward, so the probe + must report ``not_supported`` and say why in ``first_difference``. + """ + evidence = measure_branch_support(WobblyEnv, seed=7, prefix_actions=PREFIX) + assert evidence.support is BranchingSupport.NOT_SUPPORTED + assert not evidence.prefix_equal + assert not evidence.identical_fork_digests + assert "seed" in evidence.first_difference + + +def test_within_tolerance_is_prefix_replay_not_measured_fork() -> None: + """Float-noise engines reach the middle rung only, and only when they say so. + + With no tolerance the 1e-9 jitter is a divergence (byte equality is the + price of the top rung, so nothing excuses it silently); with an explicit + tolerance the probe records a faithful-but-noisy replay and still withholds + the equality-certifying rung. + """ + strict = measure_branch_support(JitterEnv, seed=1, prefix_actions=PREFIX[:2]) + assert strict.support is BranchingSupport.NOT_SUPPORTED + assert strict.max_float_delta == pytest.approx(1e-9, abs=1e-12) + + lenient = measure_branch_support(JitterEnv, seed=1, prefix_actions=PREFIX[:2], tolerance=1e-6) + assert lenient.support is BranchingSupport.PREFIX_REPLAY + assert lenient.prefix_equal + assert not lenient.identical_fork_digests + + +def test_empty_prefix_still_proves_the_rung_it_claims() -> None: + """Zero shared steps is a legal degenerate branch point (step 0). + + The twin reset record is inside the trace, so ``measured_fork`` still + requires the two instances to agree on their reset information. + """ + evidence = measure_branch_support(DeterministicEnv, seed=4, prefix_actions=[]) + assert evidence.support is BranchingSupport.MEASURED_FORK + wobbly = measure_branch_support(WobblyEnv, seed=4, prefix_actions=[]) + assert wobbly.support is BranchingSupport.NOT_SUPPORTED + + +def test_measurement_outranks_a_loud_declaration() -> None: + """``claimed`` records the strongest standing claim so callers see the gap. + + A capability block claiming ``measured_fork`` is contradicted by a world + whose twins diverge: the evidence must keep ``claimed`` at the claim and + ``support`` at the measurement. + """ + loud = WorldCapabilities( + physics=True, closed_loop=True, branching=BranchingSupport.MEASURED_FORK + ) + evidence = measure_branch_support(WobblyEnv, seed=9, prefix_actions=PREFIX, capabilities=loud) + assert evidence.support is BranchingSupport.NOT_SUPPORTED + assert evidence.claimed is BranchingSupport.MEASURED_FORK + + +@pytest.mark.parametrize("bad", [math.nan, math.inf, -math.inf, -0.5]) +def test_tolerance_must_be_a_real_number(bad: float) -> None: + with pytest.raises(TaskContractError): + measure_branch_support(DeterministicEnv, seed=1, prefix_actions=PREFIX, tolerance=bad) + + +def test_failing_env_factory_is_a_contract_error_not_a_crash() -> None: + def broken() -> DeterministicEnv: + raise RuntimeError("driver gone") + + with pytest.raises(TaskContractError): + measure_branch_support(broken, seed=1, prefix_actions=PREFIX) + + +# -------------------------------------------------------------------------- +# branch_from: compare from a common state, or refuse +# -------------------------------------------------------------------------- + + +def test_branches_diverge_from_a_shared_prefix() -> None: + result = branch_from( + DeterministicEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A, CAND_B] + ) + assert result["supported"] + assert result["refused"] == "" + assert result["evidence"].candidate_diverged + assert len({arm["digest"] for arm in result["arms"]}) == 2 + assert all(arm["prefix_matches_reference"] and arm["replayed_equal"] for arm in result["arms"]) + # the arms really share the prefix: same plan starts, different endings + assert result["arms"][0]["plan"][: len(PREFIX)] == result["arms"][1]["plan"][: len(PREFIX)] + + +def test_a_world_that_cannot_replay_its_prefix_is_refused_not_compared() -> None: + """The core F3 refusal: no candidate verdicts are emitted at all. + + An honest answer on a world that cannot fork is "I cannot tell you", so the + result must carry zero arm digests — reporting comparisons from states that + already diverged before the branch point would be fabricated evidence. + """ + result = branch_from(WobblyEnv, seed=3, prefix_actions=PREFIX, candidates=[CAND_A, CAND_B]) + assert not result["supported"] + assert result["arms"] == [] + assert result["evidence"].support is BranchingSupport.NOT_SUPPORTED + assert not result["evidence"].candidate_diverged + assert "seed alone does not restore" in result["refused"] + + +def test_candidate_arm_whose_prefix_drifts_is_refused() -> None: + """Fork pair agrees, but the candidate replays the prefix differently. + + The per-arm prefix check must catch drift that the fork-pair comparison + structurally cannot see (it only sees the reference pair), and the whole + branch set is refused rather than reporting one drifting arm's number. + """ + ThirdInstanceDriftEnv.constructed[0] = 0 + result = branch_from( + ThirdInstanceDriftEnv, seed=2, prefix_actions=PREFIX, candidates=[CAND_A, CAND_B] + ) + assert not result["supported"] + assert result["arms"] == [] + assert result["evidence"].support is BranchingSupport.NOT_SUPPORTED + + +def test_identical_candidate_plans_measure_nothing() -> None: + """Distinct-but-equal plans that all digest the same mean the intervention + changed nothing: ``candidate_diverged`` stays False, so a caller cannot + publish a "counterfactual" that never compared anything.""" + result = branch_from( + DeterministicEnv, + seed=7, + prefix_actions=PREFIX, + candidates=[[[0.2, 0.2]], [[0.2, 0.2]]], + ) + assert result["supported"] + assert not result["evidence"].candidate_diverged + + +def test_arm_traces_are_json_safe() -> None: + result = branch_from(DeterministicEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A]) + json.dumps(result["arms"]) # must not raise + assert result["arms"][0]["plan"] == [*PREFIX, [pytest.approx(0.9), pytest.approx(0.9)]] + + +def test_numpy_actions_digest_by_content_not_identity() -> None: + """Array actions must not split a digest by object identity. + + Two arms whose actions differ only in *representation* (array vs list of + the same numbers) have to digest equal, or an intervention encoded one way + would look like a divergence and the other way would hide one. + """ + pytest.importorskip("numpy") + import numpy as np + + with_array = branch_from( + DeterministicEnv, seed=5, prefix_actions=PREFIX, candidates=[[np.array([0.7, 0.8])]] + ) + with_list = branch_from( + DeterministicEnv, seed=5, prefix_actions=PREFIX, candidates=[[[0.7, 0.8]]] + ) + assert with_array["arms"][0]["digest"] == with_list["arms"][0]["digest"] + + +def test_branch_from_rejects_unusable_inputs() -> None: + with pytest.raises(TaskContractError): + branch_from(DeterministicEnv, seed=1, prefix_actions=PREFIX, candidates=[]) + with pytest.raises(TaskContractError): + branch_from( + DeterministicEnv, seed=1, prefix_actions=PREFIX, candidates=CAND_A, tolerance=-1.0 + ) + with pytest.raises(TaskContractError): + branch_from(cast(Any, "not-a-factory"), seed=1, prefix_actions=PREFIX, candidates=CAND_A) + + +# -------------------------------------------------------------------------- +# require_branch_support: fail closed +# -------------------------------------------------------------------------- + + +def _caps(branching: BranchingSupport, **flags: bool) -> WorldCapabilities: + return WorldCapabilities(physics=True, closed_loop=True, branching=branching, **flags) + + +def test_no_capability_source_is_a_refusal() -> None: + with pytest.raises(TaskContractError, match="capability declarations"): + require_branch_support(None, DRIVING_MECHANISM) + + +def test_unspecified_support_refuses_every_mechanism() -> None: + """The default must be the safe one: nothing declared, nothing permitted.""" + for mechanism in (DRIVING_MECHANISM, FORK_EQUALITY_MECHANISM): + with pytest.raises(TaskContractError): + require_branch_support(_caps(BranchingSupport.UNSPECIFIED), mechanism) + + +def test_a_bare_declaration_can_drive_but_never_certify() -> None: + declared = _caps(BranchingSupport.DECLARED) + require_branch_support(declared, DRIVING_MECHANISM) # must not raise + with pytest.raises(TaskContractError, match="measured evidence"): + require_branch_support(declared, FORK_EQUALITY_MECHANISM) + + +def test_measured_backing_is_the_only_route_to_certification() -> None: + require_branch_support(_caps(BranchingSupport.MEASURED_FORK), FORK_EQUALITY_MECHANISM) + # prefix_replay drives but does not certify: byte equality was never observed + replay = _caps(BranchingSupport.PREFIX_REPLAY) + require_branch_support(replay, DRIVING_MECHANISM) + with pytest.raises(TaskContractError): + require_branch_support(replay, FORK_EQUALITY_MECHANISM) + + +@pytest.mark.parametrize("state", [BranchingSupport.NOT_SUPPORTED, BranchingSupport.UNSPECIFIED]) +def test_reported_states_are_not_requestable(state: BranchingSupport) -> None: + """``not_supported``/``unspecified`` are things a world says, not asks.""" + with pytest.raises(TaskContractError, match="not a branching mechanism"): + require_branch_support(_caps(BranchingSupport.MEASURED_FORK), state) + + +def test_strength_order_is_total_and_measured_is_strongest() -> None: + order = [ + BranchingSupport.NOT_SUPPORTED, + BranchingSupport.UNSPECIFIED, + BranchingSupport.DECLARED, + BranchingSupport.PREFIX_REPLAY, + BranchingSupport.MEASURED_FORK, + ] + for i, weaker in enumerate(order): + for j, stronger in enumerate(order): + assert branching_at_least(stronger, weaker) is (j >= i) + + +# -------------------------------------------------------------------------- +# BranchEvidence guards +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize("bad", [math.nan, math.inf, -1.0]) +def test_evidence_rejects_unusable_tolerance(bad: float) -> None: + with pytest.raises(TaskContractError): + BranchEvidence( + support=BranchingSupport.PREFIX_REPLAY, + prefix_equal=True, + identical_fork_digests=False, + candidate_diverged=False, + tolerance=bad, + ) + + +def test_evidence_refuses_an_unbacked_measured_fork() -> None: + """``measured_fork`` without the two digest witnesses is a fabrication.""" + with pytest.raises(TaskContractError, match="earned from two"): + BranchEvidence( + support=BranchingSupport.MEASURED_FORK, + prefix_equal=True, + identical_fork_digests=False, + candidate_diverged=False, + ) + with pytest.raises(TaskContractError, match="earned from two"): + BranchEvidence( + support=BranchingSupport.MEASURED_FORK, + prefix_equal=False, + identical_fork_digests=True, + candidate_diverged=True, + ) + + +# -------------------------------------------------------------------------- +# Capability surface: declaration cross-checked against the adapter +# -------------------------------------------------------------------------- + + +def test_branching_rides_the_task_versus_adapter_crosscheck() -> None: + """A task may not grant itself a fork mechanism the adapter withholds. + + ``gates()`` includes the branching promotion flag, so a task declaring + ``measured_fork`` against an adapter that declared nothing must be refused + by ``resolve_world_capabilities`` — the same positional cross-check that + already guards physics/closed-loop overstating. + """ + try: + register_world_kind( + WorldKindSpec( + kind="f3-probe-world", + capabilities=WorldCapabilities(physics=True, closed_loop=True), + provider="tests", + ) + ) + greedy = WorldCapabilities( + physics=True, closed_loop=True, branching=BranchingSupport.MEASURED_FORK + ) + with pytest.raises(TaskContractError, match="disagrees with the installed"): + resolve_world_capabilities("f3-probe-world", greedy) + # under-claiming stays legal: declining a mechanism you have is safe + resolved = resolve_world_capabilities( + "f3-probe-world", WorldCapabilities(physics=True, closed_loop=True) + ) + assert resolved.branching is BranchingSupport.UNSPECIFIED + finally: + # Remove only what this test added. ``reset_default_world_kinds()`` would + # wipe the entry-point-discovered adapter identities that sibling suites + # (test_world_kinds.py) assert, turning an isolated probe into cross-file + # contamination. + _worlds._WORLD_KIND_REGISTRY.pop("f3-probe-world", None) + + +def test_branch_claims_require_a_stepped_world() -> None: + """``closed_loop=False`` means the runner cannot step the world at all, + so there is no trajectory to fork — the claim is refused at construction.""" + for mechanism in ( + BranchingSupport.DECLARED, + BranchingSupport.PREFIX_REPLAY, + BranchingSupport.MEASURED_FORK, + ): + with pytest.raises(TaskContractError, match="closed_loop"): + WorldCapabilities(physics=True, closed_loop=False, branching=mechanism) + + +def test_measured_fork_claim_requires_physics() -> None: + """A fork of a traceless synthetic stepper certifies nothing about dynamics; + pairing the top rung with non-physics state would let it ride into the + physics-oracle gate for free.""" + with pytest.raises(TaskContractError, match="physics"): + WorldCapabilities(closed_loop=True, physics=False, branching=BranchingSupport.MEASURED_FORK) + # declarations (no measurement claimed) remain allowed on non-physics worlds + WorldCapabilities(closed_loop=True, physics=False, branching=BranchingSupport.DECLARED) + + +def test_lumen_gym_declares_prefix_replay_and_nothing_louder() -> None: + """The one built-in claim is the one the pinned stack demonstrates + (tests/test_lumen_branch.py); the measured rung is never declared from a + table, and no built-in ships ``measured_fork``.""" + assert ( + BUILTIN_WORLD_CAPABILITIES[WorldKind.LUMEN_GYM].branching is BranchingSupport.PREFIX_REPLAY + ) + for kind, capabilities in BUILTIN_WORLD_CAPABILITIES.items(): + assert capabilities.branching is not BranchingSupport.MEASURED_FORK, kind + if capabilities.branching is not BranchingSupport.UNSPECIFIED: + assert capabilities.closed_loop, kind + + +def test_gates_tuple_carries_the_promotion_flag() -> None: + assert WorldCapabilities().gates().count(True) == 0 + promoted = WorldCapabilities(closed_loop=True, branching=BranchingSupport.DECLARED) + assert promoted.gates()[-1] is True + assert WorldCapabilities(closed_loop=True).gates()[-1] is False From f69fa4f8dcd75d2c37a1930b71b7c699a978fc28 Mon Sep 17 00:00:00 2001 From: Colin Son Date: Fri, 11 Sep 2026 13:47:43 -0500 Subject: [PATCH 2/3] docs: record Phase F3 branching capability in the implementation-status ledger (#68) --- docs/NEXT_STATUS.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/NEXT_STATUS.md b/docs/NEXT_STATUS.md index 9f7015a..efd9b7a 100644 --- a/docs/NEXT_STATUS.md +++ b/docs/NEXT_STATUS.md @@ -33,7 +33,7 @@ language — this table is the current status source of truth. | C | partial | Dossier + cited gate (#37), orphan-label refusal + provenance + fault robustness (#42) | C4 world-native distributions (need SOFA/GPU world revisions); C5 phantom (blocked: partner) | | D | partial | Paired comparison + CIs (#43), compare CLI (#45), split manifests with patient/site grouping (#60-62), D5 explicit cohort subgroup validation, head-covered trial bindings, clustered continuous bootstrap, and cloud parity (#64, cloud #5) | D5 benchmark-level reporting with external study data | | E | partial | Obs/action contract + A3 (#44), interactive streams (#46), LeRobot reader + MONAI (#51), honest video adapter (#54), semantic domain profiles for StreamSpec and CapabilitySpec (#66), profile-binding soundness closure: channel/schema conflation, identity pinning, geometry and calibration validation (#67) | E4 media alignment (no test media; mp4 banned from VC); real robotics policy adapter | -| F | partial | Prefix-replay branching proof + seed caveat (#47), trajectory-backed forecast task + null baseline (#56), planning utility measurement and recipe (#58) | F1 trajectory benchmark imports; F4 closed-loop simulator branching | +| F | partial | Prefix-replay branching proof + seed caveat (#47), trajectory-backed forecast task + null baseline (#56), planning utility measurement and recipe (#58), declared snapshot/restore capability with measured-fork proof and honest branch-equivalence refusal (#68) | F1 trajectory benchmark imports; F4 closed-loop simulator branching | | G | partial | SB3 PPO recipe + measurement (#48), runnable fixes (#50, #52), kernel-observed episode divergence (#57, #59), verifier-derived training rewards with divergence penalty (#65) | Prime interop | | H | partial | Episode resume (#49), crash-safe writes + partial recovery (#53), strict trial provenance + pre-write drift rejection (#59) | Fleet queue/autoscaling (blocked: hosted capacity decision); vectorized stepping (not justified by measured need) | From 85b0135fb4eb9e0b4410e8f9488928cf58546aef Mon Sep 17 00:00:00 2001 From: Colin Son Date: Fri, 11 Sep 2026 14:22:54 -0500 Subject: [PATCH 3/3] fix: branch probes digest the observation channel, closing a false-certification hole _rollout discarded both observations, so a world whose nondeterminism lives only in the policy-visible stream digested byte-equal and was certified measured_fork; two different candidate plans whose reward/info are action-independent constants could also digest equal, so a branch comparison could claim candidate divergence the policy could refute. The reset record and every step entry now carry jsonable(observation), and _first_difference names the diverging channel so a refusal says which stream failed to replay. Regression envs pin all three holes: observation-only jitter cannot certify a fork (and is refused outright as an arm replay), and identical plans on worlds that differ only in observation must digest apart. Verified counterfactually: with the observation stripped from the trace each new test fails, with it recorded each passes. --- src/or_audit/eval/branching.py | 44 ++++++++++---- tests/test_branching.py | 104 +++++++++++++++++++++++++++++++++ 2 files changed, 137 insertions(+), 11 deletions(-) diff --git a/src/or_audit/eval/branching.py b/src/or_audit/eval/branching.py index 5d41898..f041c11 100644 --- a/src/or_audit/eval/branching.py +++ b/src/or_audit/eval/branching.py @@ -212,20 +212,29 @@ def _rollout( bookkeeping meant for *scoring*. A probe must observe the engine's own bytes — anything sanitized before the digest is measured can hide the very nondeterminism being tested, and an injected perturbation would compare - perturbation schedules instead of worlds. The reset record is part of the - trace, because hidden state chosen at reset is exactly what "a seed - restores the state" would smuggle past a prefix-only comparison. - Termination honours ``terminated``/``truncated`` the way the runner does. + perturbation schedules instead of worlds. The reset record and every step + entry carry the *observation* the engine returned, not just reward and + info: a fork that hides its nondeterminism in the observation channel — the + only channel many real worlds expose state through — must not digest + byte-equal and be certified ``measured_fork``. Termination honours + ``terminated``/``truncated`` the way the runner does. """ - _, reset_info = env.reset(seed=seed, options=options) - trace: list[dict[str, Any]] = [{"reset_info": jsonable(reset_info), "seed": seed}] + observation, reset_info = env.reset(seed=seed, options=options) + trace: list[dict[str, Any]] = [ + { + "observation": jsonable(observation), + "reset_info": jsonable(reset_info), + "seed": seed, + } + ] for index, action in enumerate(actions): if index >= max_steps: break - _, reward, terminated, truncated, info = env.step(action) + observation, reward, terminated, truncated, info = env.step(action) trace.append( { "action": jsonable(_normalize_action(action)), + "observation": jsonable(observation), "reward": jsonable(reward), "terminated": bool(terminated), "truncated": bool(truncated), @@ -275,12 +284,25 @@ def _fresh_rollout( def _first_difference(first: Sequence[dict[str, Any]], second: Sequence[dict[str, Any]]) -> str: - """Canonical summary of the first step where two traces part company.""" + """Canonical summary of the first step where two traces part company. + + Names the *channel* that differs, not just the step: a divergence hidden in + the observation must be reported as an observation divergence, so the + refusal tells the author which stream their state restore is failing to + reproduce. + """ for index, (a, b) in enumerate(zip(first, second, strict=False)): - if canonical_digest(a) != canonical_digest(b): - if "action" in a: - return f"step {index} action {a['action']!r}" + if canonical_digest(a) == canonical_digest(b): + continue + if "action" not in a: return f"reset record of step {index}" + channels = [ + key + for key in ("observation", "reward", "terminated", "truncated", "info") + if canonical_digest(a.get(key)) != canonical_digest(b.get(key)) + ] + where = "/".join(channels) if channels else "step flags" + return f"step {index} action {a['action']!r} diverged in {where}" if len(first) != len(second): return f"traces end at different steps ({len(first)} vs {len(second)})" return "traces differ" diff --git a/tests/test_branching.py b/tests/test_branching.py index 82ce150..83d4d9c 100644 --- a/tests/test_branching.py +++ b/tests/test_branching.py @@ -132,6 +132,47 @@ def step(self, action): return {"x": value}, value * 0.1, False, False, {"h": value} +class ObsJitterEnv(DeterministicEnv): + """Twins differ only in the *observation* channel: reward and info are fixed. + + The observation-only sibling of ``JitterEnv``: an engine whose + nondeterminism is visible nowhere but in the bytes the policy actually + conditions on. If a probe measured only reward/info it would certify a + fork this world cannot provide. + """ + + counter: ClassVar[list[int]] = [0] + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + ObsJitterEnv.counter[0] += 1 + return ( + {"x": 1e-9 * ObsJitterEnv.counter[0]}, + 0.5, + False, + False, + {"h": 0.0}, + ) + + +class ObsBlindProbeEnv(DeterministicEnv): + """Identical reward/info streams; the observation scales by a class knob. + + Two instances configured with different multipliers are, from the + reward/info side, the *same world*: any difference in what the policy can + see lives in the observation. If arm digests were blind to observations, + these two runs would digest identically and a probe would be certifying + equality of worlds the policy can tell apart. + """ + + multiplier: ClassVar[float] = 1.0 + + def step(self, action): + self.history.append(tuple(float(v) for v in action)) + value = _score(self.history) + return {"x": self.multiplier * value}, 0.5, False, False, {"h": 0.0} + + PREFIX: list[list[float]] = [[0.1, 0.2], [0.3, 0.4], [0.5, 0.6]] CAND_A: list[list[float]] = [[0.9, 0.9]] CAND_B: list[list[float]] = [[0.1, 0.1]] @@ -164,6 +205,31 @@ def test_hidden_state_at_reset_defeats_the_fork() -> None: assert "seed" in evidence.first_difference +def test_observation_only_nondeterminism_cannot_certify_a_fork() -> None: + """The false-certification hole: noise hidden in the observation channel. + + Reward and info are constants here, so a probe that ignored the + observation would see byte-equal traces and hand out ``measured_fork`` to + a world whose policy input differs between twins run to run. With the + observation recorded, the digest mismatch is caught; and even under a + generous tolerance the world stops at ``prefix_replay`` — the noisy rung + never earns the equality-certifying rung. + """ + ObsJitterEnv.counter[0] = 0 + strict = measure_branch_support(ObsJitterEnv, seed=3, prefix_actions=PREFIX[:2]) + assert strict.support is BranchingSupport.NOT_SUPPORTED + assert not strict.identical_fork_digests + assert strict.max_float_delta == pytest.approx(2e-9, abs=1e-12) + assert "diverged in observation" in strict.first_difference + + ObsJitterEnv.counter[0] = 0 + lenient = measure_branch_support( + ObsJitterEnv, seed=3, prefix_actions=PREFIX[:2], tolerance=1e-6 + ) + assert lenient.support is BranchingSupport.PREFIX_REPLAY + assert not lenient.identical_fork_digests + + def test_within_tolerance_is_prefix_replay_not_measured_fork() -> None: """Float-noise engines reach the middle rung only, and only when they say so. @@ -286,6 +352,44 @@ def test_identical_candidate_plans_measure_nothing() -> None: assert not result["evidence"].candidate_diverged +def test_arm_digests_are_sensitive_to_the_observation_channel() -> None: + """Same plan, two worlds distinguishable only through the observation. + + Reward and info are constant and action-independent in both variants, so + an observation-blind digest would hash the two arms the same and certify + cross-world equality the policy could refute. The digests must therefore + track what the policy actually conditions on. + """ + ObsBlindProbeEnv.multiplier = 1.0 + plain = branch_from(ObsBlindProbeEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A]) + ObsBlindProbeEnv.multiplier = 4.0 + scaled = branch_from(ObsBlindProbeEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A]) + assert plain["supported"] + assert scaled["supported"] + assert plain["arms"][0]["digest"] != scaled["arms"][0]["digest"] + + +def test_observation_only_replay_drift_is_refused_not_certified() -> None: + """The false-certification hole on the arm side of ``branch_from``. + + Each replay pass of a candidate plan jitters only the observation (reward + and info are constants). If the trace ignored observations, the two passes + would digest byte-equal, every arm would report ``replayed_equal``, and + the branch set would be certified ``measured_fork`` although the policy + input differs run to run. With the observation in the digest, the arm's + prefix fails the byte-strict replay against the reference and the whole + comparison is refused with zero arms — the only honest outcome. + """ + ObsJitterEnv.counter[0] = 0 + result = branch_from( + ObsJitterEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A], tolerance=1e-6 + ) + assert not result["supported"] + assert result["arms"] == [] + assert result["evidence"].support is BranchingSupport.NOT_SUPPORTED + assert "replay" in result["refused"] + + def test_arm_traces_are_json_safe() -> None: result = branch_from(DeterministicEnv, seed=7, prefix_actions=PREFIX, candidates=[CAND_A]) json.dumps(result["arms"]) # must not raise