[CONTRACT] Add reach-indexed routing output

AGENTS.md

@@ -42,9 +42,55 @@ 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`.
+
+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,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
+```
+
+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`. `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
+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.
 There is no network. Do not attempt to download weights or data at run time;
@@ -171,6 +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 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 |
@@ -255,6 +304,28 @@ 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: [reach_q_in, q_out, channel_storage]
+needs_static: [routing_network]
+```
+
+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 `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
@@ -297,7 +368,10 @@ ht run --model <model-name>              # 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

docs/adapting-a-model.md

@@ -64,6 +64,33 @@ 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: [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`.
+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.
@@ -91,8 +118,10 @@ river/reach-inflow path whenever the column is present.
 
 ## 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())

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.

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. 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,
           "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."

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. 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",
           "items": {

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
 
@@ -233,12 +234,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

src/hydroturing/harness.py

@@ -512,6 +512,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 if probe.requires_routing else ()),
         variants=(),
         criteria=(),
     )