#221 + #223: fix access-segmentation over-credit + accessible_km parity roll-up

NAMESPACE

@@ -42,6 +42,7 @@ export(lnk_pipeline_setup)
 export(lnk_pipeline_species)
 export(lnk_points_snap)
 export(lnk_presence)
+export(lnk_rollup_wsg)
 export(lnk_rules_build)
 export(lnk_score)
 export(lnk_source)

R/lnk_compare_rollup.R

@@ -41,10 +41,13 @@
 #'   (e.g. `c("BT","CO")`) to restrict the rollup to. Default `NULL`
 #'   discovers the set from PG.
 #'
-#' @return A tibble with one row per (species, habitat_type) — 7
-#'   habitat types per species. Columns: `wsg`, `species`,
+#' @return A tibble with one row per (species, habitat_type) — 8
+#'   habitat types per species (the 7 habitat km/ha types plus
+#'   `accessible` km, link#221). Columns: `wsg`, `species`,
 #'   `habitat_type`, `unit` (`km` | `ha`), `link_value`, `ref_value`,
-#'   `diff_pct`.
+#'   `diff_pct`. `accessible`'s `ref_value` is sourced tunnel-free from
+#'   `fresh.streams_vw_bcfp` for the salmon group (CH/CM/CO/PK/SK); other
+#'   species carry `NA` until their reference path lands (link#221 Phase 3).
 #'
 #' @examples
 #' \dontrun{
@@ -120,7 +123,7 @@ lnk_compare_rollup <- function(conn, aoi, cfg,
     conn = conn, cfg = cfg, aoi = aoi, species = species)
   rollup_ref <- .lnk_compare_wsg_rollup_reference( # nolint
     reference = reference, conn_ref = conn_ref,
-    aoi = aoi, species = species)
+    aoi = aoi, species = species, conn = conn)
 
   .lnk_compare_wsg_assemble_rollup( # nolint
     aoi = aoi, species = species,
@@ -195,35 +198,43 @@ lnk_compare_rollup <- function(conn, aoi, cfg,
   et_lake_sql    <- "(1500, 1525)"
   et_wetland_sql <- "(1700)"
 
-  union_streams <- paste(vapply(species, function(sp) {
-    sp_lit <- DBI::dbQuoteLiteral(conn, sp)
-    sprintf(
-      "SELECT %s AS species_code, s.id_segment, s.length_metre,
-              s.edge_type, h.spawning, h.rearing
-         FROM %s.streams s
-         JOIN %s.streams_habitat_%s h ON s.id_segment = h.id_segment AND s.watershed_group_code = h.watershed_group_code
-        WHERE s.watershed_group_code = %s",
-      sp_lit, tn$schema, tn$schema, tolower(sp), aoi_lit)  # nolint: indentation_linter
-  }, character(1)), collapse = "\n        UNION ALL\n        ")
+  # Habitat km sums delegate to the single predicate-driven roll-up
+  # primitive (link#221) so there is one per-(WSG, species) km query
+  # builder. lnk_rollup_wsg exposes `length_metre` / `edge_type` /
+  # `access` / `spawning` / `rearing` under generic aliases; the first
+  # five metrics reproduce the historical shape, and `accessible_km`
+  # (link#221) sums link's per-species access model
+  # (`streams_access.access_<sp> IN (1,2)`, LEFT-joined by
+  # lnk_rollup_wsg). It returns `wsg` + `species` + metrics — drop
+  # `wsg` and rename `species` -> `species_code` to preserve this
+  # helper's `list(km, lake_ha, wetland_ha)` contract.
+  # COALESCE(..., 0): an empty FILTER set sums to NULL, but the
+  # historical CASE-WHEN-ELSE-0 form returned a measured 0 for a
+  # species present in the WSG with no segments in that slice. Preserve
+  # that 0 (not NA) so downstream parity diff_pct is unchanged.
+  km_metrics <- c(
+    spawning_km =
+      "round(COALESCE(sum(length_metre) FILTER (WHERE spawning), 0)::numeric / 1000, 2)", # nolint: line_length_linter
+    rearing_km =
+      "round(COALESCE(sum(length_metre) FILTER (WHERE rearing), 0)::numeric / 1000, 2)", # nolint: line_length_linter
+    rearing_stream_km = sprintf(
+      "round(COALESCE(sum(length_metre) FILTER (WHERE rearing AND edge_type IN %s), 0)::numeric / 1000, 2)", # nolint: line_length_linter
+      et_stream_sql),
+    rearing_lake_centerline_km = sprintf(
+      "round(COALESCE(sum(length_metre) FILTER (WHERE rearing AND edge_type IN %s), 0)::numeric / 1000, 2)", # nolint: line_length_linter
+      et_lake_sql),
+    rearing_wetland_centerline_km = sprintf(
+      "round(COALESCE(sum(length_metre) FILTER (WHERE rearing AND edge_type IN %s), 0)::numeric / 1000, 2)", # nolint: line_length_linter
+      et_wetland_sql),
+    accessible_km =
+      "round(COALESCE(sum(length_metre) FILTER (WHERE access IN (1, 2)), 0)::numeric / 1000, 2)") # nolint: line_length_linter
 
-  km <- DBI::dbGetQuery(conn, sprintf("
-    SELECT species_code,
-      round(SUM(CASE WHEN spawning THEN length_metre ELSE 0 END)::numeric
-        / 1000, 2) AS spawning_km,
-      round(SUM(CASE WHEN rearing  THEN length_metre ELSE 0 END)::numeric
-        / 1000, 2) AS rearing_km,
-      round(SUM(CASE WHEN rearing AND edge_type IN %s
-                     THEN length_metre ELSE 0 END)::numeric / 1000, 2)
-        AS rearing_stream_km,
-      round(SUM(CASE WHEN rearing AND edge_type IN %s
-                     THEN length_metre ELSE 0 END)::numeric / 1000, 2)
-        AS rearing_lake_centerline_km,
-      round(SUM(CASE WHEN rearing AND edge_type IN %s
-                     THEN length_metre ELSE 0 END)::numeric / 1000, 2)
-        AS rearing_wetland_centerline_km
-    FROM (%s) per_species
-    GROUP BY species_code ORDER BY species_code",
-    et_stream_sql, et_lake_sql, et_wetland_sql, union_streams))  # nolint: indentation_linter
+  km <- lnk_rollup_wsg( # nolint: object_usage_linter
+    conn = conn, aoi = aoi, species = species,
+    schema = tn$schema, metrics = km_metrics)
+  km$wsg <- NULL
+  names(km)[names(km) == "species"] <- "species_code"
+  km <- km[, c("species_code", names(km_metrics))]
 
   # Lake / wetland ha — DISTINCT waterbody_key joins to fwa polygon
   # tables avoid double-counting multi-segment lakes/wetlands.

R/lnk_compare_wsg.R

@@ -51,12 +51,13 @@
 #'   deprecation warning when supplied.
 #'
 #' @return A list with two elements:
-#'   - `rollup`: tibble with one row per (species, habitat_type) — 7
+#'   - `rollup`: tibble with one row per (species, habitat_type) — 8
 #'     habitat types: `spawning`, `rearing`, `lake_rearing`,
 #'     `wetland_rearing`, `rearing_stream`, `rearing_lake_centerline`,
-#'     `rearing_wetland_centerline`. Columns: `wsg`, `species`,
-#'     `habitat_type`, `unit` (`km` | `ha`), `link_value`,
-#'     `ref_value`, `diff_pct`.
+#'     `rearing_wetland_centerline`, `accessible` (km, link#221).
+#'     Columns: `wsg`, `species`, `habitat_type`, `unit` (`km` | `ha`),
+#'     `link_value`, `ref_value`, `diff_pct`. `accessible`'s `ref_value`
+#'     is `NA` until the tunnel-free reference path lands.
 #'   - `mapping_code`: tibble with one row per species — segment-level
 #'     match stats vs `bcfishpass.streams_mapping_code`. Columns:
 #'     `wsg`, `species`, `total_segs`, `match_pct`, `n_diffs`,
@@ -311,21 +312,63 @@ lnk_compare_wsg <- function(conn, aoi, cfg, loaded,
 #' Compute reference-side rollup queries for the requested `reference`
 #'
 #' Dispatches on `reference`. Currently only `"bcfishpass"` is wired —
-#' queries `bcfishpass.habitat_linear_<sp>` per species, joined to the
-#' same `fwa_lakes_poly` / `fwa_wetlands_poly` for the ha columns
-#' bcfp doesn't materialize natively.
+#' queries `bcfishpass.habitat_linear_<sp>` per species (on `conn_ref`,
+#' the tunnel), joined to the same `fwa_lakes_poly` / `fwa_wetlands_poly`
+#' for the ha columns bcfp doesn't materialize natively. The `accessible_km`
+#' reference is sourced separately + tunnel-free from `fresh.streams_vw_bcfp`
+#' on the local `conn` (see [.lnk_compare_wsg_accessible_ref()]); the two
+#' reference sources stay intentionally decoupled (link#221).
 #'
 #' @noRd
 .lnk_compare_wsg_rollup_reference <- function(reference, conn_ref,
-                                              aoi, species) {
+                                              aoi, species, conn) {
   if (reference == "bcfishpass") {
-    return(.lnk_compare_wsg_rollup_bcfishpass(
-      conn_ref = conn_ref, aoi = aoi, species = species))
+    ref <- .lnk_compare_wsg_rollup_bcfishpass(
+      conn_ref = conn_ref, aoi = aoi, species = species) # nolint: indentation_linter
+    ref$accessible_km <- .lnk_compare_wsg_accessible_ref(
+      conn = conn, aoi = aoi, species = ref$species_code) # nolint: indentation_linter
+    return(ref)
   }
   stop("Unknown reference '", reference, "' in dispatch.", call. = FALSE)
 }
 
 
+#' Tunnel-free `accessible_km` reference from `fresh.streams_vw_bcfp`
+#'
+#' Sums `length_metre` where the species' bcfp barrier-group column is
+#' empty (`= ''` = no barrier downstream = accessible). Sourced from the
+#' local province-wide snapshot (`conn`), NOT the tunnel — the `:63333`
+#' tunnel is dead on M1 and the snapshot is authoritative. Only the
+#' Phase-1-proven salmon group (CH/CM/CO/PK/SK →
+#' `barriers_ch_cm_co_pk_sk_dnstr`) is wired; other species short-circuit
+#' to `NA` (their reference lands in a later phase, each needing its own
+#' proof). `barrier_group` values are a trusted constant whitelist — safe
+#' to interpolate.
+#'
+#' @noRd
+.lnk_compare_wsg_accessible_ref <- function(conn, aoi, species) {
+  barrier_group <- c(
+    CH = "barriers_ch_cm_co_pk_sk_dnstr",
+    CM = "barriers_ch_cm_co_pk_sk_dnstr",
+    CO = "barriers_ch_cm_co_pk_sk_dnstr",
+    PK = "barriers_ch_cm_co_pk_sk_dnstr",
+    SK = "barriers_ch_cm_co_pk_sk_dnstr"
+  )
+  aoi_lit <- DBI::dbQuoteLiteral(conn, aoi)
+  vapply(species, function(sp) {
+    grp <- barrier_group[toupper(sp)]
+    if (is.na(grp)) return(NA_real_)
+    res <- DBI::dbGetQuery(conn, sprintf("
+      SELECT round((COALESCE(sum(length_metre), 0) / 1000)::numeric, 2)
+        AS accessible_km
+      FROM fresh.streams_vw_bcfp
+      WHERE watershed_group_code = %s AND %s = ''",
+      aoi_lit, unname(grp)))  # nolint: indentation_linter
+    res$accessible_km[1]
+  }, numeric(1), USE.NAMES = FALSE)
+}
+
+
 #' Reference dispatch: bcfishpass tunnel
 #'
 #' Mirrors the link-side methodology applied to `bcfishpass.habitat_linear_<sp>`
@@ -434,33 +477,37 @@ lnk_compare_wsg <- function(conn, aoi, cfg, loaded,
 
 #' Assemble long-format output tibble from link + ref rollup data
 #'
-#' Produces 7 rows per species (spawning, rearing, lake_rearing,
+#' Produces 8 rows per species (spawning, rearing, lake_rearing,
 #' wetland_rearing, rearing_stream, rearing_lake_centerline,
-#' rearing_wetland_centerline). `diff_pct = NA` when `ref_value` is
-#' `NA` (species not modelled by reference) or `0` (avoid div-by-zero
-#' even when the measurement is real).
+#' rearing_wetland_centerline, accessible). `diff_pct = NA` when
+#' `ref_value` is `NA` (species not modelled by reference, or — for
+#' `accessible` — the tunnel-free reference path not yet wired) or `0`
+#' (avoid div-by-zero even when the measurement is real).
 #'
 #' @noRd
 .lnk_compare_wsg_assemble_rollup <- function(aoi, species,
                                              rollup_link, rollup_ref) {
   habitat_types <- c(
     "spawning", "rearing", "lake_rearing", "wetland_rearing",
-    "rearing_stream", "rearing_lake_centerline", "rearing_wetland_centerline"
+    "rearing_stream", "rearing_lake_centerline", "rearing_wetland_centerline",
+    "accessible"
   )
   units <- c(
     spawning = "km", rearing = "km",
     lake_rearing = "ha", wetland_rearing = "ha",
     rearing_stream = "km",
     rearing_lake_centerline = "km",
-    rearing_wetland_centerline = "km"
+    rearing_wetland_centerline = "km",
+    accessible = "km"
   )
   col_suffix <- c(
     spawning = "spawning_km", rearing = "rearing_km",
     lake_rearing = "lake_rearing_ha",
     wetland_rearing = "wetland_rearing_ha",
     rearing_stream = "rearing_stream_km",
     rearing_lake_centerline = "rearing_lake_centerline_km",
-    rearing_wetland_centerline = "rearing_wetland_centerline_km"
+    rearing_wetland_centerline = "rearing_wetland_centerline_km",
+    accessible = "accessible_km"
   )
 
   sp_col <- rep(species, each = length(habitat_types))
@@ -483,7 +530,8 @@ lnk_compare_wsg <- function(conn, aoi, cfg, loaded,
     wetland_rearing            = rollup_link$wetland_ha,
     rearing_stream             = rollup_link$km,
     rearing_lake_centerline    = rollup_link$km,
-    rearing_wetland_centerline = rollup_link$km
+    rearing_wetland_centerline = rollup_link$km,
+    accessible                 = rollup_link$km
   )
 
   for (i in seq_len(nrow(out))) {

R/lnk_pipeline_break.R

@@ -21,7 +21,7 @@
 #' | name | source table | role | classify label |
 #' |---|---|---|---|
 #' | `observations` | `<schema>.observations_breaks` | fish observations from `bcfishobs.observations`, WSG- and species-filtered, exclusions applied | (informational; not a barrier) |
-#' | `gradient_minimal` | `<schema>.gradient_barriers_minimal` | minimal-reduced gradient barriers (per-model 15/20/25/30%) | classify uses the FULL set with `gradient_<NNNN>` labels |
+#' | `gradient_minimal` | `<schema>.gradient_barriers_minimal` | FULL per-model gradient barriers (15/20/25/30%) + falls — every position breaks the network so access gates at each frontier (#223); NOT minimal-reduced despite the legacy name | classify uses the same FULL set with `gradient_<NNNN>` labels |
 #' | `falls` | `<schema>.falls` | natural waterfalls from `whse_basemapping.fwa_obstacles_sp` (loaded by `prep_load_aux`); each fall is its own barrier (NOT minimal-reduced) | `blocked` |
 #' | `barriers_definite` | `<schema>.barriers_definite` | `user_barriers_definite.csv` for the AOI | `blocked` |
 #' | `subsurfaceflow` | `<schema>.barriers_subsurfaceflow` | FWA `edge_type IN (1410, 1425)` start points; honours `user_barriers_definite_control`. Opt-in (only built when listed) | `blocked` |

R/lnk_pipeline_prepare.R

@@ -15,9 +15,10 @@
 #'   `lnk_barrier_overrides()` to compute the per-species skip list.
 #'   User-definite barriers are intentionally excluded here and
 #'   consumed by later phases directly — bcfishpass parity.
-#' - Per-model barrier tables reduced to the minimal downstream-most
-#'   set via [fresh::frs_barriers_minimal()], then unioned into
-#'   `gradient_barriers_minimal` for segmentation
+#' - Per-model barrier tables (gradient class-filtered + falls) unioned
+#'   into `gradient_barriers_minimal` — the FULL per-model position set
+#'   (NOT minimal-reduced, despite the legacy name), so streams break at
+#'   every frontier and per-segment access gating is correct (#223)
 #' - Base stream segments (`fresh.streams`) loaded from FWA with
 #'   channel width, stream order parent, GENERATED gradient / measures
 #'   / length columns, and a unique `id_segment`
@@ -31,9 +32,8 @@
 #'   - `<schema>.barriers_subsurfaceflow` (only when subsurfaceflow opted in
 #'     via `cfg$pipeline$break_order`)
 #'   - `<schema>.barrier_overrides`
-#'   - `<schema>.barriers_<model>` + `<schema>.barriers_<model>_min`
-#'     per-model pre/post minimal reduction
-#'   - `<schema>.gradient_barriers_minimal` (union of minimal positions)
+#'   - `<schema>.barriers_<model>` — per-model gradient + falls positions
+#'   - `<schema>.gradient_barriers_minimal` (union of per-model raw positions)
 #'   - `fresh.streams` (base segments — not namespaced by AOI; fresh
 #'     owns its output schema)
 #'   - `<schema>.dams` (only when `conn_tunnel` is supplied) — pulled
@@ -567,9 +567,8 @@ lnk_pipeline_prepare <- function(conn, aoi, cfg, loaded, schema,
   for (model_name in names(models)) {
     class_filter <- paste(models[[model_name]], collapse = ", ")
     model_tbl <- paste0(schema, ".barriers_", model_name)
-    min_tbl <- paste0(model_tbl, "_min")
 
-    # Build pre-minimal set: gradient (class-filtered) + falls
+    # Build the per-model break set: gradient (class-filtered) + falls.
     .lnk_db_execute(conn, sprintf("DROP TABLE IF EXISTS %s", model_tbl))
     .lnk_db_execute(conn, sprintf(
       "CREATE TABLE %s AS
@@ -589,8 +588,17 @@ lnk_pipeline_prepare <- function(conn, aoi, cfg, loaded, schema,
       model_tbl, schema, class_filter,
       schema, .lnk_quote_literal(aoi)))
 
-    fresh::frs_barriers_minimal(conn, from = model_tbl, to = min_tbl)
-    minimal_tbls <- c(minimal_tbls, min_tbl)
+    # Feed the RAW per-model positions into the break network — every gradient /
+    # falls barrier must split the streams so the per-segment downstream-barrier
+    # access check (lnk_pipeline_access) can gate the reach ABOVE each frontier.
+    # Do NOT frs_barriers_minimal() here: minimal reduction keeps only the
+    # downstream-most position per flow path (correct for an access DECISION,
+    # wrong as a SEGMENTATION source — it strips the interior frontier so a single
+    # segment straddles the barrier and its whole reach is mislabelled accessible;
+    # #223). This mirrors the orphan-set treatment below and matches bcfishpass,
+    # which segments at every gradient barrier. Table name kept as
+    # gradient_barriers_minimal for now; a follow-up issue renames it.
+    minimal_tbls <- c(minimal_tbls, model_tbl)
   }
 
   # Orphan classes: thresholds in `classes` that fall below every species's