diff --git a/docs/formal_planning_contract.md b/docs/formal_planning_contract.md new file mode 100644 index 0000000..f353069 --- /dev/null +++ b/docs/formal_planning_contract.md @@ -0,0 +1,70 @@ +# Formal planning runtime contract + +This document defines the runtime side of the cross-repository boundary with +`chboishabba/dashi_agda`. + +## Ownership split + +`dashi_agda` owns: + +- semantic carriers and invariants; +- finite depth/truncation and escalation rules; +- spatial-path, conservation, calibration, Pareto, evidence, governance, and + deployment receipt shapes; +- promotion and non-promotion boundaries. + +Living Environment System owns: + +- GIS ingestion and reprojection; +- process-model adapters and numerical runs; +- latent/surrogate training and inference; +- scenario generation and optimisation; +- map and report outputs; +- serialisation of evidence into runtime receipts. + +A structurally valid runtime receipt is not proof that its external data are +true. Artifact hashes, provenance, validation, and human/domain approval remain +mandatory. + +## Python contract + +`les.contracts.planning_receipt` provides: + +- hashed `Artifact` references; +- Path A/B/C `ModelLane` values; +- fail-closed `EscalationEvidence`; +- explicit conservation balances with bounded residuals; +- unit-labelled objective vectors; +- finite Pareto dominance and front extraction; +- `PlanningRuntimeReceipt.validate()`. + +Escalation occurs when any of the following is true: + +- the input lies outside surrogate training support; +- residual error is too large; +- uncertainty is too large; +- conservation fails; +- the decision is policy-critical. + +A single escalation advances Path A to B or Path B to C. Path C is fixed. +Deployment additionally requires Path C and completed human approval. + +## Springfield pond fixture + +The Agda golden scenario is intentionally synthetic. The runtime implementation +should eventually supply audited artifacts for: + +1. DEM and catchment boundaries; +2. drainage and stormwater connectivity; +3. rainfall and antecedent moisture; +4. land use and nutrient-source layers; +5. pond geometry and residence time; +6. nutrient and pondweed observations; +7. candidate intervention footprints; +8. labour, machinery, fuel, maintenance, and capital costs; +9. model calibration and held-out validation; +10. community, ecological, engineering, and regulatory review. + +Expected output layers include ranked source hypotheses, candidate +interventions, Pareto membership, uncertainty, model-lane escalation, and +provenance. diff --git a/les/contracts/__init__.py b/les/contracts/__init__.py new file mode 100644 index 0000000..f7f53b6 --- /dev/null +++ b/les/contracts/__init__.py @@ -0,0 +1,71 @@ +"""LES runtime contracts.""" + +from .calibration import ( + CalibrationReceipt, + HeldOutValidation, + InputDomain, + LatentModel, + ModelIdentity, + TrainingCoverage, +) +from .knowledge import ( + EvidenceSource, + KnowledgeCatalogue, + KnowledgeEntry, + KnowledgeKind, + RegionContext, +) +from .planning_receipt import ( + Artifact, + ArtifactKind, + ConservationBalance, + EscalationEvidence, + EvaluatedPlan, + ModelLane, + ObjectiveScore, + PlanningRuntimeReceipt, + dominates, + pareto_front, +) +from .spatial import ( + MachineryProfile, + MachineryRouteAssessment, + SpatialKind, + SpatialNode, + SpatialTransportGraph, + TimeWindow, + TransportEdge, + TransportKind, +) + +__all__ = [ + "Artifact", + "ArtifactKind", + "CalibrationReceipt", + "ConservationBalance", + "EscalationEvidence", + "EvaluatedPlan", + "EvidenceSource", + "HeldOutValidation", + "InputDomain", + "KnowledgeCatalogue", + "KnowledgeEntry", + "KnowledgeKind", + "LatentModel", + "MachineryProfile", + "MachineryRouteAssessment", + "ModelIdentity", + "ModelLane", + "ObjectiveScore", + "PlanningRuntimeReceipt", + "RegionContext", + "SpatialKind", + "SpatialNode", + "SpatialTransportGraph", + "TimeWindow", + "TrainingCoverage", + "TransportEdge", + "TransportKind", + "dominates", + "pareto_front", +] diff --git a/les/contracts/calibration.py b/les/contracts/calibration.py new file mode 100644 index 0000000..8034762 --- /dev/null +++ b/les/contracts/calibration.py @@ -0,0 +1,114 @@ +"""Simulator-to-latent calibration contracts for LES.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Callable, Generic, TypeVar + +from .planning_receipt import EscalationEvidence + +InputT = TypeVar("InputT") +OutputT = TypeVar("OutputT") +LatentT = TypeVar("LatentT") + + +@dataclass(frozen=True) +class ModelIdentity: + name: str + version: str + source_revision: str + configuration_hash: str + authority_class: str + + +@dataclass(frozen=True) +class InputDomain: + name: str + variable_names: tuple[str, ...] + unit_declarations: tuple[str, ...] + spatial_extent: str + temporal_extent: str + parameter_bounds: tuple[str, ...] + exclusions: tuple[str, ...] = () + + +@dataclass(frozen=True) +class TrainingCoverage: + scenario_count: int + climate_regimes: tuple[str, ...] + soil_regimes: tuple[str, ...] + management_regimes: tuple[str, ...] + intervention_regimes: tuple[str, ...] + held_out_scenario_count: int + coverage_reference: str + + +@dataclass(frozen=True) +class HeldOutValidation: + dataset: str + sample_count: int + error_bound: float + observed_maximum_error: float + calibration_method: str + calibration_reference: str + + @property + def passes(self) -> bool: + return self.observed_maximum_error <= self.error_bound + + +@dataclass(frozen=True) +class LatentModel(Generic[InputT, LatentT, OutputT]): + encode: Callable[[InputT], LatentT] + predict: Callable[[LatentT], OutputT] + uncertainty: Callable[[LatentT], float] + inside_declared_support: Callable[[InputT], bool] + + +@dataclass(frozen=True) +class CalibrationReceipt(Generic[InputT, LatentT, OutputT]): + authoritative_model: ModelIdentity + input_domain: InputDomain + coverage: TrainingCoverage + output_names: tuple[str, ...] + output_units: tuple[str, ...] + latent_model: LatentModel[InputT, LatentT, OutputT] + held_out: HeldOutValidation + uncertainty_threshold: float + residual_threshold: float + escalation_policy: str + provenance: tuple[str, ...] + + def validate(self) -> list[str]: + errors: list[str] = [] + if not self.held_out.passes: + errors.append("held-out maximum error exceeds declared bound") + if self.coverage.scenario_count <= 0: + errors.append("training scenario count must be positive") + if self.coverage.held_out_scenario_count <= 0: + errors.append("held-out scenario count must be positive") + if len(self.output_names) != len(self.output_units): + errors.append("output names and units must align") + if not self.provenance: + errors.append("calibration provenance is required") + return errors + + def assess( + self, + input_value: InputT, + measured_residual: float, + conservation_passed: bool, + policy_critical: bool, + ) -> EscalationEvidence: + latent = self.latent_model.encode(input_value) + return EscalationEvidence( + outside_training_support=not self.latent_model.inside_declared_support( + input_value + ), + residual_too_large=measured_residual > self.residual_threshold, + uncertainty_too_large=( + self.latent_model.uncertainty(latent) > self.uncertainty_threshold + ), + conservation_failed=not conservation_passed, + policy_critical=policy_critical, + ) diff --git a/les/contracts/knowledge.py b/les/contracts/knowledge.py new file mode 100644 index 0000000..a79ad79 --- /dev/null +++ b/les/contracts/knowledge.py @@ -0,0 +1,116 @@ +"""Regional, versioned ecological evidence catalogue contracts.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum + + +class KnowledgeKind(str, Enum): + FUNCTIONAL_TRAIT = "functional-trait" + HOST_MYCORRHIZA_COMPATIBILITY = "host-mycorrhiza-compatibility" + FLOWERING_CALENDAR = "flowering-calendar" + POLLINATOR_RANGE = "pollinator-range" + POLLINATOR_COMPETITION = "pollinator-competition" + THREATENED_HABITAT_REQUIREMENT = "threatened-habitat-requirement" + CROP_SUITABILITY = "crop-suitability" + GRAZING_TOLERANCE = "grazing-tolerance" + REMEDIATION_TRAIT = "remediation-trait" + ENVIRONMENTAL_ENVELOPE = "environmental-envelope" + INVASIVE_RISK = "invasive-risk" + BIOSECURITY_RISK = "biosecurity-risk" + + +@dataclass(frozen=True) +class RegionContext: + region_name: str + jurisdiction: str + climate_classification: str + soil_classification: str + bioregion: str + custodians_or_communities: tuple[str, ...] = () + + +@dataclass(frozen=True) +class EvidenceSource: + citation_or_dataset: str + version: str + geographic_scope: str + temporal_scope: str + method_summary: str + confidence_depth: int + independently_reviewed: bool + + +@dataclass(frozen=True) +class KnowledgeEntry: + entry_id: str + kind: KnowledgeKind + subject: str + functional_groups: tuple[str, ...] + region: RegionContext + evidence: tuple[EvidenceSource, ...] + valid_from: str + reviewed_at: str + version: str + confidence_depth: int + limitations: tuple[str, ...] + active: bool = True + + def validate(self) -> list[str]: + errors: list[str] = [] + if not self.entry_id: + errors.append("knowledge entry id is required") + if not self.evidence: + errors.append(f"{self.entry_id}: at least one evidence source is required") + if self.confidence_depth < 0: + errors.append(f"{self.entry_id}: confidence depth cannot be negative") + if not self.region.jurisdiction: + errors.append(f"{self.entry_id}: jurisdiction is required") + return errors + + +@dataclass(frozen=True) +class KnowledgeCatalogue: + name: str + version: str + entries: tuple[KnowledgeEntry, ...] + schema_reference: str + provenance_manifest: tuple[str, ...] + regional_fallback_policy: str + update_policy: str + + def validate(self) -> list[str]: + errors: list[str] = [] + ids = [entry.entry_id for entry in self.entries] + if len(ids) != len(set(ids)): + errors.append("duplicate knowledge entry ids") + for entry in self.entries: + errors.extend(entry.validate()) + if not self.provenance_manifest: + errors.append("catalogue provenance manifest is required") + return errors + + def query( + self, + kind: KnowledgeKind, + subject: str, + jurisdiction: str, + minimum_confidence_depth: int = 0, + ) -> tuple[KnowledgeEntry, ...]: + """Return exact-jurisdiction entries only. + + Regional transfer is intentionally not implicit. Callers must create a + separate geographic-transfer evidence object before using entries from + another jurisdiction. + """ + + return tuple( + entry + for entry in self.entries + if entry.active + and entry.kind is kind + and entry.subject == subject + and entry.region.jurisdiction == jurisdiction + and entry.confidence_depth >= minimum_confidence_depth + ) diff --git a/les/contracts/planning_receipt.py b/les/contracts/planning_receipt.py new file mode 100644 index 0000000..f8dd5f8 --- /dev/null +++ b/les/contracts/planning_receipt.py @@ -0,0 +1,245 @@ +"""Runtime-side mirror of the DASHI LES planning receipt boundary. + +The Agda repository owns the semantic contract and promotion gates. This +module provides a dependency-free Python representation for serialising GIS, +model, calibration, conservation, optimisation, and approval evidence. + +It intentionally does not certify that referenced artifacts are scientifically +valid. It verifies structural completeness and fail-closed escalation rules. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from enum import Enum +from typing import Any, Iterable, Mapping + + +class ModelLane(str, Enum): + SCREENING = "pathA-screening" + LATENT = "pathB-latent" + AUTHORITATIVE = "pathC-authoritative" + + +class ArtifactKind(str, Enum): + RASTER = "raster-layer" + VECTOR = "vector-layer" + GRAPH = "graph-layer" + TIME_SERIES = "time-series" + MODEL_RUN = "model-run-manifest" + SURROGATE = "surrogate-artifact" + POPULATION = "optimisation-population" + REPORT = "report-artifact" + + +@dataclass(frozen=True) +class Artifact: + artifact_id: str + kind: ArtifactKind + uri: str + content_hash: str + producer_version: str + crs: str = "" + units: tuple[str, ...] = () + provenance: tuple[str, ...] = () + + def validate(self) -> list[str]: + errors: list[str] = [] + if not self.artifact_id: + errors.append("artifact_id is required") + if not self.uri: + errors.append(f"{self.artifact_id}: uri is required") + if not self.content_hash: + errors.append(f"{self.artifact_id}: content_hash is required") + if not self.producer_version: + errors.append(f"{self.artifact_id}: producer_version is required") + return errors + + +@dataclass(frozen=True) +class EscalationEvidence: + outside_training_support: bool = False + residual_too_large: bool = False + uncertainty_too_large: bool = False + conservation_failed: bool = False + policy_critical: bool = False + + @property + def requires_escalation(self) -> bool: + return any(asdict(self).values()) + + +@dataclass(frozen=True) +class ConservationBalance: + unit: str + opening_storage: float + external_input: float + external_export: float + closing_storage: float + accounted_transformation: float + unaccounted_residual: float + residual_tolerance: float + model_reference: str + evidence: tuple[str, ...] = () + + @property + def lhs(self) -> float: + return self.opening_storage + self.external_input + + @property + def rhs(self) -> float: + return ( + self.external_export + + self.closing_storage + + self.accounted_transformation + + self.unaccounted_residual + ) + + def validate(self, numerical_tolerance: float = 1e-9) -> list[str]: + errors: list[str] = [] + if not self.unit: + errors.append("conservation unit is required") + if abs(self.lhs - self.rhs) > numerical_tolerance: + errors.append( + f"{self.unit}: balance mismatch lhs={self.lhs} rhs={self.rhs}" + ) + if abs(self.unaccounted_residual) > self.residual_tolerance: + errors.append( + f"{self.unit}: residual {self.unaccounted_residual} exceeds " + f"tolerance {self.residual_tolerance}" + ) + if not self.model_reference: + errors.append(f"{self.unit}: model_reference is required") + return errors + + +@dataclass(frozen=True) +class ObjectiveScore: + objective_id: str + direction: str + value: float + unit: str + evidence_reference: str + + def validate(self) -> list[str]: + errors: list[str] = [] + if self.direction not in {"minimise", "maximise"}: + errors.append(f"{self.objective_id}: invalid direction") + if not self.unit: + errors.append(f"{self.objective_id}: unit is required") + if not self.evidence_reference: + errors.append(f"{self.objective_id}: evidence reference is required") + return errors + + +@dataclass(frozen=True) +class EvaluatedPlan: + plan_id: str + hard_constraints_satisfied: bool + objectives: tuple[ObjectiveScore, ...] + intervention_artifacts: tuple[str, ...] = () + assumptions: tuple[str, ...] = () + exclusions: tuple[str, ...] = () + + def validate(self) -> list[str]: + errors = [e for objective in self.objectives for e in objective.validate()] + ids = [objective.objective_id for objective in self.objectives] + if len(ids) != len(set(ids)): + errors.append(f"{self.plan_id}: duplicate objective ids") + if not self.plan_id: + errors.append("plan_id is required") + return errors + + +def dominates(better: EvaluatedPlan, worse: EvaluatedPlan) -> bool: + """Return finite Pareto dominance for aligned objective vectors.""" + + if not (better.hard_constraints_satisfied and worse.hard_constraints_satisfied): + return False + better_by_id = {score.objective_id: score for score in better.objectives} + worse_by_id = {score.objective_id: score for score in worse.objectives} + if better_by_id.keys() != worse_by_id.keys(): + return False + + no_worse = True + strictly_better = False + for objective_id, a in better_by_id.items(): + b = worse_by_id[objective_id] + if a.direction != b.direction or a.unit != b.unit: + return False + if a.direction == "minimise": + no_worse &= a.value <= b.value + strictly_better |= a.value < b.value + else: + no_worse &= a.value >= b.value + strictly_better |= a.value > b.value + return no_worse and strictly_better + + +def pareto_front(plans: Iterable[EvaluatedPlan]) -> tuple[EvaluatedPlan, ...]: + population = tuple(plans) + return tuple( + candidate + for candidate in population + if candidate.hard_constraints_satisfied + and not any( + other is not candidate and dominates(other, candidate) + for other in population + ) + ) + + +@dataclass(frozen=True) +class PlanningRuntimeReceipt: + scenario_id: str + starting_lane: ModelLane + resulting_lane: ModelLane + escalation: EscalationEvidence + artifacts: tuple[Artifact, ...] + conservation: tuple[ConservationBalance, ...] + candidate_plans: tuple[EvaluatedPlan, ...] + selected_plan_id: str + source_hypotheses: tuple[Mapping[str, Any], ...] = () + community_constraints: tuple[Mapping[str, Any], ...] = () + human_approval_required: bool = True + deployment_permitted: bool = False + provenance: tuple[str, ...] = () + + def validate(self) -> list[str]: + errors: list[str] = [] + errors.extend(e for artifact in self.artifacts for e in artifact.validate()) + errors.extend(e for balance in self.conservation for e in balance.validate()) + errors.extend(e for plan in self.candidate_plans for e in plan.validate()) + + plan_ids = {plan.plan_id for plan in self.candidate_plans} + if self.selected_plan_id not in plan_ids: + errors.append("selected_plan_id is not in candidate_plans") + + expected_lane = self.starting_lane + if self.escalation.requires_escalation: + expected_lane = { + ModelLane.SCREENING: ModelLane.LATENT, + ModelLane.LATENT: ModelLane.AUTHORITATIVE, + ModelLane.AUTHORITATIVE: ModelLane.AUTHORITATIVE, + }[self.starting_lane] + if self.resulting_lane != expected_lane: + errors.append( + f"resulting lane {self.resulting_lane.value} does not match " + f"fail-closed lane {expected_lane.value}" + ) + + if self.deployment_permitted and self.human_approval_required: + errors.append("deployment cannot be permitted while human approval is pending") + if self.deployment_permitted and self.resulting_lane is not ModelLane.AUTHORITATIVE: + errors.append("deployment requires authoritative Path C verification") + if not self.provenance: + errors.append("receipt provenance is required") + return errors + + def to_dict(self) -> dict[str, Any]: + data = asdict(self) + data["starting_lane"] = self.starting_lane.value + data["resulting_lane"] = self.resulting_lane.value + for artifact in data["artifacts"]: + artifact["kind"] = artifact["kind"].value + return data diff --git a/les/contracts/spatial.py b/les/contracts/spatial.py new file mode 100644 index 0000000..7366f24 --- /dev/null +++ b/les/contracts/spatial.py @@ -0,0 +1,149 @@ +"""Typed spatial and transport graph used by LES planning receipts.""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum +from collections import deque + + +class SpatialKind(str, Enum): + RASTER_CELL = "raster-cell" + PARCEL = "parcel" + CATCHMENT = "catchment" + WATERBODY = "waterbody" + STREAM_REACH = "stream-reach" + GROUNDWATER_ZONE = "groundwater-zone" + ROAD_SEGMENT = "road-segment" + HABITAT_PATCH = "habitat-patch" + INTERVENTION_FOOTPRINT = "intervention-footprint" + + +class TransportKind(str, Enum): + SURFACE_WATER = "surface-water" + GROUNDWATER = "groundwater" + SEDIMENT = "sediment" + DISSOLVED_NITROGEN = "dissolved-nitrogen" + DISSOLVED_PHOSPHORUS = "dissolved-phosphorus" + POLLEN = "pollen" + ANIMAL_MOVEMENT = "animal-movement" + SEED_DISPERSAL = "seed-dispersal" + MACHINERY_ACCESS = "machinery-access" + HUMAN_ACCESS = "human-access" + + +@dataclass(frozen=True) +class SpatialNode: + node_id: str + kind: SpatialKind + crs: str + geometry_uri: str + source_dataset: str + + +@dataclass(frozen=True) +class TimeWindow: + start: int + end: int + + def validate(self) -> list[str]: + return [] if self.start <= self.end else ["time window start exceeds end"] + + def overlaps(self, other: "TimeWindow") -> bool: + return self.start <= other.end and other.start <= self.end + + +@dataclass(frozen=True) +class TransportEdge: + source_id: str + target_id: str + kind: TransportKind + active_window: TimeWindow + direction_verified: bool + capacity_recorded: bool + evidence_reference: str + uncertainty_reference: str + + def validate(self, known_nodes: set[str]) -> list[str]: + errors = self.active_window.validate() + if self.source_id not in known_nodes: + errors.append(f"unknown source node {self.source_id}") + if self.target_id not in known_nodes: + errors.append(f"unknown target node {self.target_id}") + if not self.direction_verified: + errors.append(f"{self.source_id}->{self.target_id}: direction not verified") + if not self.evidence_reference: + errors.append(f"{self.source_id}->{self.target_id}: evidence missing") + return errors + + +@dataclass(frozen=True) +class SpatialTransportGraph: + nodes: tuple[SpatialNode, ...] + edges: tuple[TransportEdge, ...] + + def validate(self) -> list[str]: + errors: list[str] = [] + ids = [node.node_id for node in self.nodes] + if len(ids) != len(set(ids)): + errors.append("duplicate spatial node ids") + known = set(ids) + for edge in self.edges: + errors.extend(edge.validate(known)) + return errors + + def find_path( + self, + source_id: str, + target_id: str, + kind: TransportKind, + window: TimeWindow, + ) -> tuple[TransportEdge, ...] | None: + """Find an auditable directed path in one transport medium.""" + + adjacency: dict[str, list[TransportEdge]] = {} + for edge in self.edges: + if ( + edge.kind is kind + and edge.direction_verified + and edge.active_window.overlaps(window) + ): + adjacency.setdefault(edge.source_id, []).append(edge) + + queue: deque[tuple[str, tuple[TransportEdge, ...]]] = deque( + [(source_id, ())] + ) + visited = {source_id} + while queue: + node_id, path = queue.popleft() + if node_id == target_id: + return path + for edge in adjacency.get(node_id, []): + if edge.target_id not in visited: + visited.add(edge.target_id) + queue.append((edge.target_id, path + (edge,))) + return None + + +@dataclass(frozen=True) +class MachineryProfile: + name: str + maximum_slope_percent: float + turning_radius_m: float + width_m: float + wet_soil_access_allowed: bool + fuel_model_reference: str + + +@dataclass(frozen=True) +class MachineryRouteAssessment: + machine: MachineryProfile + edge_ids: tuple[str, ...] + slope_ok: bool + turning_ok: bool + width_ok: bool + seasonal_access_ok: bool + + @property + def feasible(self) -> bool: + return self.slope_ok and self.turning_ok and self.width_ok and self.seasonal_access_ok diff --git a/tests/test_planning_receipt.py b/tests/test_planning_receipt.py new file mode 100644 index 0000000..4d7c15c --- /dev/null +++ b/tests/test_planning_receipt.py @@ -0,0 +1,125 @@ +from les.contracts.planning_receipt import ( + Artifact, + ArtifactKind, + ConservationBalance, + EscalationEvidence, + EvaluatedPlan, + ModelLane, + ObjectiveScore, + PlanningRuntimeReceipt, + dominates, + pareto_front, +) + + +def _score(objective_id: str, direction: str, value: float, unit: str) -> ObjectiveScore: + return ObjectiveScore( + objective_id=objective_id, + direction=direction, + value=value, + unit=unit, + evidence_reference="fixture", + ) + + +def test_pareto_front_respects_direction_and_constraints() -> None: + combined = EvaluatedPlan( + plan_id="combined", + hard_constraints_satisfied=True, + objectives=( + _score("nutrient-load", "minimise", 4.0, "kg-P/year"), + _score("habitat", "maximise", 8.0, "index"), + ), + ) + mechanical = EvaluatedPlan( + plan_id="mechanical-only", + hard_constraints_satisfied=True, + objectives=( + _score("nutrient-load", "minimise", 7.0, "kg-P/year"), + _score("habitat", "maximise", 2.0, "index"), + ), + ) + infeasible = EvaluatedPlan( + plan_id="infeasible", + hard_constraints_satisfied=False, + objectives=( + _score("nutrient-load", "minimise", 0.0, "kg-P/year"), + _score("habitat", "maximise", 99.0, "index"), + ), + ) + + assert dominates(combined, mechanical) + assert not dominates(infeasible, combined) + assert pareto_front((combined, mechanical, infeasible)) == (combined,) + + +def test_policy_critical_receipt_escalates_latent_to_authoritative() -> None: + receipt = PlanningRuntimeReceipt( + scenario_id="springfield-pond-fixture", + starting_lane=ModelLane.LATENT, + resulting_lane=ModelLane.AUTHORITATIVE, + escalation=EscalationEvidence(policy_critical=True), + artifacts=( + Artifact( + artifact_id="catchment", + kind=ArtifactKind.VECTOR, + uri="fixture://catchment", + content_hash="sha256:fixture", + producer_version="test", + crs="EPSG:7856", + provenance=("synthetic fixture",), + ), + ), + conservation=( + ConservationBalance( + unit="kg-P", + opening_storage=0.0, + external_input=0.0, + external_export=0.0, + closing_storage=0.0, + accounted_transformation=0.0, + unaccounted_residual=0.0, + residual_tolerance=0.0, + model_reference="fixture", + ), + ), + candidate_plans=( + EvaluatedPlan( + plan_id="combined", + hard_constraints_satisfied=True, + objectives=(_score("nutrient-load", "minimise", 4.0, "kg-P/year"),), + ), + ), + selected_plan_id="combined", + human_approval_required=True, + deployment_permitted=False, + provenance=("DASHI planning boundary fixture",), + ) + + assert receipt.validate() == [] + + +def test_receipt_rejects_unverified_deployment() -> None: + receipt = PlanningRuntimeReceipt( + scenario_id="bad", + starting_lane=ModelLane.SCREENING, + resulting_lane=ModelLane.SCREENING, + escalation=EscalationEvidence(), + artifacts=(), + conservation=(), + candidate_plans=( + EvaluatedPlan( + plan_id="plan", + hard_constraints_satisfied=True, + objectives=(), + ), + ), + selected_plan_id="plan", + human_approval_required=True, + deployment_permitted=True, + provenance=("fixture",), + ) + + errors = receipt.validate() + assert "deployment cannot be permitted while human approval is pending" in errors + assert "deployment requires authoritative Path C verification" in errors