invariants: guard mainstem_id across all stages (merge + ngen + aggregated)

mainstem_id is a global reference levelpath id that must propagate intact
through every stage / derivative layer -- not a minted per-stage id. Add a
shared .hf_mainstem_check() (flags >5% NULL = dropped or mis-mapped) and wire
it into hf_check_invariants for the aggregated and ngen stages AND into
hf_check_merge_invariants. This regression (Stage-4 merge remapped mainstem_id
through the flowpath-id table -> 78% NULL) shipped silently because nothing
checked it; now any stage that drops it fails fast. Tests cover dropped + clean
at both the per-stage and merge entry points.

Author: Mike Johnson

R/hf_invariants.R

@@ -234,6 +234,7 @@ hf_check_invariants <- function(stage, ..., strict = TRUE,
       sum(duplicated(ids)) == 0L,
       sprintf("aggregated flowpath_id has %d duplicate(s)",
         sum(duplicated(ids))))
+    checks$mainstem_id_populated <- .hf_mainstem_check(flowpaths)
   }
 
   if (!is.null(divides)) {
@@ -263,6 +264,7 @@ hf_check_invariants <- function(stage, ..., strict = TRUE,
     n_bad <- sum(!is.na(ids) & !startsWith(ids, "fp-"))
     checks$ngen_fp_prefix <- .hf_ok(n_bad == 0L,
       sprintf("%d ngen flowpath_id(s) do not start with 'fp-'", n_bad))
+    checks$mainstem_id_populated <- .hf_mainstem_check(flowpaths)
   }
 
   if (!is.null(divides)) {
@@ -327,6 +329,21 @@ hf_check_invariants <- function(stage, ..., strict = TRUE,
 
 .hf_info <- function(msg) list(ok = TRUE, msg = msg, kind = "info")
 
+# Shared mainstem_id population check. mainstem_id is a global reference
+# levelpath id that must propagate intact through every stage / derivative
+# layer -- it is NOT a minted per-stage id. A high NULL fraction means it was
+# dropped or mis-mapped (e.g. remapped through an id table it does not belong
+# to). Returns a check result, or .hf_info when the column is absent.
+.hf_mainstem_check <- function(flowpaths) {
+  if (!"mainstem_id" %in% names(flowpaths)) {
+    return(.hf_info("no mainstem_id column on flowpaths"))
+  }
+  frac <- mean(is.na(flowpaths$mainstem_id))
+  .hf_ok(frac <= 0.05,
+    sprintf("%.1f%% of flowpaths have NULL mainstem_id%s", 100 * frac,
+      if (frac > 0.05) " -- dropped/mis-mapped?" else ""))
+}
+
 .hf_ring_counts <- function(geoms) {
   vapply(geoms, function(g) {
     if (length(g) == 0L || sf::st_is_empty(g)) return(0L)
@@ -493,6 +510,9 @@ hf_check_merge_invariants <- function(merged, expected = NULL,
         if (frac_accum < 0.10) " -- WARNING: looks unaccumulated (topology resolved?)" else ""))
   } else checks$drainage_area_populated <- .hf_ok(FALSE, "total_dasqkm column absent from flowpaths")
 
+  # mainstem_id (a global reference levelpath id) must survive the merge intact.
+  checks$mainstem_id_populated <- .hf_mainstem_check(fp)
+
   # single CRS across all merged layers
   crs_of <- function(x) if (inherits(x, "sf")) sf::st_crs(x)$epsg else NA
   crs <- unique(stats::na.omit(c(crs_of(fp), crs_of(dv), crs_of(nx))))

tests/testthat/test-invariants.R

@@ -60,6 +60,20 @@ test_that("ngen stage enforces fp-/cat- id prefixes", {
     "do not start with 'fp-'")
 })
 
+test_that("ngen stage flags dropped/mis-mapped mainstem_id", {
+  drop_fp <- data.frame(flowpath_id = paste0("fp-", 1:5),
+    flowpath_toid = "0", mainstem_id = c(10, rep(NA_real_, 4)),  # 80% NULL
+    stringsAsFactors = FALSE)
+  res <- suppressMessages(hf_check_invariants("ngen", flowpaths = drop_fp,
+    strict = FALSE))
+  expect_false(res$checks$mainstem_id_populated$ok)
+
+  full_fp <- transform(drop_fp, mainstem_id = 11:15)
+  res2 <- suppressMessages(hf_check_invariants("ngen", flowpaths = full_fp,
+    strict = FALSE))
+  expect_true(res2$checks$mainstem_id_populated$ok)
+})
+
 test_that("aggregated stage catches duplicate flowpath_id", {
   fp <- data.frame(flowpath_id = c("1", "1"), stringsAsFactors = FALSE)
   expect_error(

tests/testthat/test-merge-invariants.R

@@ -39,3 +39,22 @@ test_that("hf_check_merge_invariants strict stops on failure", {
     hf_check_merge_invariants(list(flowpaths = fp, divides = dv, nexus = NULL),
       strict = TRUE)))
 })
+
+test_that("hf_check_merge_invariants flags dropped/mis-mapped mainstem_id", {
+  # mostly-NULL mainstem_id -- the merge-drop regression this guards against
+  fp <- data.frame(flowpath_id = paste0("fp-", 1:10), vpuid = "01",
+    total_dasqkm = 1:10, areasqkm = rep(0.5, 10),
+    mainstem_id = c(101, rep(NA_real_, 9)))            # 90% NULL
+  dv <- data.frame(divide_id = paste0("cat-", 1:10), vpuid = "01")
+  res <- suppressMessages(
+    hf_check_merge_invariants(list(flowpaths = fp, divides = dv, nexus = NULL),
+      strict = FALSE))
+  expect_false(res$checks$mainstem_id_populated$ok)
+
+  # fully-populated mainstem_id passes
+  fp$mainstem_id <- 100 + (1:10)
+  res2 <- suppressMessages(
+    hf_check_merge_invariants(list(flowpaths = fp, divides = dv, nexus = NULL),
+      strict = FALSE))
+  expect_true(res2$checks$mainstem_id_populated$ok)
+})