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/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/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/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/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..39d8f65b --- /dev/null +++ b/R/lnk_wsg_downstream_check.R @@ -0,0 +1,458 @@ +# 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("(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) { + 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 = ", "), ")") + # 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 = ", ")) + } + + 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)) + ) +} + + +# --------------------------------------------------------------------------- +# 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( + 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(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) + ) +} 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/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/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/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/archive/2026-08-issue-227-downstream-state-guard/findings.md b/planning/archive/2026-08-issue-227-downstream-state-guard/findings.md new file mode 100644 index 00000000..312d89c1 --- /dev/null +++ b/planning/archive/2026-08-issue-227-downstream-state-guard/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/archive/2026-08-issue-227-downstream-state-guard/progress.md b/planning/archive/2026-08-issue-227-downstream-state-guard/progress.md new file mode 100644 index 00000000..64223319 --- /dev/null +++ b/planning/archive/2026-08-issue-227-downstream-state-guard/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/archive/2026-08-issue-227-downstream-state-guard/task_plan.md b/planning/archive/2026-08-issue-227-downstream-state-guard/task_plan.md new file mode 100644 index 00000000..d9b0b88f --- /dev/null +++ b/planning/archive/2026-08-issue-227-downstream-state-guard/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) + +- [x] Golden test for current `.lnk_pipeline_prep_dams` (row count + `dam_id` set + psc + distribution for one dam-bearing WSG) +- [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) + +- [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 +- [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 + +- [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 +- [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 +- [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 + +- [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` +- [x] `study_area_run.sh`: export `LNK_GUARD_DOWNSTREAM=warn` on the dispatcher leg **and + inside the ssh string** +- [x] Update header contract + the L272-290 comment block + +## Phase 4 — Post-condition + +- [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 + +- [x] `RUNBOOK.md` §8c — membership ≠ path, with the BULK counterexample; note the + single-outlet-per-group bound +- [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 + +- [x] `devtools::test()` green; `lintr::lint_package()` no new lint classes +- [ ] `/code-check` clean on each commit +- [x] PWF checkboxes match landed work +- [ ] `/planning-archive` then `/gh-pr-push` diff --git a/tests/testthat/test-lnk_pipeline_prepare.R b/tests/testthat/test-lnk_pipeline_prepare.R index 1d263b0b..e2103e8c 100644 --- a/tests/testthat/test-lnk_pipeline_prepare.R +++ b/tests/testthat/test-lnk_pipeline_prepare.R @@ -872,3 +872,153 @@ 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) +}) + + +# --- 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) + } + ) +}) 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)) +})