Downstream-state guard: fail loud when downstream dams aren't persisted

CLAUDE.md

@@ -19,6 +19,12 @@ Experimental package — breaking all the time and loving the learning curve. St
 
 **Closures are now tighter** — `PARS + BULK` gives 9 WSGs, not 15 (drops LKEL, MSKE, USKE, LBTN, MORR, FINA, all correctly). Study-area buckets built before and after are **not comparable**; earlier runs were over-inclusive rather than short, so their results stand. **Gotchas:** `lnk_db_conn()` lands on the tunnel (`:63333`, `bcfishpass`), while `data-raw/study_area_wsgs.R` hardcodes docker `fwapg` (`:5432`) — two databases, different load states (#222); docker `fwapg`'s `fwa_watershed_groups_poly` has NULL code columns, so derive from `fwa_stream_networks_sp`, which is populated in both. **#227 re-scoped** to just the single-WSG downstream-state guard — its body carries a three-tier design (auto-pass via a 2.4 s `cabd.dams` spatial check, fail loud, override recorded in the #127 run log) and is buildable now the closure is correct.
 
+## Status (2026-08-28) — v0.46.0 shipped (#227 downstream-state guard)
+
+`data-raw/wsg_run_one.R` had stated the DS-first precondition in its header since #175 and **nothing enforced it** — accessibility reads the *already-persisted* barriers of downstream WSGs, so an out-of-order run marks dammed-off segments accessible and exits 0. New exported **`lnk_wsg_downstream_check()`** verifies it. **The predicate is PATH, not membership** — that distinction is the whole design: flagging when a downstream *group contains* a blocking dam fires on BULK (18 dams across LSKE/KISP/KLUM, none below its outlet), which is the issue's own example, and would train operators to override reflexively. Testing each dam with measure-aware `fwa_downstream()` from `fresh::frs_wsg_outlets()` gives PARS its three real dams (Peace Canyon, Site C, Bennett), SLOC the Brilliant Dam, BULK nothing. ~0.5 s. **Do not "simplify" it back to membership** — `RUNBOOK.md` §8c carries the measurement for exactly that reason.
+
+The guard applies link's filters by **sharing** the pipeline's SQL: the `cabd` / `matched` CTEs live once and are consumed by both `.lnk_pipeline_prep_dams()` and the probe (parameterized on source — staged tables vs inline `VALUES`). Refactor verified behaviour-preserving against a golden capture taken first (ADMS/KOTL/PARS byte-identical). Override needs a written justification (bare `TRUE` rejected) and lands in `<persist>.log.notes` beside `wsg_upstream`. `study_area_run.sh` exports `LNK_GUARD_DOWNSTREAM=warn` on **both** legs (local *and* inside the ssh string — missing the second means cyphers hard-fail and WSGs get *skipped*, which `lnk_access(merge=TRUE)` cannot repair); `wsg_recompute_one.R` re-runs it in `error` mode post-consolidate, which is what makes `warn` a deferral rather than a hole. **Also fixed a v0.45.0 defect:** `.lnk_log_create_tables()` built the log tables but never the schema, so a brand-new persist schema failed — the log opens before `persist_init` by design, and every schema tested until now already existed. Follow-up **#244** (`cabd_additions` dams carry `barrier_ind = t` but psc NULL, so they can never become barriers).
+
 ## Status (2026-07-11) — #231 closed misdirected; #232 opened (crossings parity)
 
 **Key correction (do NOT re-rabbit-hole):** the pipeline builds `<schema>.crossings` **from DB primitives** (`lnk_pipeline_crossings` → `.lnk_crossings_union`: PSCIS + `fresh.modelled_stream_crossings` + CABD), **not** from `crossings.csv`. The CSV read at `lnk_pipeline_load.R:100` is **vestigial** — the union drops + rebuilds the table before break/classify/mapping_code touch it. So #231 ("consume weekly crossings.csv; repoint pipeline off fresh") was **closed as misdirected**; lessons in `planning/archive/2026-07-issue-231-crossings-from-primitives/README.md` (+ memory `crossings-built-from-primitives`). Freshness lever = `data-raw/snapshot_bcfp.sh` reloading the primitives into the **`fwapg`** DB (last load ~2026-05-26; `lnk_db_conn()` defaults to a `bcfishpass` DB that LACKS them). Opened **#232** — confirm link's built crossings ≈ bcfp's complete `crossings_vw` (the parity reference). Aside: `crossings.csv` was published to `s3://newgraph` (db_newgraph#15, smnorris PR #57) before we realized the models don't consume it — **db_newgraph#16** tracks reconsidering that dump.

DESCRIPTION

@@ -1,6 +1,6 @@
 Package: link
 Title: Stream Network Habitat Interpretation (Experimental)
-Version: 0.45.3
+Version: 0.46.0
 Date: 2026-08-28
 Authors@R: c(
     person("Allan", "Irvine", , "[email protected]",

NAMESPACE

@@ -50,6 +50,7 @@ export(lnk_source)
 export(lnk_stamp)
 export(lnk_stamp_finish)
 export(lnk_thresholds)
+export(lnk_wsg_downstream_check)
 export(lnk_wsg_resolve)
 import(DBI)
 importFrom(RPostgres,Postgres)

NEWS.md

@@ -1,3 +1,15 @@
+# link 0.46.0
+
+Enforces the DS-first precondition that `data-raw/wsg_run_one.R` has stated in its own header since link#175 and that nothing checked ([#227](https://github.com/NewGraphEnvironment/link/issues/227)). Accessibility is not computed from the focal watershed group alone — link reads the **already-persisted** barriers of the groups downstream — so modelling a WSG out of order writes `streams_access` / `streams_mapping_code` marking segments accessible that are in fact dammed off, prints `done in N min`, and exits 0. A wrong answer indistinguishable from a right one. New exported `lnk_wsg_downstream_check()` verifies it instead: find the blocking dams on the focal WSG's downstream flow path, confirm each is already persisted as a barrier, and fail loud naming them when it is not.
+
+**The predicate is path, not membership**, and that is the difference between a guard and a nuisance. The issue proposed flagging when a downstream group *contains* a blocking dam; measured against the live database that fires on BULK — whose closure holds 18 blocking dams across LSKE/KISP/KLUM, none of them below its outlet — which is the issue's own motivating example. Operators would learn to reach for the override and the guard would stop meaning anything. Testing each dam with the measure-aware `whse_basemapping.fwa_downstream()` from the outlet shipped by `fresh::frs_wsg_outlets()` gives PARS its three real dams (Peace Canyon, Site C, W.A.C. Bennett), SLOC the Brilliant Dam, and BULK nothing. It is complete rather than merely cheaper: access walks downstream from every segment and every focal segment exits through the focal outlet, so the out-of-WSG barriers reachable from *any* focal segment are exactly those below it. ~0.5 s against a 5 s budget.
+
+The guard applies link's own filters by **sharing** the pipeline's SQL, not copying it: the `cabd` and `matched` CTE bodies now live once and are consumed by both `.lnk_pipeline_prep_dams()` and the probe, parameterized on source so the pipeline reads its staged tables while the guard inlines `(VALUES …)` and writes nothing — the pattern fresh 0.33.0 used to retire `public.wsg_outlet`. A guard that snapped or filtered differently would flag dams the pipeline treats as passable, which is the issue's stated make-or-break. The refactor is behaviour-preserving and was verified so, against a golden capture taken first: byte-identical for ADMS, KOTL and PARS.
+
+Three tiers. Auto-pass is the common case. Failure names the dams, their passability codes and the DS-first order to model them in, and offers `wsg_recompute_one.R` as a correct escape that is not the override. The override requires a written justification — a bare `TRUE` is rejected, because the justification *is* the mechanism — and lands in `<persist>.log.notes` beside `wsg_upstream` (#127), so `lnk_log_read()` reports afterwards that a network was built on a stated assumption. `study_area_run.sh` exports `LNK_GUARD_DOWNSTREAM=warn` on **both** legs, since a downstream group can legitimately be mid-flight on another cypher; that is a deferral rather than a hole because `wsg_recompute_one.R` re-runs the guard in `error` mode after consolidate, when everything must be persisted. A hard pre-flight there would be worse than the bug: per-WSG failures soft-fail, so the WSG would be *skipped*, and `lnk_access(merge = TRUE)` cannot repair a group that was never modelled.
+
+Also fixes a defect shipped in 0.45.0: `.lnk_log_create_tables()` built the run-log tables but never the schema, so a brand-new persist schema failed with `schema does not exist`. The log is opened before `lnk_persist_init` by design — the open row must predate any write so `wsg_upstream` reflects the state the run started from — and every schema tested until now already existed. Found by running the guard's smoke tests into scratch schemas. New `RUNBOOK.md` §8c carries the membership-vs-path measurement so it is not "simplified" back, and records the guard's known bound: it inherits `frs_wsg_drainage()`'s one-outlet-per-group model. Follow-up [#244](https://github.com/NewGraphEnvironment/link/issues/244) records that `cabd_additions` dams carry `barrier_ind = t` yet can never become barriers.
+
 # link 0.45.3
 
 Purges stale `public.wsg_outlet` references and documents the ltree-closure trap ([#243](https://github.com/NewGraphEnvironment/link/pull/243)). #227/v0.45.1 removed the table as a *concept* — `[email protected]` ships outlets and passes them as a `VALUES` list — but references survived in places that still steered behaviour, including one shipped in v0.45.0: `R/lnk_log.R` listed `public.wsg_outlet` as a pipeline **input primitive**, so the run-provenance log fingerprinted a table that should not exist (real numbers on a machine carrying the leftover, NULLs on a clean one). `research/study_areas.md` was worse, documenting the **superseded** technique as current — "closure and downstream-first order are derived from `wscode_ltree` ancestry (`@>`)" — and citing #227 as tracking the table's reproducible build, when #227 had been re-scoped to delete it. `data-raw/study_area_wsgs.R` was already a shim around `lnk_wsg_resolve()`, so no live code path computed a wrong closure; the damage was documentation steering a reader into the trap, which is exactly what it did. New `RUNBOOK.md` §8b records the mechanism: two watershed groups on the same stream **share an outlet code**, so `a.outlet @> b.outlet` is true in *both* directions and reports each as downstream of the other — closure is measure-aware, not code-aware. The worked example is the Kootenay, where FWA carries the out-and-back through Montana and Idaho under one continuous `wscode_ltree = 300.625474`: an ltree test puts KOTR/SMAR/BULL below Kootenay Lake when they sit above it via the US loop, and the true closure of `LARL/KOTL/SLOC` is just those three, LARL being the terminal BC group. §8b also warns that a `public.wsg_outlet` table found in any database is a leftover that still answers queries, wrongly. Found by falling into it — a hand-rolled `@>` query invented a five-WSG "outstanding downstream closure" that does not exist.

R/lnk_log.R

@@ -274,6 +274,11 @@ cols_log_input <- c(
 #'
 #' @noRd
 .lnk_log_create_tables <- function(conn, schema) {
+  # The log is opened BEFORE lnk_persist_init runs (the open row must predate
+  # any write, so wsg_upstream reflects the state the run started from), so on
+  # a brand-new persist schema nothing has created it yet.
+  .lnk_db_execute(conn, sprintf("CREATE SCHEMA IF NOT EXISTS %s", schema))
+
   specs <- list(
     list(table = "log", cols = cols_log, pk = "run_id"),
     list(table = "log_input", cols = cols_log_input,

R/lnk_pipeline_prepare.R

@@ -798,52 +798,17 @@ lnk_pipeline_prepare <- function(conn, aoi, cfg, loaded, schema,
   #    - UNION ALL with cabd_additions where feature_type='dams' (US placeholders)
   .lnk_db_execute(conn, sprintf(
     "DROP TABLE IF EXISTS %1$s.dams", schema))
+  # The `cabd` and `matched` CTE bodies are shared verbatim with the link#227
+  # downstream guard (`R/lnk_wsg_downstream_check.R`). A guard that snapped or
+  # filtered differently would flag dams this pipeline treats as passable, so
+  # the SQL is defined once and parameterized on its source rather than copied.
   .lnk_db_execute(conn, sprintf(
     "CREATE TABLE %1$s.dams AS
      WITH cabd AS (
-       SELECT d.cabd_id::text  AS dam_id,
-              blk.blue_line_key,
-              ST_GeomFromEWKB(d.geom_ewkb) AS geom,
-              d.dam_name_en, d.height_m, d.owner, d.dam_use,
-              d.operating_status,
-              COALESCE(u.passability_status_code,
-                       d.passability_status_code) AS passability_status_code
-       FROM %1$s.cabd_dams_raw d
-       LEFT OUTER JOIN %1$s.cabd_exclusions x ON d.cabd_id = x.cabd_id
-       LEFT OUTER JOIN %1$s.cabd_blkey_xref blk ON d.cabd_id = blk.cabd_id
-       LEFT OUTER JOIN %1$s.cabd_passability_status_updates u
-         ON d.cabd_id = u.cabd_id
-       WHERE x.cabd_id IS NULL
+       %2$s
      ),
      matched AS (
-       SELECT DISTINCT ON (c.dam_id)
-              c.dam_id,
-              str.linear_feature_id,
-              str.blue_line_key,
-              str.wscode_ltree,
-              str.localcode_ltree,
-              str.watershed_group_code,
-              ST_Distance(str.geom, c.geom) AS distance_to_stream,
-              ST_InterpolatePoint(str.geom, c.geom) AS downstream_route_measure,
-              c.dam_name_en, c.height_m, c.owner, c.dam_use,
-              c.operating_status, c.passability_status_code,
-              str.geom AS line_geom
-       FROM cabd c
-       CROSS JOIN LATERAL (
-         SELECT linear_feature_id, blue_line_key, wscode_ltree, localcode_ltree,
-                watershed_group_code, geom
-         FROM whse_basemapping.fwa_stream_networks_sp str
-         WHERE str.localcode_ltree IS NOT NULL
-           AND NOT str.wscode_ltree <@ '999'::ltree
-           AND (
-             (c.blue_line_key IS NULL)
-             OR (c.blue_line_key = str.blue_line_key)
-           )
-         ORDER BY str.geom <-> c.geom
-         LIMIT 1
-       ) str
-       WHERE ST_Distance(str.geom, c.geom) <= 65
-       ORDER BY c.dam_id, ST_Distance(str.geom, c.geom), str.linear_feature_id
+       %3$s
      ),
      placed AS (
        SELECT m.dam_id,
@@ -891,7 +856,17 @@ lnk_pipeline_prepare <- function(conn, aoi, cfg, loaded, schema,
      SELECT * FROM placed
      UNION ALL
      SELECT * FROM usa;",
-    schema))
+    schema,
+    .lnk_dams_cabd_sql(
+      dams_expr = sprintf(
+        "(SELECT cabd_id, passability_status_code, dam_name_en, height_m,
+                 owner, dam_use, operating_status,
+                 ST_GeomFromEWKB(geom_ewkb) AS geom
+            FROM %s.cabd_dams_raw)", schema),
+      excl_ref = sprintf("%s.cabd_exclusions", schema),
+      xref_ref = sprintf("%s.cabd_blkey_xref", schema),
+      upd_ref  = sprintf("%s.cabd_passability_status_updates", schema)),
+    .lnk_dams_matched_sql()))
 
   # 4. Filter the local <schema>.dams to the AOI (per-WSG locality).
   .lnk_db_execute(conn, sprintf(