diff --git a/tests/research/test_native_method_firewall.py b/tests/research/test_native_method_firewall.py new file mode 100644 index 00000000..47fd9e2d --- /dev/null +++ b/tests/research/test_native_method_firewall.py @@ -0,0 +1,290 @@ +"""Executable audit of the research-local native-method firewall (issue #156).""" + +from __future__ import annotations + +from dataclasses import replace +import importlib.util +import json +from pathlib import Path +import sys + +import pytest + + +MODULE_PATH = ( + Path(__file__).parents[2] + / "workstreams/native_method_firewall/native_method_firewall.py" +) +SPEC = importlib.util.spec_from_file_location("native_method_firewall", MODULE_PATH) +assert SPEC and SPEC.loader +module = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = module +SPEC.loader.exec_module(module) + + +def contract(*, with_lowering: bool = False) -> object: + task = module.TaskContract( + task_id="escape-value", + observer="normalized escape value at one declared state", + deliverable="value with a retained tail certificate", + regime="degree >= 2, interaction > 0, inverse state in [0, 1]", + accuracy="caller tolerance for the analytic tail", + claim_mode=module.ClaimMode.CERTIFIED_APPROXIMATE, + failure_semantics=("outside-domain", "budget-exhausted"), + ) + lowering = () + if with_lowering: + lowering = ( + module.LoweringWitness( + witness_id="fixed-chart-series", + mechanism=module.MethodMechanism.POWER_SERIES, + source_presentation="native inverse-state recurrence", + target_presentation="finite fixed-chart coefficient vector", + task_scope=("escape-value",), + allowed_lanes=(module.MethodLane.NATIVE_EVALUATION,), + adequacy_grade=module.AdequacyGrade.TASK_APPROXIMATE, + preserved_information=("escape-value", "first omitted residual"), + forgotten_information=("full process history",), + residual="first omitted degree and coefficient", + decoder="degree-ray Horner evaluation in the frozen chart", + certificate="replay the finite eigenrelation and residual", + failure_semantics=("chart-failure", "order-budget-exhausted"), + ), + ) + return module.MethodContract( + contract_id="amp-escape-method", + problem="evaluate one AMP escape process without changing task semantics", + primitive_processes=("inverse-state recurrence",), + tasks=(task,), + native_charts=("q = exp(-y)",), + retained_fibres=("tail residual", "domain witness"), + native_function_family="finite composition of power and log1p atoms", + native_composition="chronological inverse-state recurrence", + native_operators=("raise to degree", "divide by 1+t*q^d"), + claim_boundary="one scalar task; no generic AMP solver or complexity theorem", + allowed_lowerings=lowering, + baselines=( + module.BaselineSpec( + baseline_id="direct-iterate", + mechanism=module.MethodMechanism.BLACK_BOX_NUMERICS, + task_scope=("escape-value",), + purpose="independent same-information numerical comparison", + independent_reference="direct normalized log iteration", + ), + module.BaselineSpec( + baseline_id="series-red-team", + mechanism=module.MethodMechanism.POWER_SERIES, + task_scope=("escape-value",), + purpose="detect fixed-chart truncation failure", + independent_reference="explicit finite coefficient evaluation", + ), + ), + ) + + +def event_kwargs() -> dict[str, object]: + return { + "task_id": "escape-value", + "action": "evaluate the declared recurrence", + "input_semantics": "inverse state and tolerance", + "output_semantics": "escape value and tail residual", + } + + +def test_incomplete_contract_fails_before_a_trace_exists() -> None: + incomplete = replace(contract(), primitive_processes=()) + with pytest.raises(module.MethodContractError, match="primitive_processes"): + module.MethodTrace(incomplete) + + +@pytest.mark.parametrize( + "mechanism", + sorted(module.CLASSICAL_MECHANISMS, key=lambda item: item.value), +) +def test_every_classical_mechanism_fails_in_native_discovery(mechanism: object) -> None: + trace = module.MethodTrace(contract()) + with pytest.raises(module.PrematureLoweringError, match="lowering witness"): + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_DISCOVERY, + mechanism=mechanism, + ) + + +def test_declared_classical_baseline_remains_legal() -> None: + trace = module.MethodTrace(contract()) + event = trace.record( + **event_kwargs(), + lane=module.MethodLane.BASELINE, + mechanism=module.MethodMechanism.POWER_SERIES, + baseline_id="series-red-team", + cost=module.CostLedger(evaluation_steps=20, stored_history_units=10), + ) + assert event.lane is module.MethodLane.BASELINE + assert trace.audit_report()["summary"]["lane_counts"]["baseline"] == 1 + + +@pytest.mark.parametrize( + "mechanism", + sorted(module.CLASSICAL_MECHANISMS, key=lambda item: item.value), +) +def test_each_classical_mechanism_is_legal_when_declared_as_baseline( + mechanism: object, +) -> None: + declared = tuple( + module.BaselineSpec( + baseline_id=f"{item.value}-baseline", + mechanism=item, + task_scope=("escape-value",), + purpose="independent red team", + independent_reference=f"declared {item.value} implementation", + ) + for item in sorted(module.CLASSICAL_MECHANISMS, key=lambda item: item.value) + ) + trace = module.MethodTrace(replace(contract(), baselines=declared)) + event = trace.record( + **event_kwargs(), + lane=module.MethodLane.BASELINE, + mechanism=mechanism, + baseline_id=f"{mechanism.value}-baseline", + ) + assert event.mechanism is mechanism + + +def test_undeclared_baseline_fails_closed() -> None: + trace = module.MethodTrace(contract()) + with pytest.raises(module.MethodContractError, match="undeclared baseline"): + trace.record( + **event_kwargs(), + lane=module.MethodLane.BASELINE, + mechanism=module.MethodMechanism.POWER_SERIES, + baseline_id="friendly-name-only", + ) + + +def test_task_scoped_lowering_enters_only_its_declared_native_lane() -> None: + trace = module.MethodTrace(contract(with_lowering=True)) + accepted = trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_EVALUATION, + mechanism=module.MethodMechanism.POWER_SERIES, + lowering_id="fixed-chart-series", + ) + assert accepted.lowering_id == "fixed-chart-series" + + with pytest.raises(module.PrematureLoweringError, match="outside lane"): + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_DISCOVERY, + mechanism=module.MethodMechanism.POWER_SERIES, + lowering_id="fixed-chart-series", + ) + + +def test_wrong_mechanism_cannot_borrow_a_lowering_witness() -> None: + trace = module.MethodTrace(contract(with_lowering=True)) + with pytest.raises(module.PrematureLoweringError, match="does not certify"): + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_EVALUATION, + mechanism=module.MethodMechanism.MATRIX_LINEARIZATION, + lowering_id="fixed-chart-series", + ) + + +def test_alias_string_cannot_evade_the_typed_mechanism_vocabulary() -> None: + trace = module.MethodTrace(contract()) + with pytest.raises(module.MethodContractError, match="unknown mechanism"): + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_DISCOVERY, + mechanism="spectral-diagonalization", + ) + + +def test_baseline_event_cannot_be_relabelled_as_native_evidence() -> None: + trace = module.MethodTrace(contract()) + native = trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_EVALUATION, + mechanism=module.MethodMechanism.NATIVE_PROCESS, + ) + baseline = trace.record( + **event_kwargs(), + lane=module.MethodLane.BASELINE, + mechanism=module.MethodMechanism.BLACK_BOX_NUMERICS, + baseline_id="direct-iterate", + ) + with pytest.raises(module.EvidenceLaneError, match="cannot be relabelled"): + trace.claim_native_result( + task_id="escape-value", + statement="native recurrence evaluates the task", + evidence_event_ids=(native.event_id, baseline.event_id), + ) + + +def test_certificate_may_support_but_not_replace_native_evidence() -> None: + trace = module.MethodTrace(contract()) + certificate = trace.record( + **event_kwargs(), + lane=module.MethodLane.CERTIFICATE, + mechanism=module.MethodMechanism.GENERIC_CAS, + ) + with pytest.raises(module.EvidenceLaneError, match="certificate-only"): + trace.claim_native_result( + task_id="escape-value", + statement="a CAS output is not a native derivation", + evidence_event_ids=(certificate.event_id,), + ) + + native = trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_EVALUATION, + mechanism=module.MethodMechanism.NATIVE_PROCESS, + ) + claim = trace.claim_native_result( + task_id="escape-value", + statement="native recurrence with an independent symbolic certificate", + evidence_event_ids=(native.event_id, certificate.event_id), + ) + assert claim.evidence_event_ids == (native.event_id, certificate.event_id) + + +def test_json_audit_preserves_lane_and_multi_axis_cost() -> None: + trace = module.MethodTrace(contract()) + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_DISCOVERY, + mechanism=module.MethodMechanism.TASK_FIBRE, + cost=module.CostLedger( + discovery_steps=3, + live_state_units=2, + residual_units=1, + ), + ) + trace.record( + **event_kwargs(), + lane=module.MethodLane.NATIVE_EVALUATION, + mechanism=module.MethodMechanism.NATIVE_PROCESS, + cost=module.CostLedger(evaluation_steps=5, decoder_steps=1), + ) + report = json.loads(trace.to_json()) + assert report["summary"] == { + "cost_scalarization": "not-authorized", + "lane_counts": { + "baseline": 0, + "certificate": 0, + "native-discovery": 1, + "native-evaluation": 1, + }, + "mechanism_counts": {"native-process": 1, "task-fibre": 1}, + "total_cost": { + "compilation_steps": 0, + "decoder_steps": 1, + "discovery_steps": 3, + "evaluation_steps": 5, + "live_state_units": 2, + "residual_units": 1, + "stored_history_units": 0, + }, + } diff --git a/workstreams/native_method_firewall/README.md b/workstreams/native_method_firewall/README.md new file mode 100644 index 00000000..e191cafb --- /dev/null +++ b/workstreams/native_method_firewall/README.md @@ -0,0 +1,63 @@ +# Native-method firewall + +Research-local executable method contract for issue #156. This workstream +prevents a Process Geometry calculation from silently changing its ontology +when a familiar backend is convenient. + +The firewall keeps four lanes distinct: + +```text +native discovery primitive histories, task, chart, fibres, native family +native evaluation calculation in that declared language +certificate exact or independent checks of a declared claim +baseline conventional competing method and red team +``` + +Taylor/power series, ordinary polynomial algebra, matrix linearization, +Fourier/spectral methods, Koopman/Carleman lifting, generic CAS calls, and +black-box numerics are not prohibited. They may appear freely in a declared +baseline, and may support certificates. To enter a native lane they need a +`LoweringWitness` scoped to the exact task and lane. The witness records the +source and target presentation, adequacy grade, preserved and forgotten +information, residual, decoder, certificate, and failure semantics. + +## Solver plan + +```text +Problem and task: + Keep method semantics auditable while Process Geometry prototypes are run. +Primitive process / constraints: + A caller-declared MethodContract; no universal Process carrier is assumed. +Mathematical Core relation: + History -> task -> presentation -> retained fibre/residual -> decoder. + +Chosen algorithm: + Exact enum/type checks on an append-only in-memory trace. +Claim mode: + Exact finite validation of the declared record, not proof of source honesty. +Failure semantics: + Incomplete contract, unknown mechanism/lane/task, premature lowering, + undeclared baseline, or evidence-lane mismatch all fail closed. +Cost: + Multi-axis integer counts; scalarization is explicitly unauthorized. +Baseline: + Caller-declared and task-scoped; never relabelled as native evidence. + +Current software layer: + Research workstream only. +Mathematical Core / Theory Map / Public API effect: + Unchanged. +``` + +## Important boundary + +This is a trace and contract checker, not a Python sandbox, theorem prover, +source-code classifier, solver, or universal calculus. A dishonest caller can +mislabel an algorithm as `native-process`; correctness still requires +mathematical review, executable certificates, and independent baselines. The +typed mechanism vocabulary closes accidental aliasing, not adversarial code +execution. + +The Brownian scale/fibre Sonnet is the first intended downstream consumer. It +will declare Gaussian, heat-kernel, Fourier, and PDE machinery as lowering or +baseline evidence rather than supplying them to native scale discovery. diff --git a/workstreams/native_method_firewall/native_method_firewall.py b/workstreams/native_method_firewall/native_method_firewall.py new file mode 100644 index 00000000..bdeef61d --- /dev/null +++ b/workstreams/native_method_firewall/native_method_firewall.py @@ -0,0 +1,615 @@ +"""Research-local lane and lowering audit for Process Geometry calculations. + +This module is deliberately outside ``src/process_geometry``. It records and +checks a research method; it is not a generic solver or a package API. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import Enum +import json +from typing import Iterable + + +class MethodContractError(ValueError): + """A method contract or trace violates its declared semantics.""" + + +class PrematureLoweringError(MethodContractError): + """A classical mechanism entered a native lane without a scoped witness.""" + + +class EvidenceLaneError(MethodContractError): + """Evidence from one lane was claimed as evidence from another lane.""" + + +class MethodLane(str, Enum): + NATIVE_DISCOVERY = "native-discovery" + NATIVE_EVALUATION = "native-evaluation" + CERTIFICATE = "certificate" + BASELINE = "baseline" + + +class MethodMechanism(str, Enum): + """Mechanisms are typed so renaming cannot evade the lane audit.""" + + NATIVE_PROCESS = "native-process" + TASK_FIBRE = "task-fibre" + NATIVE_FUNCTION_FAMILY = "native-function-family" + EXACT_FINITE_ENUMERATION = "exact-finite-enumeration" + ORDINARY_POLYNOMIAL = "ordinary-polynomial" + POWER_SERIES = "power-series" + MATRIX_LINEARIZATION = "matrix-linearization" + FOURIER_SPECTRAL = "fourier-spectral" + KOOPMAN_CARLEMAN = "koopman-carleman" + GENERIC_CAS = "generic-cas" + BLACK_BOX_NUMERICS = "black-box-numerics" + + +CLASSICAL_MECHANISMS = frozenset( + { + MethodMechanism.ORDINARY_POLYNOMIAL, + MethodMechanism.POWER_SERIES, + MethodMechanism.MATRIX_LINEARIZATION, + MethodMechanism.FOURIER_SPECTRAL, + MethodMechanism.KOOPMAN_CARLEMAN, + MethodMechanism.GENERIC_CAS, + MethodMechanism.BLACK_BOX_NUMERICS, + } +) + + +class AdequacyGrade(str, Enum): + TASK_EXACT = "task-exact" + TASK_APPROXIMATE = "task-approximate" + + +class ClaimMode(str, Enum): + EXACT_SYMBOLIC = "exact-symbolic" + EXACT_FINITE = "exact-finite" + CERTIFIED_APPROXIMATE = "certified-approximate" + NUMERICAL = "numerical" + STOCHASTIC = "stochastic" + SEARCH_ONLY = "search-only" + + +def _require_text(value: str, field_name: str) -> None: + if not isinstance(value, str) or not value.strip(): + raise MethodContractError(f"{field_name} must be non-empty text") + + +def _require_texts(values: Iterable[str], field_name: str) -> None: + values = tuple(values) + if not values: + raise MethodContractError(f"{field_name} must not be empty") + for value in values: + _require_text(value, field_name) + + +def _enum_value(value: object, enum_type: type[Enum], field_name: str) -> Enum: + try: + return enum_type(value) + except (TypeError, ValueError) as exc: + raise MethodContractError(f"unknown {field_name}: {value!r}") from exc + + +@dataclass(frozen=True) +class TaskContract: + task_id: str + observer: str + deliverable: str + regime: str + accuracy: str + claim_mode: ClaimMode + failure_semantics: tuple[str, ...] + + def validate(self) -> None: + for name in ("task_id", "observer", "deliverable", "regime", "accuracy"): + _require_text(getattr(self, name), f"task.{name}") + _enum_value(self.claim_mode, ClaimMode, "claim mode") + _require_texts(self.failure_semantics, "task.failure_semantics") + + def as_dict(self) -> dict[str, object]: + return { + "task_id": self.task_id, + "observer": self.observer, + "deliverable": self.deliverable, + "regime": self.regime, + "accuracy": self.accuracy, + "claim_mode": ClaimMode(self.claim_mode).value, + "failure_semantics": list(self.failure_semantics), + } + + +@dataclass(frozen=True) +class LoweringWitness: + witness_id: str + mechanism: MethodMechanism + source_presentation: str + target_presentation: str + task_scope: tuple[str, ...] + allowed_lanes: tuple[MethodLane, ...] + adequacy_grade: AdequacyGrade + preserved_information: tuple[str, ...] + forgotten_information: tuple[str, ...] + residual: str + decoder: str + certificate: str + failure_semantics: tuple[str, ...] + + def validate(self, task_ids: frozenset[str]) -> None: + for name in ( + "witness_id", + "source_presentation", + "target_presentation", + "residual", + "decoder", + "certificate", + ): + _require_text(getattr(self, name), f"lowering.{name}") + mechanism = _enum_value(self.mechanism, MethodMechanism, "mechanism") + if mechanism not in CLASSICAL_MECHANISMS: + raise MethodContractError( + "a lowering witness must name a classical/lowered mechanism" + ) + _enum_value(self.adequacy_grade, AdequacyGrade, "adequacy grade") + _require_texts(self.task_scope, "lowering.task_scope") + unknown_tasks = set(self.task_scope) - task_ids + if unknown_tasks: + raise MethodContractError( + f"lowering {self.witness_id!r} has unknown tasks: " + f"{sorted(unknown_tasks)!r}" + ) + if not self.allowed_lanes: + raise MethodContractError("lowering.allowed_lanes must not be empty") + lanes = { + _enum_value(lane, MethodLane, "method lane") for lane in self.allowed_lanes + } + native_lanes = { + MethodLane.NATIVE_DISCOVERY, + MethodLane.NATIVE_EVALUATION, + } + if not lanes <= native_lanes: + raise MethodContractError( + "lowering witnesses apply only inside declared native lanes" + ) + _require_texts(self.preserved_information, "lowering.preserved_information") + _require_texts(self.forgotten_information, "lowering.forgotten_information") + _require_texts(self.failure_semantics, "lowering.failure_semantics") + + def as_dict(self) -> dict[str, object]: + return { + "witness_id": self.witness_id, + "mechanism": MethodMechanism(self.mechanism).value, + "source_presentation": self.source_presentation, + "target_presentation": self.target_presentation, + "task_scope": list(self.task_scope), + "allowed_lanes": [MethodLane(lane).value for lane in self.allowed_lanes], + "adequacy_grade": AdequacyGrade(self.adequacy_grade).value, + "preserved_information": list(self.preserved_information), + "forgotten_information": list(self.forgotten_information), + "residual": self.residual, + "decoder": self.decoder, + "certificate": self.certificate, + "failure_semantics": list(self.failure_semantics), + } + + +@dataclass(frozen=True) +class BaselineSpec: + baseline_id: str + mechanism: MethodMechanism + task_scope: tuple[str, ...] + purpose: str + independent_reference: str + + def validate(self, task_ids: frozenset[str]) -> None: + for name in ("baseline_id", "purpose", "independent_reference"): + _require_text(getattr(self, name), f"baseline.{name}") + _enum_value(self.mechanism, MethodMechanism, "mechanism") + _require_texts(self.task_scope, "baseline.task_scope") + unknown_tasks = set(self.task_scope) - task_ids + if unknown_tasks: + raise MethodContractError( + f"baseline {self.baseline_id!r} has unknown tasks: " + f"{sorted(unknown_tasks)!r}" + ) + + def as_dict(self) -> dict[str, object]: + return { + "baseline_id": self.baseline_id, + "mechanism": MethodMechanism(self.mechanism).value, + "task_scope": list(self.task_scope), + "purpose": self.purpose, + "independent_reference": self.independent_reference, + } + + +@dataclass(frozen=True) +class CostLedger: + """Non-scalarized, unit-free operation counts for one trace event.""" + + discovery_steps: int = 0 + compilation_steps: int = 0 + evaluation_steps: int = 0 + live_state_units: int = 0 + stored_history_units: int = 0 + residual_units: int = 0 + decoder_steps: int = 0 + + def validate(self) -> None: + for name in self.__dataclass_fields__: + value = getattr(self, name) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise MethodContractError(f"cost.{name} must be a non-negative integer") + + def as_dict(self) -> dict[str, int]: + return {name: getattr(self, name) for name in self.__dataclass_fields__} + + def __add__(self, other: "CostLedger") -> "CostLedger": + return CostLedger( + **{ + name: getattr(self, name) + getattr(other, name) + for name in self.__dataclass_fields__ + } + ) + + +@dataclass(frozen=True) +class MethodContract: + contract_id: str + problem: str + primitive_processes: tuple[str, ...] + tasks: tuple[TaskContract, ...] + native_charts: tuple[str, ...] + retained_fibres: tuple[str, ...] + native_function_family: str + native_composition: str + native_operators: tuple[str, ...] + claim_boundary: str + forbidden_premature_lowerings: frozenset[MethodMechanism] = field( + default_factory=lambda: CLASSICAL_MECHANISMS + ) + allowed_lowerings: tuple[LoweringWitness, ...] = () + baselines: tuple[BaselineSpec, ...] = () + + def validate(self) -> "MethodContract": + for name in ( + "contract_id", + "problem", + "native_function_family", + "native_composition", + "claim_boundary", + ): + _require_text(getattr(self, name), name) + _require_texts(self.primitive_processes, "primitive_processes") + _require_texts(self.native_charts, "native_charts") + _require_texts(self.retained_fibres, "retained_fibres") + _require_texts(self.native_operators, "native_operators") + if not self.tasks: + raise MethodContractError("tasks must not be empty") + for task in self.tasks: + task.validate() + task_ids = tuple(task.task_id for task in self.tasks) + if len(set(task_ids)) != len(task_ids): + raise MethodContractError("task ids must be unique") + known_tasks = frozenset(task_ids) + + forbidden = { + _enum_value(mechanism, MethodMechanism, "mechanism") + for mechanism in self.forbidden_premature_lowerings + } + if not forbidden <= CLASSICAL_MECHANISMS: + raise MethodContractError( + "forbidden_premature_lowerings may contain only classical mechanisms" + ) + if forbidden != CLASSICAL_MECHANISMS: + missing = sorted( + mechanism.value for mechanism in CLASSICAL_MECHANISMS - forbidden + ) + raise MethodContractError( + "the research firewall must enumerate every classical mechanism; " + f"missing {missing!r}" + ) + + lowering_ids = tuple(witness.witness_id for witness in self.allowed_lowerings) + if len(set(lowering_ids)) != len(lowering_ids): + raise MethodContractError("lowering witness ids must be unique") + for witness in self.allowed_lowerings: + witness.validate(known_tasks) + + baseline_ids = tuple(baseline.baseline_id for baseline in self.baselines) + if len(set(baseline_ids)) != len(baseline_ids): + raise MethodContractError("baseline ids must be unique") + for baseline in self.baselines: + baseline.validate(known_tasks) + return self + + @property + def task_ids(self) -> frozenset[str]: + return frozenset(task.task_id for task in self.tasks) + + def lowering(self, witness_id: str) -> LoweringWitness: + for witness in self.allowed_lowerings: + if witness.witness_id == witness_id: + return witness + raise PrematureLoweringError(f"undeclared lowering witness: {witness_id!r}") + + def baseline_for( + self, task_id: str, mechanism: MethodMechanism, baseline_id: str + ) -> BaselineSpec: + for baseline in self.baselines: + if baseline.baseline_id == baseline_id: + if task_id not in baseline.task_scope: + raise MethodContractError( + f"baseline {baseline_id!r} is outside task {task_id!r}" + ) + if MethodMechanism(baseline.mechanism) != mechanism: + raise MethodContractError( + f"baseline {baseline_id!r} does not certify {mechanism.value!r}" + ) + return baseline + raise MethodContractError(f"undeclared baseline: {baseline_id!r}") + + def as_dict(self) -> dict[str, object]: + self.validate() + return { + "contract_id": self.contract_id, + "problem": self.problem, + "primitive_processes": list(self.primitive_processes), + "tasks": [task.as_dict() for task in self.tasks], + "native_charts": list(self.native_charts), + "retained_fibres": list(self.retained_fibres), + "native_function_family": self.native_function_family, + "native_composition": self.native_composition, + "native_operators": list(self.native_operators), + "claim_boundary": self.claim_boundary, + "forbidden_premature_lowerings": sorted( + MethodMechanism(mechanism).value + for mechanism in self.forbidden_premature_lowerings + ), + "allowed_lowerings": [ + witness.as_dict() for witness in self.allowed_lowerings + ], + "baselines": [baseline.as_dict() for baseline in self.baselines], + } + + +@dataclass(frozen=True) +class MethodEvent: + event_id: int + task_id: str + lane: MethodLane + mechanism: MethodMechanism + action: str + input_semantics: str + output_semantics: str + cost: CostLedger + lowering_id: str | None = None + baseline_id: str | None = None + + def as_dict(self) -> dict[str, object]: + return { + "event_id": self.event_id, + "task_id": self.task_id, + "lane": self.lane.value, + "mechanism": self.mechanism.value, + "action": self.action, + "input_semantics": self.input_semantics, + "output_semantics": self.output_semantics, + "cost": self.cost.as_dict(), + "lowering_id": self.lowering_id, + "baseline_id": self.baseline_id, + } + + +@dataclass(frozen=True) +class NativeClaim: + claim_id: int + task_id: str + statement: str + evidence_event_ids: tuple[int, ...] + + def as_dict(self) -> dict[str, object]: + return { + "claim_id": self.claim_id, + "task_id": self.task_id, + "statement": self.statement, + "evidence_event_ids": list(self.evidence_event_ids), + } + + +class MethodTrace: + """Fail-closed audit trail for one validated method contract.""" + + def __init__(self, contract: MethodContract): + self.contract = contract.validate() + self._events: list[MethodEvent] = [] + self._claims: list[NativeClaim] = [] + + @property + def events(self) -> tuple[MethodEvent, ...]: + return tuple(self._events) + + @property + def claims(self) -> tuple[NativeClaim, ...]: + return tuple(self._claims) + + def record( + self, + *, + task_id: str, + lane: MethodLane, + mechanism: MethodMechanism, + action: str, + input_semantics: str, + output_semantics: str, + cost: CostLedger = CostLedger(), + lowering_id: str | None = None, + baseline_id: str | None = None, + ) -> MethodEvent: + if task_id not in self.contract.task_ids: + raise MethodContractError(f"unknown task: {task_id!r}") + lane = _enum_value(lane, MethodLane, "method lane") # type: ignore[assignment] + mechanism = _enum_value( # type: ignore[assignment] + mechanism, MethodMechanism, "mechanism" + ) + _require_text(action, "event.action") + _require_text(input_semantics, "event.input_semantics") + _require_text(output_semantics, "event.output_semantics") + cost.validate() + + native_lanes = { + MethodLane.NATIVE_DISCOVERY, + MethodLane.NATIVE_EVALUATION, + } + if lane in native_lanes and mechanism in CLASSICAL_MECHANISMS: + if lowering_id is None: + raise PrematureLoweringError( + f"{mechanism.value} cannot enter {lane.value} without a " + "task-scoped lowering witness" + ) + witness = self.contract.lowering(lowering_id) + if MethodMechanism(witness.mechanism) != mechanism: + raise PrematureLoweringError( + f"lowering {lowering_id!r} does not certify {mechanism.value!r}" + ) + if task_id not in witness.task_scope: + raise PrematureLoweringError( + f"lowering {lowering_id!r} is outside task {task_id!r}" + ) + if lane not in {MethodLane(value) for value in witness.allowed_lanes}: + raise PrematureLoweringError( + f"lowering {lowering_id!r} is outside lane {lane.value!r}" + ) + elif lowering_id is not None: + raise MethodContractError( + "a lowering witness may be attached only to a classical mechanism " + "inside a native lane" + ) + + if lane is MethodLane.BASELINE: + if baseline_id is None: + raise MethodContractError( + "baseline events require a declared baseline_id" + ) + self.contract.baseline_for(task_id, mechanism, baseline_id) + elif baseline_id is not None: + raise EvidenceLaneError( + "baseline evidence cannot be attached outside the baseline lane" + ) + + event = MethodEvent( + event_id=len(self._events), + task_id=task_id, + lane=lane, + mechanism=mechanism, + action=action, + input_semantics=input_semantics, + output_semantics=output_semantics, + cost=cost, + lowering_id=lowering_id, + baseline_id=baseline_id, + ) + self._events.append(event) + return event + + def claim_native_result( + self, + *, + task_id: str, + statement: str, + evidence_event_ids: Iterable[int], + ) -> NativeClaim: + if task_id not in self.contract.task_ids: + raise MethodContractError(f"unknown task: {task_id!r}") + _require_text(statement, "claim.statement") + event_ids = tuple(evidence_event_ids) + if not event_ids: + raise EvidenceLaneError("a native claim requires evidence") + selected: list[MethodEvent] = [] + for event_id in event_ids: + if isinstance(event_id, bool) or not isinstance(event_id, int): + raise EvidenceLaneError("evidence ids must be integer event ids") + try: + event = self._events[event_id] + except IndexError as exc: + raise EvidenceLaneError(f"unknown evidence event: {event_id}") from exc + if event.event_id != event_id: + raise EvidenceLaneError(f"unknown evidence event: {event_id}") + if event.task_id != task_id: + raise EvidenceLaneError( + f"event {event_id} belongs to task {event.task_id!r}" + ) + if event.lane is MethodLane.BASELINE: + raise EvidenceLaneError( + f"baseline event {event_id} cannot be relabelled as native evidence" + ) + selected.append(event) + if not any( + event.lane + in {MethodLane.NATIVE_DISCOVERY, MethodLane.NATIVE_EVALUATION} + for event in selected + ): + raise EvidenceLaneError( + "certificate-only evidence cannot establish a native result" + ) + + claim = NativeClaim( + claim_id=len(self._claims), + task_id=task_id, + statement=statement, + evidence_event_ids=event_ids, + ) + self._claims.append(claim) + return claim + + def total_cost(self) -> CostLedger: + total = CostLedger() + for event in self._events: + total = total + event.cost + return total + + def audit_report(self) -> dict[str, object]: + lane_counts = { + lane.value: sum(event.lane is lane for event in self._events) + for lane in MethodLane + } + mechanism_counts = { + mechanism.value: sum( + event.mechanism is mechanism for event in self._events + ) + for mechanism in MethodMechanism + if any(event.mechanism is mechanism for event in self._events) + } + return { + "contract": self.contract.as_dict(), + "events": [event.as_dict() for event in self._events], + "native_claims": [claim.as_dict() for claim in self._claims], + "summary": { + "lane_counts": lane_counts, + "mechanism_counts": mechanism_counts, + "total_cost": self.total_cost().as_dict(), + "cost_scalarization": "not-authorized", + }, + } + + def to_json(self, *, indent: int | None = 2) -> str: + return json.dumps( + self.audit_report(), + ensure_ascii=False, + indent=indent, + sort_keys=True, + ) + + +def contract_json(contract: MethodContract, *, indent: int | None = 2) -> str: + """Return a deterministic, machine-readable validated contract.""" + + return json.dumps( + contract.as_dict(), + ensure_ascii=False, + indent=indent, + sort_keys=True, + )