From ed85a5d126287491f159cf218684d630dc778a67 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 00:53:25 -0700 Subject: [PATCH 01/50] Add routing requirements to probe schema Allow probes to request reach-indexed variables supplied through output/routing.csv. --- schemas/probe.schema.json | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/schemas/probe.schema.json b/schemas/probe.schema.json index a6e14475..435f5e04 100644 --- a/schemas/probe.schema.json +++ b/schemas/probe.schema.json @@ -103,6 +103,14 @@ "uniqueItems": true, "description": "Diagnostics read by criteria, such as a surface or layer temperature, or the liquid share of a snowpack. Never added to any water-storage sum: `lwsnl` is water, but it is water already counted inside `snw`, so summing it beside `snw` would count the same kilogram twice. Each criterion defines their time handling." }, + "routing": { + "type": "array", + "items": { + "type": "string" + }, + "uniqueItems": true, + "description": "Reach-indexed variables read from output/routing.csv. q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + }, "forcing": { "type": "array", "items": { From e9805c12425905d24029b58c27c44a7dc6faa937 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 00:56:10 -0700 Subject: [PATCH 02/50] Add routing emissions to model schema Allow routing-capable models to declare reach-indexed output written to output/routing.csv. --- schemas/model.schema.json | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/schemas/model.schema.json b/schemas/model.schema.json index be1eb500..aad89523 100644 --- a/schemas/model.schema.json +++ b/schemas/model.schema.json @@ -40,6 +40,10 @@ "type": "array", "items": {"type": "string"}, "uniqueItems": true, "description": "Absolute storage values, never tendencies. The harness differences them." }, + "routing": { + "type": "array", "items": {"type": "string"}, "uniqueItems": true, + "description": "Reach-indexed variables written to output/routing.csv. q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + }, "diagnostics": { "type": "array", "items": {"type": "string"}, "uniqueItems": true, "description": "Diagnostics such as a surface or layer temperature, or the liquid share of a snowpack. Never added to any water-storage sum: `lwsnl` is water, but it is water already counted inside `snw`, so summing it beside `snw` would count the same kilogram twice. Each criterion defines their time handling." From 7038a45b9cd7f9dff96b6471ccd3031fe414ddd7 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 01:05:23 -0700 Subject: [PATCH 03/50] Load reach-indexed routing capabilities Define the routing variables and units, load requires.routing and emits.routing, and include routing capabilities in model-probe compatibility checks. --- src/hydroturing/spec.py | 27 +++++++++++++++++++++++++-- 1 file changed, 25 insertions(+), 2 deletions(-) diff --git a/src/hydroturing/spec.py b/src/hydroturing/spec.py index 8ef87254..e6332d09 100644 --- a/src/hydroturing/spec.py +++ b/src/hydroturing/spec.py @@ -39,11 +39,18 @@ # pack rather than evidence of a phase change. DIAG_VARS = ("ts", "tsoil_layer", "stage", "lwsnl", "csnow") +# Reach-indexed routing variables live in output/routing.csv rather than the +# one-row-per-timestep result table. +ROUTING_VARS = ("q_in", "q_out", "channel_storage") + UNITS = { "pr": "mm day-1", "evspsbl": "mm day-1", "mrro": "mm day-1", "dis": "m3 s-1", + "q_in": "m3 s-1", + "q_out": "m3 s-1", + "channel_storage": "m3", "gwex": "mm day-1", # Net river-aquifer exchange, positive into the aquifer. Unlike gwex # (a source or sink crossing the catchment boundary), this moves water @@ -243,6 +250,8 @@ class ProbeSpec: min_window_days: int = 0 # Missing diagnostic outputs cause INCOMPLETE, as for missing fluxes. requires_diagnostics: tuple[str, ...] = () + # Reach-indexed variables requested from output/routing.csv. + requires_routing: tuple[str, ...] = () # Case-supplied inputs the verdict rests on: forcing columns and # static.json keys a model must declare it consumes, or it is judged # against values it never read and is INCOMPATIBLE instead. @@ -346,6 +355,8 @@ class ModelManifest: window_days: int | str | None = None # Diagnostics the model reports in addition to its fluxes and states. emits_diagnostics: tuple[str, ...] = () + # Reach-indexed variables written to output/routing.csv. + emits_routing: tuple[str, ...] = () # Static inputs the adapter cannot run without. needs_static: tuple[str, ...] = () # Optional inputs the adapter consumes whenever the case supplies them. @@ -371,7 +382,11 @@ def missing_for(self, probe: ProbeSpec) -> list[str]: INCOMPLETE: the model cannot demonstrate conservation because it never says enough to be checked, and it has not violated it either. """ - return [v for v in probe.required_vars if v not in self.emitted] + missing = [v for v in probe.required_vars if v not in self.emitted] + missing.extend( + v for v in probe.requires_routing if v not in self.emits_routing + ) + return missing def load_probe(path: str | Path) -> ProbeSpec: @@ -457,7 +472,12 @@ def load_probe(path: str | Path) -> ProbeSpec: # asked for as a state would be met by the flux and never noticed. unknown = [ name - for key, known in (("fluxes", FLUX_VARS), ("states", STATE_VARS), ("diagnostics", DIAG_VARS)) + for key, known in ( + ("fluxes", FLUX_VARS), + ("states", STATE_VARS), + ("diagnostics", DIAG_VARS), + ("routing", ROUTING_VARS), + ) for name in requires.get(key, []) if name not in known ] @@ -474,6 +494,7 @@ def load_probe(path: str | Path) -> ProbeSpec: requires_fluxes=tuple(requires.get("fluxes", [])), requires_states=tuple(requires.get("states", [])), requires_diagnostics=tuple(requires.get("diagnostics", [])), + requires_routing=tuple(requires.get("routing", [])), requires_forcing=tuple(requires.get("forcing", [])), requires_static=tuple(requires.get("static", [])), generator=case["generator"], @@ -549,6 +570,7 @@ def load_model(path: str | Path) -> ModelManifest: unknown = [v for v in raw["emits"]["fluxes"] if v not in FLUX_VARS] unknown += [v for v in raw["emits"]["states"] if v not in STATE_VARS] unknown += [v for v in raw["emits"].get("diagnostics", []) if v not in DIAG_VARS] + unknown += [v for v in raw["emits"].get("routing", []) if v not in ROUTING_VARS] if unknown: raise SpecError(f"{spec_file}: unknown variables in emits: {unknown}") @@ -565,6 +587,7 @@ def load_model(path: str | Path) -> ModelManifest: emits_fluxes=tuple(raw["emits"]["fluxes"]), emits_states=tuple(raw["emits"]["states"]), emits_diagnostics=tuple(raw["emits"].get("diagnostics", [])), + emits_routing=tuple(raw["emits"].get("routing", [])), runner=runner, needs_forcing=tuple(raw.get("needs_forcing", [])), needs_static=tuple(raw.get("needs_static", [])), From 250a4b25cc51051921f3492e53320926c92e0719 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 01:21:05 -0700 Subject: [PATCH 04/50] Add reach-indexed routing output protocol Expose output/routing.csv in request.json, accept it in the output allow-list, and validate reach IDs, time coverage, requested columns, duplicates, and finite values. --- src/hydroturing/protocol.py | 121 ++++++++++++++++++++++++++++++++++-- 1 file changed, 116 insertions(+), 5 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 192eb424..cc1c2008 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -9,6 +9,7 @@ input/forcing.csv <- harness writes, read only input/static.json <- harness writes, read only output/result.csv -> adapter writes + output/routing.csv -> adapter writes, when reach-indexed output is requested output/run.json -> adapter writes CSV is the required format. NetCDF is accepted when present but never @@ -35,10 +36,11 @@ STATIC_FILE = "input/static.json" RESULT_CSV = "output/result.csv" RESULT_NC = "output/result.nc" +ROUTING_CSV = "output/routing.csv" RUN_FILE = "output/run.json" TIME_COL = "time" - +REACH_COL = "reach_id" class ProtocolError(RuntimeError): """The adapter did not honour the contract.""" @@ -101,6 +103,7 @@ class RunResult: table: pd.DataFrame meta: dict[str, Any] wall_seconds: float + routing: pd.DataFrame | None = None def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> Path: @@ -130,10 +133,19 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P "fluxes": list(model.emits_fluxes), "states": list(model.emits_states), "diagnostics": list(model.emits_diagnostics), + "routing": list(model.emits_routing), }, "input": {"forcing": FORCING_FILE, "static": STATIC_FILE}, - "output": {"table": RESULT_CSV, "run": RUN_FILE}, - "units": {v: UNITS[v] for v in model.emitted if v in UNITS}, + "output": { + "table": RESULT_CSV, + "routing": ROUTING_CSV, + "run": RUN_FILE, + }, + "units": { + v: UNITS[v] + for v in (*model.emitted, *model.emits_routing) + if v in UNITS + }, "notes": ( "States are absolute storages, not tendencies. The harness " "differences them itself. Row i's ts and rlus are instantaneous " @@ -141,6 +153,8 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P "when supplied. Row i's tsoil_layer is the mean temperature of " "the specified layer at the interval end; hfg and hfg_bottom " "are interval-mean boundary fluxes for the soil-storage check." + "Routing q_in and q_out are interval-mean m3 s-1; " + "channel_storage is absolute end-of-step m3." ), } request_path = io_dir / REQUEST_FILE @@ -155,7 +169,7 @@ def _validate_output_files(io_dir: Path, probe: ProbeSpec) -> None: if not output_dir.exists(): return - allowed = {RESULT_CSV, RESULT_NC, RUN_FILE} + allowed = {RESULT_CSV, RESULT_NC, ROUTING_CSV, RUN_FILE} total = 0 for path in output_dir.iterdir(): relative = f"output/{path.name}" @@ -199,6 +213,96 @@ def _validate_time_axis(table: pd.DataFrame, case: Case) -> None: ) +def _read_routing( + io_dir: Path, + case: Case, + probe: ProbeSpec, +) -> pd.DataFrame | None: + """Read and validate the optional reach-indexed routing result.""" + path = io_dir / ROUTING_CSV + if not path.exists(): + if probe.requires_routing: + raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") + return None + + table = pd.read_csv(path) + required_columns = [TIME_COL, REACH_COL, *probe.requires_routing] + missing = [ + column for column in required_columns if column not in table.columns + ] + if missing: + raise ProtocolError(f"routing result is missing columns: {missing}") + + if table[REACH_COL].isna().any(): + raise ProtocolError(f"routing result has missing '{REACH_COL}' values") + table[REACH_COL] = table[REACH_COL].astype(str) + + duplicate = table.duplicated([TIME_COL, REACH_COL]) + if duplicate.any(): + raise ProtocolError( + "routing result has duplicate (time, reach_id) rows: " + f"{int(duplicate.sum())} duplicates" + ) + + network = case.static.get("routing_network") + declared_reaches = ( + network.get("reaches") if isinstance(network, dict) else None + ) + if probe.requires_routing and not declared_reaches: + raise ProtocolError( + "routing output requires static.json routing_network.reaches" + ) + + if declared_reaches: + expected_reaches = tuple(str(reach) for reach in declared_reaches) + if len(expected_reaches) != len(set(expected_reaches)): + raise ProtocolError( + "static.json routing_network.reaches contains duplicates" + ) + + actual_reaches = set(table[REACH_COL]) + expected_set = set(expected_reaches) + if actual_reaches != expected_set: + missing_reaches = sorted(expected_set - actual_reaches) + extra_reaches = sorted(actual_reaches - expected_set) + raise ProtocolError( + "routing reach IDs differ from static.json: " + f"missing={missing_reaches}, extra={extra_reaches}" + ) + else: + expected_reaches = tuple(dict.fromkeys(table[REACH_COL])) + + expected_rows = case.n_steps * len(expected_reaches) + if len(table) != expected_rows: + raise ProtocolError( + f"routing result has {len(table)} rows, expected {expected_rows} " + "(one row per forcing step and reach, spinup included)" + ) + + for reach in expected_reaches: + reach_table = table.loc[ + table[REACH_COL] == reach + ].reset_index(drop=True) + if len(reach_table) != case.n_steps: + raise ProtocolError( + f"routing reach '{reach}' has {len(reach_table)} rows, " + f"expected {case.n_steps}" + ) + _validate_time_axis(reach_table, case) + + for var in probe.requires_routing: + column = pd.to_numeric(table[var], errors="coerce") + invalid = ~np.isfinite(column) + if invalid.any(): + raise ProtocolError( + f"routing column '{var}' has " + f"{int(invalid.sum())} non-finite values" + ) + table[var] = column + + return table + + def read_result(io_dir: Path, case: Case, probe: ProbeSpec, wall_seconds: float) -> RunResult: """Read and validate what the adapter produced.""" _validate_output_files(io_dir, probe) @@ -229,6 +333,7 @@ def read_result(io_dir: Path, case: Case, probe: ProbeSpec, wall_seconds: float) ) _validate_time_axis(table, case) + routing = _read_routing(io_dir, case, probe) missing = [v for v in probe.required_vars if v not in table.columns] if missing: @@ -263,4 +368,10 @@ def read_result(io_dir: Path, case: Case, probe: ProbeSpec, wall_seconds: float) f"run.json reports n_steps={meta['n_steps']!r}, expected {case.n_steps}" ) - return RunResult(case=case, table=table, meta=meta, wall_seconds=wall_seconds) + return RunResult( + case=case, + table=table, + meta=meta, + wall_seconds=wall_seconds, + routing=routing, + ) From 6120eb90c99252e4772bc20efcc4bb9b84c9df91 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 01:22:34 -0700 Subject: [PATCH 05/50] Verify declared routing outputs Commit directly to the contract/reach-indexed-routing branch --- src/hydroturing/harness.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/hydroturing/harness.py b/src/hydroturing/harness.py index 3da292e4..90a5db04 100644 --- a/src/hydroturing/harness.py +++ b/src/hydroturing/harness.py @@ -436,6 +436,7 @@ def verify_adapter_contract( requires_fluxes=model.emits_fluxes, requires_states=model.emits_states, requires_diagnostics=model.emits_diagnostics, + requires_routing=model.emits_routing, variants=(), criteria=(), ) From 76f0cc481750b6e9e1e574e812c7aa5f60fb24e4 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 01:41:42 -0700 Subject: [PATCH 06/50] Document the reach-indexed routing contract Define output/routing.csv, its long-table shape and time semantics, routing variable units, emits.routing declarations, and verify-adapter expectations. --- AGENTS.md | 48 +++++++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 47 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index bca697da..c23a5e1f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -39,9 +39,33 @@ Paths inside `request.json` are relative to the request file's directory. and sw_stage_m (m) instead /io/input/static.json read-only: catchment attributes /io/output/result.csv write: one row per forcing row, spinup included +/io/output/routing.csv write when requested: one row per forcing row and reach /io/output/run.json write: {"status": "ok"} ``` +When `request.request.routing` is non-empty, the adapter must write +`output/routing.csv` as a long table with one row for every forcing time and +every reach declared by `static.json` under `routing_network.reaches`. The +required identifier columns are `time` and `reach_id`; the remaining required +columns are the names listed in `request.request.routing`. For example: + +```csv +time,reach_id,q_in,q_out,channel_storage +2001-01-01,A,12.3,8.4,336960 +2001-01-01,B,7.1,4.9,190080 +2001-01-01,C,13.3,9.2,354240 +``` + +Each `(time, reach_id)` pair must be unique. Every declared reach must have +exactly one row for every forcing row, including spinup, and each reach's time +axis must match `forcing.csv`. `q_in` and `q_out` are interval-mean discharge +rates in m3/s. `channel_storage` is the absolute volume stored in that reach at +the end of the interval, in m3. It is not a tendency and must not be converted +to catchment-average depth. A missing requested routing column, reach or time +row is a contract error. Models that do not declare `emits.routing` do not +write `routing.csv`. + + The model seed is deterministic for reproducible stochastic inference, but it is not the generator seed recorded in the host-side report. Exit 0 on success. There is no network. Do not attempt to download weights or data at run time; @@ -148,6 +172,9 @@ numbers in both runs; keep it that way and do not reseed from the clock. | `evspsbl` | evapotranspiration | mm/day | | `mrro` | total runoff | mm/day | | `dis` | river discharge | m3/s | +| `q_in` | interval-mean inflow to the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | +| `q_out` | interval-mean outflow from the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | +| `channel_storage` | absolute water volume stored in the identified reach at the end of the row's time interval; reach-indexed in `routing.csv`, not a tendency and not catchment-average `channel` storage | m3 | | `gwex` | a declared exchange with the outside: regional groundwater, inter-basin transfer; positive into the catchment | mm/day | | `gw_sw_exchange` | net river-aquifer exchange, positive into the aquifer; unlike `gwex`, this moves water between two stores inside the control volume (`gw` and `channel`), so it is never added to `gwex` or counted as a `closure` source | mm/day | | `gw_to_sw` | groundwater-to-river exchange component: the aquifer losing to the river, so it is never positive; a component of `gw_sw_exchange`, not of `gwex`, and not an addition to it | mm/day | @@ -231,6 +258,22 @@ emits: states: [] ``` +A model that also exposes native reach-indexed routing output declares it +separately from the catchment-level result columns: + +```yaml +emits: + fluxes: [mrro, dis] + states: [channel] + routing: [q_in, q_out, channel_storage] +``` + +Variables declared under `emits.routing` belong in `routing.csv`, not in +`result.csv`, `emits.fluxes` or `emits.states`. Declaring them means the +adapter must write the requested long table; do not construct `q_in` or +`channel_storage` from a budget residual merely to satisfy the contract. + + Every probe that needs more than that will be `N/A` with reason `INCOMPLETE`, which is the honest outcome: not a fail, but a probe that cannot be put to the model, so it counts neither way. Fabricating an @@ -269,7 +312,10 @@ ht run --model # the actual evaluation ``` `verify-adapter` runs a single seed, on the same window the evaluation will -use, and checks the shape of what came back. Get that green before looking +use, and checks the shape of what came back. For a model that declares +`emits.routing`, this includes the presence of `routing.csv`, its requested +columns, reach IDs, complete time coverage, unique `(time, reach_id)` pairs +and finite numeric values. Get that green before looking at any residual. Without `--probe` it checks the closure probe, or the first probe the model can consume when it cannot consume that one: a step it does not declare, a forcing or static input the probe does not generate, or a window that drops a From 27fe3fb91c9fad22de7db131a908a899d02772f7 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 01:47:49 -0700 Subject: [PATCH 07/50] Test reach-indexed routing contract Cover routing declarations, request metadata, valid long-table output, missing files and columns, reach and time coverage, duplicate rows, non-finite values, and verify-adapter enforcement. --- tests/test_routing_protocol.py | 222 +++++++++++++++++++++++++++++++++ 1 file changed, 222 insertions(+) create mode 100644 tests/test_routing_protocol.py diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py new file mode 100644 index 00000000..119e1068 --- /dev/null +++ b/tests/test_routing_protocol.py @@ -0,0 +1,222 @@ +"""Reach-indexed routing declarations and /io tables obey the contract.""" + +from __future__ import annotations + +import json +from dataclasses import replace + +import numpy as np +import pandas as pd +import pytest +import yaml + +from hydroturing import registry +from hydroturing.harness import verify_adapter_contract +from hydroturing.protocol import Case, ProtocolError, read_result, stage +from hydroturing.spec import SpecError, load_model, load_probe + + +@pytest.fixture +def base_probe(): + return registry.find_probe("mass/catchment-closure") + + +@pytest.fixture +def routing_probe(base_probe): + return replace( + base_probe, + requires_fluxes=(), + requires_states=(), + requires_diagnostics=(), + requires_routing=("q_in", "q_out", "channel_storage"), + ) + + +@pytest.fixture +def case(): + return Case( + probe_id="mass/routing-contract-test", + seed=11, + forcing=pd.DataFrame( + { + "time": pd.date_range("2000-01-01", periods=3, freq="D"), + "pr": [0.0, 1.0, 0.0], + } + ), + static={"routing_network": {"reaches": ["A", "B"]}}, + spinup_steps=1, + ) + + +def _write_probe(path, base_probe, routing): + path.mkdir(parents=True) + raw = yaml.safe_load((base_probe.path / "probe.yaml").read_text()) + raw["id"] = "mass/routing-contract-test" + raw["requires"] = {"routing": routing} + (path / "generate.py").write_text("# Only manifest loading is tested.\n") + (path / "probe.yaml").write_text(yaml.safe_dump(raw)) + + +def _write_model(path, routing): + path.mkdir() + manifest = { + "name": path.name, + "version": "1", + "entrypoint": ["python", "adapter.py"], + "timestep": "PT1D", + "emits": {"fluxes": [], "states": [], "routing": routing}, + } + (path / "model.yaml").write_text(yaml.safe_dump(manifest)) + + +def _routing_table(case): + rows = [] + for time in case.forcing["time"]: + rows.extend( + [ + { + "time": time.strftime("%Y-%m-%d"), + "reach_id": "A", + "q_in": 2.0, + "q_out": 1.0, + "channel_storage": 86400.0, + }, + { + "time": time.strftime("%Y-%m-%d"), + "reach_id": "B", + "q_in": 3.0, + "q_out": 2.0, + "channel_storage": 86400.0, + }, + ] + ) + return pd.DataFrame(rows) + + +def _write_outputs(path, case, routing=None): + output = path / "output" + output.mkdir(parents=True, exist_ok=True) + pd.DataFrame({"time": case.forcing["time"]}).to_csv( + output / "result.csv", index=False + ) + if routing is not None: + routing.to_csv(output / "routing.csv", index=False) + (output / "run.json").write_text(json.dumps({"status": "ok"})) + + +def test_routing_declarations_load_and_old_defaults_remain_empty( + base_probe, tmp_path +): + probe_dir = tmp_path / "mass" / "routing-contract-test" + variables = ["q_in", "q_out", "channel_storage"] + _write_probe(probe_dir, base_probe, variables) + probe = load_probe(probe_dir) + + model_dir = tmp_path / "routing_model" + _write_model(model_dir, variables) + model = load_model(model_dir) + + assert probe.requires_routing == model.emits_routing == tuple(variables) + assert model.missing_for(probe) == [] + assert base_probe.requires_routing == () + assert registry.find_model("reference_bucket").emits_routing == () + + +@pytest.mark.parametrize("kind", ["probe", "model"]) +def test_unknown_routing_variable_is_rejected(base_probe, tmp_path, kind): + if kind == "probe": + path = tmp_path / "mass" / "routing-contract-test" + _write_probe(path, base_probe, ["not_a_routing_variable"]) + loader = load_probe + else: + path = tmp_path / "routing_model" + _write_model(path, ["not_a_routing_variable"]) + loader = load_model + + with pytest.raises(SpecError, match="unknown variables"): + loader(path) + + +def test_stage_requests_routing_path_and_units(case, base_probe, tmp_path): + model = replace( + registry.find_model("reference_bucket"), + emits_routing=("q_in", "q_out", "channel_storage"), + ) + request = json.loads(stage(tmp_path, case, base_probe, model).read_text()) + + assert request["request"]["routing"] == [ + "q_in", + "q_out", + "channel_storage", + ] + assert request["output"]["routing"] == "output/routing.csv" + assert request["units"]["q_in"] == "m3 s-1" + assert request["units"]["q_out"] == "m3 s-1" + assert request["units"]["channel_storage"] == "m3" + + +def test_valid_routing_table_roundtrips(case, routing_probe, tmp_path): + expected = _routing_table(case) + _write_outputs(tmp_path, case, expected) + + result = read_result(tmp_path, case, routing_probe, 0.0) + + assert result.routing is not None + pd.testing.assert_frame_equal(result.routing, expected) + + +def test_missing_required_routing_file_is_rejected(case, routing_probe, tmp_path): + _write_outputs(tmp_path, case) + with pytest.raises(ProtocolError, match="required output/routing.csv"): + read_result(tmp_path, case, routing_probe, 0.0) + + +@pytest.mark.parametrize( + ("mutation", "message"), + [ + (lambda table: table.drop(columns="q_out"), "missing columns"), + ( + lambda table: pd.concat([table, table.iloc[[0]]], ignore_index=True), + "duplicate", + ), + (lambda table: table.loc[table["reach_id"] != "B"], "reach IDs"), + ], +) +def test_malformed_routing_shape_is_rejected( + case, routing_probe, tmp_path, mutation, message +): + _write_outputs(tmp_path, case, mutation(_routing_table(case))) + with pytest.raises(ProtocolError, match=message): + read_result(tmp_path, case, routing_probe, 0.0) + + +def test_routing_time_axis_must_match_forcing(case, routing_probe, tmp_path): + table = _routing_table(case) + table.loc[table["reach_id"] == "A", "time"] = pd.date_range( + "2001-01-01", periods=case.n_steps, freq="D" + ).strftime("%Y-%m-%d") + _write_outputs(tmp_path, case, table) + + with pytest.raises(ProtocolError, match="time axis"): + read_result(tmp_path, case, routing_probe, 0.0) + + +@pytest.mark.parametrize("bad_value", [np.nan, np.inf, "not-a-number"]) +def test_routing_values_must_be_finite( + case, routing_probe, tmp_path, bad_value +): + table = _routing_table(case) + table.loc[0, "q_in"] = bad_value + _write_outputs(tmp_path, case, table) + + with pytest.raises(ProtocolError, match="non-finite"): + read_result(tmp_path, case, routing_probe, 0.0) + + +def test_verify_adapter_checks_declared_routing(base_probe, tmp_path): + model = replace( + registry.find_model("reference_bucket"), + emits_routing=("q_in",), + ) + with pytest.raises(ProtocolError, match="routing.csv"): + verify_adapter_contract(model, base_probe, 11, workdir=tmp_path) From 0e162170666a93d5a8d28a7b19a96efeef8c0e7f Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:24:05 -0700 Subject: [PATCH 08/50] Fix routing test compatibility with pandas 3 --- tests/test_routing_protocol.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 119e1068..01f0ead7 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -206,6 +206,7 @@ def test_routing_values_must_be_finite( case, routing_probe, tmp_path, bad_value ): table = _routing_table(case) + table["q_in"] = table["q_in"].astype(object) table.loc[0, "q_in"] = bad_value _write_outputs(tmp_path, case, table) From e1b390f26475f0018f6483fbcf52711734d03386 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:25:39 -0700 Subject: [PATCH 09/50] Enforce routing variables in documentation sync --- tests/test_docs_in_sync.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/test_docs_in_sync.py b/tests/test_docs_in_sync.py index 96acf8fe..729b645e 100644 --- a/tests/test_docs_in_sync.py +++ b/tests/test_docs_in_sync.py @@ -15,7 +15,7 @@ import pytest import yaml -from hydroturing.spec import DIAG_VARS, FLUX_VARS, REPO_ROOT, STATE_VARS, load_model, load_probe +from hydroturing.spec import DIAG_VARS, FLUX_VARS, REPO_ROOT, ROUTING_VARS, STATE_VARS, load_model, load_probe PROBES = {p.id: p for p in (load_probe(x) for x in REPO_ROOT.glob("probes/*/*/probe.yaml"))} PROBE_IDS = sorted(PROBES) @@ -197,7 +197,7 @@ def test_site_flowchart_shows_every_probe(): assert html.count(f'"{key}":') == 3, f"{key} lacks a translation in one of the three languages" -@pytest.mark.parametrize("variable", FLUX_VARS + STATE_VARS + DIAG_VARS) +@pytest.mark.parametrize("variable", FLUX_VARS + STATE_VARS + DIAG_VARS + ROUTING_VARS) def test_agents_doc_defines_every_variable(variable): assert f"| `{variable}` |" in read("AGENTS.md"), ( f"{variable} is accepted by spec.py but the table in AGENTS.md does not define it" From 0a6a046789ef04cf2411d18a9b923a63e7293dcd Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:27:40 -0700 Subject: [PATCH 10/50] Fix spacing in routing request notes --- src/hydroturing/protocol.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index cc1c2008..be3c3bc3 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -153,7 +153,7 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P "when supplied. Row i's tsoil_layer is the mean temperature of " "the specified layer at the interval end; hfg and hfg_bottom " "are interval-mean boundary fluxes for the soil-storage check." - "Routing q_in and q_out are interval-mean m3 s-1; " + " Routing q_in and q_out are interval-mean m3 s-1; " "channel_storage is absolute end-of-step m3." ), } From 72a95750d12d8c318d4da0d6670e6a4ad2a2da04 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:31:12 -0700 Subject: [PATCH 11/50] Document reach-indexed adapter outputs --- docs/adapting-a-model.md | 21 +++++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/docs/adapting-a-model.md b/docs/adapting-a-model.md index 858cb214..83e286af 100644 --- a/docs/adapting-a-model.md +++ b/docs/adapting-a-model.md @@ -64,6 +64,21 @@ diagnostic but omitting it from the output is a protocol error. The separate `ts` diagnostic is instantaneous surface temperature for radiative checks; it cannot replace the interval-end mean-layer `tsoil_layer`. +Reach-indexed river-network outputs use the optional `routing` group and are +written separately from the catchment-level result table: + +```yaml +emits: + fluxes: [mrro, dis] + states: [channel] + routing: [q_in, q_out, channel_storage] +``` + +Variables declared under `emits.routing` belong in `output/routing.csv`, not +`output/result.csv`. The harness requests them only on a probe that declares +`requires.routing` and supplies `routing_network.reaches` in `static.json`. +The routing table contains one row for every forcing time and declared reach. + For the soil-storage probe, the control volume extends from the surface to `soil_layer_depth_m`, with prescribed `soil_heat_capacity_areal` and `soil_temperature_initial`. Configure that layer before running the model. @@ -79,8 +94,10 @@ prove that the model used it. See the [soil-storage case](../probes/energy/soil- ## 2. Write the adapter -Read `/io/request.json`, read the forcing, call your model, write the table. -The plumbing is about thirty lines and does not depend on your model. +Read `/io/request.json`, read the forcing and static inputs, call your model, +and write `output/result.csv`. When `request.request.routing` is non-empty, +also write `output/routing.csv` using the paths declared under +`request.output`. The plumbing does not otherwise depend on your model. ```python request = json.loads(Path(sys.argv[-1]).read_text()) From 85ff1c2e6de0e6c49823e8547d8405053a9ec241 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:34:13 -0700 Subject: [PATCH 12/50] Require routing probes to declare network input --- src/hydroturing/spec.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/src/hydroturing/spec.py b/src/hydroturing/spec.py index e6332d09..ac7a1b15 100644 --- a/src/hydroturing/spec.py +++ b/src/hydroturing/spec.py @@ -483,6 +483,13 @@ def load_probe(path: str | Path) -> ProbeSpec: ] if unknown: raise SpecError(f"{spec_file}: unknown variables in requires: {unknown}") + if requires.get("routing") and "routing_network" not in requires.get( + "static", [] + ): + raise SpecError( + f"{spec_file}: requires.routing also needs requires.static " + "to include routing_network" + ) return ProbeSpec( id=raw["id"], title=raw["title"], From e7be972823fc40e7324459ba4be5d542412f3d2b Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:38:17 -0700 Subject: [PATCH 13/50] Test routing probes require network input --- tests/test_routing_protocol.py | 16 +++++++++++++++- 1 file changed, 15 insertions(+), 1 deletion(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 01f0ead7..041c44f1 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -29,6 +29,8 @@ def routing_probe(base_probe): requires_states=(), requires_diagnostics=(), requires_routing=("q_in", "q_out", "channel_storage"), + requires_static=("routing_network",), + variants=(), ) @@ -52,7 +54,7 @@ def _write_probe(path, base_probe, routing): path.mkdir(parents=True) raw = yaml.safe_load((base_probe.path / "probe.yaml").read_text()) raw["id"] = "mass/routing-contract-test" - raw["requires"] = {"routing": routing} + raw["requires"] = {"routing": routing, "static": ["routing_network"]} (path / "generate.py").write_text("# Only manifest loading is tested.\n") (path / "probe.yaml").write_text(yaml.safe_dump(raw)) @@ -122,6 +124,18 @@ def test_routing_declarations_load_and_old_defaults_remain_empty( assert registry.find_model("reference_bucket").emits_routing == () + +def test_routing_probe_must_require_network(base_probe, tmp_path): + path = tmp_path / "mass" / "routing-contract-test" + _write_probe(path, base_probe, ["q_out"]) + raw = yaml.safe_load((path / "probe.yaml").read_text()) + raw["requires"].pop("static") + (path / "probe.yaml").write_text(yaml.safe_dump(raw)) + + with pytest.raises(SpecError, match="routing_network"): + load_probe(path) + + @pytest.mark.parametrize("kind", ["probe", "model"]) def test_unknown_routing_variable_is_rejected(base_probe, tmp_path, kind): if kind == "probe": From 67711e3c333807db7f09ae902cec7673cab08961 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:42:39 -0700 Subject: [PATCH 14/50] Validate declared routing reach identifiers --- src/hydroturing/protocol.py | 52 +++++++++++++++++++++---------------- 1 file changed, 30 insertions(+), 22 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index be3c3bc3..313bcb77 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -225,7 +225,7 @@ def _read_routing( raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") return None - table = pd.read_csv(path) + table = pd.read_csv(path, dtype={REACH_COL: str}) required_columns = [TIME_COL, REACH_COL, *probe.requires_routing] missing = [ column for column in required_columns if column not in table.columns @@ -245,32 +245,40 @@ def _read_routing( ) network = case.static.get("routing_network") - declared_reaches = ( - network.get("reaches") if isinstance(network, dict) else None - ) - if probe.requires_routing and not declared_reaches: + if not isinstance(network, dict) or "reaches" not in network: raise ProtocolError( "routing output requires static.json routing_network.reaches" ) - if declared_reaches: - expected_reaches = tuple(str(reach) for reach in declared_reaches) - if len(expected_reaches) != len(set(expected_reaches)): - raise ProtocolError( - "static.json routing_network.reaches contains duplicates" - ) + declared_reaches = network["reaches"] + if ( + not isinstance(declared_reaches, list) + or not declared_reaches + or any( + not isinstance(reach, str) or not reach + for reach in declared_reaches + ) + ): + raise ProtocolError( + "static.json routing_network.reaches must be a non-empty " + "list of non-empty string IDs" + ) - actual_reaches = set(table[REACH_COL]) - expected_set = set(expected_reaches) - if actual_reaches != expected_set: - missing_reaches = sorted(expected_set - actual_reaches) - extra_reaches = sorted(actual_reaches - expected_set) - raise ProtocolError( - "routing reach IDs differ from static.json: " - f"missing={missing_reaches}, extra={extra_reaches}" - ) - else: - expected_reaches = tuple(dict.fromkeys(table[REACH_COL])) + expected_reaches = tuple(declared_reaches) + if len(expected_reaches) != len(set(expected_reaches)): + raise ProtocolError( + "static.json routing_network.reaches contains duplicates" + ) + + actual_reaches = set(table[REACH_COL]) + expected_set = set(expected_reaches) + if actual_reaches != expected_set: + missing_reaches = sorted(expected_set - actual_reaches) + extra_reaches = sorted(actual_reaches - expected_set) + raise ProtocolError( + "routing reach IDs differ from static.json: " + f"missing={missing_reaches}, extra={extra_reaches}" + ) expected_rows = case.n_steps * len(expected_reaches) if len(table) != expected_rows: From 5198d3cff8881bdf886f41e814b006af4f4c6d5c Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:46:07 -0700 Subject: [PATCH 15/50] Test routing reach identifier validation --- tests/test_routing_protocol.py | 50 ++++++++++++++++++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 041c44f1..2f50be79 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -179,6 +179,56 @@ def test_valid_routing_table_roundtrips(case, routing_probe, tmp_path): pd.testing.assert_frame_equal(result.routing, expected) +def test_routing_output_requires_declared_reaches( + case, routing_probe, tmp_path +): + case = replace(case, static={}) + _write_outputs(tmp_path, case, _routing_table(case)) + + with pytest.raises(ProtocolError, match="routing_network.reaches"): + read_result(tmp_path, case, routing_probe, 0.0) + + +@pytest.mark.parametrize( + ("reaches", "message"), + [ + ([], "non-empty"), + (["A", "A"], "duplicates"), + ([{"id": "A"}, {"id": "B"}], "string IDs"), + ], +) +def test_invalid_declared_reach_ids_are_rejected( + case, routing_probe, tmp_path, reaches, message +): + case = replace( + case, + static={"routing_network": {"reaches": reaches}}, + ) + _write_outputs(tmp_path, case, _routing_table(case)) + + with pytest.raises(ProtocolError, match=message): + read_result(tmp_path, case, routing_probe, 0.0) + + +def test_leading_zero_reach_ids_are_preserved( + case, routing_probe, tmp_path +): + case = replace( + case, + static={"routing_network": {"reaches": ["007", "012"]}}, + ) + table = _routing_table(case) + table["reach_id"] = table["reach_id"].replace( + {"A": "007", "B": "012"} + ) + _write_outputs(tmp_path, case, table) + + result = read_result(tmp_path, case, routing_probe, 0.0) + + assert result.routing is not None + assert set(result.routing["reach_id"]) == {"007", "012"} + + def test_missing_required_routing_file_is_rejected(case, routing_probe, tmp_path): _write_outputs(tmp_path, case) with pytest.raises(ProtocolError, match="required output/routing.csv"): From a9a2946ac88b8ce0fb2bb6e7ec7edd5eaf986128 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:49:26 -0700 Subject: [PATCH 16/50] Request routing output only for routing probes --- src/hydroturing/protocol.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 313bcb77..cfe26eaa 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -122,18 +122,21 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P json.dump(case.static, fh, indent=2) case_id, model_seed = _opaque_case_metadata(case) + requested_routing = ( + list(model.emits_routing) if probe.requires_routing else [] + ) request = { "case_id": case_id, "seed": model_seed, "timestep": case.timestep, "n_steps": case.n_steps, "request": { - # Asking for every declared output keeps this part of the request - # invariant across probes and prevents it identifying the criterion. + # Catchment-level groups remain invariant across probes. + # Routing is requested only when the probe supplies a network. "fluxes": list(model.emits_fluxes), "states": list(model.emits_states), "diagnostics": list(model.emits_diagnostics), - "routing": list(model.emits_routing), + "routing": requested_routing, }, "input": {"forcing": FORCING_FILE, "static": STATIC_FILE}, "output": { @@ -143,7 +146,7 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P }, "units": { v: UNITS[v] - for v in (*model.emitted, *model.emits_routing) + for v in (*model.emitted, *requested_routing) if v in UNITS }, "notes": ( From c9143745dc6c627df1531a67e6737830dc247320 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:53:08 -0700 Subject: [PATCH 17/50] Test conditional routing requests --- tests/test_routing_protocol.py | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 2f50be79..deb920dd 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -151,12 +151,12 @@ def test_unknown_routing_variable_is_rejected(base_probe, tmp_path, kind): loader(path) -def test_stage_requests_routing_path_and_units(case, base_probe, tmp_path): +def test_stage_requests_routing_path_and_units(case, routing_probe, tmp_path): model = replace( registry.find_model("reference_bucket"), emits_routing=("q_in", "q_out", "channel_storage"), ) - request = json.loads(stage(tmp_path, case, base_probe, model).read_text()) + request = json.loads(stage(tmp_path, case, routing_probe, model).read_text()) assert request["request"]["routing"] == [ "q_in", @@ -169,6 +169,21 @@ def test_stage_requests_routing_path_and_units(case, base_probe, tmp_path): assert request["units"]["channel_storage"] == "m3" +def test_stage_omits_routing_for_nonrouting_probe( + case, base_probe, tmp_path +): + model = replace( + registry.find_model("reference_bucket"), + emits_routing=("q_in",), + ) + request = json.loads( + stage(tmp_path, case, base_probe, model).read_text() + ) + + assert request["request"]["routing"] == [] + assert "q_in" not in request["units"] + + def test_valid_routing_table_roundtrips(case, routing_probe, tmp_path): expected = _routing_table(case) _write_outputs(tmp_path, case, expected) From 43f16f3e6d2da459d857f0d342dc4a8ab8c9a535 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:55:30 -0700 Subject: [PATCH 18/50] Verify routing only on routing-capable probes --- src/hydroturing/harness.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/hydroturing/harness.py b/src/hydroturing/harness.py index 90a5db04..d582439c 100644 --- a/src/hydroturing/harness.py +++ b/src/hydroturing/harness.py @@ -436,7 +436,7 @@ def verify_adapter_contract( requires_fluxes=model.emits_fluxes, requires_states=model.emits_states, requires_diagnostics=model.emits_diagnostics, - requires_routing=model.emits_routing, + requires_routing=(model.emits_routing if probe.requires_routing else ()), variants=(), criteria=(), ) From 7475ca4ac845a911416d4ca9315124ad0bd7f5b1 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 12:58:03 -0700 Subject: [PATCH 19/50] Test network-aware adapter verification --- tests/test_routing_protocol.py | 39 ++++++++++++++++++++++++++++++++-- 1 file changed, 37 insertions(+), 2 deletions(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index deb920dd..34f0036d 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -293,10 +293,45 @@ def test_routing_values_must_be_finite( read_result(tmp_path, case, routing_probe, 0.0) -def test_verify_adapter_checks_declared_routing(base_probe, tmp_path): +def test_verify_adapter_skips_routing_on_nonrouting_probe( + base_probe, tmp_path +): + model = replace( + registry.find_model("reference_bucket"), + emits_routing=("q_in",), + ) + + result = verify_adapter_contract( + model, base_probe, 11, workdir=tmp_path + ) + + assert result.routing is None + + +def test_verify_adapter_checks_routing_on_network_probe( + routing_probe, tmp_path, monkeypatch +): + import hydroturing.harness as harness + + original_build_case = harness.build_case + + def build_case_with_network(*args, **kwargs): + built = original_build_case(*args, **kwargs) + static = dict(built.static) + static["routing_network"] = {"reaches": ["A"]} + return replace(built, static=static) + + monkeypatch.setattr( + harness, + "build_case", + build_case_with_network, + ) model = replace( registry.find_model("reference_bucket"), emits_routing=("q_in",), ) + with pytest.raises(ProtocolError, match="routing.csv"): - verify_adapter_contract(model, base_probe, 11, workdir=tmp_path) + verify_adapter_contract( + model, routing_probe, 11, workdir=tmp_path + ) From 461dfd76890edf47c4b6c8074b3ab02ad38760d9 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 13:01:06 -0700 Subject: [PATCH 20/50] Prefer routing probes for adapter verification --- src/hydroturing/cli.py | 29 +++++++++++++++++++++++------ 1 file changed, 23 insertions(+), 6 deletions(-) diff --git a/src/hydroturing/cli.py b/src/hydroturing/cli.py index c0fcb0ba..471f7166 100644 --- a/src/hydroturing/cli.py +++ b/src/hydroturing/cli.py @@ -233,12 +233,29 @@ def cmd_verify_adapter(args) -> int: if args.probe: candidates = [registry.find_probe(args.probe, roots)] else: - # The closure probe is the reference implementation and the natural - # smoke test. A model that cannot consume it, at its step, with its - # forcing or on its window, is checked on the first probe it can - # consume instead: one it cannot is N/A for it and asks the adapter - # nothing. - candidates = sorted(probes, key=lambda p: (p.id != "mass/catchment-closure", p.id)) + if model.emits_routing: + # Reach-indexed output can only be verified on a probe that + # supplies a routing network. Prefer one when the model declares + # routing output, then fall back to the ordinary probe order. + candidates = sorted( + probes, + key=lambda p: ( + not bool(p.requires_routing), + p.id != "mass/catchment-closure", + p.id, + ), + ) + else: + # The closure probe is the reference implementation and the + # natural smoke test. A model that cannot consume it is checked + # on the first probe it can consume instead. + candidates = sorted( + probes, + key=lambda p: ( + p.id != "mass/catchment-closure", + p.id, + ), + ) workdir = Path(args.workdir) if args.workdir else None result = None From 95deef4f486b3b4ec94fbcb90cc4037b7f5196cf Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 13:06:08 -0700 Subject: [PATCH 21/50] Test routing probe preference in verify-adapter --- tests/test_routing_protocol.py | 44 +++++++++++++++++++++++++++++++++- 1 file changed, 43 insertions(+), 1 deletion(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 34f0036d..5ccda4ea 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -4,6 +4,7 @@ import json from dataclasses import replace +from types import SimpleNamespace import numpy as np import pandas as pd @@ -124,7 +125,6 @@ def test_routing_declarations_load_and_old_defaults_remain_empty( assert registry.find_model("reference_bucket").emits_routing == () - def test_routing_probe_must_require_network(base_probe, tmp_path): path = tmp_path / "mass" / "routing-contract-test" _write_probe(path, base_probe, ["q_out"]) @@ -335,3 +335,45 @@ def build_case_with_network(*args, **kwargs): verify_adapter_contract( model, routing_probe, 11, workdir=tmp_path ) + + +def test_verify_adapter_prefers_routing_probe( + base_probe, routing_probe, monkeypatch +): + from hydroturing import cli + + model = replace( + registry.find_model("reference_bucket"), + emits_routing=("q_in",), + ) + attempted = [] + + def stop_after_selection(model, probe, seed, **kwargs): + attempted.append(probe) + raise RuntimeError("stop after selecting the probe") + + monkeypatch.setattr( + cli.registry, + "find_model", + lambda name: model, + ) + monkeypatch.setattr( + cli, + "_all_probes", + lambda args: [base_probe, routing_probe], + ) + monkeypatch.setattr( + cli, + "verify_adapter_contract", + stop_after_selection, + ) + args = SimpleNamespace( + model="reference_bucket", + probe=None, + workdir=None, + window=None, + csv=None, + ) + + assert cli.cmd_verify_adapter(args) == 2 + assert attempted == [routing_probe] From ec304f8a1ff13ee1d0d295d969720e37642a1136 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 13:10:48 -0700 Subject: [PATCH 22/50] Define the routing network input contract --- AGENTS.md | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index c23a5e1f..a11d4b71 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -47,7 +47,25 @@ When `request.request.routing` is non-empty, the adapter must write `output/routing.csv` as a long table with one row for every forcing time and every reach declared by `static.json` under `routing_network.reaches`. The required identifier columns are `time` and `reach_id`; the remaining required -columns are the names listed in `request.request.routing`. For example: +columns are the names listed in `request.request.routing`. + +A probe that declares `requires.routing` must also declare +`requires.static: [routing_network]`. `routing_network` is an object whose +`reaches` member is a non-empty list of unique, non-empty string IDs: + +```json +{ + "routing_network": { + "reaches": ["A", "B", "C"] + } +} +``` + +Keep topology metadata such as junctions, headwaters, outlets, lengths or +areas in separate fields under `routing_network`; do not place reach objects +inside `reaches`. Reach IDs are strings and leading zeros are significant. + +For example, the routing table is: ```csv time,reach_id,q_in,q_out,channel_storage @@ -65,6 +83,10 @@ to catchment-average depth. A missing requested routing column, reach or time row is a contract error. Models that do not declare `emits.routing` do not write `routing.csv`. +A routing table has `n_steps * n_reaches` rows. Probe authors must set +`case.max_output_mb` high enough for that table; the normal output size +allow-list and limit still apply. + The model seed is deterministic for reproducible stochastic inference, but it is not the generator seed recorded in the host-side report. Exit 0 on success. From 0d9ead66e6fcbe88ad28f6bcad58396a31a11ce1 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 13:17:12 -0700 Subject: [PATCH 23/50] Identify reaches in routing time-axis errors --- src/hydroturing/protocol.py | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index cfe26eaa..1359c562 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -191,7 +191,12 @@ def _validate_output_files(io_dir: Path, probe: ProbeSpec) -> None: ) -def _validate_time_axis(table: pd.DataFrame, case: Case) -> None: +def _validate_time_axis( + table: pd.DataFrame, + case: Case, + *, + label: str = "result", +) -> None: expected = case.forcing[TIME_COL].reset_index(drop=True) actual = table[TIME_COL].reset_index(drop=True) @@ -211,7 +216,7 @@ def _validate_time_axis(table: pd.DataFrame, case: Case) -> None: if not equal: first = int(mismatch.to_numpy().nonzero()[0][0]) raise ProtocolError( - f"result time axis differs from the forcing at row {first}: " + f"{label} time axis differs from the forcing at row {first}: " f"got {actual.iloc[first]!r}, expected {expected.iloc[first]!r}" ) @@ -299,7 +304,11 @@ def _read_routing( f"routing reach '{reach}' has {len(reach_table)} rows, " f"expected {case.n_steps}" ) - _validate_time_axis(reach_table, case) + _validate_time_axis( + reach_table, + case, + label=f"routing.csv reach {reach!r}", + ) for var in probe.requires_routing: column = pd.to_numeric(table[var], errors="coerce") From ba6a1e253afdcda62476999402f764c0634c1726 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 20 Sep 2026 13:18:35 -0700 Subject: [PATCH 24/50] Test reach-specific routing time errors --- tests/test_routing_protocol.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 5ccda4ea..0f282945 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -276,7 +276,7 @@ def test_routing_time_axis_must_match_forcing(case, routing_probe, tmp_path): ).strftime("%Y-%m-%d") _write_outputs(tmp_path, case, table) - with pytest.raises(ProtocolError, match="time axis"): + with pytest.raises(ProtocolError, match=r"routing\.csv reach 'A' time axis"): read_result(tmp_path, case, routing_probe, 0.0) From 49632f5d7deae342b7a1df09192dfaa5c27f5391 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:46:49 -0700 Subject: [PATCH 25/50] Rename reach inflow routing variable --- src/hydroturing/spec.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/hydroturing/spec.py b/src/hydroturing/spec.py index ddc0b525..6bab51f0 100644 --- a/src/hydroturing/spec.py +++ b/src/hydroturing/spec.py @@ -41,7 +41,7 @@ # Reach-indexed routing variables live in output/routing.csv rather than the # one-row-per-timestep result table. -ROUTING_VARS = ("q_in", "q_out", "channel_storage") +ROUTING_VARS = ("reach_q_in", "q_out", "channel_storage") UNITS = { "pr": "mm day-1", @@ -49,6 +49,7 @@ "mrro": "mm day-1", "dis": "m3 s-1", "q_in": "m3 s-1", + "reach_q_in": "m3 s-1", "q_out": "m3 s-1", "channel_storage": "m3", "gwex": "mm day-1", From 70f388505f7529a38d2d997cb1e905951cda8fed Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:48:50 -0700 Subject: [PATCH 26/50] Rename reach inflow in protocol notes --- src/hydroturing/protocol.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 2bdc80e9..1747dd31 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -162,7 +162,7 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P "when supplied. Row i's tsoil_layer is the mean temperature of " "the specified layer at the interval end; hfg and hfg_bottom " "are interval-mean boundary fluxes for the soil-storage check." - " Routing q_in and q_out are interval-mean m3 s-1; " + " Routing reach_q_in and q_out are interval-mean m3 s-1; " "channel_storage is absolute end-of-step m3." ), } From f9955d99c17eb10e3628ca73a954b66535fe9893 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:52:45 -0700 Subject: [PATCH 27/50] Distinguish forcing and reach inflow variables --- AGENTS.md | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 6a244956..59f2022d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -71,7 +71,7 @@ inside `reaches`. Reach IDs are strings and leading zeros are significant. For example, the routing table is: ```csv -time,reach_id,q_in,q_out,channel_storage +time,reach_id,reach_q_in,q_out,channel_storage 2001-01-01,A,12.3,8.4,336960 2001-01-01,B,7.1,4.9,190080 2001-01-01,C,13.3,9.2,354240 @@ -79,7 +79,7 @@ time,reach_id,q_in,q_out,channel_storage Each `(time, reach_id)` pair must be unique. Every declared reach must have exactly one row for every forcing row, including spinup, and each reach's time -axis must match `forcing.csv`. `q_in` and `q_out` are interval-mean discharge +axis must match `forcing.csv`. `reach_q_in` and `q_out` are interval-mean discharge rates in m3/s. `channel_storage` is the absolute volume stored in that reach at the end of the interval, in m3. It is not a tendency and must not be converted to catchment-average depth. A missing requested routing column, reach or time @@ -217,7 +217,7 @@ numbers in both runs; keep it that way and do not reseed from the clock. | `evspsbl` | evapotranspiration | mm/day | | `mrro` | total runoff | mm/day | | `dis` | river discharge | m3/s | -| `q_in` | interval-mean inflow to the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | +| `reach_q_in` | interval-mean inflow to the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | | `q_out` | interval-mean outflow from the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | | `channel_storage` | absolute water volume stored in the identified reach at the end of the row's time interval; reach-indexed in `routing.csv`, not a tendency and not catchment-average `channel` storage | m3 | | `gwex` | a declared exchange with the outside: regional groundwater, inter-basin transfer; positive into the catchment | mm/day | @@ -311,12 +311,12 @@ separately from the catchment-level result columns: emits: fluxes: [mrro, dis] states: [channel] - routing: [q_in, q_out, channel_storage] + routing: [reach_q_in, q_out, channel_storage] ``` Variables declared under `emits.routing` belong in `routing.csv`, not in `result.csv`, `emits.fluxes` or `emits.states`. Declaring them means the -adapter must write the requested long table; do not construct `q_in` or +adapter must write the requested long table; do not construct `reach_q_in` or `channel_storage` from a budget residual merely to satisfy the contract. From 69e69972241c6c0b3941f7f7912b52658816f177 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:55:39 -0700 Subject: [PATCH 28/50] Rename reach inflow in model schema --- schemas/model.schema.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/schemas/model.schema.json b/schemas/model.schema.json index aad89523..c575f150 100644 --- a/schemas/model.schema.json +++ b/schemas/model.schema.json @@ -42,7 +42,7 @@ }, "routing": { "type": "array", "items": {"type": "string"}, "uniqueItems": true, - "description": "Reach-indexed variables written to output/routing.csv. q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + "description": "Reach-indexed variables written to output/routing.csv. reach_q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." }, "diagnostics": { "type": "array", "items": {"type": "string"}, "uniqueItems": true, From 5305aa79720313ec4c65617c8ef7499ab702b150 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:56:51 -0700 Subject: [PATCH 29/50] Rename reach inflow in probe schema --- schemas/probe.schema.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/schemas/probe.schema.json b/schemas/probe.schema.json index 1564daa4..89559f6e 100644 --- a/schemas/probe.schema.json +++ b/schemas/probe.schema.json @@ -109,7 +109,7 @@ "type": "string" }, "uniqueItems": true, - "description": "Reach-indexed variables read from output/routing.csv. q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + "description": "Reach-indexed variables read from output/routing.csv. reach_q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." }, "forcing": { "type": "array", From 272bf078400ad0ba7c5bf29562484289131ba2df Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Fri, 16 Oct 2026 16:58:10 -0700 Subject: [PATCH 30/50] Update routing protocol tests for reach_q_in --- tests/test_routing_protocol.py | 28 ++++++++++++++-------------- 1 file changed, 14 insertions(+), 14 deletions(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 0f282945..9e95096b 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -29,7 +29,7 @@ def routing_probe(base_probe): requires_fluxes=(), requires_states=(), requires_diagnostics=(), - requires_routing=("q_in", "q_out", "channel_storage"), + requires_routing=("reach_q_in", "q_out", "channel_storage"), requires_static=("routing_network",), variants=(), ) @@ -80,14 +80,14 @@ def _routing_table(case): { "time": time.strftime("%Y-%m-%d"), "reach_id": "A", - "q_in": 2.0, + "reach_q_in": 2.0, "q_out": 1.0, "channel_storage": 86400.0, }, { "time": time.strftime("%Y-%m-%d"), "reach_id": "B", - "q_in": 3.0, + "reach_q_in": 3.0, "q_out": 2.0, "channel_storage": 86400.0, }, @@ -111,7 +111,7 @@ def test_routing_declarations_load_and_old_defaults_remain_empty( base_probe, tmp_path ): probe_dir = tmp_path / "mass" / "routing-contract-test" - variables = ["q_in", "q_out", "channel_storage"] + variables = ["reach_q_in", "q_out", "channel_storage"] _write_probe(probe_dir, base_probe, variables) probe = load_probe(probe_dir) @@ -154,17 +154,17 @@ def test_unknown_routing_variable_is_rejected(base_probe, tmp_path, kind): def test_stage_requests_routing_path_and_units(case, routing_probe, tmp_path): model = replace( registry.find_model("reference_bucket"), - emits_routing=("q_in", "q_out", "channel_storage"), + emits_routing=("reach_q_in", "q_out", "channel_storage"), ) request = json.loads(stage(tmp_path, case, routing_probe, model).read_text()) assert request["request"]["routing"] == [ - "q_in", + "reach_q_in", "q_out", "channel_storage", ] assert request["output"]["routing"] == "output/routing.csv" - assert request["units"]["q_in"] == "m3 s-1" + assert request["units"]["reach_q_in"] == "m3 s-1" assert request["units"]["q_out"] == "m3 s-1" assert request["units"]["channel_storage"] == "m3" @@ -174,14 +174,14 @@ def test_stage_omits_routing_for_nonrouting_probe( ): model = replace( registry.find_model("reference_bucket"), - emits_routing=("q_in",), + emits_routing=("reach_q_in",), ) request = json.loads( stage(tmp_path, case, base_probe, model).read_text() ) assert request["request"]["routing"] == [] - assert "q_in" not in request["units"] + assert "reach_q_in" not in request["units"] def test_valid_routing_table_roundtrips(case, routing_probe, tmp_path): @@ -285,8 +285,8 @@ def test_routing_values_must_be_finite( case, routing_probe, tmp_path, bad_value ): table = _routing_table(case) - table["q_in"] = table["q_in"].astype(object) - table.loc[0, "q_in"] = bad_value + table["reach_q_in"] = table["reach_q_in"].astype(object) + table.loc[0, "reach_q_in"] = bad_value _write_outputs(tmp_path, case, table) with pytest.raises(ProtocolError, match="non-finite"): @@ -298,7 +298,7 @@ def test_verify_adapter_skips_routing_on_nonrouting_probe( ): model = replace( registry.find_model("reference_bucket"), - emits_routing=("q_in",), + emits_routing=("reach_q_in",), ) result = verify_adapter_contract( @@ -328,7 +328,7 @@ def build_case_with_network(*args, **kwargs): ) model = replace( registry.find_model("reference_bucket"), - emits_routing=("q_in",), + emits_routing=("reach_q_in",), ) with pytest.raises(ProtocolError, match="routing.csv"): @@ -344,7 +344,7 @@ def test_verify_adapter_prefers_routing_probe( model = replace( registry.find_model("reference_bucket"), - emits_routing=("q_in",), + emits_routing=("reach_q_in",), ) attempted = [] From 4886751083e4b51a5193ff97615603b42f887087 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 22:48:46 -0700 Subject: [PATCH 31/50] Require routing models to declare routing_network --- src/hydroturing/spec.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/hydroturing/spec.py b/src/hydroturing/spec.py index 6bab51f0..3a36c591 100644 --- a/src/hydroturing/spec.py +++ b/src/hydroturing/spec.py @@ -624,6 +624,15 @@ def load_model(path: str | Path) -> ModelManifest: if unknown: raise SpecError(f"{spec_file}: unknown variables in emits: {unknown}") + if raw["emits"].get("routing") and "routing_network" not in { + *raw.get("needs_static", []), + *raw.get("uses_static", []), + }: + raise SpecError( + f"{spec_file}: emits.routing also needs needs_static or " + "uses_static to include routing_network" + ) + declared = raw["timestep"] timesteps = tuple(declared) if isinstance(declared, list) else (declared,) if len(set(timesteps)) != len(timesteps): From db78121da6109c21ced6e94cbbbe4973bb2347f4 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 22:51:38 -0700 Subject: [PATCH 32/50] Test routing model network declaration --- tests/test_routing_protocol.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 9e95096b..3412a882 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -68,6 +68,7 @@ def _write_model(path, routing): "entrypoint": ["python", "adapter.py"], "timestep": "PT1D", "emits": {"fluxes": [], "states": [], "routing": routing}, + "needs_static": ["routing_network"], } (path / "model.yaml").write_text(yaml.safe_dump(manifest)) @@ -136,6 +137,17 @@ def test_routing_probe_must_require_network(base_probe, tmp_path): load_probe(path) +def test_routing_model_must_declare_network(tmp_path): + path = tmp_path / "routing_model" + _write_model(path, ["q_out"]) + raw = yaml.safe_load((path / "model.yaml").read_text()) + raw.pop("needs_static") + (path / "model.yaml").write_text(yaml.safe_dump(raw)) + + with pytest.raises(SpecError, match="routing_network"): + load_model(path) + + @pytest.mark.parametrize("kind", ["probe", "model"]) def test_unknown_routing_variable_is_rejected(base_probe, tmp_path, kind): if kind == "probe": From 67e2a52591b6fcc45df970de62de71fa14e98168 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 22:54:12 -0700 Subject: [PATCH 33/50] Document routing model network dependency --- AGENTS.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 59f2022d..2914c7ce 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -312,6 +312,7 @@ emits: fluxes: [mrro, dis] states: [channel] routing: [reach_q_in, q_out, channel_storage] +needs_static: [routing_network] ``` Variables declared under `emits.routing` belong in `routing.csv`, not in @@ -320,6 +321,11 @@ adapter must write the requested long table; do not construct `reach_q_in` or `channel_storage` from a budget residual merely to satisfy the contract. +A model that declares `emits.routing` must also list `routing_network` in +`needs_static` or `uses_static`; otherwise the manifest is invalid because the +routing output has no declared network topology. + + Every probe that needs more than that will be `N/A` with reason `INCOMPLETE`, which is the honest outcome: not a fail, but a probe that cannot be put to the model, so it counts neither way. Fabricating an From 119fa786ce8a8c4fe801ead60326ac513c041fde Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 22:55:36 -0700 Subject: [PATCH 34/50] Document routing network input for adapters --- docs/adapting-a-model.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/adapting-a-model.md b/docs/adapting-a-model.md index 4869d460..f1f01def 100644 --- a/docs/adapting-a-model.md +++ b/docs/adapting-a-model.md @@ -72,8 +72,13 @@ emits: fluxes: [mrro, dis] states: [channel] routing: [reach_q_in, q_out, channel_storage] +needs_static: [routing_network] ``` +A model that declares `emits.routing` must also list `routing_network` in +`needs_static` or `uses_static`, so its adapter has the network topology that +defines the declared reaches. + Variables declared under `emits.routing` belong in `output/routing.csv`, not `output/result.csv`. The harness requests them only on a probe that declares `requires.routing` and supplies `routing_network.reaches` in `static.json`. From 1f7467c03c40d4c81ca121de864f1c694b0a4ae5 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:03:45 -0700 Subject: [PATCH 35/50] Ignore routing output when it was not requested --- src/hydroturing/protocol.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 1747dd31..2820efe9 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -233,12 +233,13 @@ def _read_routing( probe: ProbeSpec, ) -> pd.DataFrame | None: """Read and validate the optional reach-indexed routing result.""" - path = io_dir / ROUTING_CSV - if not path.exists(): - if probe.requires_routing: - raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") + if not probe.requires_routing: return None + path = io_dir / ROUTING_CSV + if not path.exists(): + raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") + table = pd.read_csv(path, dtype={REACH_COL: str}) required_columns = [TIME_COL, REACH_COL, *probe.requires_routing] missing = [ From 2b7119efad8a598ae9bfc47901036cdc8071c558 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:06:52 -0700 Subject: [PATCH 36/50] Test ignoring unrequested routing output --- tests/test_routing_protocol.py | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 3412a882..6f53b702 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -196,6 +196,22 @@ def test_stage_omits_routing_for_nonrouting_probe( assert "reach_q_in" not in request["units"] +def test_unrequested_routing_output_is_ignored( + case, routing_probe, tmp_path +): + probe = replace( + routing_probe, + requires_routing=(), + requires_static=(), + ) + case = replace(case, static={}) + _write_outputs(tmp_path, case, _routing_table(case)) + + result = read_result(tmp_path, case, probe, 0.0) + + assert result.routing is None + + def test_valid_routing_table_roundtrips(case, routing_probe, tmp_path): expected = _routing_table(case) _write_outputs(tmp_path, case, expected) From 3a73b3ab33136a98180571cbc87c3083be281d01 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:10:39 -0700 Subject: [PATCH 37/50] Define routing control volume semantics --- AGENTS.md | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 2914c7ce..8ca7996f 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -217,9 +217,9 @@ numbers in both runs; keep it that way and do not reseed from the clock. | `evspsbl` | evapotranspiration | mm/day | | `mrro` | total runoff | mm/day | | `dis` | river discharge | m3/s | -| `reach_q_in` | interval-mean inflow to the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | -| `q_out` | interval-mean outflow from the identified reach for the row's time interval; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | -| `channel_storage` | absolute water volume stored in the identified reach at the end of the row's time interval; reach-indexed in `routing.csv`, not a tendency and not catchment-average `channel` storage | m3 | +| `reach_q_in` | interval-mean total inflow into the identified reach control volume, including upstream, lateral, and prescribed boundary inflows; positive into the reach; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | +| `q_out` | interval-mean flow across the identified reach's downstream boundary; positive downstream and negative for reverse or backwater flow; reach-indexed in `routing.csv`, never a catchment-average flux | m3/s | +| `channel_storage` | absolute water volume stored in the routed river channel of the identified reach at the end of the row's time interval; excludes land-surface, soil, groundwater, floodplain, and overland storage; reach-indexed in `routing.csv`, not a tendency and not catchment-average `channel` storage | m3 | | `gwex` | a declared exchange with the outside: regional groundwater, inter-basin transfer; positive into the catchment | mm/day | | `gw_sw_exchange` | net river-aquifer exchange, positive into the aquifer; unlike `gwex`, this moves water between two stores inside the control volume (`gw` and `channel`), so it is never added to `gwex` or counted as a `closure` source | mm/day | | `gw_to_sw` | groundwater-to-river exchange component: the aquifer losing to the river, so it is never positive; a component of `gw_sw_exchange`, not of `gwex`, and not an addition to it | mm/day | From 810ebb16031d9f4d387dd3b05a80e2982e2e4e82 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:13:22 -0700 Subject: [PATCH 38/50] Describe routing control volume in request notes --- src/hydroturing/protocol.py | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 2820efe9..85d7f8cb 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -162,8 +162,13 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P "when supplied. Row i's tsoil_layer is the mean temperature of " "the specified layer at the interval end; hfg and hfg_bottom " "are interval-mean boundary fluxes for the soil-storage check." - " Routing reach_q_in and q_out are interval-mean m3 s-1; " - "channel_storage is absolute end-of-step m3." + " Routing reach_q_in is total inflow into the reach control " + "volume (upstream, lateral, and prescribed), positive into the " + "reach; q_out is flow across the downstream boundary, positive " + "downstream and negative for reverse or backwater flow. Both are " + "interval-mean m3 s-1. channel_storage is absolute end-of-step " + "channel water volume in m3 and excludes land-surface, soil, " + "groundwater, floodplain, and overland storage." ), } request_path = io_dir / REQUEST_FILE From d10e83a908f7e93361a1e852013f936bac7c0ac7 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:15:59 -0700 Subject: [PATCH 39/50] Define routing semantics in model schema --- schemas/model.schema.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/schemas/model.schema.json b/schemas/model.schema.json index c575f150..ea9a1d50 100644 --- a/schemas/model.schema.json +++ b/schemas/model.schema.json @@ -42,7 +42,7 @@ }, "routing": { "type": "array", "items": {"type": "string"}, "uniqueItems": true, - "description": "Reach-indexed variables written to output/routing.csv. reach_q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + "description": "Reach-indexed variables written to output/routing.csv. reach_q_in is total inflow into the reach control volume (upstream, lateral, and prescribed), positive into the reach; q_out crosses the downstream boundary, positive downstream and negative for reverse or backwater flow. Both are interval-mean discharges in m3 s-1. channel_storage is absolute end-of-step channel water volume in m3 and excludes land-surface, soil, groundwater, floodplain, and overland storage." }, "diagnostics": { "type": "array", "items": {"type": "string"}, "uniqueItems": true, From 25723f85e2bbb4596cad9e3e658dfe75cc2751a1 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:18:42 -0700 Subject: [PATCH 40/50] Define routing semantics in probe schema --- schemas/probe.schema.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/schemas/probe.schema.json b/schemas/probe.schema.json index 89559f6e..c08463c4 100644 --- a/schemas/probe.schema.json +++ b/schemas/probe.schema.json @@ -109,7 +109,7 @@ "type": "string" }, "uniqueItems": true, - "description": "Reach-indexed variables read from output/routing.csv. reach_q_in and q_out are interval-mean discharges in m3 s-1; channel_storage is absolute end-of-step storage in m3." + "description": "Reach-indexed variables read from output/routing.csv. reach_q_in is total inflow into the reach control volume (upstream, lateral, and prescribed), positive into the reach; q_out crosses the downstream boundary, positive downstream and negative for reverse or backwater flow. Both are interval-mean discharges in m3 s-1. channel_storage is absolute end-of-step channel water volume in m3 and excludes land-surface, soil, groundwater, floodplain, and overland storage." }, "forcing": { "type": "array", From 9a1a45628e0229b0edc6edb264eaf96ba6c18aaa Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:21:23 -0700 Subject: [PATCH 41/50] Document routing control volume for adapters --- docs/adapting-a-model.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/adapting-a-model.md b/docs/adapting-a-model.md index f1f01def..5d63167e 100644 --- a/docs/adapting-a-model.md +++ b/docs/adapting-a-model.md @@ -84,6 +84,13 @@ Variables declared under `emits.routing` belong in `output/routing.csv`, not `requires.routing` and supplies `routing_network.reaches` in `static.json`. The routing table contains one row for every forcing time and declared reach. +`reach_q_in` is the interval-mean rate of total inflow into the reach control +volume, including upstream, lateral, and prescribed boundary inflows, and is +positive into the reach. `q_out` is the interval-mean flow across the reach's +downstream boundary, positive downstream and negative for reverse or backwater +flow. `channel_storage` is the absolute end-of-step channel water volume and +excludes land-surface, soil, groundwater, floodplain, and overland storage. + The `stage` diagnostic is a water-surface elevation in metres on the fixed vertical datum declared by the case. `bed_elevation_m` is the rigid reach bed on that same datum, so water depth is `stage - bed_elevation_m` at every step. From 286795dfbc3ed1bf9ad922b9640bed6e9ee53ba6 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:23:43 -0700 Subject: [PATCH 42/50] Preserve string reach IDs when reading routing output --- src/hydroturing/protocol.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 85d7f8cb..5e97d029 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -245,7 +245,7 @@ def _read_routing( if not path.exists(): raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") - table = pd.read_csv(path, dtype={REACH_COL: str}) + table = pd.read_csv(path, dtype={REACH_COL: str}, keep_default_na=False) required_columns = [TIME_COL, REACH_COL, *probe.requires_routing] missing = [ column for column in required_columns if column not in table.columns From c371eda1de6f6307f2a55c7482ceb0bef88468b1 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:32:06 -0700 Subject: [PATCH 43/50] Test preservation of NA-like reach IDs --- tests/test_routing_protocol.py | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 6f53b702..31c8b970 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -272,6 +272,27 @@ def test_leading_zero_reach_ids_are_preserved( assert set(result.routing["reach_id"]) == {"007", "012"} +@pytest.mark.parametrize( + "reach_id", + ["NA", "N/A", "n/a", "nan", "None", "null", "NULL"], +) +def test_na_like_reach_ids_are_preserved( + case, routing_probe, tmp_path, reach_id +): + case = replace( + case, + static={"routing_network": {"reaches": [reach_id, "B"]}}, + ) + table = _routing_table(case) + table["reach_id"] = table["reach_id"].replace({"A": reach_id}) + _write_outputs(tmp_path, case, table) + + result = read_result(tmp_path, case, routing_probe, 0.0) + + assert result.routing is not None + assert set(result.routing["reach_id"]) == {reach_id, "B"} + + def test_missing_required_routing_file_is_rejected(case, routing_probe, tmp_path): _write_outputs(tmp_path, case) with pytest.raises(ProtocolError, match="required output/routing.csv"): From 1ad3e9bec20ac68d6d950f156a96c6b925750966 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:52:09 -0700 Subject: [PATCH 44/50] Validate routing networks before adapter execution --- src/hydroturing/protocol.py | 59 +++++++++++++++++++++---------------- 1 file changed, 33 insertions(+), 26 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index 5e97d029..d5d6d324 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -95,6 +95,33 @@ def after_spinup(self, frame: pd.DataFrame) -> pd.DataFrame: return frame.iloc[self.spinup_steps :].reset_index(drop=True) +def _routing_reaches(case: Case) -> tuple[str, ...]: + """Return and validate the reach IDs declared by a routing case.""" + network = case.static.get("routing_network") + if not isinstance(network, dict) or "reaches" not in network: + raise ValueError( + "routing probes require static.json routing_network.reaches" + ) + + reaches = network["reaches"] + if ( + not isinstance(reaches, list) + or not reaches + or any(not isinstance(reach, str) or not reach for reach in reaches) + ): + raise ValueError( + "static.json routing_network.reaches must be a non-empty " + "list of non-empty string IDs" + ) + + declared = tuple(reaches) + if len(declared) != len(set(declared)): + raise ValueError( + "static.json routing_network.reaches contains duplicates" + ) + return declared + + @dataclass class RunResult: """What a model returned for one case, after contract validation.""" @@ -114,6 +141,10 @@ class RunResult: def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> Path: """Write request.json and the read-only inputs. Returns the request path.""" + if probe.requires_routing: + _routing_reaches(case) + + (io_dir / "input").mkdir(parents=True, exist_ok=True) (io_dir / "output").mkdir(parents=True, exist_ok=True) @@ -264,32 +295,8 @@ def _read_routing( f"{int(duplicate.sum())} duplicates" ) - network = case.static.get("routing_network") - if not isinstance(network, dict) or "reaches" not in network: - raise ProtocolError( - "routing output requires static.json routing_network.reaches" - ) - - declared_reaches = network["reaches"] - if ( - not isinstance(declared_reaches, list) - or not declared_reaches - or any( - not isinstance(reach, str) or not reach - for reach in declared_reaches - ) - ): - raise ProtocolError( - "static.json routing_network.reaches must be a non-empty " - "list of non-empty string IDs" - ) - - expected_reaches = tuple(declared_reaches) - if len(expected_reaches) != len(set(expected_reaches)): - raise ProtocolError( - "static.json routing_network.reaches contains duplicates" - ) - + expected_reaches = _routing_reaches(case) + actual_reaches = set(table[REACH_COL]) expected_set = set(expected_reaches) if actual_reaches != expected_set: From 44e5e8d0a0ea3dd37c3d05b606a9dce317a16f41 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Sun, 18 Oct 2026 23:59:18 -0700 Subject: [PATCH 45/50] Test pre-run routing network validation --- tests/test_routing_protocol.py | 17 +++++++++++++++-- 1 file changed, 15 insertions(+), 2 deletions(-) diff --git a/tests/test_routing_protocol.py b/tests/test_routing_protocol.py index 31c8b970..3f311dfc 100644 --- a/tests/test_routing_protocol.py +++ b/tests/test_routing_protocol.py @@ -228,7 +228,7 @@ def test_routing_output_requires_declared_reaches( case = replace(case, static={}) _write_outputs(tmp_path, case, _routing_table(case)) - with pytest.raises(ProtocolError, match="routing_network.reaches"): + with pytest.raises(ValueError, match="routing_network.reaches"): read_result(tmp_path, case, routing_probe, 0.0) @@ -249,10 +249,23 @@ def test_invalid_declared_reach_ids_are_rejected( ) _write_outputs(tmp_path, case, _routing_table(case)) - with pytest.raises(ProtocolError, match=message): + with pytest.raises(ValueError, match=message): read_result(tmp_path, case, routing_probe, 0.0) +def test_stage_rejects_malformed_network(case, routing_probe, tmp_path): + case = replace( + case, + static={"routing_network": {"reaches": [1, 2]}}, + ) + model = registry.find_model("reference_bucket") + + with pytest.raises(ValueError, match="string IDs"): + stage(tmp_path, case, routing_probe, model) + + assert not (tmp_path / "request.json").exists() + + def test_leading_zero_reach_ids_are_preserved( case, routing_probe, tmp_path ): From 3a6749212dafc1ac42cb8e3d46d3a70374cdde95 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Mon, 19 Oct 2026 00:05:30 -0700 Subject: [PATCH 46/50] Show routing variables in ht list --- src/hydroturing/cli.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/src/hydroturing/cli.py b/src/hydroturing/cli.py index c8d7dfb2..cf90d69a 100644 --- a/src/hydroturing/cli.py +++ b/src/hydroturing/cli.py @@ -170,7 +170,7 @@ def cmd_list(args) -> int: models = registry.all_models() print(f"probes ({len(probes)}):") for p in probes: - req = ", ".join(p.required_vars) + req = ", ".join((*p.required_vars, *p.requires_routing)) steps = "/".join(p.timesteps) print(f" {p.id:<34} {p.law:<9} {p.track:<10} {steps:<10} seeds={p.n_seeds} needs: {req}") print(f"\nmodels ({len(models)}):") @@ -179,9 +179,10 @@ def cmd_list(args) -> int: days = resolve_window_days(m, matching) if matching else m.window_days window = "full" if days is None else f"{days}d" steps = "/".join(m.timesteps) + emits = ", ".join((*m.emitted, *m.emits_routing)) print( f" {m.name:<24} v{m.version:<8} runner={m.runner:<11} {steps:<22} " - f"window={window:<5} emits: {', '.join(m.emitted)}" + f"window={window:<5} emits: {emits}" ) return 0 From 9907b464777c5de8d9ace99c0ee7c88b409c4315 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Mon, 19 Oct 2026 00:11:11 -0700 Subject: [PATCH 47/50] Show routing requirements in probe template --- templates/probe.template.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/templates/probe.template.yaml b/templates/probe.template.yaml index ccda50b5..4b404525 100644 --- a/templates/probe.template.yaml +++ b/templates/probe.template.yaml @@ -53,12 +53,14 @@ description: > #! fluxes: pr, evspsbl, mrro, dis (rates) #! states: mrso, snw, canopy (absolute storages, never tendencies) #! diagnostics: ts (read by a criterion, never integrated) +#! routing: reach_q_in, q_out, channel_storage (reach-indexed outputs) #! forcing, static: case inputs the verdict rests on; a model must declare #! it consumes them (needs_* or uses_*) or it is #! N/A with reason INCOMPATIBLE requires: fluxes: [pr, evspsbl, mrro] states: [mrso, snw, canopy] + routing: [] case: #! Must define generate(seed) -> (DataFrame, dict) and be deterministic. From 37273912cb41ce9ea4ef38bd592140de1f8f74f4 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Mon, 19 Oct 2026 00:14:15 -0700 Subject: [PATCH 48/50] Document routing requirements for probe authors --- docs/writing-a-probe.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/docs/writing-a-probe.md b/docs/writing-a-probe.md index aa5fbbd9..9277a8bf 100644 --- a/docs/writing-a-probe.md +++ b/docs/writing-a-probe.md @@ -85,6 +85,26 @@ Requirements: Determinism is not a nicety. A benchmark whose failures cannot be reproduced is unusable the first time a result is disputed. +## Required variables + +`requires` names the inputs and outputs that the verdict depends on. Ask for +only what the probe actually uses: a model that cannot supply a requested +output is N/A with reason INCOMPLETE, not a physics failure. + +A reach-indexed routing probe declares its routing outputs separately from the +catchment-level `fluxes` and `states`: + +```yaml +requires: + routing: [reach_q_in, q_out, channel_storage] + static: [routing_network] +``` + +Its generator must supply `static["routing_network"]["reaches"]` as a +non-empty list of unique, non-empty string IDs. The adapter writes the +requested variables to `output/routing.csv`, with one row for every forcing +time and declared reach. + ## Criteria Each entry in `criteria` names a registered criterion and its parameters. From 49ac5fe8ba8751db3fa1598524937df172c0e0a7 Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Mon, 19 Oct 2026 00:23:06 -0700 Subject: [PATCH 49/50] Update spec.py --- src/hydroturing/spec.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/hydroturing/spec.py b/src/hydroturing/spec.py index 3a36c591..28456e52 100644 --- a/src/hydroturing/spec.py +++ b/src/hydroturing/spec.py @@ -624,7 +624,7 @@ def load_model(path: str | Path) -> ModelManifest: if unknown: raise SpecError(f"{spec_file}: unknown variables in emits: {unknown}") - if raw["emits"].get("routing") and "routing_network" not in { + if raw["emits"].get("routing") and "routing_network" not in { *raw.get("needs_static", []), *raw.get("uses_static", []), }: From 6cc4db9166ff068ac250c213cb3f98bec56486dc Mon Sep 17 00:00:00 2001 From: Joa_Wang Date: Mon, 19 Oct 2026 00:25:59 -0700 Subject: [PATCH 50/50] Update protocol.py --- src/hydroturing/protocol.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/hydroturing/protocol.py b/src/hydroturing/protocol.py index d5d6d324..3eaed4be 100644 --- a/src/hydroturing/protocol.py +++ b/src/hydroturing/protocol.py @@ -144,7 +144,7 @@ def stage(io_dir: Path, case: Case, probe: ProbeSpec, model: ModelManifest) -> P if probe.requires_routing: _routing_reaches(case) - + (io_dir / "input").mkdir(parents=True, exist_ok=True) (io_dir / "output").mkdir(parents=True, exist_ok=True) @@ -275,7 +275,7 @@ def _read_routing( path = io_dir / ROUTING_CSV if not path.exists(): raise ProtocolError(f"adapter did not write required {ROUTING_CSV}") - + table = pd.read_csv(path, dtype={REACH_COL: str}, keep_default_na=False) required_columns = [TIME_COL, REACH_COL, *probe.requires_routing] missing = [ @@ -296,7 +296,7 @@ def _read_routing( ) expected_reaches = _routing_reaches(case) - + actual_reaches = set(table[REACH_COL]) expected_set = set(expected_reaches) if actual_reaches != expected_set: