"""The orchestrator: executes a scenario and assembles its evidence. The runner is the only component that sees everything. Actors see their own credentials and memory; drivers see mechanics; observers see the system; oracles see the observation snapshot. Keeping those views separate is what makes the independence claim structural rather than procedural. """ from __future__ import annotations import uuid from dataclasses import dataclass from datetime import datetime, timezone from typing import Any from .actions import SurfaceNotPermitted from .drivers import Driver from .energy import EnergyEvent, EnergyEventType from .evidence import EvidencePack, Observation, Stratum from .observers import StateObserver from .oracles import Judgment, Oracle, Verdict, overall from .scenario import Scenario, VerificationAsset from .world import World class CollectorIndependenceError(Exception): """An actor was about to be recorded as the collector of S2/S3 evidence.""" @dataclass(slots=True) class RunResult: run_id: str verdict: Verdict judgments: list[Judgment] evidence: EvidencePack def judgment(self, assertion_id: str) -> Judgment: for j in self.judgments: if j.assertion_id == assertion_id: return j raise KeyError(assertion_id) class Runner: def __init__( self, world: World, driver: Driver, observer: StateObserver, oracle: Oracle | None = None, ) -> None: self._world = world self._driver = driver self._observer = observer self._oracle = oracle or Oracle() # -- independence guards --------------------------------------------- def _isolation_violations(self) -> list[str]: """Does any actor hold another's canary? Run on every scenario, not only on ones written to test isolation. """ canaries = {actor.canary: actor.id for actor in self._world.cast} violations: list[str] = [] for actor in self._world.cast: for key in actor.known_keys(): value = actor.recall(key) owner = canaries.get(value) if isinstance(value, str) else None if owner is not None and owner != actor.id: violations.append( f"actor {actor.id!r} holds the private marker of {owner!r} " f"under key {key!r}" ) return violations def _assert_collector_independence(self, stratum: Stratum, collector: str) -> None: """S2 and S3 evidence may never be attributed to an actor. This is the check that makes oracle independence falsifiable rather than merely asserted: wiring an actor in as an observer fails loudly here. """ if stratum is Stratum.SURFACE: return if collector in self._world.cast.actors: raise CollectorIndependenceError( f"{stratum.value} evidence cannot be collected by actor " f"{collector!r}; actors do not judge their own outcomes" ) def _record( self, pack: EvidencePack, stratum: Stratum, collector: str, kind: str, data: dict[str, Any], step_id: str | None, ) -> None: self._assert_collector_independence(stratum, collector) pack.record( Observation( id=f"obs-{len(pack.observations) + 1:04d}", stratum=stratum, collector=collector, step_id=step_id, kind=kind, data=data, ) ) # -- execution -------------------------------------------------------- def run(self, asset: VerificationAsset) -> RunResult: scenario: Scenario = asset.scenario run_id = f"run-{uuid.uuid4().hex[:12]}" pack = EvidencePack( run_id=run_id, scenario_id=scenario.id, use_case_id=scenario.use_case.id, sut_version=self._world.sut_version, ) pack.provenance_index = { scenario.use_case.id: scenario.use_case.provenance.value, **{c.id: c.provenance.value for c in scenario.use_case.claims}, **{i.id: i.provenance.value for i in scenario.use_case.invariants}, } pack.energy_events.append( EnergyEvent(asset.id, run_id, EnergyEventType.EXECUTED).as_dict() ) isolation = self._isolation_violations() self._record( pack, Stratum.JUDGMENT, self._observer.name, "actor_isolation", {"violations": isolation, "actors": sorted(self._world.cast.actors)}, None, ) judgments: list[Judgment] = [] scenario_sound = True claims_by_step: dict[str, list] = {} for claim in scenario.use_case.claims: claims_by_step.setdefault(claim.after_step, []).append(claim) for step in scenario.steps: actor = self._world.cast[step.actor_id] # --- S1: how it was done ------------------------------------- try: realization = self._driver.realize(actor, step.action) except SurfaceNotPermitted as exc: # D-05: routing around a control is a finding, not a recovery. self._record( pack, Stratum.REALIZATION, self._observer.name, "surface_violation", {"step": step.id, "action": step.action.describe(), "error": str(exc)}, step.id, ) break self._record( pack, Stratum.SURFACE, actor.id, "realization", { "action": step.action.describe(), "surface": realization.surface_id, "mechanics": realization.mechanics, "raised": realization.raised, }, step.id, ) # --- S3: what is now true ------------------------------------ snapshot = self._observer.snapshot() self._record( pack, Stratum.JUDGMENT, self._observer.name, "state_snapshot", dict(snapshot), step.id, ) # --- S2: did the action actually take effect ------------------ refused = realization.raised is not None postcondition_met: bool | None = None if step.action.postcondition is not None: try: postcondition_met = step.action.postcondition(snapshot) except KeyError: postcondition_met = None self._record( pack, Stratum.REALIZATION, self._observer.name, "realization_check", { "step": step.id, "action": step.action.name, "surface_used": realization.surface_id, "refused_by_sut": refused, "postcondition_met": postcondition_met, }, step.id, ) # Only an action that did not *happen* makes the scenario unsound. # An action that was accepted but did not take effect did happen — # and that is a statement about the system, judged below, not a # reason to stop judging. if refused: scenario_sound = False # --- invariants after every step ----------------------------- # Invariants are judged regardless: they are supposed to hold at all # times, however far the scenario got. for invariant in scenario.use_case.invariants: judgments.append(self._oracle.judge(invariant, snapshot, step.id)) # --- claims attached to this step ---------------------------- # Claims describe the state a *completed* scenario should reach. If a # step did not happen, a claim about the state after it is not # failing — it is unevaluable. Reporting FAIL there would accuse the # system of a defect on the strength of the test's own inability to # act, which is the mirror image of a false adaptation and just as # dishonest. for claim in claims_by_step.get(step.id, ()): if scenario_sound: judgments.append(self._oracle.judge(claim, snapshot, step.id)) else: judgments.append(Judgment( claim.id, claim.text, Verdict.INCONCLUSIVE, step.id, {"reason": "an earlier step in this scenario did not complete, " "so the state this claim describes was never reached"}, )) pack.verdicts = [j.as_dict() for j in judgments] pack.finished_at = datetime.now(timezone.utc).isoformat() result_verdict = overall(judgments) if result_verdict is Verdict.FAIL: pack.energy_events.append( EnergyEvent(asset.id, run_id, EnergyEventType.DEFECT_DETECTED).as_dict() ) return RunResult(run_id, result_verdict, judgments, pack)