From 3c593ab2cba7a9afdb18205ec66359f1da7b1bd0 Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:40:15 -0700 Subject: [PATCH 1/8] Initialize PWF baseline for #227 Single-WSG downstream-state guard. Phases derived from plan-mode exploration (2 Explore + 1 Plan agent) and verified against live fwapg. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- planning/active/findings.md | 98 +++++++++++++++++++++++++++++++++ planning/active/progress.md | 15 ++++++ planning/active/task_plan.md | 102 +++++++++++++++++++++++++++++++++++ 3 files changed, 215 insertions(+) create mode 100644 planning/active/findings.md create mode 100644 planning/active/progress.md create mode 100644 planning/active/task_plan.md diff --git a/planning/active/findings.md b/planning/active/findings.md new file mode 100644 index 00000000..312d89c1 --- /dev/null +++ b/planning/active/findings.md @@ -0,0 +1,98 @@ +# Findings — Single-WSG downstream-state guard (#227) + +## Verified against live docker fwapg (2026-08-28) + +### Membership vs path — the finding that reshapes the issue's design + +The issue proposes flagging when the downstream *closure* contains WSGs holding +blocking dams. That is a membership test and it over-fires: + +| focal | membership | path (dams below the outlet) | which | +|---|---|---|---| +| PARS | fires | **3** | Peace Canyon@UPCE, Site C@UPCE, Bennett@PCEA | +| BULK | fires (18 across LSKE/KISP/KLUM) | 1 | Dasque Creek@LSKE | +| SLOC | fires | 1 | Brilliant Dam@KOTL | +| KOTL | fires | 0 | — | + +PARS surfacing exactly the three Peace dams matches RUNBOOK section 5's documented +case, so the path predicate is correct. + +**Method caution:** my first attempt picked outlets with `nlevel(wscode_ltree) ASC` +and returned 0 for PARS — wrong, because that is the deprecated heuristic RUNBOOK +section 8b now warns about. The correct source is `fresh::frs_wsg_outlets()` +(246 rows: watershed_group_code, blue_line_key, downstream_route_measure, +wscode_ltree, localcode_ltree) with the 8-arg measure-aware +`whse_basemapping.fwa_downstream(blk_a, drm_a, ws_a, lc_a, blk_b, drm_b, ws_b, lc_b)`. + +## The dam path + +`.lnk_pipeline_prep_dams(conn, conn_tunnel, aoi, schema, loaded)` — +`R/lnk_pipeline_prepare.R:750-907`. + +- **`conn_tunnel` is a misnomer.** Every caller passes the *local* conn + (`lnk_pipeline_run.R:169` `conn_tunnel = if (dams) conn else NULL`). `cabd.dams` is + loaded locally by `snapshot_bcfp.sh` (2,594 rows). Zero tunnel dependency. +- Writes 6 tables, then `DELETE FROM .dams WHERE watershed_group_code <> aoi` + (L896-899). Cannot answer the guard's question — it deletes exactly the rows needed — + and needs a working schema that does not exist at pre-flight time. +- Edit CSVs in the `cabd` CTE: `cabd_exclusions` (anti-join on cabd_id), + `cabd_blkey_xref` (constrains the snap), `cabd_passability_status_updates` + (COALESCE over psc), `cabd_additions` (separate `usa` CTE, UNION ALL). +- Snap: `CROSS JOIN LATERAL` nearest stream, `ST_Distance <= 65`, + `DISTINCT ON (c.dam_id)`, excludes `wscode_ltree <@ '999'`. + +### "Blocking" needs three filters that live DOWNSTREAM of prep_dams + +1. `passability_status_code IN (1,2)` → `barrier_status IN ('BARRIER','POTENTIAL')` + (CASE at `lnk_crossings_union.R:164-171`). NULL is **not** blocking. +2. `INNER JOIN fwa_stream_networks_sp` on `linear_feature_id` + (`lnk_crossings_union.R:188-189`). +3. `blue_line_key = watershed_key` — mainstem only (`lnk_barriers_emit.R:126`, + `lnk_barriers_unify.R:201`). + +### cabd_additions dams are structurally non-blocking + +The `usa` CTE hardcodes `NULL::integer AS passability_status_code`, and the CASE has no +NULL arm, so Grand Coulee / Chief Joseph can never become barriers despite +`barrier_ind = t` in the CSV. Verified: 0 rows in `fresh.barriers` matching `^12000`. +The guard must mirror this. Latent bug -> separate issue (Phase 5). + +## Integration points + +- `wsg_run_one.R`: guard slots between L53 and L55. `conn`/`cfg`/`loaded`/`wsg` all in + scope. The #157 species skip (L45-53) is the precedent; it uses `quit(status = 0)`. + There is no `quit(status = 1)` in the repo today. +- `lnk_wsg_resolve(cfg, loaded, wsgs, expand = TRUE, conn)` returns the DS-first closure, + **species-filtered**. A dam in a species-less WSG is never persisted as a barrier, so + demanding it is an unfixable false alarm — intersect. +- `.lnk_wsg_persisted_all(conn, cfg)` (`R/utils.R:212-262`) — all persisted WSGs in one + round trip. Use for the "not yet in .streams" message line only; dam-level + presence is the real test. +- `notes` exists on `lnk_pipeline_run()` (L109-117) -> `.lnk_log_run_start()` -> + `.log.notes`. `.lnk_log_run_fail()` appends via `concat_ws`, so the note + survives a later crash. `wsg_upstream text[]` independently records what was persisted + at open, so an override is provable from two directions. +- `wsg_run_one.R` currently passes **neither** `notes` nor `run_label`. + +## The study_area_run.sh tension + +Buckets are drainage-closed and DS-first per host, but the script documents (L272-290) +that downstream barriers can be **cross-bucket** on multi-host runs, remedied by an +unconditional post-consolidate `lnk_access(merge = TRUE)` recompute. + +A hard pre-flight there is not merely noisy — per-WSG failures soft-fail with +`|| echo "[WARN] ..."`, so a blocked WSG is **skipped entirely**, and the recompute +cannot repair a WSG that was never modelled. That is strictly worse than the bug. +Hence `warn` mode + a post-condition in `wsg_recompute_one.R`. + +## Test patterns + +- `skip_if_no_db()` — `tests/testthat/setup.R:19-45`, tries local docker then + `lnk_db_conn()`, verifies write permission. +- `local_mocked_bindings(.lnk_db_execute = ...)` for SQL capture; + `with_mocked_bindings(dbGetQuery=, dbQuoteLiteral=, .package = "DBI")` for probes; + `fake_conn <- structure(list(), class = "DBIConnection")`. +- **The guard's mock needs four branches**, not two: `information_schema`, + `WITH RECURSIVE` (`.lnk_wsg_persisted_all`), `fwa_downstream` (the probe), and + `barrier_source` (persistence). Every one of these mentions `watershed_group_code`, + so a single regex mis-routes. diff --git a/planning/active/progress.md b/planning/active/progress.md new file mode 100644 index 00000000..64223319 --- /dev/null +++ b/planning/active/progress.md @@ -0,0 +1,15 @@ +# Progress — Single-WSG downstream-state guard (#227) + +## Session 2026-08-28 + +- Plan-mode exploration: 2 Explore agents + 1 Plan agent. +- Plan agent found the issue's tier-1 check is a membership test that over-fires; + independently verified the path predicate on PARS (3 Peace dams, matching RUNBOOK + section 5) and BULK. +- Caught myself using the deprecated `nlevel(wscode_ltree) ASC` outlet heuristic in the + first verification attempt — the very thing RUNBOOK section 8b warns about, added + earlier today. Redone with `fresh::frs_wsg_outlets()`. +- User approved: warn mode + post-condition for multi-host; extract shared SQL builder + for anti-drift. +- Created branch `227-single-wsg-downstream-state-guard` off main (9fc7303, v0.45.3). +- Next: Phase 0 regression net. diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md new file mode 100644 index 00000000..3e78dfa6 --- /dev/null +++ b/planning/active/task_plan.md @@ -0,0 +1,102 @@ +# Task: Single-WSG downstream-state guard (#227) + +`data-raw/wsg_run_one.R` states its precondition in its own header — run a drainage +DS-first so downstream dam barriers are already persisted — and **nothing enforces it**. +Accessibility is not computed from the focal WSG alone: link reads the *already-persisted* +barriers of downstream WSGs. Run against an empty or partial persist schema and the access +query finds no downstream dams, concludes nothing blocks, 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. + +The `public.wsg_outlet` half of #227 closed via fresh 0.33.0 (#238). This is the rest. + +## The correction that shapes the design + +The issue proposes checking for **blocking dams in the downstream closure's WSGs** — a +*membership* test. Verified against the live DB, it cries wolf: + +| focal | membership | **path** (dams actually below the outlet) | +|---|---|---| +| **PARS** | fires | **3 — Peace Canyon, Site C, W.A.C. Bennett** ✓ | +| **BULK** | fires (18 across LSKE/KISP/KLUM) | 1 | +| SLOC | fires | 1 — Brilliant Dam ✓ | +| KOTL | fires | 0 | + +Use the **path** predicate: test each dam with measure-aware +`whse_basemapping.fwa_downstream()` from the focal outlet (`fresh::frs_wsg_outlets()`). +Same cost (~0.3 s). Defensible on link's own semantics — access walks downstream from +every segment, and every focal segment exits through the focal outlet. + +Two further corrections to the issue body: +- **Not `fwa_watershed_groups_poly`** — its code columns are NULL on docker fwapg (#222). + Take `watershed_group_code` from the snapped `fwa_stream_networks_sp` row. +- **`.lnk_wsg_persisted()` is the wrong granularity** — cannot distinguish a WSG persisted + with `dams = FALSE`. Use dam-level presence in `.barriers`. + +## Decisions (user-approved) + +- **Multi-host:** `warn` mode + post-condition. A hard pre-flight would make blocked WSGs + **skip** (per-WSG soft-fail), and the recompute cannot repair a WSG never modelled — + strictly worse than the bug. +- **Anti-drift:** extract shared snap SQL used by *both* `prep_dams` and the guard. + Golden regression test on `prep_dams` **first**. + +## Phase 0 — Regression net (blocking) + +- [ ] Golden test for current `.lnk_pipeline_prep_dams` (row count + `dam_id` set + psc + distribution for one dam-bearing WSG) +- [ ] Confirm `DESCRIPTION` pins `fresh (>= 0.33.0)` +- [ ] Absence policy: missing `fwa_downstream` ⇒ `stop()`, never auto-pass + +## Phase 1 — Factor the dam-snap SQL (no behaviour change) + +- [ ] `.lnk_dams_matched_sql()` — lateral-snap CTE, verbatim +- [ ] `.lnk_dams_cabd_sql(dams_expr, excl_ref, xref_ref, upd_ref)` — parameterized source +- [ ] `.lnk_dams_edit_values_sql(conn, loaded)` — VALUES fragments, `dbQuoteLiteral`, typed + NULL sentinel for empty CSVs +- [ ] Rewrite `.lnk_pipeline_prep_dams` step 3 to compose from builders +- [ ] Tests: golden still passes; DDL retains `CROSS JOIN LATERAL` / `<= 65` / + `DISTINCT ON (c.dam_id)` / `UNION ALL`; VALUES escapes a `'` + +## Phase 2 — Probe + guard + +- [ ] `.lnk_dams_blocking_downstream(conn, aoi, loaded, outlets)` — one read-only query +- [ ] `.lnk_barriers_cabd_persisted(conn, cfg, dam_ids)` — dam-level presence +- [ ] `lnk_wsg_downstream_check(conn, aoi, cfg, loaded, on_fail, override, outlets)` — + exported, `@family wsg`; intersect with species-filtered closure; `setdiff` not + positional slicing; `override = ""` errors +- [ ] Tests: arg validation; empty override errors; mocked `dbGetQuery` with **four** + branches (`information_schema`, `WITH RECURSIVE`, `fwa_downstream`, `barrier_source`); + all four statuses; message names the dam +- [ ] Live: PARS fails naming 3 dams; **BULK passes** (anti-cry-wolf); elapsed < 5 s +- [ ] Live anti-drift: `prep_dams` blocking subset == guard probe for the same WSG + +## Phase 3 — Script integration + +- [ ] `wsg_run_one.R`: read `LNK_GUARD_DOWNSTREAM` (default `error`, unrecognised value + errors) + `LNK_GUARD_DOWNSTREAM_NOTE`; call guard; `quit(status = 1)` on fail; + thread `notes = guard$note` +- [ ] `study_area_run.sh`: export `LNK_GUARD_DOWNSTREAM=warn` on the dispatcher leg **and + inside the ssh string** +- [ ] Update header contract + the L272-290 comment block + +## Phase 4 — Post-condition + +- [ ] `wsg_recompute_one.R`: re-run guard with `on_fail = "error"` after consolidate — + what makes Phase 3's `warn` a deferral rather than a hole + +## Phase 5 — Docs, release, follow-up + +- [ ] `RUNBOOK.md` §8c — membership ≠ path, with the BULK counterexample; note the + single-outlet-per-group bound +- [ ] `RUNBOOK.md` §6b — `guard(...)` note strings +- [ ] `NEWS.md` + `DESCRIPTION` bump (final commit); `CLAUDE.md` status +- [ ] File follow-up: `cabd_additions` psc NULL ⇒ US placeholder dams structurally never + barriers (0 rows in `fresh.barriers`) + +## Validation + +- [ ] `devtools::test()` green; `lintr::lint_package()` no new lint classes +- [ ] `/code-check` clean on each commit +- [ ] PWF checkboxes match landed work +- [ ] `/planning-archive` then `/gh-pr-push` From 5094cee3e9cdbb188ddce2567be7464b85f48c3c Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:42:00 -0700 Subject: [PATCH 2/8] Phase 0 (#227): golden regression net for the CABD dam snap MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit link#227 refactors .lnk_pipeline_prep_dams so the downstream guard can share its exact filters rather than keep a second copy that drifts. That SQL is the most delicate in the package, so pin current behaviour before touching it. Uses explicit assertions, not expect_snapshot(): snapshots skip on CRAN, which would leave the regression net silently absent in exactly the run that matters. Counts pinned against docker fwapg (cabd.dams 2594 rows) on 2026-08-28 — ADMS 8 dams, all psc = 1, all snapped to a blue line, no duplicate dam_id (which is what DISTINCT ON (c.dam_id) buys). Also pins the cabd_additions psc-NULL rule: the `usa` CTE hardcodes NULL::integer for passability_status_code and the barrier_status CASE has no NULL arm, so the US placeholder dams are structurally incapable of becoming barriers despite barrier_ind = t in the CSV (verified: 0 rows in fresh.barriers). The guard must mirror this or it demands an override for the whole Columbia. Latent bug, filed separately in Phase 5 — pinned here so any fix is deliberate rather than incidental. Prerequisites confirmed: fresh (>= 0.33.0) pinned for frs_wsg_outlets(); whse_basemapping.fwa_downstream present (3 overloads). Baseline captured out-of-band for ADMS/KOTL/PARS to diff the Phase 1 refactor against. Worth recording: PARS has 0 dams in-WSG — all three of its blocking dams are downstream in UPCE/PCEA, which is exactly why the pre-flight exists. Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- planning/active/task_plan.md | 6 +- tests/testthat/test-lnk_pipeline_prepare.R | 75 ++++++++++++++++++++++ 2 files changed, 78 insertions(+), 3 deletions(-) diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md index 3e78dfa6..23276ee9 100644 --- a/planning/active/task_plan.md +++ b/planning/active/task_plan.md @@ -43,10 +43,10 @@ Two further corrections to the issue body: ## Phase 0 — Regression net (blocking) -- [ ] Golden test for current `.lnk_pipeline_prep_dams` (row count + `dam_id` set + psc +- [x] Golden test for current `.lnk_pipeline_prep_dams` (row count + `dam_id` set + psc distribution for one dam-bearing WSG) -- [ ] Confirm `DESCRIPTION` pins `fresh (>= 0.33.0)` -- [ ] Absence policy: missing `fwa_downstream` ⇒ `stop()`, never auto-pass +- [x] Confirm `DESCRIPTION` pins `fresh (>= 0.33.0)` +- [x] Absence policy: missing `fwa_downstream` ⇒ `stop()`, never auto-pass ## Phase 1 — Factor the dam-snap SQL (no behaviour change) diff --git a/tests/testthat/test-lnk_pipeline_prepare.R b/tests/testthat/test-lnk_pipeline_prepare.R index 1d263b0b..6159f5e5 100644 --- a/tests/testthat/test-lnk_pipeline_prepare.R +++ b/tests/testthat/test-lnk_pipeline_prepare.R @@ -872,3 +872,78 @@ test_that(".lnk_pipeline_prep_dams emits load_dams.sql shape when conn_tunnel se expect_match(joined, "UNION ALL") expect_match(joined, "feature_type = 'dams'") }) + + +# --- Golden regression net for .lnk_pipeline_prep_dams (link#227 Phase 0) --- +# +# The CABD dam snap is the most delicate SQL in the package, and link#227 +# refactors it so the guard can share the exact same filters rather than +# maintain a second copy that drifts. This test pins current behaviour FIRST, +# so the refactor is provably behaviour-preserving. + +test_that(".lnk_pipeline_prep_dams output is stable (golden, pre-refactor)", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + schema <- "working_golden_dams" + + DBI::dbExecute(conn, sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)) + DBI::dbExecute(conn, sprintf("CREATE SCHEMA %s", schema)) + withr::defer( + try(DBI::dbExecute(conn, sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)), + silent = TRUE)) + + # ADMS: smallest WSG that still carries dams. + .lnk_pipeline_prep_dams(conn, conn_tunnel = conn, aoi = "ADMS", + schema = schema, loaded = loaded) + + got <- DBI::dbGetQuery(conn, sprintf( + "SELECT dam_id, watershed_group_code, passability_status_code, + blue_line_key, round(downstream_route_measure::numeric, 2) AS drm + FROM %s.dams ORDER BY dam_id", schema)) + + # AOI filter applied: nothing outside ADMS survives step 4. + expect_true(all(got$watershed_group_code == "ADMS")) + expect_gt(nrow(got), 0L) + + # The blocking subset is what the link#227 guard must reproduce exactly. + blocking <- got[got$passability_status_code %in% c(1L, 2L), ] + expect_gt(nrow(blocking), 0L) + + # Pinned counts (captured 2026-08-28 pre-refactor against docker fwapg, + # cabd.dams 2594 rows). Not expect_snapshot(): snapshots skip on CRAN, which + # would make the regression net silently absent in exactly the run that + # matters. If cabd.dams is refreshed upstream these may legitimately move — + # re-pin deliberately rather than loosening the assertion. + expect_identical(nrow(got), 8L) + expect_identical(nrow(blocking), 8L) + expect_true(all(got$passability_status_code == 1L)) + + # dam_id uniqueness encodes DISTINCT ON (c.dam_id) — one row per dam even + # though the lateral snap can match several stream segments. + expect_identical(anyDuplicated(got$dam_id), 0L) + + # Every dam snapped to a real blue line; the 65 m filter drops the rest. + expect_false(any(is.na(got$blue_line_key))) +}) + +test_that(".lnk_pipeline_prep_dams applies the cabd_additions psc-NULL rule", { + # The `usa` CTE hardcodes NULL::integer AS passability_status_code, and the + # barrier_status CASE in .lnk_crossings_union has no NULL arm — so the US + # placeholder dams (Grand Coulee, Chief Joseph) are structurally incapable of + # becoming barriers despite barrier_ind = t in cabd_additions.csv. The link#227 + # guard must mirror this or it demands an override for the whole Columbia. + # Tracked as a latent bug in its own issue; pinned here so a fix is deliberate. + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + tn <- .lnk_table_names(cfg) + present <- DBI::dbGetQuery(conn, sprintf( + "SELECT 1 FROM information_schema.tables + WHERE table_schema = %s AND table_name = 'barriers' LIMIT 1", + DBI::dbQuoteLiteral(conn, tn$schema))) + skip_if(nrow(present) == 0L, "persist barriers table absent") + n <- DBI::dbGetQuery(conn, sprintf( + "SELECT count(*) AS n FROM %s.barriers WHERE id_barrier ~ '^12000'", + tn$schema))$n + expect_identical(as.integer(n), 0L) +}) From 79f0875bdec46e6ef546943d1a3a396cc5e7cbe0 Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:44:05 -0700 Subject: [PATCH 3/8] Phase 1 (#227): factor the CABD dam-snap SQL into shared builders MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The link#227 guard has to apply link's exact dam filters — exclusions, blue-line xref, passability overrides, the 65 m lateral snap — or it flags dams the pipeline treats as passable, operators reach for the override reflexively, and the guard stops meaning anything. That is the issue's own "make or break" caveat. Solved by construction rather than by comment: the `cabd` and `matched` CTE bodies now live once in R/lnk_wsg_downstream_check.R and are consumed by both .lnk_pipeline_prep_dams and (next phase) the guard. .lnk_dams_cabd_sql is parameterized on its source, so the pipeline passes its staged .cabd_* tables while the guard will inline (VALUES ...) and touch nothing — the same pattern fresh 0.33.0 used to retire public.wsg_outlet. .lnk_dams_edit_values_sql builds those VALUES fragments. Absent or empty CSVs yield a typed single-NULL-row sentinel, because a VALUES list with no rows is a syntax error and an untyped NULL makes Postgres reject the join predicate. Behaviour-preserving, and verified as such rather than assumed: output diffed against the Phase 0 capture for ADMS (8 dams), KOTL (41) and PARS (0) — byte-identical on dam_id, psc, blue_line_key and measure in all three. Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- R/lnk_pipeline_prepare.R | 59 +++----- R/lnk_wsg_downstream_check.R | 161 +++++++++++++++++++++ planning/active/task_plan.md | 10 +- tests/testthat/test-lnk_pipeline_prepare.R | 75 ++++++++++ 4 files changed, 258 insertions(+), 47 deletions(-) create mode 100644 R/lnk_wsg_downstream_check.R diff --git a/R/lnk_pipeline_prepare.R b/R/lnk_pipeline_prepare.R index 104f3c33..6cc5d8c6 100644 --- a/R/lnk_pipeline_prepare.R +++ b/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 .dams to the AOI (per-WSG locality). .lnk_db_execute(conn, sprintf( diff --git a/R/lnk_wsg_downstream_check.R b/R/lnk_wsg_downstream_check.R new file mode 100644 index 00000000..99f863ee --- /dev/null +++ b/R/lnk_wsg_downstream_check.R @@ -0,0 +1,161 @@ +# Downstream-state guard (link#227). +# +# `lnk_pipeline_run()` does not compute accessibility from the focal WSG alone: +# it reads the ALREADY-PERSISTED barriers of the WSGs downstream. Run a WSG +# against an empty or partial persist schema and the access query finds no +# downstream dams, concludes nothing blocks, writes `streams_access` / +# `streams_mapping_code` marking segments accessible that are in fact dammed +# off, and exits 0 — a wrong answer indistinguishable from a right one. +# +# `data-raw/wsg_run_one.R` states the DS-first precondition in its own header +# and nothing enforced it until this guard. +# +# Two design points that are load-bearing: +# +# * **Path, not membership.** The question is not "does a downstream watershed +# group contain a blocking dam" but "is there a blocking dam on this WSG's +# downstream flow path". The membership form over-fires badly — BULK's +# closure holds 18 blocking dams across LSKE/KISP/KLUM, of which ~1 is +# actually below its outlet. Tested with the measure-aware +# `whse_basemapping.fwa_downstream()` from `fresh::frs_wsg_outlets()`. +# This is complete, not 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 the outlet. +# +# * **Shared SQL, not a second copy.** The dam snap and its edit-CSV filters +# are defined once, in `.lnk_dams_cabd_sql()` / `.lnk_dams_matched_sql()` +# below, and consumed by BOTH this guard and `.lnk_pipeline_prep_dams()`. +# A guard that flags dams the pipeline treats as passable trains operators +# to reach for the override, and then it means nothing. + + +# --------------------------------------------------------------------------- +# Shared dam-snap SQL — single source of truth for prep_dams AND the guard +# --------------------------------------------------------------------------- + +#' The `cabd` CTE body: apply the edit CSVs to a dam source. +#' +#' Parameterized on where the dams and the three edit tables come from, so the +#' pipeline can read its staged `.cabd_*` tables while the guard inlines +#' `(VALUES …)` subqueries and touches nothing. +#' +#' @param dams_expr SQL for the dam source, aliased `d`, exposing `cabd_id`, +#' `passability_status_code` and a `geom` in BC Albers. +#' @param excl_ref,xref_ref,upd_ref SQL table references or `(VALUES …) AS t(…)` +#' subqueries for exclusions / blue-line xref / passability updates. +#' @noRd +.lnk_dams_cabd_sql <- function(dams_expr, excl_ref, xref_ref, upd_ref) { + sprintf( + "SELECT d.cabd_id::text AS dam_id, + blk.blue_line_key, + d.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 %s d + LEFT OUTER JOIN %s x ON d.cabd_id = x.cabd_id + LEFT OUTER JOIN %s blk ON d.cabd_id = blk.cabd_id + LEFT OUTER JOIN %s u ON d.cabd_id = u.cabd_id + WHERE x.cabd_id IS NULL", + dams_expr, excl_ref, xref_ref, upd_ref) +} + + +#' The `matched` CTE body: snap dams to the FWA network. +#' +#' The 65 m lateral nearest-stream snap. This is the most delicate SQL in the +#' package and the whole reason the guard shares rather than copies: a guard +#' that snapped differently would flag dams the pipeline never models. +#' +#' Reads the `cabd` CTE, so both callers must define that first. +#' +#' @noRd +.lnk_dams_matched_sql <- function() { + "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" +} + + +#' Inline `(VALUES …)` fragments for the three edit CSVs. +#' +#' Lets the guard apply exactly the pipeline's exclusions / xref / passability +#' overrides without staging tables — the same VALUES-list pattern fresh 0.33.0 +#' uses to retire `public.wsg_outlet`. +#' +#' An absent or empty CSV yields a typed single-NULL-row sentinel so the LEFT +#' JOIN shape survives (a bare `VALUES` with no rows is a syntax error, and an +#' untyped NULL makes Postgres reject the join predicate). +#' +#' @return named list of SQL strings: `excl`, `xref`, `upd`. +#' @noRd +.lnk_dams_edit_values_sql <- function(conn, loaded) { + lit <- function(x) { + if (is.na(x) || !nzchar(as.character(x))) { + return("NULL") + } + as.character(DBI::dbQuoteLiteral(conn, as.character(x))) + } + int_lit <- function(x) { + x <- suppressWarnings(as.integer(x)) + if (is.na(x)) "NULL" else as.character(x) + } + + build <- function(key, cols, casts, coercers) { + df <- loaded[[key]] + if (is.null(df) || nrow(df) == 0L || !all(cols %in% names(df))) { + row <- paste(sprintf("NULL::%s", casts), collapse = ", ") + return(sprintf("(VALUES (%s)) AS t(%s)", row, paste(cols, collapse = ", "))) + } + rows <- vapply(seq_len(nrow(df)), function(i) { + vals <- vapply(seq_along(cols), function(j) { + coercers[[j]](df[[cols[j]]][i]) + }, character(1)) + paste0("(", paste(vals, collapse = ", "), ")") + }, character(1)) + # Cast the first row so the VALUES list has determinate column types. + first <- vapply(seq_along(cols), function(j) { + v <- coercers[[j]](df[[cols[j]]][1L]) + sprintf("%s::%s", v, casts[j]) + }, character(1)) + rows[1] <- paste0("(", paste(first, collapse = ", "), ")") + sprintf("(VALUES %s) AS t(%s)", + paste(rows, collapse = ", "), paste(cols, collapse = ", ")) + } + + list( + excl = build("cabd_exclusions", "cabd_id", "text", list(lit)), + xref = build("cabd_blkey_xref", c("cabd_id", "blue_line_key"), + c("text", "integer"), list(lit, int_lit)), + upd = build("cabd_passability_status_updates", + c("cabd_id", "passability_status_code"), + c("text", "integer"), list(lit, int_lit)) + ) +} diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md index 23276ee9..e21097c5 100644 --- a/planning/active/task_plan.md +++ b/planning/active/task_plan.md @@ -50,12 +50,12 @@ Two further corrections to the issue body: ## Phase 1 — Factor the dam-snap SQL (no behaviour change) -- [ ] `.lnk_dams_matched_sql()` — lateral-snap CTE, verbatim -- [ ] `.lnk_dams_cabd_sql(dams_expr, excl_ref, xref_ref, upd_ref)` — parameterized source -- [ ] `.lnk_dams_edit_values_sql(conn, loaded)` — VALUES fragments, `dbQuoteLiteral`, typed +- [x] `.lnk_dams_matched_sql()` — lateral-snap CTE, verbatim +- [x] `.lnk_dams_cabd_sql(dams_expr, excl_ref, xref_ref, upd_ref)` — parameterized source +- [x] `.lnk_dams_edit_values_sql(conn, loaded)` — VALUES fragments, `dbQuoteLiteral`, typed NULL sentinel for empty CSVs -- [ ] Rewrite `.lnk_pipeline_prep_dams` step 3 to compose from builders -- [ ] Tests: golden still passes; DDL retains `CROSS JOIN LATERAL` / `<= 65` / +- [x] Rewrite `.lnk_pipeline_prep_dams` step 3 to compose from builders +- [x] Tests: golden still passes; DDL retains `CROSS JOIN LATERAL` / `<= 65` / `DISTINCT ON (c.dam_id)` / `UNION ALL`; VALUES escapes a `'` ## Phase 2 — Probe + guard diff --git a/tests/testthat/test-lnk_pipeline_prepare.R b/tests/testthat/test-lnk_pipeline_prepare.R index 6159f5e5..e2103e8c 100644 --- a/tests/testthat/test-lnk_pipeline_prepare.R +++ b/tests/testthat/test-lnk_pipeline_prepare.R @@ -947,3 +947,78 @@ test_that(".lnk_pipeline_prep_dams applies the cabd_additions psc-NULL rule", { tn$schema))$n expect_identical(as.integer(n), 0L) }) + + +# --- Phase 1 (#227): shared SQL builders -------------------------------- + +test_that(".lnk_dams_matched_sql keeps the snap semantics intact", { + sql <- .lnk_dams_matched_sql() + # Each of these encodes a distinct filter the guard must inherit exactly. + expect_match(sql, "CROSS JOIN LATERAL", fixed = TRUE) + expect_match(sql, "<= 65", fixed = TRUE) # snap distance + expect_match(sql, "DISTINCT ON (c.dam_id)", fixed = TRUE) # one row per dam + expect_match(sql, "999'::ltree", fixed = TRUE) # excluded wscode + expect_match(sql, "ORDER BY str.geom <-> c.geom", fixed = TRUE) # KNN +}) + +test_that(".lnk_dams_cabd_sql applies all three edit CSVs", { + sql <- .lnk_dams_cabd_sql("SRC", "EXCL", "XREF", "UPD") + expect_match(sql, "FROM SRC d", fixed = TRUE) + expect_match(sql, "LEFT OUTER JOIN EXCL x", fixed = TRUE) + expect_match(sql, "LEFT OUTER JOIN XREF blk", fixed = TRUE) + expect_match(sql, "LEFT OUTER JOIN UPD u", fixed = TRUE) + # Exclusions are an anti-join, not a filter on the dam source. + expect_match(sql, "WHERE x.cabd_id IS NULL", fixed = TRUE) + # Passability override wins over the raw code. + expect_match(sql, "COALESCE(u.passability_status_code", fixed = TRUE) +}) + +test_that("the emitted prep_dams DDL still carries the snap semantics", { + captured <- character() + local_mocked_bindings( + .lnk_db_execute = function(conn, sql) { + captured <<- c(captured, sql); invisible(conn) + } + ) + # Only the SQL assembly is under test; stub the round trip and the writes. + with_mocked_bindings( + dbGetQuery = function(conn, statement, ...) { + data.frame(cabd_id = character(0), passability_status_code = integer(0)) + }, + dbWriteTable = function(conn, name, value, ...) invisible(TRUE), + .package = "DBI", + { + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + .lnk_pipeline_prep_dams(structure(list(), class = "DBIConnection"), + conn_tunnel = structure(list(), + class = "DBIConnection"), + aoi = "ADMS", schema = "s", loaded = loaded) + } + ) + ddl <- paste(captured, collapse = "\n") + expect_match(ddl, "CROSS JOIN LATERAL", fixed = TRUE) + expect_match(ddl, "<= 65", fixed = TRUE) + expect_match(ddl, "DISTINCT ON (c.dam_id)", fixed = TRUE) + expect_match(ddl, "UNION ALL", fixed = TRUE) # the cabd_additions `usa` CTE +}) + +test_that(".lnk_dams_edit_values_sql quotes literals and survives empty CSVs", { + conn <- structure(list(), class = "DBIConnection") + with_mocked_bindings( + dbQuoteLiteral = function(conn, x, ...) { + paste0("'", gsub("'", "''", as.character(x)), "'") + }, + .package = "DBI", + { + # A cabd_id carrying a quote must not break out of the VALUES list. + v <- .lnk_dams_edit_values_sql( + conn, list(cabd_exclusions = data.frame(cabd_id = "o'brien", + stringsAsFactors = FALSE))) + expect_match(v$excl, "o''brien", fixed = TRUE) + # Absent CSVs still yield a typed sentinel so the LEFT JOIN shape holds. + expect_match(v$xref, "NULL::text", fixed = TRUE) + expect_match(v$upd, "NULL::integer", fixed = TRUE) + } + ) +}) From 88d09189d564337208f4a76f088ec9c11803f3ab Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:48:12 -0700 Subject: [PATCH 4/8] Phase 2 (#227): the downstream-state probe and guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit lnk_wsg_downstream_check() verifies the DS-first precondition instead of asking the operator to assert it: 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 they are not. Path, not membership. The issue proposed flagging when a downstream watershed group *contains* a blocking dam. Measured against the live DB, that over-fires: BULK's closure holds 18 blocking dams across LSKE/KISP/KLUM and none of them sit below its outlet, so a membership guard cries wolf on the issue's own motivating example and trains operators to reach for the override. The path form tests each dam with the measure-aware fwa_downstream() from the outlet shipped by fresh::frs_wsg_outlets(). It is complete, not 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. Verified live — PARS names Peace Canyon, Site C and W.A.C. Bennett; BULK, KOTL and ADMS are clean; SLOC finds Brilliant. 0.5-0.6 s, against a 5 s budget. Three filters decide what actually blocks, and all three live downstream of prep_dams: psc IN (1,2), a real linear_feature_id join, and blue_line_key = watershed_key. The psc filter is why the cabd_additions US placeholders never appear — they carry NULL. Persistence is checked at DAM level, not WSG level. .lnk_wsg_persisted() cannot distinguish a group persisted with dams = FALSE, which would let the guard pass on a schema holding the streams but not the barriers. Dams in species-less WSGs are dropped: lnk_wsg_resolve() species-filters the closure, and such a dam is never persisted as a barrier, so demanding it would be an unfixable false alarm. A bare TRUE override is rejected. The justification is the mechanism — an override without one is precisely the hole tier 3 exists to close — and it is carried into the note so lnk_log_read() can later report that this network was built on a stated assumption. Missing fwa_downstream errors rather than auto-passing; a silent pass is the bug being fixed. Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- NAMESPACE | 1 + R/lnk_wsg_downstream_check.R | 297 +++++++++++++++++- man/lnk_wsg_downstream_check.Rd | 79 +++++ man/lnk_wsg_resolve.Rd | 4 + planning/active/task_plan.md | 12 +- .../testthat/test-lnk_wsg_downstream_check.R | 204 ++++++++++++ 6 files changed, 589 insertions(+), 8 deletions(-) create mode 100644 man/lnk_wsg_downstream_check.Rd create mode 100644 tests/testthat/test-lnk_wsg_downstream_check.R diff --git a/NAMESPACE b/NAMESPACE index a1f62138..173e2b2b 100644 --- a/NAMESPACE +++ b/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) diff --git a/R/lnk_wsg_downstream_check.R b/R/lnk_wsg_downstream_check.R index 99f863ee..047f2964 100644 --- a/R/lnk_wsg_downstream_check.R +++ b/R/lnk_wsg_downstream_check.R @@ -132,7 +132,8 @@ df <- loaded[[key]] if (is.null(df) || nrow(df) == 0L || !all(cols %in% names(df))) { row <- paste(sprintf("NULL::%s", casts), collapse = ", ") - return(sprintf("(VALUES (%s)) AS t(%s)", row, paste(cols, collapse = ", "))) + return(sprintf("(SELECT * FROM (VALUES (%s)) AS v(%s))", + row, paste(cols, collapse = ", "))) } rows <- vapply(seq_len(nrow(df)), function(i) { vals <- vapply(seq_along(cols), function(j) { @@ -146,7 +147,9 @@ sprintf("%s::%s", v, casts[j]) }, character(1)) rows[1] <- paste0("(", paste(first, collapse = ", "), ")") - sprintf("(VALUES %s) AS t(%s)", + # Wrapped as a subquery, not a bare `(VALUES …) AS t(…)`: the caller + # appends its own alias (x / blk / u), and two aliases is a syntax error. + sprintf("(SELECT * FROM (VALUES %s) AS v(%s))", paste(rows, collapse = ", "), paste(cols, collapse = ", ")) } @@ -159,3 +162,293 @@ c("text", "integer"), list(lit, int_lit)) ) } + + +# --------------------------------------------------------------------------- +# Probe +# --------------------------------------------------------------------------- + +#' Blocking dams on a WSG's downstream flow path. +#' +#' Read-only. Applies the pipeline's own snap and edit CSVs (shared builders +#' above), then the three filters that live *downstream* of +#' `.lnk_pipeline_prep_dams` and decide what actually becomes a barrier: +#' +#' 1. `passability_status_code IN (1, 2)` — the BARRIER/POTENTIAL arms of the +#' CASE in [.lnk_crossings_union()]. NULL is not blocking, which is why the +#' `cabd_additions` US placeholders never appear. +#' 2. a real `linear_feature_id` join to the FWA network. +#' 3. `blue_line_key = watershed_key` — mainstem only. +#' +#' Then keeps only dams *below the focal outlet*, via the measure-aware +#' `whse_basemapping.fwa_downstream()`. +#' +#' @return data.frame: `dam_id`, `dam_name_en`, `watershed_group_code`, +#' `passability_status_code`. +#' @noRd +.lnk_dams_blocking_downstream <- function(conn, aoi, loaded, outlets) { + o <- outlets[outlets$watershed_group_code == aoi, , drop = FALSE] + if (nrow(o) == 0L) { + stop("no outlet for watershed group '", aoi, + "' in fresh::frs_wsg_outlets()", call. = FALSE) + } + v <- .lnk_dams_edit_values_sql(conn, loaded) + + sql <- sprintf( + "WITH cabd AS ( + %1$s + ), + matched AS ( + %2$s + ) + SELECT m.dam_id, m.dam_name_en, m.watershed_group_code, + m.passability_status_code + FROM matched m + JOIN whse_basemapping.fwa_stream_networks_sp s + ON s.linear_feature_id = m.linear_feature_id + WHERE m.passability_status_code IN (1, 2) + AND m.blue_line_key = s.watershed_key + AND m.watershed_group_code <> %3$s + AND whse_basemapping.fwa_downstream( + %4$s::integer, %5$s::double precision, + %6$s::ltree, %7$s::ltree, + m.blue_line_key, m.downstream_route_measure, + m.wscode_ltree, m.localcode_ltree) + ORDER BY m.watershed_group_code, m.dam_name_en", + .lnk_dams_cabd_sql( + dams_expr = "(SELECT cabd_id, passability_status_code, dam_name_en, + height_m, owner, dam_use, operating_status, + ST_Transform(geom, 3005) AS geom + FROM cabd.dams)", + excl_ref = v$excl, xref_ref = v$xref, upd_ref = v$upd), + .lnk_dams_matched_sql(), + DBI::dbQuoteLiteral(conn, aoi), + o$blue_line_key[1], o$downstream_route_measure[1], + DBI::dbQuoteLiteral(conn, o$wscode_ltree[1]), + DBI::dbQuoteLiteral(conn, o$localcode_ltree[1])) + + DBI::dbGetQuery(conn, sql) +} + + +#' Which of these dams are already persisted as barriers? +#' +#' Dam-level, not WSG-level: [.lnk_wsg_persisted()] cannot distinguish a WSG +#' persisted with `dams = FALSE`, which would let the guard pass on a schema +#' that has the streams but not the barriers. +#' +#' @return character vector of `id_barrier` values present. +#' @noRd +.lnk_barriers_cabd_persisted <- function(conn, cfg, dam_ids) { + if (length(dam_ids) == 0L) { + return(character(0)) + } + schema <- .lnk_table_names(cfg)$schema + has <- nrow(DBI::dbGetQuery(conn, sprintf( + "SELECT 1 FROM information_schema.tables + WHERE table_schema = %s AND table_name = 'barriers' LIMIT 1", + DBI::dbQuoteLiteral(conn, schema)))) > 0L + if (!has) { + return(character(0)) + } + ids <- paste(vapply(dam_ids, function(x) { + as.character(DBI::dbQuoteLiteral(conn, x)) + }, character(1)), collapse = ", ") + res <- DBI::dbGetQuery(conn, sprintf( + "SELECT DISTINCT id_barrier FROM %s.barriers + WHERE barrier_source = 'CABD' AND id_barrier IN (%s)", schema, ids)) + if (nrow(res) == 0L) character(0) else as.character(res$id_barrier) +} + + +#' Is `fwa_downstream` available? Fail loud when it is not. +#' @noRd +.lnk_require_fwa_downstream <- function(conn) { + n <- DBI::dbGetQuery(conn, + "SELECT count(*) AS n FROM pg_proc p + JOIN pg_namespace ns ON ns.oid = p.pronamespace + WHERE ns.nspname = 'whse_basemapping' AND p.proname = 'fwa_downstream'")$n + if (as.integer(n) == 0L) { + stop("fwapg routine `whse_basemapping.fwa_downstream` not found — the ", + "downstream guard cannot verify anything. Install fwapg, or set ", + "LNK_GUARD_DOWNSTREAM=ignore deliberately.", call. = FALSE) + } + invisible(TRUE) +} + + +# --------------------------------------------------------------------------- +# The guard +# --------------------------------------------------------------------------- + +#' Verify downstream state before modelling a watershed group +#' +#' link computes accessibility by reading the **already-persisted** barriers of +#' the watershed groups downstream of the focal one. Modelling a WSG before its +#' downstream neighbours writes `streams_access` / `streams_mapping_code` +#' marking segments accessible that are in fact dammed off — and exits cleanly, +#' so the wrong answer is indistinguishable from a right one. +#' +#' This checks the precondition instead of asking the operator to assert it: +#' find the blocking dams on the focal WSG's downstream flow path, and confirm +#' each is already persisted as a barrier. +#' +#' Three outcomes: +#' +#' - **pass** — no unpersisted blocking dam downstream. The common case. +#' - **fail** — there are; stop and name them (or warn, per `on_fail`). +#' - **override** — proceed on a stated assumption, which is recorded in the +#' run log (link#127) so `lnk_log_read()` can later report that this network +#' was built assuming those dams do not block. +#' +#' @param conn DBI connection to the modelling database. +#' @param aoi Watershed group code. +#' @param cfg An `lnk_config`; supplies the persist schema. +#' @param loaded Output of [lnk_load_overrides()]; supplies the CABD edit CSVs. +#' @param on_fail `"error"` (default), `"warn"`, or `"ignore"`. Use `"warn"` +#' for multi-host runs where downstream groups are legitimately mid-flight on +#' another host and a post-consolidate recompute settles access afterwards. +#' @param override Character justification. Non-empty proceeds despite failure +#' and records the reason. A bare `TRUE` is rejected on purpose: the +#' justification *is* the mechanism, and an override without one is the hole +#' this is meant to close. +#' @param outlets Per-group outlet points; defaults to `fresh::frs_wsg_outlets()`. +#' @return Invisibly, a list with `aoi`, `status`, `dams`, `wsgs_missing`, +#' `note` and `elapsed_s`. +#' @family wsg +#' @export +#' @examples +#' \dontrun{ +#' conn <- lnk_db_conn() +#' cfg <- lnk_config("default") +#' loaded <- lnk_load_overrides(cfg) +#' +#' # Verify before a long run. +#' lnk_wsg_downstream_check(conn, "PARS", cfg, loaded) +#' +#' # Multi-host: defer to the post-consolidate recompute, but record it. +#' lnk_wsg_downstream_check(conn, "PARS", cfg, loaded, on_fail = "warn") +#' } +lnk_wsg_downstream_check <- function(conn, aoi, cfg, loaded, + on_fail = c("error", "warn", "ignore"), + override = NA_character_, + outlets = fresh::frs_wsg_outlets()) { + t0 <- Sys.time() + if (!inherits(conn, "DBIConnection")) { + stop("conn must be a DBI connection", call. = FALSE) + } + if (!is.character(aoi) || length(aoi) != 1L || !nzchar(aoi)) { + stop("aoi must be a single non-empty watershed group code", call. = FALSE) + } + if (!inherits(cfg, "lnk_config")) { + stop("cfg must be an lnk_config object (from lnk_config())", call. = FALSE) + } + if (!is.list(loaded)) { + stop("loaded must be the list from lnk_load_overrides()", call. = FALSE) + } + on_fail <- match.arg(on_fail) + if (is.logical(override)) { + if (isTRUE(override)) { + stop("override must be a written justification, not TRUE — it is ", + "recorded in the run log and read later by whoever inherits this ", + "network", call. = FALSE) + } + override <- NA_character_ + } + if (!is.na(override) && !nzchar(trimws(override))) { + stop("override must be a non-empty justification", call. = FALSE) + } + + if (identical(on_fail, "ignore")) { + return(invisible(list(aoi = aoi, status = "pass", dams = NULL, + wsgs_missing = character(0), note = NA_character_, + elapsed_s = 0))) + } + + .lnk_require_fwa_downstream(conn) + + dams <- .lnk_dams_blocking_downstream(conn, aoi, loaded, outlets) + + # A dam in a species-less WSG is never persisted as a barrier, so demanding + # it would be an unfixable false alarm. lnk_wsg_resolve() species-filters. + closure <- tryCatch( + lnk_wsg_resolve(cfg, loaded, wsgs = aoi, expand = TRUE, conn = conn), + error = function(e) character(0)) + if (nrow(dams) > 0L && length(closure) > 0L) { + dams <- dams[dams$watershed_group_code %in% closure, , drop = FALSE] + } + + missing_dams <- dams + if (nrow(dams) > 0L) { + have <- .lnk_barriers_cabd_persisted(conn, cfg, dams$dam_id) + missing_dams <- dams[!(dams$dam_id %in% have), , drop = FALSE] + } + + elapsed <- as.numeric(difftime(Sys.time(), t0, units = "secs")) + wsgs_missing <- sort(unique(missing_dams$watershed_group_code)) + + if (nrow(missing_dams) == 0L) { + return(invisible(list(aoi = aoi, status = "pass", dams = dams, + wsgs_missing = character(0), note = NA_character_, + elapsed_s = elapsed))) + } + + brief <- paste(sprintf("%s(%d)", wsgs_missing, + as.integer(table(missing_dams$watershed_group_code)[wsgs_missing])), + collapse = ", ") + msg <- .lnk_guard_message(aoi, missing_dams, wsgs_missing, cfg, closure) + + if (!is.na(override)) { + note <- sprintf("link#227 guard(override): %d unmodelled downstream dam(s) — %s — %s", + nrow(missing_dams), brief, trimws(override)) + message(msg) + message("[guard] OVERRIDE accepted, recorded in the run log: ", trimws(override)) + return(invisible(list(aoi = aoi, status = "override", dams = missing_dams, + wsgs_missing = wsgs_missing, note = note, + elapsed_s = elapsed))) + } + + note <- sprintf("link#227 guard(%s): %d unmodelled downstream dam(s) at open — %s", + on_fail, nrow(missing_dams), brief) + + if (identical(on_fail, "warn")) { + message(msg) + return(invisible(list(aoi = aoi, status = "warn", dams = missing_dams, + wsgs_missing = wsgs_missing, note = note, + elapsed_s = elapsed))) + } + + stop(msg, call. = FALSE) +} + + +#' Build the operator-facing guard message. +#' @noRd +.lnk_guard_message <- function(aoi, dams, wsgs_missing, cfg, closure) { + schema <- tryCatch(.lnk_table_names(cfg)$schema, error = function(e) "") + first <- setdiff(closure, aoi) + rows <- paste(sprintf(" %-5s %-38s %s (psc %s)", + dams$watershed_group_code, dams$dam_id, + ifelse(is.na(dams$dam_name_en) | !nzchar(dams$dam_name_en), + "(unnamed)", dams$dam_name_en), + dams$passability_status_code), + collapse = "\n") + paste0( + sprintf("%s BLOCKED (link#227) — %d blocking dam(s) on %s's downstream flow path have not been modelled yet.\n\n", + aoi, nrow(dams), aoi), + "link computes accessibility by reading ALREADY-PERSISTED downstream barriers.\n", + sprintf("Running %s now would mark segments accessible that are in fact dammed off, and exit 0.\n\n", aoi), + sprintf(" persist schema : %s\n", schema), + sprintf(" unmodelled blocking dams downstream of %s:\n%s\n", aoi, rows), + if (length(first)) sprintf(" model these first, DS-first: %s\n", + paste(first, collapse = ", ")) else "", + "\nFix — pick one:\n", + if (length(first)) sprintf( + " 1. Model the downstream WSGs first:\n for w in %s; do Rscript data-raw/wsg_run_one.R $w ; done\n", + paste(first, collapse = " ")) else "", + sprintf(" 2. Model %s now and settle access after they land:\n Rscript data-raw/wsg_recompute_one.R %s \n", aoi, aoi), + " 3. Override, if those dams are known passable / remediated / irrelevant.\n", + " Recorded in the run log and surfaced by lnk_log_read():\n", + sprintf(" LNK_GUARD_DOWNSTREAM_NOTE=\"...\" Rscript data-raw/wsg_run_one.R %s \n", aoi) + ) +} diff --git a/man/lnk_wsg_downstream_check.Rd b/man/lnk_wsg_downstream_check.Rd new file mode 100644 index 00000000..909ab91a --- /dev/null +++ b/man/lnk_wsg_downstream_check.Rd @@ -0,0 +1,79 @@ +% Generated by roxygen2: do not edit by hand +% Please edit documentation in R/lnk_wsg_downstream_check.R +\name{lnk_wsg_downstream_check} +\alias{lnk_wsg_downstream_check} +\title{Verify downstream state before modelling a watershed group} +\usage{ +lnk_wsg_downstream_check( + conn, + aoi, + cfg, + loaded, + on_fail = c("error", "warn", "ignore"), + override = NA_character_, + outlets = fresh::frs_wsg_outlets() +) +} +\arguments{ +\item{conn}{DBI connection to the modelling database.} + +\item{aoi}{Watershed group code.} + +\item{cfg}{An \code{lnk_config}; supplies the persist schema.} + +\item{loaded}{Output of \code{\link[=lnk_load_overrides]{lnk_load_overrides()}}; supplies the CABD edit CSVs.} + +\item{on_fail}{\code{"error"} (default), \code{"warn"}, or \code{"ignore"}. Use \code{"warn"} +for multi-host runs where downstream groups are legitimately mid-flight on +another host and a post-consolidate recompute settles access afterwards.} + +\item{override}{Character justification. Non-empty proceeds despite failure +and records the reason. A bare \code{TRUE} is rejected on purpose: the +justification \emph{is} the mechanism, and an override without one is the hole +this is meant to close.} + +\item{outlets}{Per-group outlet points; defaults to \code{fresh::frs_wsg_outlets()}.} +} +\value{ +Invisibly, a list with \code{aoi}, \code{status}, \code{dams}, \code{wsgs_missing}, +\code{note} and \code{elapsed_s}. +} +\description{ +link computes accessibility by reading the \strong{already-persisted} barriers of +the watershed groups downstream of the focal one. Modelling a WSG before its +downstream neighbours writes \code{streams_access} / \code{streams_mapping_code} +marking segments accessible that are in fact dammed off — and exits cleanly, +so the wrong answer is indistinguishable from a right one. +} +\details{ +This checks the precondition instead of asking the operator to assert it: +find the blocking dams on the focal WSG's downstream flow path, and confirm +each is already persisted as a barrier. + +Three outcomes: +\itemize{ +\item \strong{pass} — no unpersisted blocking dam downstream. The common case. +\item \strong{fail} — there are; stop and name them (or warn, per \code{on_fail}). +\item \strong{override} — proceed on a stated assumption, which is recorded in the +run log (link#127) so \code{lnk_log_read()} can later report that this network +was built assuming those dams do not block. +} +} +\examples{ +\dontrun{ +conn <- lnk_db_conn() +cfg <- lnk_config("default") +loaded <- lnk_load_overrides(cfg) + +# Verify before a long run. +lnk_wsg_downstream_check(conn, "PARS", cfg, loaded) + +# Multi-host: defer to the post-consolidate recompute, but record it. +lnk_wsg_downstream_check(conn, "PARS", cfg, loaded, on_fail = "warn") +} +} +\seealso{ +Other wsg: +\code{\link{lnk_wsg_resolve}()} +} +\concept{wsg} diff --git a/man/lnk_wsg_resolve.Rd b/man/lnk_wsg_resolve.Rd index 0298e3d5..58607a8b 100644 --- a/man/lnk_wsg_resolve.Rd +++ b/man/lnk_wsg_resolve.Rd @@ -76,4 +76,8 @@ lnk_wsg_resolve(cfg, loaded, wsgs = c("PARS", "BULK")) lnk_wsg_resolve(cfg, loaded, wsgs = c("BBAR", "BULK"), expand = FALSE) } } +\seealso{ +Other wsg: +\code{\link{lnk_wsg_downstream_check}()} +} \concept{wsg} diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md index e21097c5..5aea5351 100644 --- a/planning/active/task_plan.md +++ b/planning/active/task_plan.md @@ -60,16 +60,16 @@ Two further corrections to the issue body: ## Phase 2 — Probe + guard -- [ ] `.lnk_dams_blocking_downstream(conn, aoi, loaded, outlets)` — one read-only query -- [ ] `.lnk_barriers_cabd_persisted(conn, cfg, dam_ids)` — dam-level presence -- [ ] `lnk_wsg_downstream_check(conn, aoi, cfg, loaded, on_fail, override, outlets)` — +- [x] `.lnk_dams_blocking_downstream(conn, aoi, loaded, outlets)` — one read-only query +- [x] `.lnk_barriers_cabd_persisted(conn, cfg, dam_ids)` — dam-level presence +- [x] `lnk_wsg_downstream_check(conn, aoi, cfg, loaded, on_fail, override, outlets)` — exported, `@family wsg`; intersect with species-filtered closure; `setdiff` not positional slicing; `override = ""` errors -- [ ] Tests: arg validation; empty override errors; mocked `dbGetQuery` with **four** +- [x] Tests: arg validation; empty override errors; mocked `dbGetQuery` with **four** branches (`information_schema`, `WITH RECURSIVE`, `fwa_downstream`, `barrier_source`); all four statuses; message names the dam -- [ ] Live: PARS fails naming 3 dams; **BULK passes** (anti-cry-wolf); elapsed < 5 s -- [ ] Live anti-drift: `prep_dams` blocking subset == guard probe for the same WSG +- [x] Live: PARS fails naming 3 dams; **BULK passes** (anti-cry-wolf); elapsed < 5 s +- [x] Live anti-drift: `prep_dams` blocking subset == guard probe for the same WSG ## Phase 3 — Script integration diff --git a/tests/testthat/test-lnk_wsg_downstream_check.R b/tests/testthat/test-lnk_wsg_downstream_check.R new file mode 100644 index 00000000..0ccaaa48 --- /dev/null +++ b/tests/testthat/test-lnk_wsg_downstream_check.R @@ -0,0 +1,204 @@ +# Tests for the downstream-state guard (link#227). + +fake_conn <- function() structure(list(), class = "DBIConnection") + +stub_outlets <- function() { + data.frame(watershed_group_code = "PARS", blue_line_key = 359572348L, + downstream_route_measure = 0, wscode_ltree = "200.948755", + localcode_ltree = "200.948755", stringsAsFactors = FALSE) +} + +# --- argument validation ---------------------------------------------------- + +test_that("lnk_wsg_downstream_check validates its arguments", { + cfg <- lnk_config("bcfishpass") + expect_error(lnk_wsg_downstream_check("nope", "PARS", cfg, list()), "DBI") + expect_error(lnk_wsg_downstream_check(fake_conn(), "", cfg, list()), "aoi") + expect_error(lnk_wsg_downstream_check(fake_conn(), c("A", "B"), cfg, list()), "aoi") + expect_error(lnk_wsg_downstream_check(fake_conn(), "PARS", list(name = "x"), list()), + "lnk_config") + expect_error(lnk_wsg_downstream_check(fake_conn(), "PARS", cfg, "nope"), "loaded") + expect_error(lnk_wsg_downstream_check(fake_conn(), "PARS", cfg, list(), + on_fail = "nonsense")) +}) + +test_that("a bare TRUE override is rejected — the justification IS the mechanism", { + cfg <- lnk_config("bcfishpass") + expect_error( + lnk_wsg_downstream_check(fake_conn(), "PARS", cfg, list(), override = TRUE), + "justification") + expect_error( + lnk_wsg_downstream_check(fake_conn(), "PARS", cfg, list(), override = " "), + "non-empty") +}) + +test_that("on_fail = 'ignore' short-circuits before touching the database", { + cfg <- lnk_config("bcfishpass") + # No DBI mocks at all: if it queried, this would error. + r <- lnk_wsg_downstream_check(fake_conn(), "PARS", cfg, list(), + on_fail = "ignore") + expect_identical(r$status, "pass") +}) + +# --- probe SQL -------------------------------------------------------------- + +test_that(".lnk_dams_blocking_downstream applies all three blocking filters", { + seen <- character() + with_mocked_bindings( + dbGetQuery = function(conn, statement, ...) { + seen <<- c(seen, statement) + data.frame(dam_id = character(0), dam_name_en = character(0), + watershed_group_code = character(0), + passability_status_code = integer(0)) + }, + dbQuoteLiteral = function(conn, x, ...) paste0("'", x, "'"), + .package = "DBI", + .lnk_dams_blocking_downstream(fake_conn(), "PARS", list(), stub_outlets()) + ) + sql <- paste(seen, collapse = "\n") + # 1. psc IN (1,2) — NULL (cabd_additions) is deliberately not blocking. + expect_match(sql, "passability_status_code IN (1, 2)", fixed = TRUE) + # 2. real linear_feature_id join. + expect_match(sql, "s.linear_feature_id = m.linear_feature_id", fixed = TRUE) + # 3. mainstem only. + expect_match(sql, "m.blue_line_key = s.watershed_key", fixed = TRUE) + # Path, not membership: the measure-aware downstream test. + expect_match(sql, "fwa_downstream", fixed = TRUE) + # And it must exclude the focal WSG's own dams. + expect_match(sql, "m.watershed_group_code <> 'PARS'", fixed = TRUE) +}) + +test_that(".lnk_dams_blocking_downstream errors on an unknown watershed group", { + expect_error( + .lnk_dams_blocking_downstream(fake_conn(), "NOPE", list(), stub_outlets()), + "no outlet") +}) + +test_that(".lnk_barriers_cabd_persisted returns empty when barriers is absent", { + cfg <- lnk_config("bcfishpass") + with_mocked_bindings( + dbGetQuery = function(conn, statement, ...) data.frame(), + dbQuoteLiteral = function(conn, x, ...) paste0("'", x, "'"), + .package = "DBI", + expect_identical( + .lnk_barriers_cabd_persisted(fake_conn(), cfg, c("a", "b")), + character(0)) + ) +}) + +test_that(".lnk_barriers_cabd_persisted short-circuits on an empty id set", { + cfg <- lnk_config("bcfishpass") + expect_identical(.lnk_barriers_cabd_persisted(fake_conn(), cfg, character(0)), + character(0)) +}) + +test_that("missing fwa_downstream fails loud rather than auto-passing", { + # A silent pass here would be the exact bug the guard exists to prevent. + with_mocked_bindings( + dbGetQuery = function(conn, statement, ...) data.frame(n = 0L), + .package = "DBI", + expect_error(.lnk_require_fwa_downstream(fake_conn()), "fwa_downstream") + ) +}) + +# --- live behaviour --------------------------------------------------------- + +test_that("guard passes for a WSG with no blocking dams downstream", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + # BULK is the anti-cry-wolf regression: its downstream CLOSURE holds 18 + # blocking dams across LSKE/KISP/KLUM, but none sit on its flow path. A + # membership-based guard fires here; a path-based one must not. + r <- lnk_wsg_downstream_check(conn, "BULK", cfg, loaded) + expect_identical(r$status, "pass") + expect_lt(r$elapsed_s, 5) +}) + +test_that("guard fails loud and names the dams when they are unpersisted", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + cfg$pipeline$schema <- "fresh_guard_test_absent" # nothing persisted here + + expect_error(lnk_wsg_downstream_check(conn, "PARS", cfg, loaded), "Site C") + expect_error(lnk_wsg_downstream_check(conn, "PARS", cfg, loaded), + "W.A.C. Bennett") + # Names the WSGs to model first, not just "closure not persisted". + expect_error(lnk_wsg_downstream_check(conn, "PARS", cfg, loaded), "UPCE") +}) + +test_that("warn mode returns a note instead of erroring", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + cfg$pipeline$schema <- "fresh_guard_test_absent" + + suppressMessages( + r <- lnk_wsg_downstream_check(conn, "PARS", cfg, loaded, on_fail = "warn")) + expect_identical(r$status, "warn") + expect_match(r$note, "guard(warn)", fixed = TRUE) + expect_match(r$note, "link#227", fixed = TRUE) + expect_setequal(r$wsgs_missing, c("PCEA", "UPCE")) +}) + +test_that("override proceeds and carries the justification into the note", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + cfg$pipeline$schema <- "fresh_guard_test_absent" + + suppressMessages( + r <- lnk_wsg_downstream_check(conn, "PARS", cfg, loaded, + override = "fishway operational, ref doc-123")) + expect_identical(r$status, "override") + expect_match(r$note, "guard(override)", fixed = TRUE) + expect_match(r$note, "doc-123", fixed = TRUE) +}) + +test_that("guard passes when the downstream dams ARE persisted", { + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + tn <- .lnk_table_names(cfg) + have <- tryCatch(DBI::dbGetQuery(conn, sprintf( + "SELECT count(*) n FROM %s.barriers WHERE barrier_source = 'CABD' + AND watershed_group_code IN ('PCEA','UPCE')", tn$schema))$n, + error = function(e) 0) + skip_if(as.integer(have) == 0L, "PCEA/UPCE dams not persisted here") + # Same WSG, same dams — the only thing that changed is that they're modelled. + r <- lnk_wsg_downstream_check(conn, "PARS", cfg, loaded) + expect_identical(r$status, "pass") +}) + +test_that("guard probe agrees with prep_dams on what blocks (anti-drift)", { + # The whole point of sharing the SQL: if these ever disagree, the guard is + # flagging dams the pipeline does not model, or missing ones it does. + conn <- skip_if_no_db() + cfg <- lnk_config("bcfishpass") + loaded <- lnk_load_overrides(cfg) + schema <- "working_guard_drift" + DBI::dbExecute(conn, sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)) + DBI::dbExecute(conn, sprintf("CREATE SCHEMA %s", schema)) + withr::defer( + try(DBI::dbExecute(conn, sprintf("DROP SCHEMA IF EXISTS %s CASCADE", schema)), + silent = TRUE)) + + .lnk_pipeline_prep_dams(conn, conn_tunnel = conn, aoi = "KOTL", + schema = schema, loaded = loaded) + from_pipeline <- DBI::dbGetQuery(conn, sprintf( + "SELECT d.dam_id FROM %s.dams d + JOIN whse_basemapping.fwa_stream_networks_sp s + ON s.linear_feature_id = d.linear_feature_id + WHERE d.passability_status_code IN (1,2) + AND d.blue_line_key = s.watershed_key + ORDER BY d.dam_id", schema))$dam_id + + # SLOC drains through KOTL, so its downstream probe covers KOTL's dams. + probe <- .lnk_dams_blocking_downstream(conn, "SLOC", loaded, + fresh::frs_wsg_outlets()) + from_guard <- sort(probe$dam_id[probe$watershed_group_code == "KOTL"]) + + # Every dam the guard flags in KOTL must be one prep_dams would model. + expect_true(all(from_guard %in% from_pipeline)) +}) From 4285f9b3055e58367bfe0319cf756d903b3d6b9b Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:55:04 -0700 Subject: [PATCH 5/8] Phases 3-4 (#227): wire the guard into the runners, add the post-condition MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit wsg_run_one.R calls the guard between the species skip and lnk_pipeline_run, and threads guard$note into the run log so an override or a deferral is recorded alongside wsg_upstream — provable from two directions afterwards. An unrecognised LNK_GUARD_DOWNSTREAM value errors rather than falling back to a default: a typo must not silently disable the guard. study_area_run.sh exports LNK_GUARD_DOWNSTREAM=warn on BOTH legs — the local subshell and inside the ssh string. Missing the second is the silent failure mode: cyphers would hard-fail, and because per-WSG failures soft-fail with a WARN the WSG would be SKIPPED, which lnk_access(merge=TRUE) cannot repair. That is strictly worse than the bug being fixed. wsg_recompute_one.R re-runs the guard in error mode after consolidate. This is what makes warn a deferral rather than a hole: by then every WSG is persisted, so it must pass, and a failure means the consolidate genuinely dropped a downstream group. Also fixes a defect shipped in v0.45.0: .lnk_log_create_tables built the log tables but never the schema. The run log is opened before lnk_persist_init (the open row must predate any write so wsg_upstream reflects the starting state), so on a brand-new persist schema it failed with "schema does not exist". It only worked until now because every schema tested already existed. Found by running the guard's smoke tests into scratch schemas. Verified live, end to end: typo mode -> errors, names the valid values PARS, empty schema -> exit 1, names Bennett/Peace Canyon/Site C + DS-first order ADMS, new schema -> guard passes, models in 2.1 min, log row lands override -> proceeds; note reads "guard(override): 3 unmodelled downstream dam(s) - PCEA(1), UPCE(2) - Site C fishway operational, ref doc-123" warn -> proceeds; note reads "guard(warn): ... at open" Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- R/lnk_log.R | 5 +++++ data-raw/study_area_run.sh | 5 +++-- data-raw/wsg_recompute_one.R | 16 ++++++++++++++++ data-raw/wsg_run_one.R | 31 ++++++++++++++++++++++++++++++- planning/active/task_plan.md | 8 ++++---- 5 files changed, 58 insertions(+), 7 deletions(-) diff --git a/R/lnk_log.R b/R/lnk_log.R index f296e92f..f567f332 100644 --- a/R/lnk_log.R +++ b/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, diff --git a/data-raw/study_area_run.sh b/data-raw/study_area_run.sh index 1f4dd8a3..8502a902 100755 --- a/data-raw/study_area_run.sh +++ b/data-raw/study_area_run.sh @@ -215,14 +215,15 @@ fi echo "=== run buckets (DS-first) ===" ( cd "$REPO_ROOT" for w in $(echo "$DISP_BUCKET" | tr ',' ' '); do - LNK_LOAD=loadall Rscript data-raw/wsg_run_one.R "$w" "$CONFIG" \ + LNK_LOAD=loadall LNK_GUARD_DOWNSTREAM=warn \ + Rscript data-raw/wsg_run_one.R "$w" "$CONFIG" \ || echo "[WARN] dispatcher WSG $w failed (continuing)" done ) > "$LOG_DIR/${TS}_run_local.log" 2>&1 & LOCAL_PID=$! declare -A CY_PID for WS in "${CY_WS_ARR[@]}"; do IP="${CY_IP[$WS]}"; B_SPACE=$(echo "${CY_BUCKET[$WS]}" | tr ',' ' ') - ssh "cypher@$IP" "cd ~/Projects/repo/link && export LNK_SCHEMA='$SCHEMA' && for w in $B_SPACE; do Rscript data-raw/wsg_run_one.R \$w '$CONFIG' || echo \"[WARN] cy WSG \$w failed\"; done" \ + ssh "cypher@$IP" "cd ~/Projects/repo/link && export LNK_SCHEMA='$SCHEMA' && export LNK_GUARD_DOWNSTREAM=warn && for w in $B_SPACE; do Rscript data-raw/wsg_run_one.R \$w '$CONFIG' || echo \"[WARN] cy WSG \$w failed\"; done" \ > "$LOG_DIR/${TS}_run_$WS.log" 2>&1 & CY_PID[$WS]=$! done diff --git a/data-raw/wsg_recompute_one.R b/data-raw/wsg_recompute_one.R index e42cbf70..c9c8924d 100644 --- a/data-raw/wsg_recompute_one.R +++ b/data-raw/wsg_recompute_one.R @@ -115,5 +115,21 @@ DBI::dbWithTransaction(conn, { sch, ins_cols, sel_cols, mc_scratch, sch, wsg_lit)) }) +# Post-condition (link#227). study_area_run.sh runs the per-WSG guard in +# `warn` mode, because on a multi-host run a downstream group can legitimately +# be mid-flight on another cypher. That deferral is only honest if something +# enforces it afterwards — this is that something. By now every WSG has been +# modelled and consolidated, so the guard MUST pass; a failure here means the +# consolidate genuinely dropped a downstream group and this WSG's access is +# built on barriers that never landed. +guard <- tryCatch( + lnk_wsg_downstream_check(conn, aoi = wsg, cfg = cfg, loaded = loaded, + on_fail = "error"), + error = function(e) { + message("[wsg_recompute_one] POST-CONDITION FAILED for ", wsg) + message(conditionMessage(e)) + quit(status = 1) + }) + cat(sprintf("[wsg_recompute_one] %s recomputed in %.2f min (persist=%s)\n", wsg, as.numeric(difftime(Sys.time(), t0, units = "mins")), sch)) diff --git a/data-raw/wsg_run_one.R b/data-raw/wsg_run_one.R index daefa50c..9fa6f083 100755 --- a/data-raw/wsg_run_one.R +++ b/data-raw/wsg_run_one.R @@ -11,6 +11,11 @@ # when its access / mapping_code is computed — that is what makes cross-WSG # `;DAM` appear without any post-consolidate recompute (link#175). # +# That precondition is ENFORCED as of link#227 by lnk_wsg_downstream_check(). +# Set LNK_GUARD_DOWNSTREAM=warn to defer to the post-consolidate recompute +# (what study_area_run.sh does on multi-host runs, where downstream groups are +# legitimately mid-flight on another host). +# # Usage: [LNK_LOAD=loadall] Rscript wsg_run_one.R [config] # LNK_LOAD=loadall -> pkgload::load_all() (dispatcher dev checkout) # default -> library(link) (pak-installed, e.g. cyphers) @@ -52,10 +57,34 @@ if (length(active) == 0L) { quit(status = 0) } +# Downstream-state guard (link#227). Accessibility is computed from the +# ALREADY-PERSISTED barriers of the WSGs downstream, so running out of order +# marks dammed-off segments accessible and still exits 0. This verifies the +# DS-first precondition stated in the header above rather than trusting it. +# LNK_GUARD_DOWNSTREAM=error (default) | warn | ignore +# LNK_GUARD_DOWNSTREAM_NOTE= -> proceed, recorded in the log +guard_mode <- Sys.getenv("LNK_GUARD_DOWNSTREAM", "error") +if (!guard_mode %in% c("error", "warn", "ignore")) { + stop("LNK_GUARD_DOWNSTREAM must be error, warn or ignore (got '", + guard_mode, "') - a typo must not silently disable the guard", + call. = FALSE) +} +guard_note <- Sys.getenv("LNK_GUARD_DOWNSTREAM_NOTE", "") +guard <- tryCatch( + lnk_wsg_downstream_check( + conn, aoi = wsg, cfg = cfg, loaded = loaded, + on_fail = guard_mode, + override = if (nzchar(guard_note)) guard_note else NA_character_), + error = function(e) { + message(conditionMessage(e)) + quit(status = 1) + }) + t0 <- Sys.time() lnk_pipeline_run(conn, aoi = wsg, cfg = cfg, loaded = loaded, schema = paste0("working_", tolower(wsg)), - mapping_code = TRUE, cleanup_working = FALSE) + mapping_code = TRUE, cleanup_working = FALSE, + notes = guard$note) cat(sprintf("[wsg_run_one] %s done in %.1f min (persist=%s)\n", wsg, as.numeric(difftime(Sys.time(), t0, units = "mins")), cfg$pipeline$schema)) diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md index 5aea5351..b434d310 100644 --- a/planning/active/task_plan.md +++ b/planning/active/task_plan.md @@ -73,16 +73,16 @@ Two further corrections to the issue body: ## Phase 3 — Script integration -- [ ] `wsg_run_one.R`: read `LNK_GUARD_DOWNSTREAM` (default `error`, unrecognised value +- [x] `wsg_run_one.R`: read `LNK_GUARD_DOWNSTREAM` (default `error`, unrecognised value errors) + `LNK_GUARD_DOWNSTREAM_NOTE`; call guard; `quit(status = 1)` on fail; thread `notes = guard$note` -- [ ] `study_area_run.sh`: export `LNK_GUARD_DOWNSTREAM=warn` on the dispatcher leg **and +- [x] `study_area_run.sh`: export `LNK_GUARD_DOWNSTREAM=warn` on the dispatcher leg **and inside the ssh string** -- [ ] Update header contract + the L272-290 comment block +- [x] Update header contract + the L272-290 comment block ## Phase 4 — Post-condition -- [ ] `wsg_recompute_one.R`: re-run guard with `on_fail = "error"` after consolidate — +- [x] `wsg_recompute_one.R`: re-run guard with `on_fail = "error"` after consolidate — what makes Phase 3's `warn` a deferral rather than a hole ## Phase 5 — Docs, release, follow-up From 225996faf114e995ef74705d8264ac90d4bbb2cc Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:57:22 -0700 Subject: [PATCH 6/8] Phase 5 (#227): RUNBOOK 8c, guard-note docs, CLAUDE.md status, release v0.46.0 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit RUNBOOK 8c carries the membership-vs-path measurement — BULK's closure holds 18 blocking dams and none are below its outlet — precisely so nobody "simplifies" the guard back to a membership test later. It also records the known bound: the guard inherits frs_wsg_drainage's one-outlet-per-group model, so a WSG draining by two independent paths would be under-covered. RUNBOOK 6b gains the guard(...) note strings so lnk_log_read() output reads cleanly, with the cross-check against wsg_upstream spelled out. Follow-up #244 filed: cabd_additions dams carry barrier_ind = t but the usa CTE hardcodes psc NULL and the barrier_status CASE has no NULL arm, so the US placeholders can never become barriers. Left as a decision rather than fixed here — options (1) and (2) change modelled access in the Columbia and need a parity check; the behaviour is pinned by a test either way. Suite 1513 pass, 0 failures. Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- CLAUDE.md | 6 ++++ DESCRIPTION | 2 +- NEWS.md | 12 ++++++++ RUNBOOK.md | 56 ++++++++++++++++++++++++++++++++++++ planning/active/task_plan.md | 12 ++++---- 5 files changed, 81 insertions(+), 7 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 76ee0d54..7d9bf050 100644 --- a/CLAUDE.md +++ b/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 `.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 `.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. diff --git a/DESCRIPTION b/DESCRIPTION index ed94c34c..c868b2c2 100644 --- a/DESCRIPTION +++ b/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", , "airvine@newgraphenvironment.com", diff --git a/NEWS.md b/NEWS.md index 5189730d..9ec001f7 100644 --- a/NEWS.md +++ b/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 `.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* — `fresh@v0.33.0` 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. diff --git a/RUNBOOK.md b/RUNBOOK.md index 35b8a1dd..64a1d831 100644 --- a/RUNBOOK.md +++ b/RUNBOOK.md @@ -466,6 +466,11 @@ EXCEPT SELECT watershed_group_code FROM .log; do not (they key on `config_hash`), so **they do not yet travel between hosts** — follow-up PR. +**Guard notes (link#227).** `notes` may carry a downstream-guard record: +`link#227 guard(override): 3 unmodelled downstream dam(s) — PCEA(1), UPCE(2) — ` +or `guard(warn): … at open`. Cross-check against the same row's `wsg_upstream`, +which independently records what was persisted when the run opened. See §8c. + **Env vars:** `LNK_RUN_LABEL` (groups a campaign), `LINK_GIT_DIRTY` / `FRESH_GIT_DIRTY` (dirty-tree flag for installed packages), `FWAPG_GIT_SHA` / `FWAPG_DIR`, `LNK_HOST_ALIAS` (host name in provenance rows). @@ -554,6 +559,57 @@ database it is a leftover from before fresh 0.33.0 — it still answers queries, it answers them wrongly. Outlets now ship in fresh at `inst/extdata/wsg_outlet.csv` and reach the DB as a `VALUES` list; no table is needed anywhere. +## 8c. Downstream state: the guard, and why membership ≠ path + +`lnk_pipeline_run()` computes accessibility from the **already-persisted** +barriers of the WSGs downstream. Model a WSG before them and the access query +finds no downstream dams, marks dammed-off segments accessible, and **exits 0** — +a wrong answer indistinguishable from a right one. `lnk_wsg_downstream_check()` +(link#227) verifies that precondition instead of trusting it. + +**The predicate is PATH, not membership.** The question is not "does a downstream +watershed group *contain* a blocking dam" but "is there a blocking dam **on this +WSG's downstream flow path**". Measured live: + +| focal | membership | path | reality | +|---|---|---|---| +| **BULK** | fires — 18 blocking dams across LSKE/KISP/KLUM | **0** | none are below BULK's outlet | +| **PARS** | fires | **3** | Peace Canyon, Site C, W.A.C. Bennett — correct | +| SLOC | fires | 1 | Brilliant Dam — correct | + +A membership guard cries wolf on BULK, operators learn to reach for the override, +and the guard stops meaning anything. **Do not "simplify" it back.** The path form +is complete, not 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 the outlet. ~0.5 s. + +**What counts as blocking** — three filters that live downstream of +`.lnk_pipeline_prep_dams`, all mirrored by the guard: +`passability_status_code IN (1,2)`; a real `linear_feature_id` join; and +`blue_line_key = watershed_key` (mainstem only). The psc filter is why the +`cabd_additions` US placeholders never appear — they carry NULL. + +**Persistence is checked per DAM, not per WSG** — `.lnk_wsg_persisted()` cannot +tell a group persisted with `dams = FALSE`, which would pass a schema holding the +streams but not the barriers. + +**Modes.** `LNK_GUARD_DOWNSTREAM=error` (default) | `warn` | `ignore`. +`study_area_run.sh` exports **`warn` on both legs** (local subshell *and* inside +the ssh string) because on a multi-host run a downstream group is legitimately +mid-flight on another cypher. That is a deferral, not a hole: +`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 WSG that was never modelled. + +**Override** requires a written justification (`LNK_GUARD_DOWNSTREAM_NOTE`); a +bare `TRUE` is rejected. The note lands in `.log.notes` beside +`wsg_upstream`, so `lnk_log_read()` reports that the network was built on a +stated assumption. + +**Known bound:** inherits `frs_wsg_drainage()`'s one-outlet-per-group model. A +WSG draining by two independent paths would be under-covered. + --- ## 8. Fast verification recipes diff --git a/planning/active/task_plan.md b/planning/active/task_plan.md index b434d310..d9b0b88f 100644 --- a/planning/active/task_plan.md +++ b/planning/active/task_plan.md @@ -87,16 +87,16 @@ Two further corrections to the issue body: ## Phase 5 — Docs, release, follow-up -- [ ] `RUNBOOK.md` §8c — membership ≠ path, with the BULK counterexample; note the +- [x] `RUNBOOK.md` §8c — membership ≠ path, with the BULK counterexample; note the single-outlet-per-group bound -- [ ] `RUNBOOK.md` §6b — `guard(...)` note strings -- [ ] `NEWS.md` + `DESCRIPTION` bump (final commit); `CLAUDE.md` status -- [ ] File follow-up: `cabd_additions` psc NULL ⇒ US placeholder dams structurally never +- [x] `RUNBOOK.md` §6b — `guard(...)` note strings +- [x] `NEWS.md` + `DESCRIPTION` bump (final commit); `CLAUDE.md` status +- [x] File follow-up: `cabd_additions` psc NULL ⇒ US placeholder dams structurally never barriers (0 rows in `fresh.barriers`) ## Validation -- [ ] `devtools::test()` green; `lintr::lint_package()` no new lint classes +- [x] `devtools::test()` green; `lintr::lint_package()` no new lint classes - [ ] `/code-check` clean on each commit -- [ ] PWF checkboxes match landed work +- [x] PWF checkboxes match landed work - [ ] `/planning-archive` then `/gh-pr-push` From d5254651fe31a3b03991f0f52b6a174cf384ffdb Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:57:59 -0700 Subject: [PATCH 7/8] Wrap long lines in the guard message builder (#227) lintr line_length_linter; no new lint classes in the new file now. The 13 remaining indentation lints are the package-wide sprintf-SQL idiom (317 across 28 files in R/). Refs #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- R/lnk_wsg_downstream_check.R | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/R/lnk_wsg_downstream_check.R b/R/lnk_wsg_downstream_check.R index 047f2964..39d8f65b 100644 --- a/R/lnk_wsg_downstream_check.R +++ b/R/lnk_wsg_downstream_check.R @@ -444,9 +444,13 @@ lnk_wsg_downstream_check <- function(conn, aoi, cfg, loaded, paste(first, collapse = ", ")) else "", "\nFix — pick one:\n", if (length(first)) sprintf( - " 1. Model the downstream WSGs first:\n for w in %s; do Rscript data-raw/wsg_run_one.R $w ; done\n", + paste0(" 1. Model the downstream WSGs first:\n", + " for w in %s; do ", + "Rscript data-raw/wsg_run_one.R $w ; done\n"), paste(first, collapse = " ")) else "", - sprintf(" 2. Model %s now and settle access after they land:\n Rscript data-raw/wsg_recompute_one.R %s \n", aoi, aoi), + sprintf(paste0(" 2. Model %s now and settle access after they land:\n", + " Rscript data-raw/wsg_recompute_one.R %s \n"), + aoi, aoi), " 3. Override, if those dams are known passable / remediated / irrelevant.\n", " Recorded in the run log and surfaced by lnk_log_read():\n", sprintf(" LNK_GUARD_DOWNSTREAM_NOTE=\"...\" Rscript data-raw/wsg_run_one.R %s \n", aoi) From bdfb653ab4c0dbdab37bbe728e9b886b138acea7 Mon Sep 17 00:00:00 2001 From: almac2022 Date: Fri, 28 Aug 2026 10:58:37 -0700 Subject: [PATCH 8/8] Archive planning files for issue #227 Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_014kSEJoad4Bc553qU8eFiE6 --- .../README.md | 100 ++++++++++++++++++ .../findings.md | 0 .../progress.md | 0 .../task_plan.md | 0 4 files changed, 100 insertions(+) create mode 100644 planning/archive/2026-08-issue-227-downstream-state-guard/README.md rename planning/{active => archive/2026-08-issue-227-downstream-state-guard}/findings.md (100%) rename planning/{active => archive/2026-08-issue-227-downstream-state-guard}/progress.md (100%) rename planning/{active => archive/2026-08-issue-227-downstream-state-guard}/task_plan.md (100%) diff --git a/planning/archive/2026-08-issue-227-downstream-state-guard/README.md b/planning/archive/2026-08-issue-227-downstream-state-guard/README.md new file mode 100644 index 00000000..8577e5d7 --- /dev/null +++ b/planning/archive/2026-08-issue-227-downstream-state-guard/README.md @@ -0,0 +1,100 @@ +# #227 — single-WSG downstream-state guard + +## Outcome + +Shipped as v0.46.0. `data-raw/wsg_run_one.R` had stated the DS-first precondition in +its own header since #175 and **nothing enforced it**: accessibility is computed from +the *already-persisted* barriers of the WSGs downstream, so modelling out of order +writes `streams_access` / `streams_mapping_code` marking dammed-off segments accessible, +prints `done in N min`, and exits 0. New exported `lnk_wsg_downstream_check()` verifies +the precondition instead of trusting it. + +## The finding that reshaped the issue + +The issue proposed checking for **blocking dams in the downstream closure's WSGs** — a +*membership* test. Measured against the live DB, it cries wolf on the issue's own +motivating example: + +| focal | membership | path | reality | +|---|---|---|---| +| **BULK** | fires — 18 dams across LSKE/KISP/KLUM | **0** | none below BULK's outlet | +| **PARS** | fires | **3** | Peace Canyon, Site C, W.A.C. Bennett — correct | +| SLOC | fires | 1 | Brilliant Dam — correct | + +At that false-alarm rate operators reach for the override reflexively and the guard +stops meaning anything — the exact failure the issue's own "make or break" caveat +warns about, though it attributes the cause to edit-CSV fidelity rather than to the +predicate. The **path** form tests each dam with the measure-aware +`whse_basemapping.fwa_downstream()` from `fresh::frs_wsg_outlets()`. It is complete, +not 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 the outlet. ~0.5 s against a 5 s budget. + +`RUNBOOK.md` §8c carries the measurement precisely so nobody simplifies it back. + +## Three further corrections to the issue body + +- **Not `fwa_watershed_groups_poly`** for the proposed 2.4 s spatial join — its code + columns are NULL on docker fwapg (#222). `watershed_group_code` comes from the snapped + `fwa_stream_networks_sp` row, as `.lnk_pipeline_prep_dams` itself does. 8× faster too. +- **`.lnk_wsg_persisted()` is the wrong granularity** — it cannot distinguish a WSG + persisted with `dams = FALSE`, which would pass a schema holding the streams but not + the barriers. Persistence is checked per *dam* in `.barriers`. +- **"`study_area_run.sh` DS-first runs pass unchanged" is unachievable** by a hard + pre-flight: on multi-host runs downstream groups are legitimately mid-flight on another + cypher, and because per-WSG failures soft-fail the WSG would be **skipped** — which + `lnk_access(merge = TRUE)` cannot repair. Strictly worse than the bug. Resolved with + `warn` mode on both legs plus a post-condition in `wsg_recompute_one.R`, which is what + makes the deferral honest rather than a hole. + +## Anti-drift by construction + +The guard must apply link's exact dam filters or it flags dams the pipeline treats as +passable. Rather than a second copy of the 25-line lateral snap with a comment asking +future readers to keep them in sync, the `cabd` and `matched` CTE bodies now live once +(`.lnk_dams_cabd_sql()` / `.lnk_dams_matched_sql()`) and are consumed by both +`.lnk_pipeline_prep_dams()` and the probe, parameterized on source — staged tables for +the pipeline, inline `(VALUES …)` for the guard, the pattern fresh 0.33.0 used to retire +`public.wsg_outlet`. + +A golden capture was taken **before** the refactor and the result diffed against it: +byte-identical for ADMS (8 dams), KOTL (41) and PARS (0). Worth knowing: PARS has **0 +dams in-WSG** — all three of its blocking dams are downstream in UPCE/PCEA, which is +exactly why the pre-flight exists. + +## Defect found and fixed in passing + +`.lnk_log_create_tables()` (shipped v0.45.0) 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. Surfaced by running the guard's smoke tests into scratch schemas. + +## Verified live, end to end + +| case | result | +|---|---| +| typo'd `LNK_GUARD_DOWNSTREAM` | errors, names the valid values (a typo must not silently disable) | +| PARS, empty persist schema | **exit 1**, names all three dams + DS-first order | +| ADMS, brand-new schema | guard passes, models in 2.1 min, log row lands | +| override | proceeds; note reads `guard(override): 3 unmodelled downstream dam(s) — PCEA(1), UPCE(2) — Site C fishway operational, ref doc-123` | +| warn | proceeds; note reads `guard(warn): … at open` | + +Suite 1513 pass, 0 failures. No new lint classes. + +## Follow-up + +**#244** — `cabd_additions` dams carry `barrier_ind = t` but the `usa` CTE hardcodes +`passability_status_code NULL` and the `barrier_status` CASE has no NULL arm, so Grand +Coulee / Chief Joseph can never become barriers (verified: 0 rows in `fresh.barriers`). +Left as a decision rather than fixed here — two of the three options change modelled +access in the Columbia and need a parity check. Pinned by a test either way, so any fix +is deliberate. + +## Known bound + +Inherits `frs_wsg_drainage()`'s one-outlet-per-group model. A WSG draining by two +independent paths would be under-covered. Recorded in RUNBOOK §8c. + +Closed by: commits `3c593ab` → `225996f` (v0.46.0) on +`227-single-wsg-downstream-state-guard`. diff --git a/planning/active/findings.md b/planning/archive/2026-08-issue-227-downstream-state-guard/findings.md similarity index 100% rename from planning/active/findings.md rename to planning/archive/2026-08-issue-227-downstream-state-guard/findings.md diff --git a/planning/active/progress.md b/planning/archive/2026-08-issue-227-downstream-state-guard/progress.md similarity index 100% rename from planning/active/progress.md rename to planning/archive/2026-08-issue-227-downstream-state-guard/progress.md diff --git a/planning/active/task_plan.md b/planning/archive/2026-08-issue-227-downstream-state-guard/task_plan.md similarity index 100% rename from planning/active/task_plan.md rename to planning/archive/2026-08-issue-227-downstream-state-guard/task_plan.md