feat(run): sampled runs that can never become the canonical run

README.md

@@ -170,11 +170,37 @@ Useful automation options:
 openmapstack validate project.yaml --json --output validation/cli-report.json
 openmapstack validate project.yaml --strict       # warnings also return non-zero
 openmapstack validate project.yaml --preflight    # skip not-yet-generated artifacts
-openmapstack run project.yaml --dry-run
+openmapstack run project.yaml --dry-run        # print the command, execute nothing
 openmapstack run project.yaml --json
 openmapstack inspect project.yaml --json
 ```
 
+### Sampled runs — nail it before you scale it
+
+A wide-area analysis can run for hours before a late step fails. A sampled run
+executes the same pipeline over a deliberately smaller slice, so failure
+arrives in minutes:
+
+```bash
+openmapstack run project.yaml --sample                     # the manifest's declared sample
+openmapstack run project.yaml --sample-area 26.6,58.3,26.8,58.4
+openmapstack run project.yaml --sample-rows 5000
+openmapstack run project.yaml --sample-fraction 1.0
+```
+
+Each flag binds a `runtime.implementation.parameters` entry that declares the
+matching `role`; sampling a project that declares none is refused, naming what
+the manifest must add. The canonical run still passes nothing.
+
+**A sampled run proves the pipeline executes; it does not establish the
+result.** Clipping to a test AOI breaks neighbourhood operations at the cut and
+row sampling destroys the spatial coherence a join needs, so sampled counts are
+not answers. That is enforced, not merely advised: a sampled run record is
+marked `mode: sampled`, must record what it *realized* rather than only what
+was requested, and can never become `runs.latest` — `openmapstack validate`
+reports this as `runs.sample_isolation`, and `run --sample` fails outright if a
+pipeline promotes its own sampled run. See `references/project-spec.md`.
+
 ### `openmapstack verify` — check the analysis, not just the paperwork
 
 `validate` audits the manifest and its bookkeeping. `verify` runs the check

evals/COVERAGE.md

@@ -108,6 +108,15 @@ Legend: ✅ covered · ⚠️ partially covered · ❌ not covered (tracked belo
 | Duplicate-input resistance (declared) | 015 `parcel-duplicates` (every parcel appended once more, outputs must be equal); mutation 925 `duplicates_changed_output` |
 | Invalid-precondition refusal | unit tests: count/sum semantics, missing tie-break, non-growing variant, unsupported format, source already duplicated, oversize source |
 
+## Sampled runs
+
+| Risk | Positive | Mutation |
+|---|---|---|
+| A sampled run is marked, states its **realized** sample, and is not `runs.latest` | 016 `sampled-run` | 928 `sampled-run-as-canonical` |
+| A sampled run record that states only what was *requested* | unit tests | — |
+| The canonical run stays argument-free when a sampling parameter is declared | unit tests | — |
+| A pipeline that promotes its own sampled run into `runs.latest` | unit tests (CLI guard) | — |
+
 ## Known gaps (tracked)
 
 1. **PostGIS / warehouse canary** — needs a live service; candidate design is

evals/cases/016-sampled-run/expected.yaml

@@ -0,0 +1,25 @@
+id: 016-sampled-run
+case_type: positive
+modes: [fixture]
+score_types:
+  fixture: contract_ci
+project_dir: project
+hard_gate: true
+fixture:
+  generator: "{python} {evals_dir}/fixtures/reference_pipeline/gen.py {project_dir} --sampled-run"
+  source_baseline:
+    - { source: ../../fixtures/mini-tartu/parcels.geojson, destination: data/source/parcels.geojson }
+    - { source: ../../fixtures/mini-tartu/roads.geojson, destination: data/source/roads.geojson }
+    - { source: ../../fixtures/mini-tartu/pois.geojson, destination: data/source/pois.geojson }
+
+# The control arm of the sampling contract. A sampled run record sits beside
+# the canonical one: it is marked `mode: sampled`, it states what it actually
+# realized rather than only what was requested, and it is not `runs.latest`.
+# Nothing about its presence may weaken the canonical run -- the manifest still
+# validates, the hashes still resolve, and the project is still `validated`.
+assertions:
+  - assert: project.conforms_to_schema
+  - assert: validation.sample_run_not_promoted
+  - assert: validation.run_record_matches
+  - assert: validation.required_all_present
+  - assert: project.status_agrees_with_validation_report

evals/cases/928-sampled-run-as-canonical/expected.yaml

@@ -0,0 +1,26 @@
+id: 928-sampled-run-as-canonical
+case_type: mutation
+modes: [fixture]
+score_types: { fixture: mutation_tests }
+project_dir: project
+hard_gate: true
+mutation:
+  control_generator: "{python} {evals_dir}/fixtures/reference_pipeline/gen.py {project_dir} --sampled-run"
+fixture:
+  generator: "{python} {evals_dir}/fixtures/reference_pipeline/gen.py {project_dir} --break=sampled_run_as_canonical"
+  source_baseline:
+    - { source: ../../fixtures/mini-tartu/parcels.geojson, destination: data/source/parcels.geojson }
+    - { source: ../../fixtures/mini-tartu/roads.geojson, destination: data/source/roads.geojson }
+    - { source: ../../fixtures/mini-tartu/pois.geojson, destination: data/source/pois.geojson }
+
+# Validation laundering by way of sampling: the run of record is a fast pass
+# over a clipped test AOI, presented as the analysis. Every hash resolves and
+# every declared check is present -- the record is internally consistent, it
+# simply describes a slice. `runs.latest` is what `verify`, the clean-rerun
+# protocol, and every expectation attestation bind to, so this must fail on
+# the marking alone, not on arithmetic.
+assertions:
+  - assert: project.conforms_to_schema
+  - assert: validation.sample_run_not_promoted
+    expect: failed
+    expect_code: sampled_run_promoted

evals/fixtures/reference_pipeline/gen.py

@@ -109,6 +109,8 @@ def _inventory(root: Path, paths: list[Path]) -> list[dict[str, str]]:
 
 
 ROAD_DISTANCE_CANONICAL_M = 2000
+# The test AOI a sampled run of this fixture would clip to.
+SAMPLE_AREA = "26.68,58.35,26.74,58.39"
 
 # Break modes that live in the pipeline's logic rather than in the generated
 # bookkeeping. A generated project's copied pipeline.py reproduces these on
@@ -125,6 +127,7 @@ def build(
     uncertain_completeness: bool = False,
     source_dir: Path | None = None,
     road_distance_m: float = ROAD_DISTANCE_CANONICAL_M,
+    sampled_run: bool = False,
 ) -> None:
     output_dir.mkdir(parents=True, exist_ok=True)
     (output_dir / "data" / "source").mkdir(parents=True, exist_ok=True)
@@ -645,8 +648,42 @@ def local_pin(name: str) -> dict:
         "inputs": _inventory(output_dir, input_paths),
         "outputs": _inventory(output_dir, output_paths),
     }
+    if break_mode == "sampled_run_as_canonical":
+        # The laundering this guards against: a fast smoke run over a clipped
+        # AOI presented as the analysis of record. Its numbers describe a
+        # slice, so runs.latest must never point at it.
+        run_record["mode"] = "sampled"
+        run_record["sample"] = {
+            "requested": {"sample_area": SAMPLE_AREA},
+            "realized": {"bbox": SAMPLE_AREA, "rows": max(1, row_count // 10)},
+            "scale_factor": 0.1,
+        }
     (output_dir / "runs" / f"{run_id}.json").write_text(json.dumps(run_record, indent=2), encoding="utf-8")
 
+    if sampled_run:
+        # A sampled run sitting beside the canonical one: legal, marked, and
+        # not the run of record.
+        sampled_id = "run-20260825-074500"
+        (output_dir / "runs" / f"{sampled_id}.json").write_text(
+            json.dumps(
+                {
+                    "run_id": sampled_id,
+                    "started_at": "2026-08-25T07:45:00Z",
+                    "completed_at": "2026-08-25T07:45:02Z",
+                    "status": "passed",
+                    "mode": "sampled",
+                    "sample": {
+                        "requested": {"sample_area": SAMPLE_AREA},
+                        "realized": {"bbox": SAMPLE_AREA, "rows": max(1, row_count // 10)},
+                        "scale_factor": 0.1,
+                    },
+                    "environment": {"python": platform.python_version(), "duckdb": duckdb.__version__},
+                },
+                indent=2,
+            ),
+            encoding="utf-8",
+        )
+
     if break_mode == "qgis_broken_datasource":
         qgs_xml = (
             '<?xml version="1.0"?><qgis><projectlayers>'
@@ -1178,13 +1215,15 @@ def main() -> int:
                          help="drop POI completeness counts and add a completeness warning")
     parser.add_argument("--road-distance-m", type=float, default=float(ROAD_DISTANCE_CANONICAL_M),
                          help="road-distance threshold in metres (declared runtime parameter)")
+    parser.add_argument("--sampled-run", action="store_true",
+                         help="also emit a sampled run record beside the canonical one")
     args = parser.parse_args()
 
     if args.output_dir.exists():
         shutil.rmtree(args.output_dir)
     build(args.output_dir, apply_override=not args.no_override, break_mode=args.break_mode,
           with_scenario_road=args.scenario_road, uncertain_completeness=args.uncertain_completeness,
-          road_distance_m=args.road_distance_m)
+          road_distance_m=args.road_distance_m, sampled_run=args.sampled_run)
     print(f"wrote {args.output_dir}")
     return 0
 

openmapstack/checks/validation.py

@@ -176,6 +176,51 @@ def run_record_matches(
     return passed(f"report run_id {run_id!r} matches a real run record with consistent hashes")
 
 
+def sample_run_not_promoted(
+    workspace: Path, project_dir: str = ".", runs_dir: str = "runs"
+) -> AssertionResult:
+    """A sampled run is never the canonical run of record.
+
+    Sampling clips or thins the inputs, so its outputs are evidence that the
+    pipeline executes, never evidence of the answer. `runs.latest` is what
+    `verify`, the clean-rerun protocol, and every expectation attestation bind
+    to, so a sampled record reaching it would launder a smoke test into a
+    result. A sampled record must also state what it *realized*, not only what
+    was requested.
+    """
+    from openmapstack.sampling import run_mode, run_record_errors
+
+    proj = load_project_yaml(workspace, project_dir)
+    if proj is None:
+        return failed("project.yaml missing", code="manifest_missing")
+    root = project_root(workspace, project_dir)
+    directory = root / runs_dir
+    if not directory.is_dir():
+        return not_testable(f"no {runs_dir}/ directory", code="runs_dir_missing")
+
+    latest_id = str(get_in(proj, "runs.latest.id", "") or "")
+    sampled: list[str] = []
+    problems: list[str] = []
+    for record_path in sorted(directory.glob("*.json")):
+        record = load_json(record_path)
+        if not isinstance(record, dict):
+            continue
+        problems.extend(f"{record_path.name}: {problem}" for problem in run_record_errors(record))
+        if run_mode(record) == "sampled":
+            sampled.append(record_path.stem)
+
+    if latest_id and latest_id in sampled:
+        return failed(
+            f"runs.latest is sampled run {latest_id!r}; a sampled run cannot be the canonical run",
+            code="sampled_run_promoted",
+        )
+    if problems:
+        return failed(f"invalid run mode/sample declaration: {problems}", code="sample_record_invalid")
+    if not sampled:
+        return passed("no sampled run records; runs.latest is canonical")
+    return passed(f"{len(sampled)} sampled run record(s) present and none is runs.latest")
+
+
 def no_prose_only_validation(
     workspace: Path, check_id: str, project_dir: str = ".", report_path: str = "validation/latest-report.json"
 ) -> AssertionResult: