Thread per-WSG cw/mad habitat model through classify (#286)

CLAUDE.md

@@ -11,6 +11,29 @@ Experimental package — breaking all the time and loving the learning curve. St
 **Prefix:** `lnk_`
 **Branch:** `main` (current version: `DESCRIPTION` / [`NEWS.md`](NEWS.md))
 
+## Status (2026-10-01) — per-WSG `cw`/`mad` habitat model threaded (#286)
+
+**Each bundle's `parameters_habitat_method.csv` reaches fresh as `params_method`.**
+- **The table:** all `cw`, a frozen copy of bcfishpass `example_newgraph`. Putting a
+  group on `mad` is a reviewed row edit.
+- **`mad_m3s`:** on the working streams only, never persisted.
+- **fresh pin:** v0.36.2 (floor 0.35.0). Cyphers must be re-prepped before the next
+  dispatch: the preflight asserts the argument and hard-fails otherwise.
+- **Mechanics:** RUNBOOK §7 "Channel width or discharge, per watershed group". Evidence:
+  `data-raw/logs/params_method_286/`.
+
+**Facts not worth re-deriving:**
+- **Never give a bundle file `source: https://github.com/smnorris/bcfishpass` unless you
+  want it csv-synced.** `sync_bcfishpass_csvs.R` selects on that exact string and
+  auto-merges byte drift. A frozen copy uses another `source` plus `derived_from`.
+- **A `mad` group with no discharge loses all stream habitat silently.** BULK has none.
+  Waterbody rules (L/W, `thresholds: false`) inherit nothing under either model, so BT
+  keeps wetland rearing under `mad`.
+- **fresh 0.36.0 reversed `frs_db_conn()`'s precedence** (`PG*` first). `lnk_db_conn()`
+  still reads `PG_*_SHARE` first; on a machine with both groups set they connect to
+  different databases.
+- **`lnk_habitat_validate()` is cw-only**; a follow-up is drafted in the #286 archive.
+
 ## Status (2026-09-29) — #284 step 5: BT `rear_gradient_max` 0.1349 scored and held
 
 **Threshold variants are scored on one shared segmentation, never on two full runs.**
@@ -76,7 +99,8 @@ none. The `bcfishpass` copy is a frozen parity input. Runs log the values in
 `<persist>.log_parameters_habitat_thresholds`. `default_tuned` (thin,
 `extends: default`) is where #284's calibrated CH/BT values land. RUNBOOK §7
 "Where habitat thresholds live" has the details, including which columns are
-carried but never applied on link's rules path (MAD, edge types) and that
+carried but never applied on link's rules path (edge types; MAD only in groups a
+bundle's `parameters_habitat_method.csv` puts on `mad`, #286) and that
 `rear_lake_ha_min` needs a rules rebuild.
 
 **`extends:` was broken for provenance until a bundle actually used it.**
@@ -373,7 +397,7 @@ link is connectivity-system agnostic. Column names are configurable parameters w
 
 ## Database Connection
 
-Uses `PG_*_SHARE` env vars (Docker fwapg, same as `frs_db_conn()`) with fallback to standard `PG*` vars. DB is needed for match/score/habitat functions that operate via SQL. The override loading and validation can work with any PostgreSQL.
+Uses `PG_*_SHARE` env vars (Docker fwapg) with fallback to standard `PG*` vars. From fresh 0.36.0 `frs_db_conn()` reads them the other way round (`PG*` first), so on a machine that sets both the two connect to different databases (#286). DB is needed for match/score/habitat functions that operate via SQL. The override loading and validation can work with any PostgreSQL.
 
 ```r
 conn <- lnk_db_conn()  # reads PG_DB_SHARE, PG_HOST_SHARE, etc.

DESCRIPTION

@@ -21,7 +21,7 @@ Imports:
     crate (>= 0.0.2),
     DBI,
     digest,
-    fresh (>= 0.33.0),
+    fresh (>= 0.35.0),
     httr,
     jsonlite,
     RPostgres,
@@ -32,7 +32,7 @@ Imports:
     yaml
 Remotes:
     NewGraphEnvironment/crate,
-    NewGraphEnvironment/fresh@v0.33.0,
+    NewGraphEnvironment/fresh@v0.36.2,
     NewGraphEnvironment/gq
 Suggests:
     bcdata,

R/lnk_config.R

@@ -338,6 +338,24 @@ print.lnk_config <- function(x, ...) {
   fresh_path
 }
 
+# Path to the per-watershed-group habitat model table (`cw` or `mad`) a
+# config runs with (#286). Same contract as the thresholds CSV above: the
+# bundle's own `files: parameters_habitat_method:`, else fresh's shipped copy
+# (all `cw`), with a message because that copy sits outside the config's
+# provenance.
+.lnk_habitat_method_csv <- function(cfg) {
+  path <- cfg$files$parameters_habitat_method$path
+  if (!is.null(path)) {
+    return(path)
+  }
+  fresh_path <- system.file("extdata", "parameters_habitat_method.csv",
+                            package = "fresh")
+  message("config '", cfg$name %||% "<unnamed>", "' declares no ",
+          "files$parameters_habitat_method; using fresh's copy: ",
+          fresh_path)
+  fresh_path
+}
+
 # Absolute path of a provenance entry. Keys are relative to the bundle that
 # declared them: the leaf for its own entries, `.dir` for inherited ones.
 .lnk_provenance_path <- function(cfg, rel) {

R/lnk_db_conn.R

@@ -14,10 +14,11 @@
 #' @return A [DBI::DBIConnection-class] object.
 #'
 #' @details
-#' Checks `PG_*_SHARE` first (the Docker fwapg convention shared with
-#' [fresh::frs_db_conn()]), then standard PostgreSQL variables (`PGHOST`,
-#' etc.). This means `lnk_db_conn()` works identically to `frs_db_conn()`
-#' when both packages connect to the same database.
+#' Checks `PG_*_SHARE` first, then standard PostgreSQL variables
+#' (`PGHOST`, etc.). This is the reverse of [fresh::frs_db_conn()] from
+#' fresh 0.36.0, which reads `PG*` first and `PG_*_SHARE` only when none
+#' of `PG*` is set. On a machine that sets both groups to different
+#' targets, the two functions connect to different databases.
 #'
 #' @examples
 #' \dontrun{

R/lnk_habitat_validate.R

@@ -818,8 +818,9 @@ lnk_habitat_validate <- function(conn, aoi, cfg, loaded, species, schema,
   res <- lapply(species, function(sp) {
     if (!any(seg$species_code == sp)) return(NULL)
     spp <- .lnk_hv_sp_params(params, loaded$parameters_fresh, sp)
-    # Channel-width model: the default, and the only one fresh@v0.33.0 (the
-    # pinned minimum) has; it takes no `model` argument.
+    # Channel-width model only. fresh >= 0.35.0 takes `model = "mad"`, but the
+    # miss-reason relaxation below rewrites s.channel_width, so a bundle that
+    # puts a group on mad is scored here as if it were cw (#286 follow-up).
     pr <- fresh::frs_habitat_predicates(spp)
     mins <- .lnk_hv_stage_min(spp)
     stage_pred <- list(

R/lnk_log.R

@@ -73,6 +73,10 @@
   fallback <- if (is.null(cfg$files$parameters_habitat_thresholds)) {
     suppressMessages(.lnk_habitat_thresholds_csv(cfg))
   }
+  # Same for the cw/mad method table (#286).
+  fallback_method <- if (is.null(cfg$files$parameters_habitat_method)) {
+    suppressMessages(.lnk_habitat_method_csv(cfg))
+  }
 
   # Name each file relative to the bundle that holds it, never by absolute
   # path, so the hash is the same on every host. Leaf files keep their plain
@@ -122,6 +126,16 @@
     })
   }
 
+  if (length(fallback_method) == 1L && nzchar(fallback_method)) {
+    paths <- c(paths, fallback_method)
+    rel <- c(rel, "fresh:parameters_habitat_method.csv")
+    digests <- c(digests, if (file.exists(fallback_method)) {
+      digest::digest(file = fallback_method, algo = "sha256")
+    } else {
+      "MISSING"
+    })
+  }
+
   # radix: byte order, not LC_COLLATE, or the hash differs between hosts whose
   # locales sort `_control.csv` and `.csv` differently.
   ord <- order(rel, method = "radix")
@@ -189,6 +203,9 @@
     table_name = c(
       "whse_basemapping.fwa_stream_networks_sp",
       "whse_basemapping.fwa_stream_networks_channel_width",
+      # mean annual discharge: drives classification for groups a bundle's
+      # parameters_habitat_method puts on `mad` (#286)
+      "whse_basemapping.fwa_stream_networks_discharge",
       "whse_basemapping.fwa_stream_networks_order_parent",
       "whse_basemapping.fwa_lakes_poly",
       "whse_basemapping.fwa_wetlands_poly",
@@ -206,7 +223,7 @@
       "fresh.modelled_stream_crossings"
     ),
     source = c(
-      rep(fwapg, 7L),
+      rep(fwapg, 8L),
       "bcfishobs",
       "bcdata bc2pg",
       "CABD",

R/lnk_pipeline_classify.R

@@ -34,6 +34,16 @@
 #'   `NULL` uses the config's own `files$parameters_habitat_thresholds`,
 #'   falling back (with a message) to the copy shipped with fresh when
 #'   the config declares none.
+#' @param method_csv Path to the per-watershed-group habitat model table
+#'   (`watershed_group_code`, `model` = `"cw"` or `"mad"`), passed to
+#'   [fresh::frs_habitat_classify()] as `params_method`. Default `NULL` uses
+#'   the config's own `files$parameters_habitat_method`, falling back (with a
+#'   message) to fresh's all-`cw` copy when the config declares none. A group
+#'   the table does not list classifies on channel width. A `mad` group
+#'   classifies on `mad_m3s`, which the prepare phase joins onto the working
+#'   streams table, and skips the stream-order rearing bypass. Read from the
+#'   path, not from `loaded$parameters_habitat_method`, as the thresholds are,
+#'   so editing `loaded` has no effect here.
 #'
 #' @return `conn` invisibly, for pipe chaining.
 #'
@@ -58,7 +68,8 @@
 #' }
 lnk_pipeline_classify <- function(conn, aoi, cfg, loaded, schema,
                                    species = NULL,
-                                   thresholds_csv = NULL) {
+                                   thresholds_csv = NULL,
+                                   method_csv = NULL) {
   .lnk_validate_identifier(schema, "schema")
   if (!is.character(aoi) || length(aoi) != 1L || !nzchar(aoi)) {
     stop("aoi must be a single non-empty string (watershed group code)",
@@ -76,6 +87,13 @@ lnk_pipeline_classify <- function(conn, aoi, cfg, loaded, schema,
   if (!nzchar(thresholds_csv) || !file.exists(thresholds_csv)) {
     stop("thresholds_csv not found: ", thresholds_csv, call. = FALSE)
   }
+  method_csv <- method_csv %||% .lnk_habitat_method_csv(cfg)
+  if (!nzchar(method_csv) || !file.exists(method_csv)) {
+    stop("method_csv not found: ", method_csv, call. = FALSE)
+  }
+  params_method <- utils::read.csv(method_csv, stringsAsFactors = FALSE,
+                                   colClasses = "character",
+                                   na.strings = character(0))
 
   species <- species %||% lnk_pipeline_species(cfg, loaded, aoi)
   if (length(species) == 0L) {
@@ -98,6 +116,7 @@ lnk_pipeline_classify <- function(conn, aoi, cfg, loaded, schema,
     species = species,
     params = params,
     params_fresh = loaded$parameters_fresh,
+    params_method = params_method,
     gate = TRUE,
     label_block = "blocked",
     barrier_overrides = paste0(schema, ".barrier_overrides"),
@@ -142,7 +161,12 @@ lnk_pipeline_classify <- function(conn, aoi, cfg, loaded, schema,
   # `rear[].channel_width_min_bypass = list(stream_order, stream_order_parent_min)`.
   # We detect that field's presence on any rear rule and call
   # `frs_order_child` with the embedded parent_order threshold.
-  for (sp in species) {
+  #
+  # Channel-width model only (#286): the bypass stands in for a width test,
+  # and bcfp applies it inside its cw branch alone. A `mad` group skips it.
+  aoi_model <- params_method$model[match(aoi, params_method$watershed_group_code)]
+  species_bypass <- if (identical(aoi_model, "mad")) character(0) else species
+  for (sp in species_bypass) {
     rear_rules <- params[[sp]][["rules"]][["rear"]]
     bypass <- NULL
     for (rr in rear_rules) {

R/lnk_pipeline_prepare.R

@@ -691,6 +691,15 @@ lnk_pipeline_prepare <- function(conn, aoi, cfg, loaded, schema,
     cols = c("channel_width", "channel_width_source"),
     by = "linear_feature_id")
 
+  # Mean annual discharge, for watershed groups a bundle's
+  # parameters_habitat_method puts on the `mad` model (#286). Joined for
+  # every group so the column is always there; fresh reads it only for
+  # `mad` groups. Working table only: the persist shape does not carry it.
+  fresh::frs_col_join(conn, streams_tbl,
+    from = "whse_basemapping.fwa_stream_networks_discharge",
+    cols = "mad_m3s",
+    by = "linear_feature_id")
+
   fresh::frs_col_join(conn, streams_tbl,
     from = "whse_basemapping.fwa_stream_networks_order_parent",
     cols = "stream_order_parent",

R/lnk_preflight_fresh.R

@@ -25,14 +25,20 @@
 #'   run without. Defaults to the curated list in `.lnk_fresh_required()`.
 #' @param required_internal Character vector of non-exported `fresh`
 #'   objects reached via [utils::getFromNamespace()].
+#' @param required_formals Named list mapping a `fresh` function to the
+#'   arguments link passes it that older releases lack. A symbol can be
+#'   exported and still reject the call: `frs_habitat_classify()` existed
+#'   long before it took `params_method`. Defaults to
+#'   `.lnk_fresh_required_formals()`.
 #' @param min_version Minimum acceptable `fresh` version. Defaults to the
 #'   floor declared in link's own `DESCRIPTION`, so the pin lives in one
 #'   place.
 #' @param quiet Suppress the human-readable report. The report is the
 #'   point on a cypher, where the log is all the operator gets.
 #'
 #' @return Invisibly, a list with `ok`, `version`, `version_ok`,
-#'   `missing`, `missing_internal` and `message`.
+#'   `missing`, `missing_internal`, `missing_formals` (`"fn(arg)"`
+#'   strings) and `message`.
 #'
 #' @family preflight
 #'
@@ -48,11 +54,15 @@
 #' bad$missing
 lnk_preflight_fresh <- function(required = .lnk_fresh_required(),
                                 required_internal = .lnk_fresh_required_internal(),
+                                required_formals = .lnk_fresh_required_formals(),
                                 min_version = .lnk_fresh_floor(),
                                 quiet = FALSE) {
   stopifnot(
     is.character(required), length(required) >= 1L, all(nzchar(required)),
     is.character(required_internal), all(nzchar(required_internal)),
+    is.list(required_formals),
+    length(required_formals) == 0L || !is.null(names(required_formals)),
+    all(nzchar(names(required_formals))),
     is.character(min_version), length(min_version) == 1L, nzchar(min_version),
     is.logical(quiet), length(quiet) == 1L, !is.na(quiet))
 
@@ -62,20 +72,25 @@ lnk_preflight_fresh <- function(required = .lnk_fresh_required(),
   if (is.null(ns)) {
     out <- list(ok = FALSE, version = NA_character_, version_ok = FALSE,
                 missing = required, missing_internal = required_internal,
+                missing_formals = .lnk_fresh_formals_label(required_formals),
                 message = "fresh is not installed or its namespace will not load")
   } else {
     missing <- setdiff(required, getNamespaceExports(ns))
     missing_internal <- required_internal[
       !vapply(required_internal, exists, logical(1),
               envir = ns, inherits = FALSE)]
+    missing_formals <- .lnk_fresh_missing_formals(ns, required_formals)
     version_ok <- !is.na(version) &&
       utils::compareVersion(version, min_version) >= 0L
     out <- list(
-      ok = length(missing) == 0L && length(missing_internal) == 0L && version_ok,
+      ok = length(missing) == 0L && length(missing_internal) == 0L &&
+        length(missing_formals) == 0L && version_ok,
       version = version, version_ok = version_ok,
       missing = missing, missing_internal = missing_internal,
+      missing_formals = missing_formals,
       message = .lnk_fresh_message(version, min_version, missing,
-                                   missing_internal, version_ok))
+                                   missing_internal, version_ok,
+                                   missing_formals))
   }
 
   if (!quiet) message(out$message)
@@ -103,6 +118,37 @@ lnk_preflight_fresh <- function(required = .lnk_fresh_required(),
     "frs_params", "frs_wsg_drainage", "frs_wsg_outlets")
 }
 
+# Arguments link passes that older fresh releases do not accept. The call
+# would fail with "unused argument" only once a WSG reached that phase.
+# R/lnk_pipeline_classify.R: params_method arrived in fresh 0.35.0 (#286).
+.lnk_fresh_required_formals <- function() {
+  list(frs_habitat_classify = "params_method")
+}
+
+# "fn(arg)" for each required formal the namespace does not provide. A
+# function that is itself absent reports every argument it was asked for;
+# the missing export is reported separately.
+.lnk_fresh_missing_formals <- function(ns, required_formals) {
+  out <- character(0)
+  for (fn in names(required_formals)) {
+    obj <- if (exists(fn, envir = ns, inherits = FALSE)) {
+      get(fn, envir = ns)
+    }
+    have <- if (is.function(obj)) names(formals(obj)) else character(0)
+    gone <- setdiff(required_formals[[fn]], have)
+    if (length(gone)) out <- c(out, sprintf("%s(%s)", fn, gone))
+  }
+  out
+}
+
+.lnk_fresh_formals_label <- function(required_formals) {
+  out <- character(0)
+  for (fn in names(required_formals)) {
+    out <- c(out, sprintf("%s(%s)", fn, required_formals[[fn]]))
+  }
+  out
+}
+
 # Non-exported fresh objects link reaches via getFromNamespace().
 # R/lnk_pipeline_connect.R:101.
 .lnk_fresh_required_internal <- function() {
@@ -166,10 +212,12 @@ lnk_preflight_fresh <- function(required = .lnk_fresh_required(),
 }
 
 .lnk_fresh_message <- function(version, min_version, missing,
-                               missing_internal, version_ok) {
+                               missing_internal, version_ok,
+                               missing_formals = character(0)) {
   head <- sprintf("fresh %s (floor %s)",
                   if (is.na(version)) "NOT INSTALLED" else version, min_version)
-  if (length(missing) == 0L && length(missing_internal) == 0L && version_ok) {
+  if (length(missing) == 0L && length(missing_internal) == 0L &&
+      length(missing_formals) == 0L && version_ok) {
     return(paste0("[preflight] ", head, " - OK, all required symbols present"))
   }
   parts <- character(0)
@@ -184,6 +232,10 @@ lnk_preflight_fresh <- function(required = .lnk_fresh_required(),
     parts <- c(parts, sprintf("  missing internals: %s",
                               paste(missing_internal, collapse = ", ")))
   }
+  if (length(missing_formals)) {
+    parts <- c(parts, sprintf("  missing arguments: %s",
+                              paste(missing_formals, collapse = ", ")))
+  }
   parts <- c(parts,
     "  fix: pak::pkg_install(\"NewGraphEnvironment/fresh@<ref>\") at or above the floor")
   paste(c(paste0("[preflight] ", head, " - FAILED"), parts), collapse = "\n")

R/lnk_preflight_vintage.R

@@ -8,7 +8,7 @@
 #' May inputs and three on August inputs, produced one consolidated table
 #' set, and said nothing about it anywhere (link#246).
 #'
-#' The other seven tables in `.lnk_input_primitives()` are bulk-restored
+#' The FWA tables in `.lnk_input_primitives()` are bulk-restored
 #' FWA. They are never `ANALYZE`d, so they carry no vintage at all and are
 #' not an axis this can measure — including them would mean every host
 #' failing forever on data that is not the staleness risk.