From 29a68d4286d0e0f60cdd21a1bf816d09af67a4aa Mon Sep 17 00:00:00 2001 From: Jaak Laineste Date: Thu, 3 Sep 2026 12:56:44 +0000 Subject: [PATCH] feat(run): sampled runs that can never become the canonical run 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 that failure arrives in minutes. The risk this introduces is laundering: a fast pass over a clipped AOI presented as the analysis. Sampling clips or thins the inputs, so a sampled run proves the pipeline executes and nothing more -- clipping breaks every neighbourhood operation at the cut, row sampling destroys the spatial coherence a join depends on, and raster downsampling changes areas and slopes non-linearly. So non-promotion is enforced structurally, not by convention: - a sampled run record declares `mode: sampled` and must state what it *realized*, not only what was requested (TABLESAMPLE only approximates); - `runs.latest` may never reference one. It is what `verify`, the clean-rerun protocol, and every expectation attestation bind to; - `run --sample*` re-reads the manifest afterwards and fails if the pipeline promoted itself, and reports declared outputs it overwrote in place. Sampling reuses the existing `runtime.implementation.parameters` contract rather than adding a schema block: a parameter opts in with a `role` (sample_area | sample_rows | sample_fraction) and an optional `sample:` default. `canonical` must mean "no sampling", so the canonical run still passes nothing. A sampling parameter cannot pair step/field -- it selects input, not a processing threshold. `--dry-run` is deliberately untouched: it already means "print the command, execute nothing", and redefining a published flag would be worse than adding four honest ones. Reported as `runs.sample_isolation` by validate, and exposed to external harnesses as the additive `validation.sample_run_not_promoted` check. Evidence: 40 unit tests; eval 016-sampled-run (control) and 928-sampled-run-as-canonical (mutation, isolated). Full unit suite 452 passing; fixture evals 16/16 contract_ci, 26/26 mutation. Closes #20 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01TMMy6A73zi7yGjK3wHBhju --- README.md | 28 +- evals/COVERAGE.md | 9 + evals/cases/016-sampled-run/expected.yaml | 25 + .../expected.yaml | 26 + evals/fixtures/reference_pipeline/gen.py | 41 +- openmapstack/checks/validation.py | 45 ++ openmapstack/cli.py | 191 ++++++- openmapstack/parameters.py | 66 ++- openmapstack/sampling.py | 192 +++++++ openmapstack/schemas/project-v1.schema.json | 4 +- openmapstack/validation.py | 47 ++ openmapstack/verify.py | 1 + references/project-spec.md | 60 ++- templates/project.yaml | 10 + tests/goldens/verify/district-facilities.json | 13 +- tests/goldens/verify/district-facilities.txt | 3 +- tests/goldens/verify/river-crossings.json | 11 +- tests/goldens/verify/river-crossings.txt | 3 +- tests/test_cli.py | 2 +- tests/test_sampling.py | 493 ++++++++++++++++++ 20 files changed, 1250 insertions(+), 20 deletions(-) create mode 100644 evals/cases/016-sampled-run/expected.yaml create mode 100644 evals/cases/928-sampled-run-as-canonical/expected.yaml create mode 100644 openmapstack/sampling.py create mode 100644 tests/test_sampling.py diff --git a/README.md b/README.md index 2c0060f..c0882ee 100644 --- a/README.md +++ b/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 diff --git a/evals/COVERAGE.md b/evals/COVERAGE.md index 1f3ad99..8659a42 100644 --- a/evals/COVERAGE.md +++ b/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 diff --git a/evals/cases/016-sampled-run/expected.yaml b/evals/cases/016-sampled-run/expected.yaml new file mode 100644 index 0000000..9ceb278 --- /dev/null +++ b/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 diff --git a/evals/cases/928-sampled-run-as-canonical/expected.yaml b/evals/cases/928-sampled-run-as-canonical/expected.yaml new file mode 100644 index 0000000..11da60f --- /dev/null +++ b/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 diff --git a/evals/fixtures/reference_pipeline/gen.py b/evals/fixtures/reference_pipeline/gen.py index c9241b9..9fc64ae 100755 --- a/evals/fixtures/reference_pipeline/gen.py +++ b/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 = ( '' @@ -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 diff --git a/openmapstack/checks/validation.py b/openmapstack/checks/validation.py index ab60a3f..13b02ed 100644 --- a/openmapstack/checks/validation.py +++ b/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: diff --git a/openmapstack/cli.py b/openmapstack/cli.py index 2b2f508..3cc3213 100644 --- a/openmapstack/cli.py +++ b/openmapstack/cli.py @@ -4,6 +4,7 @@ import argparse import json +import os import shlex import subprocess import sys @@ -15,7 +16,8 @@ from typing import Any from . import __version__ -from .project import ProjectError, get_in, load_project, project_path, step_outputs +from .project import ProjectError, get_in, load_json, load_project, project_path, step_outputs +from .sampling import Sample, SamplingError, declared_sample, resolve_sample, run_record_errors from .validation import ValidationResult, validate_project from .verify import VerifyResult, verify_project @@ -57,6 +59,34 @@ def build_parser() -> argparse.ArgumentParser: metavar="ARG", help="pass one argument to the pipeline; repeat as needed (use --pipeline-arg=--flag for flags)", ) + sample_group = run_parser.add_argument_group( + "sampled runs", + "Run the same pipeline over a deliberately smaller slice so a late failure " + "arrives in minutes. A sampled run proves the pipeline executes; it never " + "establishes the analysis result and can never become the canonical run.", + ) + sample_group.add_argument( + "--sample", + action="store_true", + help="sampled run using every sampling parameter's declared `sample:` value", + ) + sample_group.add_argument( + "--sample-area", + metavar="BBOX", + help="sampled run over this test AOI (binds the role: sample_area parameter)", + ) + sample_group.add_argument( + "--sample-rows", + type=int, + metavar="N", + help="sampled run capped at N rows (binds the role: sample_rows parameter)", + ) + sample_group.add_argument( + "--sample-fraction", + type=float, + metavar="PERCENT", + help="sampled run over PERCENT of rows (binds the role: sample_fraction parameter)", + ) run_parser.set_defaults(handler=_cmd_run) verify_parser = subparsers.add_parser( @@ -247,36 +277,54 @@ def _cmd_run(args: argparse.Namespace) -> int: try: project_file, project = load_project(args.project) + sample = _requested_sample(args, project) command = _pipeline_command(project_file, project, args.pipeline_args) - except ProjectError as exc: + if sample is not None: + command = command + sample.argv + except (ProjectError, SamplingError) as exc: if args.json: print(_json({"schema": "openmapstack-run-result/v1", "status": "failed", "phase": "preflight", "error": str(exc)})) else: print(f"openmapstack run: {exc}", file=sys.stderr) return 2 + mode = "sampled" if sample is not None else "canonical" display_command = shlex.join(command) if args.dry_run: payload = { "schema": "openmapstack-run-result/v1", "status": preflight.status, "phase": "dry_run", + "mode": mode, "project_file": str(project_file), "cwd": str(project_file.parent), "command": command, "validation": preflight.to_dict(), } + if sample is not None: + payload["sample"] = sample.to_dict() if args.json: print(_json(payload)) else: print(f"Preflight: {preflight.status}") - print(f"Would run: {display_command}") + print(f"Would run ({mode}): {display_command}") return 0 + environment = None + if sample is not None: + environment = dict(os.environ) + environment.update(sample.environment) + environment["OPENMAPSTACK_RUN_MODE"] = "sampled" + + # A sampled run may overwrite the declared outputs in place. Fingerprint + # them first so the clobber is reported here rather than surfacing later + # as an unexplained outputs_hash mismatch. + outputs_before = _declared_output_digests(project_file.parent, project) if sample is not None else {} + latest_before = get_in(project, "runs", "latest", "id") if not args.json: print(f"Preflight: {preflight.status}") - print(f"Running: {display_command}") + print(f"Running ({mode}): {display_command}") try: completed = subprocess.run( command, @@ -284,6 +332,7 @@ def _cmd_run(args: argparse.Namespace) -> int: check=False, text=True, capture_output=args.json, + env=environment, ) except OSError as exc: if args.json: @@ -298,6 +347,7 @@ def _cmd_run(args: argparse.Namespace) -> int: "schema": "openmapstack-run-result/v1", "status": "failed", "phase": "execute", + "mode": mode, "project_file": str(project_file), "command": command, "returncode": completed.returncode, @@ -310,11 +360,17 @@ def _cmd_run(args: argparse.Namespace) -> int: print(f"Pipeline failed with exit code {completed.returncode}.", file=sys.stderr) return 1 + if sample is not None: + return _report_sampled_run( + args, project_file, command, completed, sample, outputs_before, latest_before, preflight + ) + validation = validate_project(project_file, artifacts=True) payload = { "schema": "openmapstack-run-result/v1", "status": validation.status, "phase": "complete", + "mode": mode, "project_file": str(project_file), "command": command, "returncode": completed.returncode, @@ -329,6 +385,133 @@ def _cmd_run(args: argparse.Namespace) -> int: return 0 if validation.ok(strict=args.strict) else 1 +def _requested_sample(args: argparse.Namespace, project: dict[str, Any]) -> Sample | None: + """The sampled run the command line asked for, or ``None`` for a canonical run.""" + explicit: dict[str, Any] = {} + for role, value in ( + ("sample_area", args.sample_area), + ("sample_rows", args.sample_rows), + ("sample_fraction", args.sample_fraction), + ): + if value is not None: + explicit[role] = value + if not explicit and not args.sample: + return None + requested: dict[str, Any] = dict(declared_sample(project)) if args.sample else {} + requested.update(explicit) + if not requested: + raise SamplingError( + "--sample needs a runtime.implementation.parameters entry with a sampling " + "role and a `sample:` value, or an explicit --sample-area/--sample-rows/" + "--sample-fraction" + ) + return resolve_sample(project, requested) + + +def _declared_output_digests(root: Path, project: dict[str, Any]) -> dict[str, str]: + """SHA-256 of every declared output that exists right now.""" + from .integrity import declared_output_paths, sha256_file + + digests: dict[str, str] = {} + for relative in declared_output_paths(project): + target = root / relative + if target.is_file(): + digests[relative] = sha256_file(target) + return digests + + +def _report_sampled_run( + args: argparse.Namespace, + project_file: Path, + command: list[str], + completed: subprocess.CompletedProcess, + sample: Sample, + outputs_before: dict[str, str], + latest_before: Any, + preflight: ValidationResult, +) -> int: + """Report a sampled run, refusing to let it stand in for the canonical one. + + The artifacts now describe a slice, so re-auditing them as the finished + analysis would be asking the wrong question -- the reported validation is + the pre-run one, which is the last point at which the tree described the + canonical project. What a sampled run is graded on instead is narrow: the + pipeline must not have promoted itself into ``runs.latest``, and any record + it wrote must state what it realized. + """ + root = project_file.parent + validation = preflight + status = "passed" + problems: list[str] = [] + + try: + _, after = load_project(project_file) + except ProjectError as exc: + after = {} + problems.append(f"project.yaml is unreadable after the sampled run: {exc}") + latest_after = get_in(after, "runs", "latest", "id") if after else None + if after and latest_after != latest_before: + problems.append( + f"the sampled run moved runs.latest from {latest_before!r} to {latest_after!r}; " + f"a sampled run cannot become the canonical run of record" + ) + + # Any run record the pipeline just wrote must declare its realized sample; + # reuse the shipped check so the CLI and the check API cannot drift. + for record_path in sorted((root / "runs").glob("*.json")) if (root / "runs").is_dir() else []: + try: + record = load_json(record_path) + except ProjectError: + continue + if isinstance(record, dict): + problems.extend(f"runs/{record_path.name}: {problem}" for problem in run_record_errors(record)) + + outputs_after = _declared_output_digests(root, after or {}) + overwritten = sorted( + path + for path, digest in outputs_after.items() + if path in outputs_before and outputs_before[path] != digest + ) + if problems: + status = "failed" + + payload = { + "schema": "openmapstack-run-result/v1", + "status": status, + "phase": "complete", + "mode": "sampled", + "project_file": str(project_file), + "command": command, + "returncode": completed.returncode, + "sample": sample.to_dict(), + "canonical_outputs_overwritten": overwritten, + "promotion_problems": problems, + "validation": validation.to_dict(), + "validation_phase": "preflight", + } + if args.json: + payload["stdout"] = completed.stdout + payload["stderr"] = completed.stderr + print(_json(payload)) + else: + for problem in problems: + print(f"FAIL runs.sample_isolation: {problem}", file=sys.stderr) + if not problems: + print( + "Sampled run complete. It proves the pipeline executes; it does not " + "establish the analysis result." + ) + if overwritten: + print( + "WARN the declared outputs now hold sampled data: " + f"{', '.join(overwritten)}. Re-run without --sample before validating.", + file=sys.stderr, + ) + if problems: + return 1 + return 1 if (overwritten and args.strict) else 0 + + def _cmd_skill_snapshot(args: argparse.Namespace) -> int: from .snapshot import SnapshotError, create_skill_snapshot, find_skill_root, inspect_skill_snapshot diff --git a/openmapstack/parameters.py b/openmapstack/parameters.py index 5cfb715..712f4d7 100644 --- a/openmapstack/parameters.py +++ b/openmapstack/parameters.py @@ -30,6 +30,31 @@ processing step must exist and its field must equal ``canonical``, so a manifest cannot advertise one threshold while the step declares another. +A parameter may additionally declare a sampling ``role``, which is how +``openmapstack run --sample*`` addresses the knob that shrinks the work: + +.. code-block:: yaml + + parameters: + - id: sample_area + type: string + canonical: "" # the canonical run samples nothing + role: sample_area # sample_area | sample_rows | sample_fraction + sample: "26.68,58.35,26.76,58.39" # what bare --sample uses + binding: {argument: "--sample-area"} + +Sampling rules on top of the above: + +- at most one parameter may claim each role; +- ``canonical`` must be the role's *no sampling* value (``""`` for a string, + ``0`` for a number), because a canonical run passes nothing and must + therefore process the full inputs; +- ``sample`` is optional, must differ from ``canonical``, and is what bare + ``--sample`` binds; without it the role needs an explicit value on the + command line; +- a sampling parameter must not pair ``step``/``field``: it selects *input*, + not a processing threshold, so there is no step value to agree with. + The canonical run passes nothing: a pipeline must produce the accepted result with no arguments and no variables set. Bindings exist so a *variant* run can say "same pipeline, this one knob turned". @@ -45,11 +70,16 @@ PARAMETERS_SCHEMA = "openmapstack-parameters/v1" PARAMETER_TYPES = ("integer", "number", "string") +#: Sampling knobs ``openmapstack run`` can address by role rather than by id. +SAMPLE_ROLES = ("sample_area", "sample_rows", "sample_fraction") _IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$") _ARGUMENT = re.compile(r"^--[a-z][a-z0-9]*(?:-[a-z0-9]+)*$") _ENVIRONMENT = re.compile(r"^[A-Z][A-Z0-9_]*$") +#: The value each type carries when a sampling role is switched off. +_NO_SAMPLING: dict[str, Any] = {"integer": 0, "number": 0, "string": ""} + class ParameterError(ValueError): """The parameters block is malformed or drifts from the processing steps.""" @@ -64,6 +94,8 @@ class Parameter: environment: str | None = None step: str | None = None field: str | None = None + role: str | None = None + sample: Any = None def bind(self, value: Any) -> tuple[list[str], dict[str, str]]: """Return the argv suffix and environment additions for ``value``.""" @@ -111,13 +143,14 @@ def declared_parameters(manifest: dict[str, Any]) -> list[Parameter]: } errors: list[str] = [] seen: set[str] = set() + seen_roles: dict[str, str] = {} parameters: list[Parameter] = [] for index, entry in enumerate(raw): where = f"parameters[{index}]" if not isinstance(entry, dict): errors.append(f"{where} must be a mapping") continue - unknown = set(entry) - {"id", "type", "canonical", "binding", "step", "field", "description"} + unknown = set(entry) - {"id", "type", "canonical", "binding", "step", "field", "description", "role", "sample"} if unknown: errors.append(f"{where} has unknown keys {sorted(unknown)}") parameter_id = entry.get("id") @@ -151,9 +184,36 @@ def declared_parameters(manifest: dict[str, Any]) -> list[Parameter]: environment = None else: errors.append(f"{where}.binding must declare exactly one of argument/environment") + role = entry.get("role") + sample = entry.get("sample") + if role is not None: + if role not in SAMPLE_ROLES: + errors.append(f"{where}.role must be one of {list(SAMPLE_ROLES)}, got {role!r}") + role = None + elif role in seen_roles: + errors.append(f"{where}.role {role!r} is already claimed by parameter {seen_roles[role]!r}") + else: + seen_roles[role] = parameter_id + if entry["canonical"] != _NO_SAMPLING[type_name]: + errors.append( + f"{where}.canonical must be {_NO_SAMPLING[type_name]!r} for a sampling role: " + f"the canonical run samples nothing" + ) + if "sample" in entry: + if role is None: + errors.append(f"{where}.sample needs a sampling role; a sample value alone is unaddressable") + elif not value_has_type(sample, type_name): + errors.append(f"{where}.sample must be a {type_name}") + elif sample == entry["canonical"]: + errors.append(f"{where}.sample equals canonical, so it would not sample anything") step = entry.get("step") field = entry.get("field") - if (step is None) != (field is None): + if role is not None and (step is not None or field is not None): + errors.append( + f"{where} is a sampling parameter and must not pair step/field: " + f"it selects input, not a processing threshold" + ) + elif (step is None) != (field is None): errors.append(f"{where} must declare step and field together") elif step is not None: if not isinstance(step, str) or step not in steps_by_id: @@ -176,6 +236,8 @@ def declared_parameters(manifest: dict[str, Any]) -> list[Parameter]: environment=environment, step=step if isinstance(step, str) else None, field=field if isinstance(field, str) else None, + role=role if isinstance(role, str) else None, + sample=sample if "sample" in entry else None, ) ) if errors: diff --git a/openmapstack/sampling.py b/openmapstack/sampling.py new file mode 100644 index 0000000..49a4a54 --- /dev/null +++ b/openmapstack/sampling.py @@ -0,0 +1,192 @@ +"""Sampled ("nail it before you scale it") runs, and why they can never be canonical. + +A wide-area, high-resolution analysis can run for hours before a late step +fails. A sampled run executes the *same* pipeline over a deliberately smaller +slice so that failure arrives in minutes instead. + +What a sampled run proves and does not prove +-------------------------------------------- +It proves the pipeline *executes*: the manifest graph resolves, the sources are +reachable, CRS handling holds, schemas line up, pagination works within the +slice. It does **not** prove the numbers. Clipping to a test AOI breaks every +neighbourhood operation at the cut, row sampling destroys the spatial coherence +a join depends on, and raster downsampling changes areas and slopes +non-linearly. So a sampled run's outputs are never an answer. + +How that is enforced +-------------------- +Structurally, not by convention: + +- a sampled run legitimately produces a different ``inputs_hash`` -- clipped + inputs are different bytes -- so it cannot share the canonical hash chain; +- its run record carries ``mode: sampled`` plus the **realized** sample, not + merely the requested one (``TABLESAMPLE`` and friends only approximate); +- ``runs.latest`` may never point at a sampled record, so ``verify``, the + clean-rerun protocol, and ``validation.expectations`` attestations -- which + bind to ``runs.latest.inputs_hash`` -- cannot inherit a sampled baseline; +- ``openmapstack run --sample*`` re-reads the manifest afterwards and fails + loudly if the pipeline promoted itself into ``runs.latest``. + +If a sampled run overwrites the declared outputs in place, the existing +``outputs_hash`` machinery already refuses to call the project validated: the +files no longer hash to what the canonical run recorded. The CLI reports that +clobber explicitly so the later failure is not a surprise. + +Choosing a *representative* sample is judgment, not arithmetic -- a naive +sub-bbox of a global dataset lands in open ocean -- so this module never picks +one. It binds the value the manifest or the operator supplied. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Mapping + +from .parameters import SAMPLE_ROLES, Parameter, ParameterError, declared_parameters, value_has_type + +#: Run kinds a run record may declare. Absent means ``canonical``. +RUN_MODES = ("canonical", "sampled") + +#: Sentinel for "use the value the manifest declared for this role". +USE_DECLARED = object() + +_ROLE_FLAGS = { + "sample_area": "--sample-area", + "sample_rows": "--sample-rows", + "sample_fraction": "--sample-fraction", +} + + +class SamplingError(ValueError): + """A sampled run was asked for that the manifest cannot express.""" + + +@dataclass(frozen=True) +class Sample: + """A resolved sampled run: what to pass, and what was asked for.""" + + argv: list[str] = field(default_factory=list) + environment: dict[str, str] = field(default_factory=dict) + requested: dict[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return {"requested": dict(self.requested), "argv": list(self.argv), "environment": dict(self.environment)} + + +def sampling_parameters(manifest: dict[str, Any]) -> dict[str, Parameter]: + """The declared sampling knobs, keyed by role. + + Raises ``ParameterError`` if the parameter block is malformed; role + uniqueness is enforced there, so the mapping is unambiguous. + """ + return { + parameter.role: parameter + for parameter in declared_parameters(manifest) + if parameter.role is not None + } + + +def resolve_sample(manifest: dict[str, Any], requested: Mapping[str, Any]) -> Sample: + """Bind ``requested`` (role -> value or ``USE_DECLARED``) to pipeline arguments. + + Fails rather than guessing: an undeclared role, or a bare ``--sample`` + against a manifest that declares no default, is an error naming exactly + what the manifest is missing. + """ + try: + available = sampling_parameters(manifest) + except ParameterError as exc: + raise SamplingError(str(exc)) from exc + + unknown = sorted(role for role in requested if role not in SAMPLE_ROLES) + if unknown: + raise SamplingError(f"unknown sampling role(s): {unknown}") + if not requested: + raise SamplingError("no sampling was requested") + + argv: list[str] = [] + environment: dict[str, str] = {} + resolved: dict[str, Any] = {} + for role in SAMPLE_ROLES: + if role not in requested: + continue + parameter = available.get(role) + if parameter is None: + raise SamplingError( + f"{_ROLE_FLAGS[role]} needs a runtime.implementation.parameters entry with " + f"role: {role}; the manifest declares " + f"{sorted(available) or 'no sampling parameters'}" + ) + value = requested[role] + if value is USE_DECLARED: + if parameter.sample is None: + raise SamplingError( + f"parameter {parameter.id!r} declares role {role!r} but no sample value; " + f"pass {_ROLE_FLAGS[role]} explicitly or add `sample:` to the manifest" + ) + value = parameter.sample + elif not value_has_type(value, parameter.type): + raise SamplingError(f"{_ROLE_FLAGS[role]} must be a {parameter.type} for parameter {parameter.id!r}") + extra_argv, extra_environment = parameter.bind(value) + argv.extend(extra_argv) + environment.update(extra_environment) + resolved[role] = value + return Sample(argv=argv, environment=environment, requested=resolved) + + +def declared_sample(manifest: dict[str, Any]) -> dict[str, Any]: + """The roles bare ``--sample`` would bind, i.e. those with a ``sample`` value.""" + try: + available = sampling_parameters(manifest) + except ParameterError as exc: + raise SamplingError(str(exc)) from exc + return {role: USE_DECLARED for role, parameter in available.items() if parameter.sample is not None} + + +def run_mode(run_record: Mapping[str, Any]) -> str: + """The run kind a record declares. A record with no ``mode`` is canonical.""" + mode = run_record.get("mode") + if mode is None: + return "canonical" + return str(mode) + + +def run_record_errors(run_record: Mapping[str, Any]) -> list[str]: + """Everything wrong with a run record's sampling declaration. + + A canonical record must not carry a sample descriptor, and a sampled one + must record what it *realized* -- the actual rows, AOI, or resolution -- + because a requested fraction is a request, not a measurement. + """ + mode = run_mode(run_record) + errors: list[str] = [] + if mode not in RUN_MODES: + return [f"invalid run mode {mode!r}, expected one of {list(RUN_MODES)}"] + sample = run_record.get("sample") + if mode == "canonical": + if sample is not None: + errors.append("a canonical run record must not carry a sample descriptor") + return errors + if not isinstance(sample, dict): + errors.append("a sampled run record must carry a sample object") + return errors + requested = sample.get("requested") + if not isinstance(requested, dict) or not requested: + errors.append("sample.requested must be a non-empty object") + else: + unknown = sorted(role for role in requested if role not in SAMPLE_ROLES) + if unknown: + errors.append(f"sample.requested has unknown role(s): {unknown}") + realized = sample.get("realized") + if not isinstance(realized, dict) or not realized: + errors.append( + "sample.realized must be a non-empty object: record what the run actually " + "sampled, not only what was asked for" + ) + scale_factor = sample.get("scale_factor") + if scale_factor is not None: + if isinstance(scale_factor, bool) or not isinstance(scale_factor, (int, float)): + errors.append("sample.scale_factor must be a number") + elif not 0 < scale_factor <= 1: + errors.append(f"sample.scale_factor must be within (0, 1], got {scale_factor!r}") + return errors diff --git a/openmapstack/schemas/project-v1.schema.json b/openmapstack/schemas/project-v1.schema.json index 9853486..9e6acd7 100644 --- a/openmapstack/schemas/project-v1.schema.json +++ b/openmapstack/schemas/project-v1.schema.json @@ -48,9 +48,11 @@ }, "step": {"type": "string", "minLength": 1}, "field": {"$ref": "#/$defs/identifier"}, + "role": {"enum": ["sample_area", "sample_rows", "sample_fraction"]}, + "sample": {"type": ["integer", "number", "string"]}, "description": {"type": "string"} }, - "dependentRequired": {"step": ["field"], "field": ["step"]} + "dependentRequired": {"step": ["field"], "field": ["step"], "sample": ["role"]} }, "metamorphicRelation": { "type": "object", diff --git a/openmapstack/validation.py b/openmapstack/validation.py index 3b5cfda..b65c18a 100644 --- a/openmapstack/validation.py +++ b/openmapstack/validation.py @@ -19,6 +19,7 @@ sha256_file, ) from .project import ProjectError, get_in, load_json, load_project, project_path, step_outputs +from .sampling import run_mode, run_record_errors from .schema import project_schema_errors from .sources import assess_pin, connection_reference_error, find_inline_credentials @@ -159,6 +160,7 @@ def run(self) -> ValidationResult: self._run_record() else: self._run_record_present_files() + self._sample_isolation() self._declared_status_consistency() return ValidationResult(self.project_file, self.checks) @@ -960,6 +962,51 @@ def _verify_run_inventory( ) return verified + def _sample_isolation(self) -> None: + """A sampled run may never stand in for the canonical one. + + Sampling clips or thins the inputs, so a sampled run's numbers are not + the analysis. Keeping it out of ``runs.latest`` is what stops it from + being inherited as a baseline by ``verify``, the clean-rerun protocol, + and every ``validation.expectations`` attestation bound to + ``runs.latest.inputs_hash``. + """ + runs_dir = self.root / "runs" + latest_id = str(get_in(self.project, "runs", "latest", "id") or "") + records = sorted(runs_dir.glob("*.json")) if runs_dir.is_dir() else [] + errors: list[str] = [] + sampled: list[str] = [] + for record_path in records: + relative = record_path.relative_to(self.root).as_posix() + try: + record = load_json(record_path) + except ProjectError: + continue # unreadable records are reported by runs.latest + if not isinstance(record, dict): + continue + problems = run_record_errors(record) + if problems: + errors.extend(f"{relative}: {problem}" for problem in problems) + if run_mode(record) == "sampled": + sampled.append(record_path.stem) + if latest_id and latest_id in sampled: + errors.insert( + 0, + f"runs.latest points at sampled run {latest_id}; a sampled run cannot be " + f"the canonical run of record", + ) + if errors: + self.add("runs.sample_isolation", "failed", "; ".join(errors), path="runs", errors=errors) + elif sampled: + self.add( + "runs.sample_isolation", + "passed", + f"{len(sampled)} sampled run record(s) declare a realized sample and none is runs.latest", + path="runs", + ) + else: + self.add("runs.sample_isolation", "passed", "no sampled run records", path="runs") + def _declared_status_consistency(self) -> None: project_status = get_in(self.project, "project", "status") non_passed = [check for check in self.checks if check.status in {"warning", "not_testable"}] diff --git a/openmapstack/verify.py b/openmapstack/verify.py index 5d8ecbb..0864cce 100644 --- a/openmapstack/verify.py +++ b/openmapstack/verify.py @@ -243,6 +243,7 @@ def verify_project( ("no_implicit_pass", validation_checks.no_implicit_pass), ("warning_or_failed_propagates_to_status", validation_checks.warning_or_failed_propagates_to_status), ("run_record_matches", validation_checks.run_record_matches), + ("sample_run_not_promoted", validation_checks.sample_run_not_promoted), ): _run(runs, f"validation.{name}", fn, root) diff --git a/references/project-spec.md b/references/project-spec.md index f39dc79..938d2d2 100644 --- a/references/project-spec.md +++ b/references/project-spec.md @@ -659,6 +659,12 @@ runtime: binding: {argument: "--road-distance-m"} # or {environment: OMS_ROAD_DISTANCE_M} step: road_distance # optional pair: the step that consumes it field: max_distance_m # ... whose value must equal `canonical` + - id: sample_area # optional; the knob a sampled run turns + type: string + canonical: "" # a canonical run samples nothing + role: sample_area # sample_area | sample_rows | sample_fraction + sample: "26.68,58.35,26.76,58.39" # what bare `--sample` binds + binding: {argument: "--sample-area"} environment: python: "3.13" duckdb: "1.2.x" @@ -666,7 +672,7 @@ runtime: proj: "9.4.x" runs: - latest: + latest: # the CANONICAL run; never a sampled one id: run-20260825-081503 started_at: "..." completed_at: "..." @@ -698,6 +704,58 @@ it. When `step`/`field` are given, `openmapstack verify` fails `project.parameters_match_steps` if the step's declared value drifts from `canonical` — the same honesty rule as `presentation.controls`. +#### Sampled runs + +A wide-area, high-resolution analysis can run for hours before a late step +fails. A **sampled run** executes the same pipeline over a deliberately smaller +slice so that failure arrives in minutes. A parameter opts into this by +declaring a `role`, which is how `openmapstack run --sample`, +`--sample-area`, `--sample-rows`, and `--sample-fraction` address it. (Note +that `--dry-run` is a different thing entirely: it prints the command and +executes nothing.) + +Sampling rules on top of the parameter contract above: + +- at most one parameter may claim each role; +- `canonical` must be the role's *no sampling* value (`""` for a string, `0` + for a number) — a canonical run passes nothing, so it must process the + full inputs; +- `sample` is optional, must differ from `canonical`, and is what bare + `--sample` binds; without it the role needs an explicit value on the + command line; +- a sampling parameter must **not** pair `step`/`field`: it selects *input*, + not a processing threshold, so there is no step value to agree with. + +**A sampled run proves the pipeline executes. It never establishes the +result.** Clipping to a test AOI breaks every neighbourhood operation at the +cut; row sampling destroys the spatial coherence a join depends on; +downsampling a raster changes areas and slopes non-linearly. Sampled counts and +aggregates are therefore not answers and must not be surfaced as such or bound +into `validation.expectations`. + +That is enforced structurally rather than by convention. A sampled run +legitimately produces a different `inputs_hash` — clipped inputs are different +bytes — so it cannot share the canonical hash chain. On top of that: + +- its `runs/.json` record MUST declare `mode: sampled` (a record with no + `mode` is canonical) and a `sample` object carrying `requested` and + **`realized`**, plus an optional `scale_factor` in `(0, 1]`. Recording only + what was requested is a failure: `TABLESAMPLE` and its equivalents + approximate, so the realized rows, AOI, or resolution are the measurement; +- a canonical record MUST NOT carry a `sample` object; +- `runs.latest` MUST NOT reference a sampled record. It 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. + +`openmapstack validate` reports this as `runs.sample_isolation`; the same +invariant is exposed to external harnesses as the +`validation.sample_run_not_promoted` check. `openmapstack run --sample*` +additionally re-reads the manifest afterwards and fails if the pipeline +promoted its own sampled run, and reports any declared outputs the sampled run +overwrote in place — those files no longer hash to what the canonical run +recorded, so the project is correctly no longer `validated` until it is re-run +in full. + **Warnings** give the explicit confidence/incompleteness handling. The rendered UX surfaces them (don't imply autoconfirmed geodata is current/complete). **Runs** capture what changed between executions and let a new engineer `rerun` tomorrow. The corresponding `runs/.json` record MUST contain `inputs` and `outputs` diff --git a/templates/project.yaml b/templates/project.yaml index 66890a0..63d4607 100644 --- a/templates/project.yaml +++ b/templates/project.yaml @@ -269,6 +269,16 @@ runtime: # binding: {argument: "--buffer-m"} # step: select_candidates # field: max_distance_m + # # A sampling knob, addressed by `openmapstack run --sample-area`. It + # # selects input rather than a threshold, so it takes no step/field, and + # # `canonical` must mean "no sampling". A sampled run proves the + # # pipeline executes; it never establishes the result. + # - id: sample_area + # type: string + # canonical: "" + # role: sample_area # sample_area | sample_rows | sample_fraction + # sample: "26.68,58.35,26.76,58.39" + # binding: {argument: "--sample-area"} environment: python: "3.13" duckdb: TODO diff --git a/tests/goldens/verify/district-facilities.json b/tests/goldens/verify/district-facilities.json index af86daf..da02ada 100644 --- a/tests/goldens/verify/district-facilities.json +++ b/tests/goldens/verify/district-facilities.json @@ -95,6 +95,11 @@ "message": "report run_id 'run-20260901-000001' matches a real run record with consistent hashes", "status": "passed" }, + { + "check": "validation.sample_run_not_promoted", + "message": "no sampled run records; runs.latest is canonical", + "status": "passed" + }, { "args": { "check": "geodata.feature_field_equals", @@ -159,13 +164,13 @@ "counts": { "failed": 0, "not_testable": 1, - "passed": 24, + "passed": 25, "warning": 1 }, "coverage": { - "applicable": 26, - "executed": 25, - "execution_rate": 0.9615384615384616, + "applicable": 27, + "executed": 26, + "execution_rate": 0.9629629629629629, "not_testable": 1 }, "project_file": "$PROJECT/project.yaml", diff --git a/tests/goldens/verify/district-facilities.txt b/tests/goldens/verify/district-facilities.txt index dd1122d..5da45dd 100644 --- a/tests/goldens/verify/district-facilities.txt +++ b/tests/goldens/verify/district-facilities.txt @@ -16,6 +16,7 @@ PASS validation.required_all_present: all 3 declared checks present exactly once PASS validation.no_implicit_pass: every check has an explicit status PASS validation.warning_or_failed_propagates_to_status: overall status 'warning' correctly reflects check statuses PASS validation.run_record_matches: report run_id 'run-20260901-000001' matches a real run record with consistent hashes +PASS validation.sample_run_not_promoted: no sampled run records; runs.latest is canonical WARN expectation.d1-count [data/derived/district-counts.geojson]: expectation 'd1-count' is unverified; independent review must bind expectation_sha256 to sha256:10399dfde7e9808a58d5ff9447545d111f87f3ef4d3277f10ae21d4c69bacaaa PASS geodata.crs_not_used_for_metrics: analysis_crs EPSG:3301 is valid for 0 metric operation(s); storage/load/reprojection steps were excluded PASS geodata.geometry_all_valid [data/derived/district-counts.geojson]: data/derived/district-counts.geojson: all 2 features have valid geometry @@ -24,5 +25,5 @@ PASS presentation.layers_use_semantic_roles: all 2 layers declare a semantic_rol PASS presentation.controls_match_pipeline: 0 filter control(s) and 0 scenario control(s) consistent PASS presentation.edit_targets_reference_real_sources: no edit targets declared (vacuously true) PASS rerun.no_chat_dependency: canonical project dependencies contain no chat/transcript references -WARNING: $PROJECT/project.yaml (24 passed, 1 warnings, 1 not testable, 0 failed; 25/26 applicable checks executed) +WARNING: $PROJECT/project.yaml (25 passed, 1 warnings, 1 not testable, 0 failed; 26/27 applicable checks executed) NOTE some checks could not run here; install openmapstack[geo] for geodata checks, QGIS for the .qgz checks diff --git a/tests/goldens/verify/river-crossings.json b/tests/goldens/verify/river-crossings.json index b5f1013..a7f45c2 100644 --- a/tests/goldens/verify/river-crossings.json +++ b/tests/goldens/verify/river-crossings.json @@ -100,6 +100,11 @@ "message": "report run_id 'run-20260901-000000' matches a real run record with consistent hashes", "status": "passed" }, + { + "check": "validation.sample_run_not_promoted", + "message": "no sampled run records; runs.latest is canonical", + "status": "passed" + }, { "check": "geodata.crs_not_used_for_metrics", "message": "analysis_crs EPSG:3301 is valid for 0 metric operation(s); storage/load/reprojection steps were excluded", @@ -170,12 +175,12 @@ "counts": { "failed": 0, "not_testable": 0, - "passed": 28, + "passed": 29, "warning": 0 }, "coverage": { - "applicable": 28, - "executed": 28, + "applicable": 29, + "executed": 29, "execution_rate": 1.0, "not_testable": 0 }, diff --git a/tests/goldens/verify/river-crossings.txt b/tests/goldens/verify/river-crossings.txt index d53c6c8..7bd6acc 100644 --- a/tests/goldens/verify/river-crossings.txt +++ b/tests/goldens/verify/river-crossings.txt @@ -17,6 +17,7 @@ PASS validation.required_all_present: all 2 declared checks present exactly once PASS validation.no_implicit_pass: every check has an explicit status PASS validation.warning_or_failed_propagates_to_status: overall status 'passed' correctly reflects check statuses PASS validation.run_record_matches: report run_id 'run-20260901-000000' matches a real run record with consistent hashes +PASS validation.sample_run_not_promoted: no sampled run records; runs.latest is canonical PASS geodata.crs_not_used_for_metrics: analysis_crs EPSG:3301 is valid for 0 metric operation(s); storage/load/reprojection steps were excluded PASS geodata.geometry_all_valid [data/derived/crossing-trails.geojson]: data/derived/crossing-trails.geojson: all 3 features have valid geometry PASS geodata.dataset_crs_is [data/derived/crossing-trails.geojson]: data/derived/crossing-trails.geojson actual CRS metadata is EPSG:3301 @@ -26,4 +27,4 @@ PASS presentation.edit_targets_reference_real_sources: no edit targets declared PASS metamorphic.declarations_valid: 1 metamorphic relation(s) are well-formed PASS metamorphic.trail-order: trail-order: input_permutation_invariance holds across 1 output(s) PASS rerun.no_chat_dependency: canonical project dependencies contain no chat/transcript references -PASSED: $PROJECT/project.yaml (28 passed, 0 warnings, 0 not testable, 0 failed; 28/28 applicable checks executed) +PASSED: $PROJECT/project.yaml (29 passed, 0 warnings, 0 not testable, 0 failed; 29/29 applicable checks executed) diff --git a/tests/test_cli.py b/tests/test_cli.py index b9768fb..f87ea66 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -296,7 +296,7 @@ def test_run_dry_run_does_not_execute(self) -> None: with redirect_stdout(stdout): exit_code = main(["run", str(path), "--dry-run", "--pipeline-arg=--sample"]) self.assertEqual(exit_code, 0) - self.assertIn("Would run:", stdout.getvalue()) + self.assertIn("Would run (canonical):", stdout.getvalue()) self.assertIn("--sample", stdout.getvalue()) self.assertFalse((self.root / "data" / "derived" / "candidate.json").exists()) diff --git a/tests/test_sampling.py b/tests/test_sampling.py new file mode 100644 index 0000000..c5be8ae --- /dev/null +++ b/tests/test_sampling.py @@ -0,0 +1,493 @@ +"""Sampled runs: the parameter contract, and the non-promotion invariant. + +A sampled run exists so a wide-area analysis fails in minutes instead of +hours. The thing worth testing hardest is not that sampling works, but that a +sampled result can never be mistaken for the analysis: `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 an answer. +""" + +from __future__ import annotations + +import io +import json +import os +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path + +import yaml + +from openmapstack.api import describe_check, run_check +from openmapstack.cli import main +from openmapstack.parameters import ParameterError, declared_parameters +from openmapstack.sampling import ( + USE_DECLARED, + SamplingError, + declared_sample, + resolve_sample, + run_mode, + run_record_errors, + sampling_parameters, +) +from openmapstack.validation import validate_project +from tests.test_cli import materialize_artifacts, valid_manifest + + +def _parameters(*entries: dict) -> dict: + manifest = valid_manifest() + manifest["runtime"]["implementation"]["parameters"] = list(entries) + return manifest + + +AREA = { + "id": "sample_area", + "type": "string", + "canonical": "", + "role": "sample_area", + "sample": "26.68,58.35,26.76,58.39", + "binding": {"argument": "--sample-area"}, +} +ROWS = { + "id": "sample_rows", + "type": "integer", + "canonical": 0, + "role": "sample_rows", + "binding": {"environment": "OMS_SAMPLE_ROWS"}, +} + + +class SamplingParameterTests(unittest.TestCase): + def test_role_and_sample_round_trip(self) -> None: + parameters = declared_parameters(_parameters(AREA)) + self.assertEqual(parameters[0].role, "sample_area") + self.assertEqual(parameters[0].sample, "26.68,58.35,26.76,58.39") + + def test_two_parameters_cannot_claim_the_same_role(self) -> None: + second = dict(AREA, id="other_area", binding={"argument": "--other-area"}) + with self.assertRaises(ParameterError) as caught: + declared_parameters(_parameters(AREA, second)) + self.assertIn("already claimed", str(caught.exception)) + + def test_canonical_must_mean_no_sampling(self) -> None: + """The canonical run passes nothing, so it must process the full input.""" + with self.assertRaises(ParameterError) as caught: + declared_parameters(_parameters(dict(AREA, canonical="26.0,58.0,27.0,59.0"))) + self.assertIn("samples nothing", str(caught.exception)) + + def test_sample_without_a_role_is_unaddressable(self) -> None: + entry = { + "id": "threshold", + "type": "number", + "canonical": 2000, + "sample": 500, + "binding": {"argument": "--threshold"}, + } + with self.assertRaises(ParameterError) as caught: + declared_parameters(_parameters(entry)) + self.assertIn("needs a sampling role", str(caught.exception)) + + def test_sample_equal_to_canonical_is_rejected(self) -> None: + with self.assertRaises(ParameterError) as caught: + declared_parameters(_parameters(dict(AREA, sample=""))) + self.assertIn("would not sample anything", str(caught.exception)) + + def test_a_sampling_parameter_cannot_pair_step_and_field(self) -> None: + """Sampling selects input; it is not a processing threshold, so there is + no step value for `parameters_match_steps` to agree with.""" + entry = dict(AREA, step="load", field="source") + with self.assertRaises(ParameterError) as caught: + declared_parameters(_parameters(entry)) + self.assertIn("must not pair step/field", str(caught.exception)) + + def test_sampling_parameters_are_keyed_by_role(self) -> None: + self.assertEqual(sorted(sampling_parameters(_parameters(AREA, ROWS))), ["sample_area", "sample_rows"]) + + +class ResolveSampleTests(unittest.TestCase): + def test_explicit_value_binds_to_the_declared_argument(self) -> None: + sample = resolve_sample(_parameters(AREA), {"sample_area": "1,2,3,4"}) + self.assertEqual(sample.argv, ["--sample-area", "1,2,3,4"]) + self.assertEqual(sample.requested, {"sample_area": "1,2,3,4"}) + + def test_declared_default_is_used_for_bare_sample(self) -> None: + manifest = _parameters(AREA) + sample = resolve_sample(manifest, declared_sample(manifest)) + self.assertEqual(sample.argv, ["--sample-area", "26.68,58.35,26.76,58.39"]) + + def test_a_role_without_a_declared_sample_is_not_a_bare_sample_default(self) -> None: + """ROWS declares a role but no `sample:`, so bare --sample cannot use it.""" + self.assertEqual(declared_sample(_parameters(ROWS)), {}) + + def test_environment_binding_is_honoured(self) -> None: + sample = resolve_sample(_parameters(ROWS), {"sample_rows": 500}) + self.assertEqual(sample.argv, []) + self.assertEqual(sample.environment, {"OMS_SAMPLE_ROWS": "500"}) + + def test_undeclared_role_names_what_the_manifest_must_add(self) -> None: + with self.assertRaises(SamplingError) as caught: + resolve_sample(_parameters(AREA), {"sample_rows": 10}) + message = str(caught.exception) + self.assertIn("role: sample_rows", message) + self.assertIn("--sample-rows", message) + + def test_wrong_type_is_refused(self) -> None: + with self.assertRaises(SamplingError) as caught: + resolve_sample(_parameters(ROWS), {"sample_rows": "many"}) + self.assertIn("must be a integer", str(caught.exception)) + + def test_use_declared_without_a_declared_sample_fails(self) -> None: + with self.assertRaises(SamplingError) as caught: + resolve_sample(_parameters(ROWS), {"sample_rows": USE_DECLARED}) + self.assertIn("no sample value", str(caught.exception)) + + +class RunRecordModeTests(unittest.TestCase): + def test_a_record_without_a_mode_is_canonical(self) -> None: + self.assertEqual(run_mode({}), "canonical") + self.assertEqual(run_record_errors({}), []) + + def test_canonical_record_must_not_carry_a_sample(self) -> None: + errors = run_record_errors({"mode": "canonical", "sample": {"requested": {}}}) + self.assertEqual(len(errors), 1) + self.assertIn("must not carry a sample descriptor", errors[0]) + + def test_sampled_record_must_record_what_it_realized(self) -> None: + """A requested fraction is a request; TABLESAMPLE only approximates it.""" + errors = run_record_errors( + {"mode": "sampled", "sample": {"requested": {"sample_fraction": 1.0}}} + ) + self.assertEqual(len(errors), 1) + self.assertIn("sample.realized", errors[0]) + + def test_a_complete_sampled_record_is_accepted(self) -> None: + self.assertEqual( + run_record_errors( + { + "mode": "sampled", + "sample": { + "requested": {"sample_fraction": 1.0}, + "realized": {"rows": 987, "resolution_m": 30}, + "scale_factor": 0.0125, + }, + } + ), + [], + ) + + def test_impossible_scale_factor_is_rejected(self) -> None: + errors = run_record_errors( + { + "mode": "sampled", + "sample": {"requested": {"sample_rows": 10}, "realized": {"rows": 10}, "scale_factor": 4}, + } + ) + self.assertIn("within (0, 1]", errors[0]) + + def test_unknown_mode_is_rejected(self) -> None: + self.assertIn("invalid run mode", run_record_errors({"mode": "moist"})[0]) + + +class SampleIsolationTests(unittest.TestCase): + """`runs.latest` is the canonical baseline; a sampled run may never hold it.""" + + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory(prefix="openmapstack-sampling-test-") + self.root = Path(self.tempdir.name) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def _project(self) -> Path: + path = self.root / "project.yaml" + path.write_text(yaml.safe_dump(valid_manifest(), sort_keys=False), encoding="utf-8") + (self.root / "README.md").write_text("# Test project\n", encoding="utf-8") + (self.root / "pipeline.py").write_text("pass\n", encoding="utf-8") + materialize_artifacts(self.root) + return path + + def _check(self, path: Path, name: str = "runs.sample_isolation"): + result = validate_project(path) + return next(check for check in result.checks if check.id == name) + + def _add_sampled_record(self, run_id: str, sample: dict | None = None) -> Path: + record = { + "run_id": run_id, + "started_at": "2026-08-26T00:00:00Z", + "completed_at": "2026-08-26T00:00:01Z", + "status": "passed", + "mode": "sampled", + "sample": sample + if sample is not None + else {"requested": {"sample_area": "1,2,3,4"}, "realized": {"rows": 12}}, + } + target = self.root / "runs" / f"{run_id}.json" + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(json.dumps(record), encoding="utf-8") + return target + + def test_a_sampled_record_beside_the_canonical_one_is_fine(self) -> None: + path = self._project() + self._add_sampled_record("run-20260826-120000") + check = self._check(path) + self.assertEqual(check.status, "passed", check.message) + + def test_a_sampled_run_promoted_to_runs_latest_fails(self) -> None: + path = self._project() + manifest = yaml.safe_load(path.read_text(encoding="utf-8")) + promoted = manifest["runs"]["latest"]["id"] + record = json.loads((self.root / "runs" / f"{promoted}.json").read_text(encoding="utf-8")) + record["mode"] = "sampled" + record["sample"] = {"requested": {"sample_area": "1,2,3,4"}, "realized": {"rows": 12}} + (self.root / "runs" / f"{promoted}.json").write_text(json.dumps(record), encoding="utf-8") + check = self._check(path) + self.assertEqual(check.status, "failed") + self.assertIn("cannot be the canonical run", check.message) + + def test_a_sampled_record_stating_only_its_request_fails(self) -> None: + path = self._project() + self._add_sampled_record( + "run-20260826-120000", sample={"requested": {"sample_fraction": 1.0}} + ) + check = self._check(path) + self.assertEqual(check.status, "failed") + self.assertIn("sample.realized", check.message) + + def test_a_project_with_no_sampled_records_passes(self) -> None: + check = self._check(self._project()) + self.assertEqual(check.status, "passed") + self.assertIn("no sampled run records", check.message) + + # -- the same invariant through the public check API ------------------- + + def test_check_api_exposes_the_invariant(self) -> None: + descriptor = describe_check("validation.sample_run_not_promoted") + self.assertEqual(descriptor.dimension, "validation_integrity") + self.assertTrue(descriptor.oracle_free) + + def test_check_api_reports_a_promoted_sampled_run(self) -> None: + path = self._project() + manifest = yaml.safe_load(path.read_text(encoding="utf-8")) + promoted = manifest["runs"]["latest"]["id"] + record = json.loads((self.root / "runs" / f"{promoted}.json").read_text(encoding="utf-8")) + record["mode"] = "sampled" + record["sample"] = {"requested": {"sample_area": "1,2,3,4"}, "realized": {"rows": 12}} + (self.root / "runs" / f"{promoted}.json").write_text(json.dumps(record), encoding="utf-8") + result = run_check("validation.sample_run_not_promoted", self.root) + self.assertEqual(result["status"], "failed") + self.assertEqual(result["code"], "sampled_run_promoted") + + def test_check_api_passes_a_healthy_project(self) -> None: + self._project() + self._add_sampled_record("run-20260826-120000") + result = run_check("validation.sample_run_not_promoted", self.root) + self.assertEqual(result["status"], "passed") + + def test_check_api_pins_a_code_for_an_unrealized_sample(self) -> None: + """External harnesses grade on the code, so it is part of the contract.""" + self._project() + self._add_sampled_record( + "run-20260826-120000", sample={"requested": {"sample_rows": 100}} + ) + result = run_check("validation.sample_run_not_promoted", self.root) + self.assertEqual(result["status"], "failed") + self.assertEqual(result["code"], "sample_record_invalid") + + def test_check_api_is_not_testable_without_run_records(self) -> None: + """Absent evidence is never an implicit pass.""" + path = self._project() + for record in (self.root / "runs").glob("*.json"): + record.unlink() + (self.root / "runs").rmdir() + result = run_check("validation.sample_run_not_promoted", self.root) + self.assertEqual(result["status"], "not_testable") + self.assertEqual(result["code"], "runs_dir_missing") + self.assertTrue(path.is_file()) + + +SAMPLING_PIPELINE = """\ +import json +import os +import sys +from pathlib import Path + +root = Path(__file__).resolve().parent +argv = sys.argv[1:] +area = argv[argv.index("--sample-area") + 1] if "--sample-area" in argv else None +sampled = area is not None + +(root / "received.json").write_text(json.dumps({ + "argv": argv, + "run_mode": os.environ.get("OPENMAPSTACK_RUN_MODE"), +}), encoding="utf-8") + +output = root / "data" / "derived" / "candidate.json" +output.parent.mkdir(parents=True, exist_ok=True) +output.write_text(json.dumps({"type": "FeatureCollection", "features": [], "area": area}), encoding="utf-8") + +if sampled: + run_id = "run-20260826-120000" + record = { + "run_id": run_id, + "started_at": "2026-08-26T12:00:00Z", + "completed_at": "2026-08-26T12:00:01Z", + "status": "passed", + "mode": "sampled", + "sample": {"requested": {"sample_area": area}}, + "environment": {"python": "test"}, + } + if os.environ.get("EVAL_BAD_SAMPLE") != "1": + record["sample"]["realized"] = {"rows": 3, "bbox": area} + record["sample"]["scale_factor"] = 0.01 + run_path = root / "runs" / (run_id + ".json") + run_path.parent.mkdir(parents=True, exist_ok=True) + run_path.write_text(json.dumps(record), encoding="utf-8") + if os.environ.get("EVAL_PROMOTE_SAMPLE") == "1": + import yaml + manifest_path = root / "project.yaml" + manifest = yaml.safe_load(manifest_path.read_text(encoding="utf-8")) + manifest["runs"]["latest"]["id"] = run_id + manifest_path.write_text(yaml.safe_dump(manifest, sort_keys=False), encoding="utf-8") +""" + + +class SampledRunCliTests(unittest.TestCase): + """`openmapstack run --sample*`: bind the knob, then refuse promotion.""" + + def setUp(self) -> None: + self.tempdir = tempfile.TemporaryDirectory(prefix="openmapstack-sample-cli-") + self.root = Path(self.tempdir.name) + + def tearDown(self) -> None: + self.tempdir.cleanup() + + def _project(self, *parameters: dict) -> Path: + manifest = valid_manifest() + if parameters: + manifest["runtime"]["implementation"]["parameters"] = list(parameters) + path = self.root / "project.yaml" + path.write_text(yaml.safe_dump(manifest, sort_keys=False), encoding="utf-8") + (self.root / "README.md").write_text("# Test project\n", encoding="utf-8") + (self.root / "pipeline.py").write_text(SAMPLING_PIPELINE, encoding="utf-8") + materialize_artifacts(self.root) + return path + + def _run(self, argv: list[str]) -> tuple[int, dict]: + stdout, stderr = io.StringIO(), io.StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + code = main(argv + ["--json"]) + return code, json.loads(stdout.getvalue()) + + def _run_with_env(self, argv: list[str], **environment: str) -> tuple[int, dict]: + """Run with extra variables the fixture pipeline reads to misbehave.""" + previous = {key: os.environ.get(key) for key in environment} + os.environ.update(environment) + try: + return self._run(argv) + finally: + for key, value in previous.items(): + if value is None: + os.environ.pop(key, None) + else: + os.environ[key] = value + + def test_explicit_sample_area_reaches_the_pipeline(self) -> None: + path = self._project(AREA) + code, payload = self._run(["run", str(path), "--sample-area", "1,2,3,4"]) + self.assertEqual(code, 0, payload) + self.assertEqual(payload["mode"], "sampled") + self.assertEqual(payload["sample"]["requested"], {"sample_area": "1,2,3,4"}) + received = json.loads((self.root / "received.json").read_text(encoding="utf-8")) + self.assertEqual(received["argv"], ["--sample-area", "1,2,3,4"]) + self.assertEqual(received["run_mode"], "sampled") + + def test_bare_sample_uses_the_declared_value(self) -> None: + path = self._project(AREA) + code, payload = self._run(["run", str(path), "--sample"]) + self.assertEqual(code, 0, payload) + self.assertEqual(payload["sample"]["requested"], {"sample_area": AREA["sample"]}) + + def test_a_canonical_run_passes_nothing(self) -> None: + """Declaring a sampling parameter must not leak into the canonical run: + no argument, no environment variable, no sample descriptor.""" + path = self._project(AREA) + _, payload = self._run(["run", str(path)]) + self.assertEqual(payload["mode"], "canonical") + self.assertNotIn("sample", payload) + received = json.loads((self.root / "received.json").read_text(encoding="utf-8")) + self.assertEqual(received["argv"], []) + self.assertIsNone(received["run_mode"]) + + def test_sampling_a_manifest_that_declares_no_role_is_refused(self) -> None: + path = self._project() + stdout, stderr = io.StringIO(), io.StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + code = main(["run", str(path), "--sample-area", "1,2,3,4"]) + self.assertEqual(code, 2) + self.assertIn("role: sample_area", stderr.getvalue()) + self.assertFalse((self.root / "received.json").exists()) + + def test_bare_sample_without_a_declared_default_is_refused(self) -> None: + path = self._project(ROWS) + stdout, stderr = io.StringIO(), io.StringIO() + with redirect_stdout(stdout), redirect_stderr(stderr): + code = main(["run", str(path), "--sample"]) + self.assertEqual(code, 2) + self.assertIn("--sample needs", stderr.getvalue()) + + def test_dry_run_shows_the_sampled_command_without_executing(self) -> None: + path = self._project(AREA) + stdout = io.StringIO() + with redirect_stdout(stdout): + code = main(["run", str(path), "--sample", "--dry-run"]) + self.assertEqual(code, 0) + self.assertIn("Would run (sampled):", stdout.getvalue()) + self.assertIn(AREA["sample"], stdout.getvalue()) + self.assertFalse((self.root / "received.json").exists()) + + def test_a_pipeline_that_promotes_its_sampled_run_fails_the_command(self) -> None: + """The strongest guard: it does not rely on the pipeline behaving.""" + path = self._project(AREA) + code, payload = self._run_with_env( + ["run", str(path), "--sample-area", "1,2,3,4"], EVAL_PROMOTE_SAMPLE="1" + ) + self.assertEqual(code, 1) + self.assertEqual(payload["status"], "failed") + self.assertTrue(payload["promotion_problems"]) + self.assertIn("cannot become the canonical run", payload["promotion_problems"][0]) + + def test_clobbering_the_declared_outputs_is_reported(self) -> None: + """A sampled run that writes in place leaves the canonical outputs stale; + say so here rather than letting it surface as a hash mismatch later.""" + path = self._project(AREA) + code, payload = self._run(["run", str(path), "--sample-area", "1,2,3,4"]) + self.assertEqual(code, 0, payload) + self.assertEqual(payload["canonical_outputs_overwritten"], ["data/derived/candidate.json"]) + + def test_strict_turns_a_clobbered_output_into_a_failure(self) -> None: + path = self._project(AREA) + code, payload = self._run(["run", str(path), "--sample-area", "1,2,3,4", "--strict"]) + self.assertEqual(code, 1) + self.assertEqual(payload["canonical_outputs_overwritten"], ["data/derived/candidate.json"]) + + def test_a_record_stating_only_its_request_fails_the_command(self) -> None: + """A requested fraction is a request; the record must say what it got.""" + path = self._project(AREA) + code, payload = self._run_with_env( + ["run", str(path), "--sample-area", "1,2,3,4"], EVAL_BAD_SAMPLE="1" + ) + self.assertEqual(code, 1) + self.assertTrue(any("sample.realized" in problem for problem in payload["promotion_problems"])) + + def test_a_sampled_run_leaves_runs_latest_alone(self) -> None: + path = self._project(AREA) + before = yaml.safe_load(path.read_text(encoding="utf-8"))["runs"]["latest"]["id"] + self._run(["run", str(path), "--sample-area", "1,2,3,4"]) + after = yaml.safe_load(path.read_text(encoding="utf-8"))["runs"]["latest"]["id"] + self.assertEqual(before, after) + self.assertTrue((self.root / "runs" / "run-20260826-120000.json").is_file()) + + +if __name__ == "__main__": # pragma: no cover + unittest.main()