Add DASHI-aligned formal planning runtime contract

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.

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",
+]

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,
+        )

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
+        )