Make lateral inflow sources pluggable (Protocol + adapters)

[taddyb]

Motivation

DDR currently couples lateral inflow ingestion to a specific storage convention (icechunk stores with Qr variable, (divide_id, time) dims, mm/day units, 1980-01-01 origin). Adding any new inflow source — a different LSTM variant, NWM output, a NetCDF file, on-the-fly model output, parquet, etc. — requires editing src/ddr/io/readers.py directly.

This is a concrete blocker for the ARC-GR paper work, which evaluates parameter transferability across lateral inflow models (dHBV vs LSTM vs others). Each new inflow source we want to compare currently means a code change in core DDR rather than a drop-in module.

Current state

Friction points in the existing design:

1. StreamflowReader is tightly coupled to icechunk + Qr. src/ddr/io/readers.py hardcodes the variable name, dim names, units, and time origin.
2. Daily/hourly is a boolean flag, not a resolution abstraction. is_hourly: bool branches the read path; sub-daily-but-not-hourly inputs have no clean home.
3. Divide ID convention is dataset-specific and implicit. MERIT integers vs Lynker "cat-{id}" — readers must know which BaseGeoDataset is active.
4. Unit and origin assumptions are implicit. mm/day→m³/s conversion uses catchment area from attributes; 1980-01-01 origin is hardcoded. Neither is documented at the interface boundary.
5. No registry / plugin pattern. Config selects a store path, not a reader implementation.

Proposed approach (staged)

Stage 1 — Specify the canonical contract (docs only)

Document the invariants any lateral inflow source must satisfy when handed off to dmc:

• Returns a torch.Tensor of shape (num_segments, num_hours), float32, m³/s.
• Segment ordering matches the batch adjacency in RoutingDataclass.
• Hourly resolution (dt = 3600s), aligned to Dates.batch_hourly_time_range.
• Source is responsible for its own unit conversion, temporal interpolation, and segment alignment.

Add this to CLAUDE.md and a new docs/inflow_sources.md. No code changes. This unblocks contributors writing one-off adapters today.

Stage 2 — Introduce LateralInflowReader Protocol

# src/ddr/io/protocols.py
from typing import Protocol
import torch
from ddr.geodatazoo.dataclasses import RoutingDataclass

class LateralInflowReader(Protocol):
    def __call__(self, routing_dataclass: RoutingDataclass) -> torch.Tensor: ...

• Existing StreamflowReader already matches this shape — annotate it as the reference implementation, no behavior change.
• dmc and training/testing/routing scripts type-hint against the Protocol, not the concrete class.

Stage 3 — Config-driven reader selection

Extend DataSources to accept either the current path-based shorthand (preserves all existing configs) or an explicit reader spec:

data_sources:
  streamflow:
    reader: ddr.io.readers.StreamflowReader   # dotted path
    kwargs:
      store: ${oc.env:DDR_STREAMFLOW_STORE}
      is_hourly: false

Resolution happens in validate_config() via importlib. Existing configs continue to work via a back-compat branch that constructs StreamflowReader from a bare path string.

Stage 4 — Reference adapters (as needed)

Ship one or two worked examples in src/ddr/io/adapters/:

• NetCDFInflowAdapter — reads a generic NetCDF file with configurable variable/dim/unit/origin.
• CallableInflowAdapter — wraps any Callable[[RoutingDataclass], Tensor] for users streaming from a live model.

Each gets a test fixture and a short example in examples/.

Acceptance criteria

• docs/inflow_sources.md exists and documents the canonical contract.
• LateralInflowReader Protocol exported from ddr.io.
• StreamflowReader annotated against the Protocol with no behavior change.
• Existing YAML configs in config/ continue to work unchanged (regression test).
• A new inflow source can be added in <50 LOC without modifying src/ddr/io/readers.py or src/ddr/io/__init__.py.
• At least one reference adapter (NetCDFInflowAdapter or CallableInflowAdapter) lands with tests.
• scripts/train.py, scripts/test.py, scripts/router.py resolve readers via config, not hardcoded import.

Out of scope

• Pluggable observation readers (IcechunkUSGSReader). Same shape of problem, separate issue.
• MCP server / Claude skill tooling for adapter scaffolding. Reasonable v2 once the Protocol lands; bundling muddies scope.
• Sub-hourly routing. dt = 3600s is hardcoded in mmc.py:192 and orthogonal to this issue.

Risk / concerns

• Back-compat for configs. Existing YAML uses a bare path string; the new spec is a dict. Resolution logic in validate_config() must accept both. Risk: silent breakage of user configs. Mitigation: explicit regression test loading every YAML in config/.
• Dotted-path imports in config. Lets users execute arbitrary code via config. Acceptable for a research tool but worth flagging in docs.
• Protocol vs ABC. Using typing.Protocol keeps it duck-typed and avoids forcing inheritance on third-party readers. Tradeoff: weaker static enforcement; mypy will catch most cases.

Notes for whoever picks this up

• Start with Stage 1 only and open a PR. Stages 2–4 should be separate PRs.
• The existing StreamflowReader.__call__ signature is the de facto contract — read it before drafting the Protocol.
• ARC-GR work (parameter transferability across inflow models) is the forcing function; if you need a worked example to validate the design, swapping daily LSTM for dHBV2.0 is the canonical case.