docs: survey_weight is col 28 under schema v6 (was col 26 in v5)

Rebase onto the v6 main (bd53230) shifted survey_weight from col 26 to
col 28: v5→v6 inserted cog_legacy_state/cog_legacy_county at positions
10-11 (26→28 cols). Position confirmed against the regenerated v6
fixture and both corpus docs (reader-specification.md §3 'Long parquet
schema (28 columns)' row 28; data_dictionary.md '28-column schema v6'
row 28).

DESCRIPTION

@@ -32,4 +32,4 @@ Config/testthat/edition: 3
 VignetteBuilder: knitr
 RoxygenNote: 7.3.3
 MinCorpusSchema: 4
-MaxCorpusSchema: 4
+MaxCorpusSchema: 5

NAMESPACE

@@ -10,5 +10,6 @@ export(cog_gov_search)
 export(cog_manifest)
 export(cog_mirror)
 export(cog_peer_compare)
+export(cog_recipes)
 export(cog_revenue)
 export(cog_spending)

R/basis.R

@@ -0,0 +1,78 @@
+# R/basis.R
+# basis= resolution (harmonized/raw, with v4/v5 dual-accept) and the
+# harmonization exclusion-count block attached to provenance.
+
+#' Resolve the requested `basis` against the active corpus's schema_version.
+#'
+#' On a `schema_version >= 5` corpus, the requested basis is used as-is. On
+#' an older (`schema_version == 4`) corpus, which has no harmonization
+#' tables: a caller who left `basis` at its default (`"harmonized"`, so
+#' `explicit` is `FALSE`) silently gets `"raw"` back, with a note recorded
+#' for provenance; a caller who explicitly asked for
+#' `basis = "harmonized"` gets a hard abort instead of a silent downgrade.
+#'
+#' @param basis `"harmonized"` or `"raw"` (already resolved via `match.arg`).
+#' @param explicit `TRUE` if the caller passed `basis` explicitly (as
+#'   opposed to relying on the default `c("harmonized", "raw")`).
+#' @param manifest The active session's parsed manifest list.
+#' @return List with `basis` (the resolved value) and `note` (character or
+#'   `NA_character_`).
+#' @noRd
+.resolve_basis <- function(basis, explicit, manifest) {
+  schema_version <- suppressWarnings(as.integer(manifest$schema_version %||% 0L))
+
+  if (schema_version >= 5L) {
+    return(list(basis = basis, note = NA_character_))
+  }
+
+  if (identical(basis, "harmonized") && explicit) {
+    cli::cli_abort(c(
+      "basis = \"harmonized\" requires corpus schema_version >= 5.",
+      x = "Active corpus has schema_version {schema_version}.",
+      i = "Use basis = \"raw\" (the default on this corpus), or point USCOGDATA_URL at a schema_version >= 5 corpus."
+    ), class = "uscogdata_basis_unsupported")
+  }
+
+  list(
+    basis = "raw",
+    note = sprintf(
+      "basis resolved to \"raw\": corpus schema_version %d < 5 (harmonization tables unavailable)",
+      schema_version
+    )
+  )
+}
+
+#' Count + sum item-level rows that basis="harmonized" excludes because they
+#' carry no harmonized_code (discontinued / not-yet-ruled codes) within the
+#' requested flow type (spending or revenue), govids, and years. Only
+#' meaningful when the resolved basis is "harmonized"; returns an
+#' applied = FALSE stub otherwise (raw basis never excludes rows this way).
+#' @noRd
+.build_harmonization_block <- function(con, govid, years, resolved, flow_prefixes) {
+  if (!identical(resolved$basis, "harmonized")) {
+    return(list(
+      applied = FALSE,
+      na_rows_excluded = 0L,
+      na_amount_excluded = 0,
+      note = resolved$note
+    ))
+  }
+
+  sql <- sprintf(
+    "SELECT COUNT(*) AS n, COALESCE(SUM(amt), 0) * 1000.0 AS amt
+     FROM long
+     WHERE canonical_govid IN (%s) AND year IN (%s)
+       AND NOT is_aggregate AND harmonized_code IS NULL
+       AND LEFT(item_code, 1) IN (%s)",
+    .sql_lit_chr(govid), paste(as.integer(years), collapse = ","),
+    .sql_lit_chr(flow_prefixes)
+  )
+  na <- DBI::dbGetQuery(con, sql)
+
+  list(
+    applied = TRUE,
+    na_rows_excluded = as.integer(na$n),
+    na_amount_excluded = as.numeric(na$amt),
+    note = resolved$note
+  )
+}

R/explain.R

@@ -51,6 +51,15 @@ cog_explain <- function(result, format = c("print", "list")) {
     cli::cli_text("Category: (all)")
   }
 
+  if (!is.null(prov$basis)) {
+    note <- if (!is.null(prov$basis_note) && !is.na(prov$basis_note)) {
+      sprintf(" (%s)", prov$basis_note)
+    } else {
+      ""
+    }
+    cli::cli_text("Basis: {prov$basis}{note}")
+  }
+
   cli::cli_h2("Codes observed")
   codes <- prov$codes_summed$observed
   if (length(codes) == 0L) {
@@ -66,6 +75,39 @@ cog_explain <- function(result, format = c("print", "list")) {
     )
   }
 
+  h <- prov$harmonization
+  if (!is.null(h) && isTRUE(h$applied)) {
+    cli::cli_h2("Harmonization")
+    cli::cli_text(
+      "Excluded {h$na_rows_excluded} row(s) with no harmonized_code (${format(h$na_amount_excluded, big.mark = ',')})"
+    )
+  }
+
+  rc <- prov$recipe
+  if (!is.null(rc)) {
+    cli::cli_h2("Recipe")
+    cli::cli_text("{rc$recipe_id}: {rc$label}")
+    comp_lines <- vapply(rc$components, function(x) {
+      sprintf("%s (%s, %s-%s, weight=%s)", x$component_code, x$gov_type_scope,
+              x$year_min, x$year_max, x$weight)
+    }, character(1))
+    cli::cli_ul(comp_lines)
+  }
+
+  if (length(prov$suggestions) > 0L) {
+    cli::cli_h2("Suggestions")
+    sugg_lines <- vapply(prov$suggestions, function(s) {
+      sprintf("%s -- %s (years %s-%s): %s", s$recipe_id, s$label,
+              s$available_years[1], s$available_years[2], s$hint)
+    }, character(1))
+    cli::cli_ul(sugg_lines)
+  }
+
+  if (length(prov$series_break_refs) > 0L) {
+    cli::cli_h2("Series breaks")
+    cli::cli_ul(.series_break_story_lines(prov$series_break_refs))
+  }
+
   cli::cli_h2("Transformations")
   uc <- prov$transformations$units_conversion
   if (isTRUE(uc$applied)) {
@@ -109,6 +151,28 @@ cog_explain <- function(result, format = c("print", "list")) {
   invisible(NULL)
 }
 
+# One "break-story" line per referenced break_id: "SB109 (2005): <join_advice>".
+# Re-queries series_breaks_pq for the detail (break_year, join_advice) that
+# provenance$series_break_refs deliberately doesn't carry (the schema keeps
+# that field to a plain id array). Falls back to bare ids if no session is
+# available (e.g. explaining a result after cog_close()) rather than
+# erroring cog_explain() over a cosmetic detail.
+#' @noRd
+.series_break_story_lines <- function(break_ids) {
+  con <- tryCatch(.ensure_session(), error = function(e) NULL)
+  if (is.null(con) || !DBI::dbIsValid(con)) return(break_ids)
+  detail <- tryCatch(
+    DBI::dbGetQuery(con, sprintf(
+      "SELECT break_id, break_year, join_advice FROM series_breaks_pq
+       WHERE break_id IN (%s) ORDER BY break_id",
+      .sql_lit_chr(break_ids)
+    )),
+    error = function(e) NULL
+  )
+  if (is.null(detail) || nrow(detail) == 0L) return(break_ids)
+  sprintf("%s (%s): %s", detail$break_id, detail$break_year, detail$join_advice)
+}
+
 # Expand a 2-digit Census popyear (e.g. 19) to a 4-digit calendar year (2019).
 # F-33 metadata stores popyear as 2 digits; pivot at 70 to handle a future
 # corpus that ever spans pre-1970 vintages, though current scope is 2000+.

R/manifest.R

@@ -128,11 +128,21 @@
 }
 
 #' @noRd
-.validate_schema <- function(manifest, expected_version) {
+#' Schema v6 (FIPS geography harmonization, 2026-07-22) is accepted alongside
-  if (manifest$schema_version != expected_version) {
+#' 4/5. v6 renamed the long table's fips_state_code/fips_county_code to
+#' fips_state_asof/fips_county_asof and added cog_legacy_state/
+#' cog_legacy_county (26 -> 28 cols); this package references NONE of those
+#' columns, so no code change was needed. NOTE the SILENT semantic change for
+#' any consumer of the raw long table: long fips_state/fips_county are now
+#' PRESENT/harmonized geography (current county identity carried back to every
+#' year, matching canonical_fips_xwalk) rather than as-of-year; as-of-year
+#' moved to the *_asof columns. This package's own geography always came from
+#' the xwalk (already present-based), so behaviour is unchanged.
+.validate_schema <- function(manifest, supported = c(4L, 5L, 6L)) {
+  if (!manifest$schema_version %in% supported) {
     cli::cli_abort(c(
       "Corpus schema version mismatch.",
-      x = "Package expects schema_version = {expected_version}; corpus has {manifest$schema_version}.",
+      x = "Package supports schema_version in {paste(supported, collapse = ', ')}; corpus has {manifest$schema_version}.",
       i = "Update uscogdata (install.packages or pak::pkg_install) or re-publish corpus."
     ))
   }

R/provenance.R

@@ -4,7 +4,10 @@
 #' @noRd
 .build_provenance <- function(verb, call, govid, years, category,
                               per_capita, adjust_to_year, result, sql,
-                              subtype_col) {
+                              subtype_col, basis = NA_character_,
+                              basis_note = NA_character_,
+                              harmonization = NULL, recipe = NULL,
+                              suggestions = list()) {
   manifest <- .uscogdata_env$manifest
 
   codes <- result[["codes_included"]]
@@ -29,6 +32,14 @@
     unique(result$gov_name)
   }
 
+  schema_version <- suppressWarnings(as.integer(manifest$schema_version %||% 0L))
+  con <- .uscogdata_env$con
+  break_refs <- if (!is.null(con) && DBI::dbIsValid(con)) {
+    .build_series_break_refs(con, codes_observed, years, schema_version)
+  } else {
+    character(0)
+  }
+
   list(
     verb = verb,
     call = paste(deparse(call), collapse = " "),
@@ -38,6 +49,14 @@
     ),
     years = as.integer(years),
     category = category,
+    basis = basis,
+    basis_note = basis_note,
+    harmonization = harmonization %||% list(
+      applied = FALSE, na_rows_excluded = 0L, na_amount_excluded = 0,
+      note = NA_character_
+    ),
+    recipe = recipe,
+    suggestions = suggestions,
     scope = list(
       gov_types_included = as.integer(unlist(manifest$scope$gov_types_included)),
       gov_types_excluded = as.integer(unlist(manifest$scope$gov_types_excluded)),
@@ -90,7 +109,7 @@
         index = if (is.null(adjust_to_year)) NA_character_ else "CPI-U (BLS CPIAUCSL annual average, bundled)"
       )
     ),
-    series_break_refs = character(0),
+    series_break_refs = break_refs,
     manifest = list(
       schema_version = as.integer(manifest$schema_version),
       pipeline_commit = manifest$pipeline_commit %||% NA_character_,

R/recipes.R

@@ -0,0 +1,167 @@
+# R/recipes.R
+# Harmonization recipes: multi-code, cross-vintage series built by summing a
+# fixed set of component item codes with per-component weights and
+# year/gov-type scoping (see the `harmonization_recipes` view, registered
+# from data/harmonization_recipes.parquet, schema_version >= 5 only).
+#
+# Recipes exist because some cross-vintage series can't be expressed as a
+# 1:1 harmonized_code mapping (basis = "harmonized"): the wide era (pre-2012)
+# publishes only a combined aggregate row for these families (e.g.
+# corrections functions 04+05), while the modern era splits them into leaf
+# codes. A recipe's generic join sums whichever of its component codes are
+# present for a given year, so the resulting series is continuous across
+# that format boundary.
+
+#' List available harmonization recipes
+#'
+#' Recipes are multi-code cross-vintage series (see [cog_spending()]'s
+#' `recipe` argument) catalogued in the corpus's `harmonization_recipes`
+#' table. Use this to discover valid `recipe` ids.
+#'
+#' @param pattern Optional regex matched case-insensitively against
+#'   `recipe_id` or `label`.
+#' @return Tibble with columns `recipe_id`, `label`, `n_components`,
+#'   `year_min`, `year_max` (the min/max component year coverage), sorted by
+#'   `recipe_id`.
+#' @export
+cog_recipes <- function(pattern = NULL) {
+  if (!is.null(pattern) &&
+      (!is.character(pattern) || length(pattern) != 1L)) {
+    cli::cli_abort("`pattern` must be a length-1 character string or NULL.")
+  }
+  con <- .ensure_session()
+  .require_schema_v5(con, .uscogdata_env$manifest, "cog_recipes()")
+
+  where <- if (is.null(pattern)) {
+    ""
+  } else {
+    sprintf(
+      "WHERE regexp_matches(recipe_id, %1$s, 'i') OR regexp_matches(label, %1$s, 'i')",
+      .sql_lit_chr(pattern)
+    )
+  }
+  sql <- paste(
+    "SELECT recipe_id, any_value(label) AS label,
+            COUNT(*) AS n_components,
+            MIN(year_min) AS year_min, MAX(year_max) AS year_max
+     FROM harmonization_recipes",
+    where,
+    "GROUP BY recipe_id
+     ORDER BY recipe_id"
+  )
+  out <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  out$year_min <- as.integer(out$year_min)
+  out$year_max <- as.integer(out$year_max)
+  out$n_components <- as.integer(out$n_components)
+  out
+}
+
+#' Abort unless the active corpus has schema_version >= 5.
+#' @noRd
+.require_schema_v5 <- function(con, manifest, what) {
+  sv <- suppressWarnings(as.integer(manifest$schema_version %||% 0L))
+  if (sv < 5L) {
+    cli::cli_abort(c(
+      sprintf("%s requires corpus schema_version >= 5.", what),
+      x = "Active corpus has schema_version {sv}.",
+      i = "Point USCOGDATA_URL at a schema_version >= 5 corpus to use harmonization recipes."
+    ), class = "uscogdata_schema_unsupported")
+  }
+  invisible(sv)
+}
+
+#' Abort with the valid id list unless `recipe_id` exists in the catalog.
+#' @noRd
+.validate_recipe_id <- function(con, recipe_id) {
+  ids <- DBI::dbGetQuery(
+    con, "SELECT DISTINCT recipe_id FROM harmonization_recipes"
+  )$recipe_id
+  if (!recipe_id %in% ids) {
+    cli::cli_abort(c(
+      "Unknown recipe = {.val {recipe_id}}.",
+      i = "Valid ids: {paste(sort(ids), collapse = ', ')}",
+      i = "See cog_recipes() for labels and year coverage."
+    ), class = "uscogdata_unknown_recipe")
+  }
+  invisible(TRUE)
+}
+
+#' Fetch the component rows for one recipe (label, component codes, scope,
+#' year ranges, weights) -- both for running the recipe and for the
+#' `recipe` provenance block.
+#' @noRd
+.recipe_components <- function(con, recipe_id) {
+  sql <- sprintf(
+    "SELECT recipe_id, label, component_code, gov_type_scope,
+            year_min, year_max, weight, source_break_ids, notes
+     FROM harmonization_recipes
+     WHERE recipe_id = %s
+     ORDER BY component_code",
+    .sql_lit_chr(recipe_id)
+  )
+  tibble::as_tibble(DBI::dbGetQuery(con, sql))
+}
+
+#' Run a recipe's generic join: sum `amt * weight` across whichever
+#' component codes are present for each (year, canonical_govid), scoped by
+#' gov_type_scope. Deliberately does NOT filter `NOT is_aggregate`: in the
+#' wide era (<= 2011) these families' component codes exist ONLY as
+#' aggregate rows (leaves first appear 2012), so excluding aggregates would
+#' zero out the wide-era half of every recipe. This is safe by corpus
+#' construction -- wide-era rows for these codes are aggregate-only, modern
+#' rows are leaf-only, and every component row is year-scoped via
+#' `year_min`/`year_max` -- so there is no double-counting. (Checkpoint
+#' review docs/phase_r_harmonization_review.md § 0.2.)
+#' @noRd
+.run_recipe <- function(con, recipe_id, govid, years) {
+  sql <- sprintf(
+    "SELECT l.year, l.canonical_govid,
+            COALESCE(x.gov_name, l.gov_name) AS gov_name,
+            SUM(l.amt * r.weight) * 1000.0 AS amt_nominal,
+            string_agg(DISTINCT l.item_code, ',' ORDER BY l.item_code) AS codes_included
+     FROM long l
+     JOIN harmonization_recipes r
+       ON l.item_code = r.component_code
+      AND l.year BETWEEN r.year_min AND r.year_max
+      AND (r.gov_type_scope = 'all'
+           OR (r.gov_type_scope = 'state' AND l.type = 0)
+           OR (r.gov_type_scope = 'local' AND l.type BETWEEN 1 AND 3))
+     LEFT JOIN canonical_fips_xwalk x USING (canonical_govid)
+     WHERE r.recipe_id = %1$s
+       AND l.canonical_govid IN (%2$s)
+       AND l.year IN (%3$s)
+     GROUP BY 1, 2, 3
+     ORDER BY 1, 2",
+    .sql_lit_chr(recipe_id), .sql_lit_chr(govid),
+    paste(as.integer(years), collapse = ",")
+  )
+  result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  attr(result, "sql_query") <- sql
+  result
+}
+
+#' Shape a raw .run_recipe() result into the standard cog_spending()/
+#' cog_revenue() column layout: subtype = "recipe", category = the recipe's
+#' label, aggregate_fallback = FALSE (recipes resolve coverage gaps by
+#' construction, not by falling back to an aggregate row).
+#' @noRd
+.shape_recipe_result <- function(result, subtype_col, label) {
+  sql_query <- attr(result, "sql_query")
+  n <- nrow(result)
+  result[[subtype_col]] <- rep("recipe", n)
+  result$category <- rep(label, n)
+  result$aggregate_fallback <- rep(FALSE, n)
+  result <- result[, c(
+    "year", "canonical_govid", "gov_name", subtype_col, "category",
+    "amt_nominal", "codes_included", "aggregate_fallback"
+  ), drop = FALSE]
+  attr(result, "sql_query") <- sql_query
+  result
+}
+
+#' Turn a small data.frame into a list-of-lists (one list per row), the
+#' shape used for the `recipe$components` provenance block.
+#' @noRd
+.df_to_row_list <- function(df) {
+  lapply(seq_len(nrow(df)), function(i) as.list(df[i, , drop = FALSE]))
+}

R/revenue.R

@@ -14,16 +14,20 @@
 #'   optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
 #' @export
 cog_revenue <- function(govid, years, category = NULL,
-                        per_capita = FALSE, adjust_to_year = NULL) {
+                        per_capita = FALSE, adjust_to_year = NULL,
+                        basis = c("harmonized", "raw"), recipe = NULL) {
   .verb_spendrev(
     verb           = "cog_revenue",
-    view           = "revenue_annotated",
+    view_base      = "revenue_annotated",
     subtype_col    = "revenue_subtype",
+    flow_prefixes  = c("T", "A", "U", "B", "C", "D"),
     call           = match.call(),
     govid          = govid,
     years          = years,
     category       = category,
     per_capita     = per_capita,
-    adjust_to_year = adjust_to_year
+    adjust_to_year = adjust_to_year,
+    basis          = basis,
+    recipe         = recipe
   )
 }

R/series_breaks.R

@@ -0,0 +1,22 @@
+# R/series_breaks.R
+# Populates prov$series_break_refs (schema in inst/schemas/provenance-v1.json
+# defines the field; it was always present but always empty pre-Phase-R2)
+# with the ids of any catalogued series break whose fin_code appears among
+# the result's observed item codes and whose break_year falls inside the
+# requested year span -- the "break warnings in the provenance envelope"
+# spec § 5 promises downstream consumers (cog-api passes provenance through
+# verbatim). schema_version >= 5 only: series_breaks_pq isn't registered on
+# an older corpus.
+
+#' @noRd
+.build_series_break_refs <- function(con, codes_observed, years, schema_version) {
+  if (schema_version < 5L || length(codes_observed) == 0L) return(character(0))
+  sql <- sprintf(
+    "SELECT DISTINCT break_id
+     FROM series_breaks_pq
+     WHERE fin_code IN (%s) AND break_year BETWEEN %d AND %d
+     ORDER BY break_id",
+    .sql_lit_chr(codes_observed), min(as.integer(years)), max(as.integer(years))
+  )
+  DBI::dbGetQuery(con, sql)$break_id
+}

R/session.R

@@ -12,7 +12,7 @@ cog_open <- function(url = .resolve_url(),
   DBI::dbExecute(con, "INSTALL httpfs; LOAD httpfs;")
 
   manifest <- .fetch_or_cache_manifest(url, cache_dir)
-  .validate_schema(manifest, expected_version = 4L)
+  .validate_schema(manifest, supported = c(4L, 5L, 6L))
   .validate_scope(manifest)
 
   .register_views(con, url, manifest)

R/spending.R

@@ -20,6 +20,28 @@
 #'   in that year).
 #' @param adjust_to_year Integer base year for CPI-U real-dollar conversion,
 #'   or `NULL` for nominal only.
+#' @param basis `"harmonized"` (default) sums item codes through the
+#'   cross-vintage harmonization mapping (folding series-break-affected
+#'   codes onto a comparable target and excluding aggregate / discontinued
+#'   rows -- see the `harmonization` block in `cog_explain()`); `"raw"`
+#'   reproduces the pre-Phase-R2 behavior (published item codes, no
+#'   folding). On a corpus with `schema_version < 5` (no harmonization
+#'   tables), `basis` silently resolves to `"raw"` when left at its default
+#'   and the resolution is recorded in the provenance; explicitly passing
+#'   `basis = "harmonized"` on such a corpus aborts. Ignored when `recipe`
+#'   is set (see below).
+#' @param recipe Optional harmonization recipe id (see [cog_recipes()]) for
+#'   multi-code cross-vintage series that a 1:1 harmonized_code mapping
+#'   can't express (e.g. a wide-era aggregate that only splits into leaf
+#'   codes in the modern era). Mutually exclusive with `category`. The
+#'   result's subtype column reads `"recipe"` and `category` reads the
+#'   recipe's label. Requires `schema_version >= 5`. A recipe query bypasses
+#'   `basis` entirely (it joins `long` directly rather than going through
+#'   the `*_annotated`/`*_annotated_harmonized` views), so the `basis`
+#'   argument is ignored and the result's provenance reports
+#'   `basis = "recipe"` with an inert `harmonization` block (`applied =
+#'   FALSE`, pointing at the `recipe` block instead) rather than a
+#'   possibly-misleading `"harmonized"`/`"raw"` value.
 #' @return Tibble with columns `year`, `canonical_govid`, `gov_name`,
 #'   `spend_subtype`, `category`, `amt_nominal`, optional `amt_real`,
 #'   optional `amt_per_capita_nominal`, optional `amt_per_capita_real`,
@@ -27,35 +49,65 @@
 #'   Carries a `provenance` attribute matching `inst/schemas/provenance-v1.json`.
 #' @export
 cog_spending <- function(govid, years, category = NULL,
-                         per_capita = FALSE, adjust_to_year = NULL) {
+                         per_capita = FALSE, adjust_to_year = NULL,
+                         basis = c("harmonized", "raw"), recipe = NULL) {
   .verb_spendrev(
     verb          = "cog_spending",
-    view          = "spending_annotated",
+    view_base     = "spending_annotated",
     subtype_col   = "spend_subtype",
+    flow_prefixes = c("E", "F", "G", "K"),
     call          = match.call(),
     govid         = govid,
     years         = years,
     category      = category,
     per_capita    = per_capita,
-    adjust_to_year = adjust_to_year
+    adjust_to_year = adjust_to_year,
+    basis         = basis,
+    recipe        = recipe
   )
 }
 
 #' @noRd
-.verb_spendrev <- function(verb, view, subtype_col, call,
+.verb_spendrev <- function(verb, view_base, subtype_col, flow_prefixes, call,
                            govid, years, category,
-                           per_capita, adjust_to_year) {
+                           per_capita, adjust_to_year,
+                           basis = c("harmonized", "raw"), recipe = NULL) {
+  basis_explicit <- length(basis) == 1L
+  basis <- match.arg(basis, c("harmonized", "raw"))
+
   govid <- .coerce_govid_input(govid, arg = "govid")
-  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year)
+  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year,
+                        recipe)
 
   years   <- as.integer(years)
   if (!is.null(adjust_to_year)) adjust_to_year <- as.integer(adjust_to_year)
 
   con <- .ensure_session()
+  manifest <- .uscogdata_env$manifest
   scope <- .check_govids_in_scope(govid)
 
+  resolved <- .resolve_basis(basis, basis_explicit, manifest)
+
+  recipe_block <- NULL
+  category_for_prov <- category
+  if (!is.null(recipe)) {
+    .require_schema_v5(con, manifest, "recipe =")
+    .validate_recipe_id(con, recipe)
+    comps <- .recipe_components(con, recipe)
+    recipe_label <- comps$label[[1]]
+    result <- .run_recipe(con, recipe, govid, years)
+    sql <- attr(result, "sql_query")
+    result <- .shape_recipe_result(result, subtype_col, recipe_label)
+    recipe_block <- list(
+      recipe_id = recipe, label = recipe_label,
+      components = .df_to_row_list(comps)
+    )
+    category_for_prov <- recipe_label
+  } else {
+    view <- .select_view(view_base, resolved$basis)
     sql <- .build_verb_sql(view, subtype_col, govid, years, category)
     result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  }
 
   if (per_capita) result <- .attach_per_capita(result, con, govid)
   if (!is.null(adjust_to_year)) {
@@ -64,28 +116,64 @@ cog_spending <- function(govid, years, category = NULL,
 
   result$notes <- .notes_column(result)
 
+  # A recipe result doesn't go through spending_annotated(_harmonized) /
+  # revenue_annotated(_harmonized) at all -- .run_recipe()'s generic join
+  # reads `long` directly -- so `basis` and the `harmonization` exclusion
+  # count (which is itself computed from `long`, independent of which view
+  # a non-recipe query used) would describe a code path this result never
+  # took. Rather than report a technically-still-computed but misleading
+  # basis = "harmonized"/"raw" + harmonization$applied combo, recipe
+  # results report basis = "recipe" and an explicit, inert harmonization
+  # block pointing at the `recipe` block instead. Task 12 (cog-api) passes
+  # provenance through verbatim, so this needs to be unambiguous rather
+  # than technically-defensible-but-confusing.
+  if (!is.null(recipe)) {
+    basis_for_prov <- "recipe"
+    basis_note_for_prov <- NA_character_
+    harmonization <- list(
+      applied = FALSE, na_rows_excluded = 0L, na_amount_excluded = 0,
+      note = "basis/harmonization not applicable to recipe results; see the recipe block instead"
+    )
+    suggestions <- list()
+  } else {
+    basis_for_prov <- resolved$basis
+    basis_note_for_prov <- resolved$note
+    harmonization <- .build_harmonization_block(
+      con, govid, years, resolved, flow_prefixes
+    )
+    suggestions <- .build_suggestions(con, govid, years, category, resolved$basis)
+  }
+
   prov <- .build_provenance(
     verb           = verb,
     call           = call,
     govid          = govid,
     years          = years,
-    category       = category,
+    category       = category_for_prov,
     per_capita     = per_capita,
     adjust_to_year = adjust_to_year,
     result         = result,
     sql            = sql,
-    subtype_col    = subtype_col
+    subtype_col    = subtype_col,
+    basis          = basis_for_prov,
+    basis_note     = basis_note_for_prov,
+    harmonization  = harmonization,
+    recipe         = recipe_block,
+    suggestions    = suggestions
   )
   prov$scope$govids_found   <- scope$found
   prov$scope$govids_missing <- scope$missing
   attr(result, "provenance") <- prov
   attr(result, ".popyear_range") <- NULL
+
+  if (length(suggestions) > 0L) .inform_suggestions(suggestions)
+
   result
 }
 
 #' @noRd
 .validate_verb_inputs <- function(govid, years, category,
-                                  per_capita, adjust_to_year) {
+                                  per_capita, adjust_to_year, recipe = NULL) {
   if (!is.character(govid) || length(govid) == 0L) {
     cli::cli_abort("`govid` must be a non-empty character vector.")
   }
@@ -104,9 +192,25 @@ cog_spending <- function(govid, years, category = NULL,
       cli::cli_abort("`adjust_to_year` must be NULL or a length-1 integer.")
     }
   }
+  if (!is.null(recipe)) {
+    if (!is.character(recipe) || length(recipe) != 1L) {
+      cli::cli_abort("`recipe` must be NULL or a length-1 character string.")
+    }
+    if (!is.null(category)) {
+      cli::cli_abort(c(
+        "`recipe` and `category` are mutually exclusive.",
+        i = "Pass one or the other, not both."
+      ), class = "uscogdata_recipe_category_conflict")
+    }
+  }
   invisible(TRUE)
 }
 
+#' @noRd
+.select_view <- function(view_base, basis) {
+  if (identical(basis, "harmonized")) paste0(view_base, "_harmonized") else view_base
+}
+
 #' @noRd
 .sql_lit_chr <- function(x) {
   safe <- gsub("'", "''", x, fixed = TRUE)

R/suggestions.R

@@ -0,0 +1,215 @@
+# R/suggestions.R
+# Recipe-component-driven signposting: when a basis = "harmonized" query for
+# a category asks for a code that is itself a harmonization recipe
+# component, and that specific code has no rows in some requested years
+# while the recipe's own generic join would still fill those years for this
+# government, surface that recipe as a suggestion.
+#
+# This is deliberately keyed off the recipe catalog's component codes, not
+# off harmonization_map rows: no live map row carries a non-blank
+# suggested_recipe_id (the corpus's wide era exposes split families like
+# corrections functions 04+05 ONLY as aggregate rows, which basis =
+# "harmonized" excludes by construction -- there's no NA ruling to hang a
+# suggestion off of, just a leaf-code absence a recipe happens to fill).
+# See docs/phase_r_harmonization_review.md § 0.3.
+#
+# Scope is deliberately narrow in one respect and, as of Phase R3 Task 19c,
+# deliberately WIDE in another: signposting only runs when the caller
+# supplied a `category` (an un-scoped, all-categories query has no single
+# coverage question to answer), but within that category it now checks
+# EACH recipe component that is itself a category member individually,
+# rather than asking whether the whole category *result* has zero rows
+# that year. A recipe fires when one of its own components has zero rows
+# for this government in a requested year, AND SOME OTHER component of that
+# SAME recipe -- excluding the gapped one itself -- has a row (same join
+# .run_recipe() uses, aggregate rows included) for that year. This is the
+# literal review-doc § 0.3 criterion: "...has no rows ... but other
+# components do." A component's OWN aggregate-only row does not satisfy
+# its own gap (self-coverage is not "other components"); only a genuinely
+# different sibling component can. This fires even if OTHER, unrelated
+# codes in the same category have full data that year and the overall
+# result looks complete. That is a deliberate narrowing of the R2-era
+# false-positive guard: most governments don't use every sibling code in a
+# multi-code category every year, and per-code detection WILL flag some of
+# that as a "gap" even though it's really just a government not having
+# that particular sub-type of spending, not a format-boundary artifact.
+# The remaining guard against ordinary reporting variance is the
+# per-government, per-OTHER-component `covered` check below (a component
+# is only flagged when a DIFFERENT component of the SAME recipe -- not
+# some unrelated code, and not the gapped component's own aggregate row --
+# actually has something to offer in that year); it no longer tries to
+# avoid noise from sibling *codes*, only from a recipe with genuinely
+# nothing else to contribute. The acceptable noise level this trade
+# produces is a product decision, measured (not tuned here) by
+# data-raw/measure_signposting_rate.R and ruled on at Checkpoint R3.
+
+#' Recipe components that are classified under the requested category --
+#' the codes a category-scoped query actually "requests". A recipe can
+#' have components outside the category (e.g. general_gov_e89_wide's E85
+#' leg has no category assignment); those never trigger on their own, they
+#' just were never part of what this query asked for.
+#' @noRd
+.category_recipe_components <- function(con, category) {
+  DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT r.recipe_id, r.component_code, r.year_min, r.year_max,
+            r.gov_type_scope
+     FROM harmonization_recipes r
+     JOIN summary_categories sc
+       ON sc.item_code = r.component_code AND sc.category IN (%s)",
+    .sql_lit_chr(category)
+  ))
+}
+
+#' Label + overall year coverage for a set of recipe ids (the suggestion's
+#' `label`/`available_years`).
+#' @noRd
+.recipe_meta <- function(con, candidates) {
+  tibble::as_tibble(DBI::dbGetQuery(con, sprintf(
+    "SELECT recipe_id, any_value(label) AS label,
+            MIN(year_min) AS year_min, MAX(year_max) AS year_max
+     FROM harmonization_recipes
+     WHERE recipe_id IN (%s)
+     GROUP BY recipe_id",
+    .sql_lit_chr(candidates)
+  )))
+}
+
+#' Which (recipe_id, component_code, year) triples have at least one
+#' NOT-aggregate row for these governments -- i.e. that specific requested
+#' code itself has data, scoped exactly like .run_recipe()'s join
+#' (component year_min/year_max + gov_type_scope). NOT-aggregate mirrors
+#' what basis = "harmonized" itself excludes: an aggregate-only year is a
+#' gap for that code exactly as it would be in a plain category query.
+#' @noRd
+.component_presence <- function(con, candidates, govid, years_lit) {
+  DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT r.recipe_id, r.component_code, l.year
+     FROM long l
+     JOIN harmonization_recipes r
+       ON l.item_code = r.component_code
+      AND l.year BETWEEN r.year_min AND r.year_max
+      AND (r.gov_type_scope = 'all'
+           OR (r.gov_type_scope = 'state' AND l.type = 0)
+           OR (r.gov_type_scope = 'local' AND l.type BETWEEN 1 AND 3))
+     WHERE NOT l.is_aggregate
+       AND r.recipe_id IN (%s)
+       AND l.canonical_govid IN (%s)
+       AND l.year IN (%s)",
+    .sql_lit_chr(candidates), .sql_lit_chr(govid), years_lit
+  ))
+}
+
+#' Which (recipe_id, component_code, year) triples have at least one row
+#' (aggregate rows included) for these governments -- the same scoping
+#' .run_recipe()'s join uses (component year_min/year_max + gov_type_scope),
+#' just checking existence instead of summing. Kept at per-component grain
+#' (not unioned across the whole recipe, unlike the R2/R3-pre-fix version of
+#' this function) so a gap check can require the covering evidence to come
+#' from a DIFFERENT component -- review-doc § 0.3's "other components", not
+#' the gapped component's own aggregate row. This is the per-government
+#' guard against ordinary reporting variance: a recipe with genuinely
+#' nothing to offer from any OTHER component (aggregate or leaf) never
+#' fires.
+#' @noRd
+.recipe_coverage <- function(con, candidates, govid, years_lit) {
+  DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT r.recipe_id, r.component_code, l.year
+     FROM long l
+     JOIN harmonization_recipes r
+       ON l.item_code = r.component_code
+      AND l.year BETWEEN r.year_min AND r.year_max
+      AND (r.gov_type_scope = 'all'
+           OR (r.gov_type_scope = 'state' AND l.type = 0)
+           OR (r.gov_type_scope = 'local' AND l.type BETWEEN 1 AND 3))
+     WHERE r.recipe_id IN (%s)
+       AND l.canonical_govid IN (%s)
+       AND l.year IN (%s)",
+    .sql_lit_chr(candidates), .sql_lit_chr(govid), years_lit
+  ))
+}
+
+#' TRUE if recipe `rid` has at least one requested component with an
+#' in-scope requested year that has no data (`present`), in a year some
+#' OTHER component of the same recipe is otherwise fillable (`covered`,
+#' excluding the component under test) -- the per-code gap the R2
+#' whole-result check couldn't see, covered by another component the way
+#' review-doc § 0.3 specifies (not by the gapped component's own aggregate
+#' row -- that is self-coverage, not "other components", and must not
+#' count).
+#' @noRd
+.recipe_component_gapped <- function(rid, requested, present, covered, years) {
+  comps <- requested[requested$recipe_id == rid, , drop = FALSE]
+  for (i in seq_len(nrow(comps))) {
+    this_code <- comps$component_code[i]
+    in_scope <- years[years >= comps$year_min[i] & years <= comps$year_max[i]]
+    if (length(in_scope) == 0L) next
+    has_data <- present$year[
+      present$recipe_id == rid & present$component_code == this_code
+    ]
+    gap_years <- setdiff(in_scope, has_data)
+    if (length(gap_years) == 0L) next
+    other_covered_years <- covered$year[
+      covered$recipe_id == rid & covered$component_code != this_code
+    ]
+    if (any(gap_years %in% other_covered_years)) return(TRUE)
+  }
+  FALSE
+}
+
+#' Build the `prov$suggestions` list for a (non-recipe) basis = "harmonized"
+#' verb call: recipes whose generic join would fill a real per-code gap for
+#' the requested category.
+#'
+#' @param con Active DuckDB connection.
+#' @param govid Character vector of canonical_govid values (the verb's raw
+#'   `govid`).
+#' @param years Integer vector of requested years.
+#' @param category `category` argument as passed to the verb (character
+#'   vector or `NULL`; suggestions are only computed when non-NULL).
+#' @param basis The *resolved* basis (`"harmonized"` or `"raw"`).
+#' @return List of `list(recipe_id, label, available_years, hint)`, possibly
+#'   empty.
+#' @noRd
+.build_suggestions <- function(con, govid, years, category, basis) {
+  if (!identical(basis, "harmonized") || is.null(category)) return(list())
+
+  requested <- .category_recipe_components(con, category)
+  if (nrow(requested) == 0L) return(list())
+  candidates <- unique(requested$recipe_id)
+  years_int  <- as.integer(years)
+  years_lit  <- paste(years_int, collapse = ",")
+
+  meta    <- .recipe_meta(con, candidates)
+  present <- .component_presence(con, candidates, govid, years_lit)
+  covered <- .recipe_coverage(con, candidates, govid, years_lit)
+
+  suggestions <- list()
+  for (rid in candidates) {
+    if (!.recipe_component_gapped(rid, requested, present, covered, years_int)) next
+    m <- meta[meta$recipe_id == rid, ]
+    suggestions[[length(suggestions) + 1L]] <- list(
+      recipe_id = rid,
+      label = m$label[[1]],
+      available_years = c(as.integer(m$year_min), as.integer(m$year_max)),
+      hint = sprintf("re-run with recipe = '%s'", rid)
+    )
+  }
+  suggestions
+}
+
+#' Emit the single cli::cli_inform() message summarizing all suggestions
+#' for a verb call (the brief's "one message", not one per suggestion).
+#' Bullet text is pre-formatted plain text (no cli/glue `{}` markup) since
+#' recipe ids/labels are untrusted-ish data values, not literal call-site
+#' expressions.
+#' @noRd
+.inform_suggestions <- function(suggestions) {
+  bullets <- vapply(suggestions, function(s) {
+    sprintf("%s (%d-%d): %s", s$recipe_id,
+            s$available_years[1], s$available_years[2], s$hint)
+  }, character(1))
+  cli::cli_inform(c(
+    i = "Coverage gap detected for the requested years; a harmonization recipe may fill it:",
+    stats::setNames(bullets, rep("*", length(bullets)))
+  ))
+}

R/views.R

@@ -1,11 +1,33 @@
 # R/views.R
 
+# SQL files whose view definitions read schema-v5-only parquet tables
+# (harmonization_map.parquet, harmonization_recipes.parquet,
+# series_breaks.parquet) or select from views built on top of them. DuckDB's
+# read_parquet() resolves the file at CREATE VIEW time (even for a view, it
+# still needs the source schema) and errors immediately -- "IO Error: No
+# files found" -- if the path doesn't exist, so these cannot be registered
+# unconditionally against a v4 corpus the way the rest of inst/sql/ is.
+# Registration is therefore gated on manifest$schema_version >= 5; verb-level
+# *usage* of the resulting views is separately gated by .resolve_basis() /
+# .require_schema_v5().
+.harmonization_view_files <- c(
+  "22-spending_long_harmonized.sql",
+  "23-revenue_long_harmonized.sql",
+  "33-harmonization_map.sql",
+  "34-harmonization_recipes.sql",
+  "35-series_breaks_pq.sql",
+  "42-spending_annotated_harmonized.sql",
+  "43-revenue_annotated_harmonized.sql"
+)
+
 #' Register DuckDB views from inst/sql/ SQL files
 #' @noRd
 .register_views <- function(con, url, manifest) {
   sql_dir <- system.file("sql", package = "uscogdata")
-  files   <- list.files(sql_dir, pattern = "\\.sql$", full.names = TRUE)
+  files   <- sort(list.files(sql_dir, pattern = "\\.sql$", full.names = TRUE))
+  schema_version <- suppressWarnings(as.integer(manifest$schema_version %||% 0L))
   for (f in files) {
+    if (basename(f) %in% .harmonization_view_files && schema_version < 5L) next
     sql <- paste(readLines(f, warn = FALSE), collapse = "\n")
     sql <- gsub("\\{url\\}", url, sql, fixed = FALSE)
     DBI::dbExecute(con, sql)

README.md

@@ -25,6 +25,22 @@ package implements.
 - `USCOGDATA_CACHE_DIR` — optional override for the manifest cache directory
 - `USCOGDATA_MANIFEST_TTL_SECS` — optional manifest re-fetch TTL (default 3600)
 
+## Raw-parquet caveat: `survey_weight` is not an aggregation weight
+
+Users reading the corpus parquet directly (DuckDB, arrow) will see a
+`survey_weight` column (schema v6, col 28). It is legacy Census IndFin
+sample-design **metadata passed through verbatim** — the Census Bureau's own
+source documentation says it "is for informational purposes only and should
+not be used to derive any other statistics" (`_ReadMe_First_IndFin.txt`;
+likewise `UserGuide.xls` Data User Note 8: "Do not use the weight field to
+derive state or national totals"). The raw encoding is also inconsistent
+across vintages (reciprocal scale most years, direct scale in 2003, a `1`
+placeholder in 1967/70/71/73/2001, all-`0` in 2007–2012, `NA` for all
+modern-source rows), so `sum(amt * survey_weight/10000)`-style expressions
+produce silently wrong totals — including exact zeros for 2007–2012. Sum
+`amt` unweighted; no uscogdata function reads this column. Full evidence:
+`cog_pipeline/.superpowers/sdd/weight-semantics-findings.md`.
+
 ## Developer notes
 
 ### Testing

data-raw/measure_signposting_rate.R

@@ -0,0 +1,563 @@
+# data-raw/measure_signposting_rate.R
+#
+# Phase R3 Task 19c: measures the harmonization-signposting suggestion rate
+# under THREE `.build_suggestions()` implementations, over a realistic query
+# battery:
+#   every summary_categories category
+#     x a 3-year pre/post-2012 span (the wide-aggregate -> modern-leaf
+#       format-boundary window; falls back to the widest span the corpus
+#       actually supports if it can't fill a full 3+3 design -- see
+#       .measure_year_span())
+#     x up to N_GOV sampled governments (seeded, deterministic)
+#
+# The three arms, oldest to newest:
+#   - "coarse"  (git ref b0df1ec, the merged R2 tip): a year counts as
+#     gapped only when the WHOLE category result has zero rows that year.
+#   - "selfcov" (git ref da72bf3, Task 19c's first per-code pass, since
+#     amended after review): per-code, but a component's gap could be
+#     satisfied by ANY component of the recipe INCLUDING ITSELF -- so a
+#     code whose only representation in a year was its own wide-era
+#     aggregate row satisfied its own coverage check. Flagged in review as
+#     not matching review-doc S: 0.3's literal criterion ("... has no rows
+#     ... but OTHER components do") and fixed in the next commit.
+#   - "percode" (live code): per-code, requiring a genuinely DIFFERENT
+#     sibling component to supply the covering evidence -- the shipped,
+#     corrected implementation.
+#
+# READ THIS BEFORE QUOTING ANY DELTA FROM THIS SCRIPT
+# -----------------------------------------------------
+# The coarse and per-code checks are PARTLY DISJOINT, not nested. Per-code
+# is NOT a strict widening of coarse: there are queries coarse fires on that
+# per-code does not, so moving coarse -> percode both ADDS and REMOVES
+# signposting. Every `*_delta_pp` figure this script reports -- overall and
+# per category -- is therefore a NET of those two flows and can mask a
+# coverage loss in either direction. A headline "+X pp" can sit on top of
+# categories that lost coverage outright (a NEGATIVE corrected_delta_pp),
+# and a category-level zero can be an add and a loss cancelling. Read
+# `$subset_relation` (printed under "Subset relation" below) alongside any
+# delta; that section is where the two flows are separated.
+#
+# The disjointness is structural, not a sampling artifact. Both arms pair a
+# gap test with a coverage test, and it is the COVERAGE test that differs:
+#   - coarse:  gap = the WHOLE category result has zero rows that year;
+#              covered = the recipe's generic join has ANY row that year
+#              (unioned across all components -- a component's own
+#              aggregate row counts).
+#   - percode: gap = one specific component has no non-aggregate row that
+#              year; covered = a DIFFERENT component of the SAME recipe has
+#              a row that year (self-coverage explicitly excluded, per
+#              review-doc S: 0.3's "...but OTHER components do").
+# So when a whole category is empty in a year -- exactly coarse's trigger --
+# and the only covering evidence is the gapped component's own wide-era
+# aggregate row, coarse fires and per-code CANNOT: there is by construction
+# no other component to supply the evidence. That case is already pinned as
+# intended behaviour in tests/testthat/test-recipes.R ("per-code gap does
+# NOT fire when a code's only coverage is its own aggregate row"). This
+# script's job is to say how often it costs coverage, not to relitigate it.
+#
+# This script MEASURES the deltas; it does not decide whether the resulting
+# signposting trade -- added "noise" in one direction, lost whole-category
+# gap coverage in the other -- is acceptable. That is Jared's ruling at
+# Checkpoint R3 (see
+# cog_pipeline/.superpowers/sdd/phase-r-task-19c-brief.md). The selfcov arm
+# exists purely to answer a narrower, mechanical question for that ruling:
+# how much of the coarse -> percode delta was ever attributable to the
+# self-coverage bug (selfcov -> percode), as opposed to genuine
+# other-component coverage (coarse -> percode directly)?
+#
+# All three arms no longer coexist in R/suggestions.R (each superseded the
+# last in place), so this script pulls each VERBATIM from git history and
+# evaluates it in an isolated environment parented on the uscogdata
+# namespace, so each still resolves the unchanged sibling helpers it
+# depends on (.sql_lit_chr()) exactly as the live package did at that
+# commit. This guarantees every non-live arm is the actual shipped code at
+# that point, not a hand-reconstruction that could silently drift from what
+# really shipped.
+#
+# Usage (from the uscogdata package root; a git checkout, not a tarball):
+#   Rscript data-raw/measure_signposting_rate.R
+#   USCOGDATA_URL=<staged-corpus-url> Rscript data-raw/measure_signposting_rate.R
+#
+# Or from R:
+#   source("data-raw/measure_signposting_rate.R")
+#   res <- measure_signposting_rate(corpus_url = "<url>")
+#   res$summary; res$by_category
+
+#' Pull a historical `.build_suggestions()` (and whatever helpers it uses)
+#' verbatim from git history and evaluate it in an isolated environment
+#' parented on the uscogdata namespace, so it resolves unchanged sibling
+#' helpers (`.sql_lit_chr()`) the same way the live package does.
+#' @noRd
+.measure_load_git_impl <- function(git_ref, git_path = "R/suggestions.R") {
+  old_src <- tryCatch(
+    system2("git", c("show", sprintf("%s:%s", git_ref, git_path)),
+            stdout = TRUE, stderr = TRUE),
+    error = function(e) NULL
+  )
+  status <- attr(old_src, "status")
+  if (is.null(old_src) || (!is.null(status) && status != 0L) ||
+      !any(grepl("^\\.build_suggestions", old_src))) {
+    stop(
+      "Could not retrieve the .build_suggestions() implementation from ",
+      "git ref '", git_ref, "' at '", git_path, "'. Run this script from ",
+      "inside the uscogdata git checkout (not a tarball/installed copy).",
+      call. = FALSE
+    )
+  }
+  env <- new.env(parent = asNamespace("uscogdata"))
+  # eval(parse()) here is safe: `old_src` is not external/untrusted input --
+  # it is this repo's OWN historical R/suggestions.R, fetched via `git show`
+  # from a fixed, hardcoded internal commit ref (overridable only by a
+  # caller who already has R-level code execution in this dev-only
+  # measurement script). No network or user-supplied data reaches this call.
+  eval(parse(text = old_src), envir = env)
+  stopifnot(is.function(env$.build_suggestions))
+  env
+}
+
+#' Resolve the query battery's year span: a `pre_n`-year window immediately
+#' before `boundary_year` unioned with a `post_n`-year window starting at
+#' `boundary_year` (default 3+3 around 2012, the wide-aggregate ->
+#' modern-leaf format boundary). Falls back to every distinct year the
+#' corpus actually has in `long` when it can't fill that full design, and
+#' says so explicitly in `$note` rather than silently padding or
+#' fabricating years.
+#' @noRd
+.measure_year_span <- function(con, boundary_year = 2012L,
+                                pre_n = 3L, post_n = 3L) {
+  available <- sort(as.integer(
+    DBI::dbGetQuery(con, "SELECT DISTINCT year FROM long")$year
+  ))
+  desired_pre  <- (boundary_year - pre_n):(boundary_year - 1L)
+  desired_post <- boundary_year:(boundary_year + post_n - 1L)
+  actual_pre   <- intersect(desired_pre, available)
+  actual_post  <- intersect(desired_post, available)
+  full_design  <- length(actual_pre) == pre_n && length(actual_post) == post_n
+
+  if (full_design) {
+    years <- sort(c(actual_pre, actual_post))
+    note <- sprintf(
+      "Full %d-year pre/%d-year post-%d design available -- using years: %s.",
+      pre_n, post_n, boundary_year, paste(years, collapse = ", ")
+    )
+  } else {
+    years <- available
+    note <- sprintf(paste(
+      "Corpus does NOT support a full %d-year pre/%d-year post-%d span",
+      "(desired pre-window %s -> only %s present; desired post-window %s",
+      "-> only %s present). Falling back to the WIDEST span this corpus",
+      "supports: all %d distinct year(s) actually in `long`: %s.",
+      "This is NOT a 3-year pre/post-%d design -- reported as measured,",
+      "not padded or fabricated."
+    ),
+    pre_n, post_n, boundary_year,
+    paste(desired_pre, collapse = ","),
+    if (length(actual_pre)) paste(actual_pre, collapse = ",") else "none",
+    paste(desired_post, collapse = ","),
+    if (length(actual_post)) paste(actual_post, collapse = ",") else "none",
+    length(available), paste(available, collapse = ", "),
+    boundary_year)
+  }
+  list(years = years, full_design = full_design, note = note,
+       available = available)
+}
+
+#' Deterministically sample up to `n` distinct governments that actually
+#' report *something* in the battery's year span (querying a government
+#' with zero presence in every measured year isn't a realistic query).
+#' @noRd
+.measure_sample_govids <- function(con, years, n = 20L, seed = 19L) {
+  pool <- DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT canonical_govid FROM long WHERE year IN (%s)
+     ORDER BY canonical_govid",
+    paste(years, collapse = ",")
+  ))$canonical_govid
+  if (length(pool) <= n) return(sort(pool))
+  set.seed(seed)
+  sort(sample(pool, n))
+}
+
+#' Comma-join a suggestion list's recipe ids (stable order) for the detail
+#' frame's audit columns; `""` when nothing fired.
+#' @noRd
+.measure_recipe_ids <- function(suggestions) {
+  if (length(suggestions) == 0L) return("")
+  paste(sort(vapply(suggestions, function(s) s$recipe_id, character(1))),
+        collapse = ",")
+}
+
+#' Run one (category, government) query through the coarse (`coarse_env`),
+#' self-coverage-allowed (`selfcov_env`), and live per-code
+#' `.build_suggestions()` and return a one-row summary of what each fired.
+#'
+#' Also records the coarse arm's OWN trigger evidence -- `coarse_gap_years`,
+#' the requested years in which the whole category result has zero rows --
+#' so a coarse-fired/per-code-silent disagreement can be named down to
+#' (category, government, year) instead of just counted. Per-code's gap
+#' years are deliberately NOT re-derived here: that would mean
+#' reimplementing `.recipe_component_gapped()`'s set arithmetic in the
+#' measurement harness, where it could silently drift from the code under
+#' measurement. Per-code rows are identified by the recipe ids they fired.
+#' @noRd
+.measure_one_query <- function(con, coarse_env, selfcov_env, category,
+                                category_type, govid, years) {
+  view        <- if (identical(category_type, "revenue")) {
+    "revenue_annotated_harmonized"
+  } else {
+    "spending_annotated_harmonized"
+  }
+  subtype_col <- if (identical(category_type, "revenue")) {
+    "revenue_subtype"
+  } else {
+    "spend_subtype"
+  }
+  sql    <- .build_verb_sql(view, subtype_col, govid, years, category)
+  result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+
+  coarse_sugg  <- coarse_env$.build_suggestions(
+    con, govid, years, category, result, "harmonized"
+  )
+  selfcov_sugg <- selfcov_env$.build_suggestions(
+    con, govid, years, category, "harmonized"
+  )
+  percode_sugg <- .build_suggestions(con, govid, years, category, "harmonized")
+
+  result_years <- if (nrow(result) == 0L) integer(0) else unique(as.integer(result$year))
+  gap_years    <- sort(setdiff(as.integer(years), result_years))
+
+  data.frame(
+    category         = category,
+    category_type    = category_type,
+    canonical_govid  = govid,
+    n_result_rows    = nrow(result),
+    coarse_gap_years = paste(gap_years, collapse = ","),
+    n_coarse         = length(coarse_sugg),
+    n_selfcov        = length(selfcov_sugg),
+    n_percode        = length(percode_sugg),
+    fired_coarse     = length(coarse_sugg) > 0L,
+    fired_selfcov    = length(selfcov_sugg) > 0L,
+    fired_percode    = length(percode_sugg) > 0L,
+    coarse_recipes   = .measure_recipe_ids(coarse_sugg),
+    percode_recipes  = .measure_recipe_ids(percode_sugg),
+    stringsAsFactors = FALSE
+  )
+}
+
+#' Columns that identify a disagreeing query well enough for a human to go
+#' and inspect it, in print order. Intersected with what `detail` actually
+#' has, so this works on a minimal hand-built frame too.
+#' @noRd
+.MEASURE_IDENTITY_COLS <- c(
+  "category", "category_type", "canonical_govid", "coarse_gap_years",
+  "n_result_rows", "coarse_recipes", "percode_recipes"
+)
+
+#' Separate the two flows the `*_delta_pp` figures net together.
+#'
+#' Per-code is NOT a widening of coarse (see this file's header): the two
+#' checks pair different gap tests with different coverage tests, so moving
+#' coarse -> percode both adds and removes firings. This splits the
+#' disagreement into:
+#'   - `violations`: coarse fired, per-code did NOT -- signposting coverage
+#'     LOST. These are what a net delta hides. `holds` is FALSE whenever
+#'     this is non-empty, i.e. whenever coarse is not a subset of per-code.
+#'   - `additions`: per-code fired, coarse did NOT -- the expected gain.
+#'
+#' Deliberately returns the offending rows, not just counts, so the
+#' Checkpoint R3 ruling can be made against named (category, government,
+#' year) cases. Deliberately does NOT assert -- the violation set is really
+#' non-empty on the staged corpus, and a hard assertion here would only
+#' break the harness that is supposed to report it.
+#' @noRd
+.measure_subset_relation <- function(detail) {
+  required <- c("category", "canonical_govid", "fired_coarse", "fired_percode")
+  absent   <- if (is.data.frame(detail)) setdiff(required, names(detail)) else required
+  if (!is.data.frame(detail) || length(absent) > 0L) {
+    stop("`detail` must be a data frame with columns ",
+         paste(required, collapse = ", "), " (missing: ",
+         paste(absent, collapse = ", "), ").", call. = FALSE)
+  }
+  fired_coarse  <- as.logical(detail$fired_coarse)
+  fired_percode <- as.logical(detail$fired_percode)
+  if (anyNA(fired_coarse) || anyNA(fired_percode)) {
+    stop("`fired_coarse`/`fired_percode` must be non-NA logicals.", call. = FALSE)
+  }
+
+  keep     <- intersect(.MEASURE_IDENTITY_COLS, names(detail))
+  viol_idx <- which(fired_coarse & !fired_percode)
+  add_idx  <- which(fired_percode & !fired_coarse)
+
+  list(
+    holds          = length(viol_idx) == 0L,
+    n_queries      = nrow(detail),
+    n_coarse_fired = sum(fired_coarse),
+    n_percode_fired = sum(fired_percode),
+    n_both         = sum(fired_coarse & fired_percode),
+    n_violations   = length(viol_idx),
+    n_additions    = length(add_idx),
+    violations     = detail[viol_idx, keep, drop = FALSE],
+    additions      = detail[add_idx, keep, drop = FALSE]
+  )
+}
+
+#' Render a data frame of disagreeing queries as indented report lines,
+#' capped at `max_rows` with an explicit note about what was withheld (the
+#' full set is always in the returned `$subset_relation`).
+#' @noRd
+.measure_fmt_rows <- function(df, max_rows = 50L) {
+  if (nrow(df) == 0L) return("    (none)")
+  shown <- utils::head(df, max_rows)
+  out   <- paste0("    ", utils::capture.output(print(shown, row.names = FALSE)))
+  if (nrow(df) > max_rows) {
+    out <- c(out, sprintf("    ... %d more row(s) not shown; full set in $subset_relation.",
+                          nrow(df) - max_rows))
+  }
+  out
+}
+
+#' Format `.measure_subset_relation()` as a prominent, clearly-labelled
+#' report section. States plainly whether coarse is a subset of per-code
+#' and, when it is not, exactly where it breaks.
+#' @noRd
+.measure_format_subset_report <- function(rel, max_rows = 50L) {
+  lines <- c(
+    "==== Subset relation: is COARSE a subset of PER-CODE? ====",
+    sprintf("Queries: %d  |  coarse fired: %d  |  per-code fired: %d  |  both: %d",
+            rel$n_queries, rel$n_coarse_fired, rel$n_percode_fired, rel$n_both)
+  )
+  if (rel$holds) {
+    lines <- c(lines, sprintf(paste(
+      "HOLDS: coarse IS a subset of per-code -- 0 of %d queries fire under",
+      "coarse but not per-code. On THIS battery the delta is a pure",
+      "addition of %d query/queries, with no coverage lost."
+    ), rel$n_queries, rel$n_additions))
+  } else {
+    lines <- c(lines,
+      "*** VIOLATED: coarse is NOT a subset of per-code. ***",
+      sprintf(paste(
+        "%d of %d queries fire under COARSE but NOT under PER-CODE:",
+        "signposting coverage the move LOSES."
+      ), rel$n_violations, rel$n_queries),
+      sprintf(paste(
+        "Every delta reported above is therefore a NET of %d addition(s)",
+        "MINUS %d loss(es), and understates both. Do not read it as",
+        "'per-code fires wherever coarse did, plus more'."
+      ), rel$n_additions, rel$n_violations),
+      "",
+      paste("  COVERAGE LOST -- coarse fired, per-code silent.",
+            "`coarse_gap_years` is the requested year(s) in which the whole",
+            "category result was empty (coarse's own trigger evidence):"),
+      .measure_fmt_rows(rel$violations, max_rows)
+    )
+  }
+  c(lines, "",
+    sprintf("  COVERAGE ADDED -- per-code fired, coarse silent (%d query/queries):",
+            rel$n_additions),
+    .measure_fmt_rows(rel$additions, max_rows))
+}
+
+#' Measure the coarse-vs-per-code signposting suggestion rate over a
+#' realistic query battery (every category x a pre/post-boundary_year span
+#' x up to n_gov sampled governments).
+#'
+#' @param corpus_url Corpus to measure against. Defaults to
+#'   `Sys.getenv("USCOGDATA_URL")`; if that's unset, falls back to the
+#'   bundled v5 fixture (so the script runs out of the box). Re-run with
+#'   `USCOGDATA_URL` pointed at the staged/full corpus later.
+#' @param n_gov Governments to sample (deterministically). "Up to" -- if
+#'   the corpus has fewer distinct governments in the measured years than
+#'   this, every one of them is used.
+#' @param seed Sampling seed (fixed for reproducibility).
+#' @param boundary_year,pre_years_n,post_years_n Define the desired query
+#'   span: `pre_years_n` years immediately before `boundary_year`, unioned
+#'   with `post_years_n` years starting at `boundary_year`. Falls back to
+#'   the corpus's widest actually-available span when this can't be filled
+#'   (see `.measure_year_span()`).
+#' @param coarse_ref Git ref to pull the R2 coarse `.build_suggestions()`
+#'   from.
+#' @param selfcov_ref Git ref to pull Task 19c's first, self-coverage-
+#'   allowed per-code `.build_suggestions()` from (amended after review).
+#' @param verbose Print progress/notes as the battery runs.
+#' @return Invisibly, a list with `corpus_url`, `years`, `span_note`,
+#'   `full_design`, `govids`, `seed`, `detail` (one row per query),
+#'   `by_category`, and `summary`.
+#' @noRd
+measure_signposting_rate <- function(corpus_url = Sys.getenv("USCOGDATA_URL", unset = NA),
+                                      n_gov = 20L,
+                                      seed = 19L,
+                                      boundary_year = 2012L,
+                                      pre_years_n = 3L,
+                                      post_years_n = 3L,
+                                      coarse_ref = "b0df1ec",
+                                      selfcov_ref = "da72bf3",
+                                      verbose = TRUE) {
+  pkgload::load_all(".", quiet = TRUE)
+
+  if (is.na(corpus_url) || !nzchar(corpus_url)) {
+    corpus_url <- paste0(
+      system.file("extdata/fixture_corpus", package = "uscogdata"), "/"
+    )
+    if (verbose) {
+      message("No USCOGDATA_URL set; defaulting to the bundled v5 fixture: ",
+              corpus_url)
+    }
+  }
+
+  old_url <- Sys.getenv("USCOGDATA_URL", unset = NA)
+  cog_close()
+  Sys.setenv(USCOGDATA_URL = corpus_url)
+  on.exit({
+    cog_close()
+    if (is.na(old_url)) Sys.unsetenv("USCOGDATA_URL") else Sys.setenv(USCOGDATA_URL = old_url)
+  }, add = TRUE)
+
+  con <- cog_open()
+  coarse_env  <- .measure_load_git_impl(git_ref = coarse_ref)
+  selfcov_env <- .measure_load_git_impl(git_ref = selfcov_ref)
+
+  span   <- .measure_year_span(con, boundary_year, pre_years_n, post_years_n)
+  years  <- span$years
+  if (verbose) message(span$note)
+
+  govids <- .measure_sample_govids(con, years, n = n_gov, seed = seed)
+  if (verbose) {
+    message(sprintf("Sampled %d government(s) (seed = %d) from %d present in years %s.",
+                     length(govids), seed,
+                     length(DBI::dbGetQuery(con, sprintf(
+                       "SELECT DISTINCT canonical_govid FROM long WHERE year IN (%s)",
+                       paste(years, collapse = ",")))$canonical_govid),
+                     paste(years, collapse = ", ")))
+  }
+
+  categories <- DBI::dbGetQuery(con,
+    "SELECT DISTINCT category, category_type FROM summary_categories
+     WHERE category IS NOT NULL ORDER BY category_type, category")
+  if (verbose) {
+    message(sprintf("Battery: %d categories x %d governments = %d queries.",
+                     nrow(categories), length(govids),
+                     nrow(categories) * length(govids)))
+  }
+
+  rows <- vector("list", nrow(categories) * length(govids))
+  k <- 0L
+  for (ci in seq_len(nrow(categories))) {
+    for (gv in govids) {
+      k <- k + 1L
+      rows[[k]] <- .measure_one_query(
+        con, coarse_env, selfcov_env,
+        category = categories$category[ci],
+        category_type = categories$category_type[ci],
+        govid = gv, years = years
+      )
+    }
+  }
+  detail <- do.call(rbind, rows)
+  # percode only ever fires where selfcov also fires (percode is a strict
+  # narrowing of selfcov: same gap detection, plus the self-coverage path
+  # removed) -- this is what makes the decomposition below exact rather
+  # than approximate. Checked, not assumed.
+  stopifnot(all(detail$fired_percode <= detail$fired_selfcov))
+  detail$fired_selfcov_only <- detail$fired_selfcov & !detail$fired_percode
+
+  # coarse vs percode is NOT a subset relation the way percode vs selfcov
+  # is (see header). Measured and REPORTED, never asserted: the violation
+  # set is genuinely non-empty on the staged corpus, and a stopifnot() here
+  # would break the harness whose whole job is to surface it.
+  subset_relation <- .measure_subset_relation(detail)
+  detail$coarse_only  <- detail$fired_coarse & !detail$fired_percode
+  detail$percode_only <- detail$fired_percode & !detail$fired_coarse
+
+  by_category <- dplyr::summarise(
+    dplyr::group_by(detail, category, category_type),
+    n_queries      = dplyr::n(),
+    coarse_rate    = mean(fired_coarse),
+    selfcov_rate   = mean(fired_selfcov),
+    percode_rate   = mean(fired_percode),
+    original_delta_pp  = (mean(fired_selfcov) - mean(fired_coarse)) * 100,
+    corrected_delta_pp = (mean(fired_percode) - mean(fired_coarse)) * 100,
+    selfcov_share_pp    = mean(fired_selfcov_only) * 100,
+    # the two flows corrected_delta_pp nets together, per category
+    n_coarse_only  = sum(coarse_only),
+    n_percode_only = sum(percode_only),
+    .groups = "drop"
+  )
+  by_category <- dplyr::arrange(by_category, dplyr::desc(corrected_delta_pp))
+
+  summary_overall <- data.frame(
+    n_queries     = nrow(detail),
+    n_categories  = nrow(categories),
+    n_governments = length(govids),
+    coarse_fired  = sum(detail$fired_coarse),
+    selfcov_fired = sum(detail$fired_selfcov),
+    percode_fired = sum(detail$fired_percode),
+    coarse_rate   = mean(detail$fired_coarse),
+    selfcov_rate  = mean(detail$fired_selfcov),
+    percode_rate  = mean(detail$fired_percode)
+  )
+  summary_overall$original_delta_pp  <- (summary_overall$selfcov_rate - summary_overall$coarse_rate) * 100
+  summary_overall$corrected_delta_pp <- (summary_overall$percode_rate - summary_overall$coarse_rate) * 100
+  summary_overall$selfcov_share_pp   <- mean(detail$fired_selfcov_only) * 100
+  summary_overall$relative_increase  <- if (summary_overall$coarse_rate > 0) {
+    summary_overall$percode_rate / summary_overall$coarse_rate - 1
+  } else {
+    NA_real_
+  }
+
+  if (verbose) {
+    message(sprintf(
+      "Coarse rate: %.4f (%d/%d)  |  Self-cov-allowed rate: %.4f (%d/%d)  |  Corrected per-code rate: %.4f (%d/%d)",
+      summary_overall$coarse_rate, summary_overall$coarse_fired, summary_overall$n_queries,
+      summary_overall$selfcov_rate, summary_overall$selfcov_fired, summary_overall$n_queries,
+      summary_overall$percode_rate, summary_overall$percode_fired, summary_overall$n_queries
+    ))
+    message(sprintf(
+      "Original delta (selfcov - coarse): %+.2f pp  |  Corrected delta (percode - coarse): %+.2f pp  |  Self-coverage share of original delta: %.2f pp (%d/%d queries fired ONLY via self-coverage)",
+      summary_overall$original_delta_pp, summary_overall$corrected_delta_pp,
+      summary_overall$selfcov_share_pp,
+      sum(detail$fired_selfcov_only), summary_overall$n_queries
+    ))
+    message(if (subset_relation$holds) {
+      sprintf("Subset relation coarse <= percode HOLDS (0 coarse-only firings); delta is a pure addition of %d.",
+              subset_relation$n_additions)
+    } else {
+      sprintf("*** Subset relation coarse <= percode VIOLATED: %d coarse-only firing(s) LOST vs %d percode-only added. Deltas above are NETS. ***",
+              subset_relation$n_violations, subset_relation$n_additions)
+    })
+  }
+
+  invisible(list(
+    corpus_url      = corpus_url,
+    years           = years,
+    span_note       = span$note,
+    full_design     = span$full_design,
+    govids          = govids,
+    seed            = seed,
+    n_categories    = nrow(categories),
+    detail          = detail,
+    by_category     = by_category,
+    summary         = summary_overall,
+    subset_relation = subset_relation
+  ))
+}
+
+if (identical(environment(), globalenv()) && sys.nframe() == 0L) {
+  res <- measure_signposting_rate()
+  cat("\n==== Query battery ====\n")
+  cat("Corpus:", res$corpus_url, "\n")
+  cat("Years:", paste(res$years, collapse = ", "), "\n")
+  cat("Full 3-year pre/post-2012 design achieved:", res$full_design, "\n")
+  cat(res$span_note, "\n")
+  cat("Governments sampled:", length(res$govids), "\n\n")
+
+  cat("==== Overall summary (coarse / self-coverage-allowed / corrected per-code) ====\n")
+  print(res$summary)
+
+  cat("\n==== By category (sorted by corrected delta, descending) ====\n")
+  cat("NOTE: corrected_delta_pp is a NET. n_coarse_only = firings LOST going\n")
+  cat("coarse -> percode; n_percode_only = firings ADDED. A category can be\n")
+  cat("negative (net coverage loss) even when the overall figure is positive.\n")
+  print(as.data.frame(res$by_category), row.names = FALSE)
+
+  cat("\n")
+  cat(paste(.measure_format_subset_report(res$subset_relation), collapse = "\n"), "\n")
+}

data-raw/regenerate_fixture_corpus.R

@@ -3,12 +3,20 @@
 # Regenerate inst/extdata/fixture_corpus/ from a cog_pipeline publish tree.
 #
 # What this does:
-#   1. Copies the year=2019 and year=2020 long partitions as-is (byte-for-
+#   1. Copies each requested year's long partition as-is (byte-for-byte)
-#      byte) from <publish_cache>/data/long/ into the fixture.
+#      from <publish_cache>/data/long/ into the fixture. Default years are
+#      c(2011L, 2012L, 2019L, 2020L): 2011/2012 straddle the wide-aggregate
+#      -> modern-leaf format boundary (the harmonization/recipe seam), and
+#      2019/2020 are the pre-existing per-capita/CPI regression anchors.
+#      Each partition is a full year (all states/govs) as published, so
+#      Broward County FL and every other previously-pinned government stay
+#      covered without any per-gov slicing logic.
 #   2. Copies the full canonical_fips_xwalk.parquet, canonical_alias.parquet,
-#      and summary_categories.parquet metadata tables as-is (these are small
+#      summary_categories.parquet, harmonization_map.parquet,
-#      cross-vintage registries, not partitioned by year, so the fixture
+#      harmonization_recipes.parquet, and series_breaks.parquet metadata
-#      ships the complete tables rather than a year-scoped subset).
+#      tables as-is (these are small cross-vintage registries, not
+#      partitioned by year, so the fixture ships the complete tables rather
+#      than a year-scoped subset).
 #   3. Resyncs the four reference docs (data_dictionary.md,
 #      reader-specification.md, README.md, series_breaks.md) from the
 #      publish tree's docs/.
@@ -35,7 +43,7 @@ regenerate_fixture_corpus <- function(
       "..", "cog_pipeline", "_targets", "publish_cache"
     ),
     fixture_dir = file.path("inst", "extdata", "fixture_corpus"),
-    fixture_years = c(2019L, 2020L)) {
+    fixture_years = c(2011L, 2012L, 2019L, 2020L)) {
   stopifnot(
     requireNamespace("digest", quietly = TRUE),
     requireNamespace("jsonlite", quietly = TRUE),
@@ -92,14 +100,18 @@ regenerate_fixture_corpus <- function(
   invisible(NULL)
 }
 
-# Copy the full (not year-scoped) canonical_fips_xwalk, canonical_alias, and
+# Copy the full (not year-scoped) canonical_fips_xwalk, canonical_alias,
-# summary_categories parquet tables.
+# summary_categories, and (schema v5+) harmonization_map/
+# harmonization_recipes/series_breaks parquet tables.
 #' @noRd
 .copy_metadata_parquets <- function(publish_cache_dir, fixture_dir) {
   files <- c(
     "canonical_fips_xwalk.parquet",
     "canonical_alias.parquet",
-    "summary_categories.parquet"
+    "summary_categories.parquet",
+    "harmonization_map.parquet",
+    "harmonization_recipes.parquet",
+    "series_breaks.parquet"
   )
   for (f in files) {
     src <- file.path(publish_cache_dir, "data", f)
@@ -170,7 +182,10 @@ regenerate_fixture_corpus <- function(
   metadata_files <- c(
     "canonical_alias.parquet",
     "canonical_fips_xwalk.parquet",
-    "summary_categories.parquet"
+    "summary_categories.parquet",
+    "harmonization_map.parquet",
+    "harmonization_recipes.parquet",
+    "series_breaks.parquet"
   )
   metadata <- lapply(metadata_files, function(f) {
     rel <- file.path("data", f)
@@ -187,11 +202,15 @@ regenerate_fixture_corpus <- function(
     built_at       = format(Sys.time(), "%Y-%m-%dT%H:%M:%SZ", tz = "UTC"),
     pipeline_commit = source_manifest$pipeline_commit,
     fixture_note = paste(
-      "Two-year (2019-2020) fixture for uscogdata tests. Full corpus",
+      "Four-year (2011, 2012, 2019, 2020) fixture for uscogdata tests. Full",
-      "available via USCOGDATA_URL. Regenerated for Phase P",
+      "corpus available via USCOGDATA_URL. Regenerated for Phase R2",
-      "(schema_version 4, uniformly 12-char canonical_govid) with the full",
+      "(schema_version 5, harmonization_map/harmonization_recipes/",
-      "canonical_fips_xwalk master and the new canonical_alias lookup",
+      "series_breaks parquet tables added). 2011/2012 straddle the",
-      "table via data-raw/regenerate_fixture_corpus.R."
+      "wide-aggregate -> modern-leaf format boundary exercised by basis=",
+      "\"harmonized\" and recipe= queries; 2019/2020 retain the prior",
+      "per-capita/CPI regression anchors. Full canonical_fips_xwalk master",
+      "and canonical_alias lookup table included via",
+      "data-raw/regenerate_fixture_corpus.R."
     ),
     data_vintage    = source_manifest$data_vintage,
     scope           = source_manifest$scope,

inst/extdata/fixture_corpus/manifest.json

@@ -1,11 +1,24 @@
 {
-  "schema_version": 4,
+  "schema_version": 6,
-  "built_at": "2026-07-13T23:35:09Z",
+  "built_at": "2026-07-23T16:14:30Z",
-  "pipeline_commit": "a082b26",
+  "pipeline_commit": "4f992a0",
-  "fixture_note": "Two-year (2019-2020) fixture for uscogdata tests. Full corpus available via USCOGDATA_URL. Regenerated for Phase P (schema_version 4, uniformly 12-char canonical_govid) with the full canonical_fips_xwalk master and the new canonical_alias lookup table via data-raw/regenerate_fixture_corpus.R.",
+  "fixture_note": "Four-year (2011, 2012, 2019, 2020) fixture for uscogdata tests. Full corpus available via USCOGDATA_URL. Regenerated for Phase R2 (schema_version 5, harmonization_map/harmonization_recipes/ series_breaks parquet tables added). 2011/2012 straddle the wide-aggregate -> modern-leaf format boundary exercised by basis= \"harmonized\" and recipe= queries; 2019/2020 retain the prior per-capita/CPI regression anchors. Full canonical_fips_xwalk master and canonical_alias lookup table included via data-raw/regenerate_fixture_corpus.R.",
   "data_vintage": {
-    "census_source_downloaded": "unknown",
+    "source_vintages": {
-    "cpi_vintage": "FRED CPIAUCSL",
+      "2012": "10162019",
+      "2013": "10162019",
+      "2014": "10162019",
+      "2015": "10162019",
+      "2016": "10162019",
+      "2017": "06102021",
+      "2018": "06102021",
+      "2019": "06102021",
+      "2020": "06122023",
+      "2021": "06122023",
+      "2022": "06052025",
+      "2023": "06052025"
+    },
+    "registry_rows": 148,
     "acs_vintage": "ACS 2018-2022 5-year"
   },
   "scope": {
@@ -14,42 +27,71 @@
     "scope_note": "v0.1 covers state, county, city/municipality, and township governments. Special districts (type 4) and school districts (type 5) are excluded pending validation in a future cycle."
   },
   "schema": {
-    "long_column_count": 24,
+    "long_column_count": 28,
-    "long_columns": ["fips_state", "type", "fips_county", "govid", "gov_blank", "gov_name", "county_name", "fips_state_code", "fips_county_code", "fips_place_code", "population", "popyear", "enrollment", "enrollyear", "function_code", "sch_level_code", "fiscal_year_end", "srvy_year", "item_code", "amt", "srv_data", "impute_flag", "is_aggregate", "canonical_govid"],
+    "long_columns": ["fips_state", "type", "fips_county", "govid", "gov_blank", "gov_name", "county_name", "fips_state_asof", "fips_county_asof", "cog_legacy_state", "cog_legacy_county", "fips_place_code", "population", "popyear", "enrollment", "enrollyear", "function_code", "sch_level_code", "fiscal_year_end", "srvy_year", "item_code", "amt", "srv_data", "impute_flag", "is_aggregate", "canonical_govid", "harmonized_code", "survey_weight"],
     "data_dictionary": "docs/data_dictionary.md"
   },
   "files": {
     "long_partitions": [
+      {
+        "year": 2011,
+        "path": "data/long/year=2011/part-0.parquet",
+        "sha256": "84302ab364dc9fc3b3fbbc3c3f8b826e3508b4d73ff7c42d094d3863cd1e37b5",
+        "row_count": 2864212,
+        "size_bytes": 3845911
+      },
+      {
+        "year": 2012,
+        "path": "data/long/year=2012/part-0.parquet",
+        "sha256": "b82ac82d5e35f844b26c887445601f3748438c52c998ba4e403b025941a6f170",
+        "row_count": 1163338,
+        "size_bytes": 5929917
+      },
       {
         "year": 2019,
         "path": "data/long/year=2019/part-0.parquet",
-        "sha256": "c0a2bf0758af129d5dfddb6ff6665cc435ddee87fd6879e788fb56ed53ab22b8",
+        "sha256": "5cbd4726dcc7d0dab5c2a05a64702e979533ae119ed0587073cd31c089e0d737",
         "row_count": 318139,
-        "size_bytes": 1441404
+        "size_bytes": 1719548
       },
       {
         "year": 2020,
         "path": "data/long/year=2020/part-0.parquet",
-        "sha256": "92570b9d55ec3425d034db37838f91c3b8359d0454d3d98730a6016b62e4bb48",
+        "sha256": "ee548fec80bf1beda844fe03916ac145f10dd34c45968407cc330ec260935f00",
         "row_count": 317500,
-        "size_bytes": 1444011
+        "size_bytes": 1722918
       }
     ],
     "metadata": [
       {
         "path": "data/canonical_alias.parquet",
-        "sha256": "feb8d01a640fb16c9a4b4ad66726b50b8fe8a1ce2a771bed8c1190fec51d5c8d",
+        "sha256": "3f617051c23a99bea322889857f7106df0c92954564afeec181df7083ee6698e",
         "description": "canonical_alias.parquet"
       },
       {
         "path": "data/canonical_fips_xwalk.parquet",
-        "sha256": "1ae47981531c7389f69eff3f7656045428564bfbe8032200eb9c039d32739a7e",
+        "sha256": "f98742f941269dacf8f7de5c273aa4dd4e75017a5bb70c054da35852a95a8d46",
         "description": "canonical_fips_xwalk.parquet"
       },
       {
         "path": "data/summary_categories.parquet",
-        "sha256": "dd59e7f58a022679ad43511c8c8e938b8dd4be81196bbeeee21e67bdcca2295b",
+        "sha256": "8e6fcd4dd9bb4723841a67233b19388c9762dfc23b4479501183cebf7ea3c1b5",
         "description": "summary_categories.parquet"
+      },
+      {
+        "path": "data/harmonization_map.parquet",
+        "sha256": "4cf32d0f817079ba4f28dc0ce65450d3247ebbf08d94c0c26c0d02af597bf812",
+        "description": "harmonization_map.parquet"
+      },
+      {
+        "path": "data/harmonization_recipes.parquet",
+        "sha256": "1133e9a0b02f8f34f5f936e55c5ecd596bb8a55d8425dcce76767f0f3203581c",
+        "description": "harmonization_recipes.parquet"
+      },
+      {
+        "path": "data/series_breaks.parquet",
+        "sha256": "b0b6794b6887a4f300079adfa10029c2a77109faa4952fbff1c5a270793cc02b",
+        "description": "series_breaks.parquet"
       }
     ]
   },

inst/schemas/provenance-v1.json

@@ -10,6 +10,11 @@
     "target":         { "type": "object" },
     "years":          { "type": "array", "items": { "type": "integer" } },
     "category":       { "type": ["string", "array", "null"] },
+    "basis":          { "type": ["string", "null"] },
+    "basis_note":     { "type": ["string", "null"] },
+    "harmonization":  { "type": "object" },
+    "recipe":         { "type": ["object", "null"] },
+    "suggestions":    { "type": "array" },
     "scope":          { "type": "object" },
     "codes_summed":   { "type": "object" },
     "aggregate_fallback": { "type": ["object", "null"] },

inst/sql/22-spending_long_harmonized.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW spending_long_harmonized AS
+SELECT * REPLACE (harmonized_code AS item_code)
+FROM long
+WHERE NOT is_aggregate
+  AND harmonized_code IS NOT NULL
+  AND LEFT(harmonized_code, 1) IN ('E', 'F', 'G', 'K');

inst/sql/23-revenue_long_harmonized.sql

@@ -0,0 +1,6 @@
+CREATE OR REPLACE VIEW revenue_long_harmonized AS
+SELECT * REPLACE (harmonized_code AS item_code)
+FROM long
+WHERE NOT is_aggregate
+  AND harmonized_code IS NOT NULL
+  AND LEFT(harmonized_code, 1) IN ('T', 'A', 'U', 'B', 'C', 'D');

inst/sql/33-harmonization_map.sql

@@ -0,0 +1,3 @@
+CREATE OR REPLACE VIEW harmonization_map AS
+SELECT *
+FROM read_parquet('{url}data/harmonization_map.parquet');

inst/sql/34-harmonization_recipes.sql

@@ -0,0 +1,3 @@
+CREATE OR REPLACE VIEW harmonization_recipes AS
+SELECT *
+FROM read_parquet('{url}data/harmonization_recipes.parquet');

inst/sql/35-series_breaks_pq.sql

@@ -0,0 +1,3 @@
+CREATE OR REPLACE VIEW series_breaks_pq AS
+SELECT *
+FROM read_parquet('{url}data/series_breaks.parquet');

inst/sql/42-spending_annotated_harmonized.sql

@@ -0,0 +1,16 @@
+CREATE OR REPLACE VIEW spending_annotated_harmonized AS
+SELECT
+  s.*,
+  x.gov_name       AS xwalk_gov_name,
+  x.govs_type,
+  x.type_label,
+  x.fips_state     AS xwalk_fips_state,
+  x.fips_county    AS xwalk_fips_county,
+  x.fips_place,
+  x.population_acs,
+  c.category,
+  c.category_type,
+  c.spend_subtype
+FROM spending_long_harmonized s
+LEFT JOIN canonical_fips_xwalk x USING (canonical_govid)
+LEFT JOIN summary_categories   c USING (item_code);

inst/sql/43-revenue_annotated_harmonized.sql

@@ -0,0 +1,16 @@
+CREATE OR REPLACE VIEW revenue_annotated_harmonized AS
+SELECT
+  s.*,
+  x.gov_name       AS xwalk_gov_name,
+  x.govs_type,
+  x.type_label,
+  x.fips_state     AS xwalk_fips_state,
+  x.fips_county    AS xwalk_fips_county,
+  x.fips_place,
+  x.population_acs,
+  c.category,
+  c.category_type,
+  c.revenue_subtype
+FROM revenue_long_harmonized s
+LEFT JOIN canonical_fips_xwalk x USING (canonical_govid)
+LEFT JOIN summary_categories   c USING (item_code);

man/cog_recipes.Rd

@@ -0,0 +1,22 @@
+% Generated by roxygen2: do not edit by hand
+% Please edit documentation in R/recipes.R
+\name{cog_recipes}
+\alias{cog_recipes}
+\title{List available harmonization recipes}
+\usage{
+cog_recipes(pattern = NULL)
+}
+\arguments{
+\item{pattern}{Optional regex matched case-insensitively against
+`recipe_id` or `label`.}
+}
+\value{
+Tibble with columns `recipe_id`, `label`, `n_components`,
+  `year_min`, `year_max` (the min/max component year coverage), sorted by
+  `recipe_id`.
+}
+\description{
+Recipes are multi-code cross-vintage series (see [cog_spending()]'s
+`recipe` argument) catalogued in the corpus's `harmonization_recipes`
+table. Use this to discover valid `recipe` ids.
+}

man/cog_revenue.Rd

@@ -9,7 +9,9 @@ cog_revenue(
   years,
   category = NULL,
   per_capita = FALSE,
-  adjust_to_year = NULL
+  adjust_to_year = NULL,
+  basis = c("harmonized", "raw"),
+  recipe = NULL
 )
 }
 \arguments{
@@ -29,6 +31,30 @@ in that year).}
 
 \item{adjust_to_year}{Integer base year for CPI-U real-dollar conversion,
 or `NULL` for nominal only.}
+
+\item{basis}{`"harmonized"` (default) sums item codes through the
+cross-vintage harmonization mapping (folding series-break-affected
+codes onto a comparable target and excluding aggregate / discontinued
+rows -- see the `harmonization` block in `cog_explain()`); `"raw"`
+reproduces the pre-Phase-R2 behavior (published item codes, no
+folding). On a corpus with `schema_version < 5` (no harmonization
+tables), `basis` silently resolves to `"raw"` when left at its default
+and the resolution is recorded in the provenance; explicitly passing
+`basis = "harmonized"` on such a corpus aborts. Ignored when `recipe`
+is set (see below).}
+
+\item{recipe}{Optional harmonization recipe id (see [cog_recipes()]) for
+multi-code cross-vintage series that a 1:1 harmonized_code mapping
+can't express (e.g. a wide-era aggregate that only splits into leaf
+codes in the modern era). Mutually exclusive with `category`. The
+result's subtype column reads `"recipe"` and `category` reads the
+recipe's label. Requires `schema_version >= 5`. A recipe query bypasses
+`basis` entirely (it joins `long` directly rather than going through
+the `*_annotated`/`*_annotated_harmonized` views), so the `basis`
+argument is ignored and the result's provenance reports
+`basis = "recipe"` with an inert `harmonization` block (`applied =
+FALSE`, pointing at the `recipe` block instead) rather than a
+possibly-misleading `"harmonized"`/`"raw"` value.}
 }
 \value{
 Tibble with columns `year`, `canonical_govid`, `gov_name`,

man/cog_spending.Rd

@@ -9,7 +9,9 @@ cog_spending(
   years,
   category = NULL,
   per_capita = FALSE,
-  adjust_to_year = NULL
+  adjust_to_year = NULL,
+  basis = c("harmonized", "raw"),
+  recipe = NULL
 )
 }
 \arguments{
@@ -29,6 +31,30 @@ in that year).}
 
 \item{adjust_to_year}{Integer base year for CPI-U real-dollar conversion,
 or `NULL` for nominal only.}
+
+\item{basis}{`"harmonized"` (default) sums item codes through the
+cross-vintage harmonization mapping (folding series-break-affected
+codes onto a comparable target and excluding aggregate / discontinued
+rows -- see the `harmonization` block in `cog_explain()`); `"raw"`
+reproduces the pre-Phase-R2 behavior (published item codes, no
+folding). On a corpus with `schema_version < 5` (no harmonization
+tables), `basis` silently resolves to `"raw"` when left at its default
+and the resolution is recorded in the provenance; explicitly passing
+`basis = "harmonized"` on such a corpus aborts. Ignored when `recipe`
+is set (see below).}
+
+\item{recipe}{Optional harmonization recipe id (see [cog_recipes()]) for
+multi-code cross-vintage series that a 1:1 harmonized_code mapping
+can't express (e.g. a wide-era aggregate that only splits into leaf
+codes in the modern era). Mutually exclusive with `category`. The
+result's subtype column reads `"recipe"` and `category` reads the
+recipe's label. Requires `schema_version >= 5`. A recipe query bypasses
+`basis` entirely (it joins `long` directly rather than going through
+the `*_annotated`/`*_annotated_harmonized` views), so the `basis`
+argument is ignored and the result's provenance reports
+`basis = "recipe"` with an inert `harmonization` block (`applied =
+FALSE`, pointing at the `recipe` block instead) rather than a
+possibly-misleading `"harmonized"`/`"raw"` value.}
 }
 \value{
 Tibble with columns `year`, `canonical_govid`, `gov_name`,

tests/testthat/helper-fixture.R

@@ -26,3 +26,34 @@ with_fixture_corpus <- function(code) {
   }, add = TRUE)
   force(code)
 }
+
+# Copy the bundled fixture to a temp dir with manifest.json's schema_version
+# patched to `version`, then run `code` against it with a clean session
+# (mirrors with_fixture_corpus()). Used to exercise the v4/v5 dual-accept
+# path without a second physical fixture tree: a real v4 corpus has no
+# harmonization_map/harmonization_recipes/series_breaks parquet files, but
+# .register_views() only *reads* those when schema_version >= 5 (see
+# R/views.R), so a doctored copy of the (v5) bundled fixture with the
+# manifest's schema_version knocked down to 4 is a faithful stand-in.
+with_doctored_schema_version <- function(version, code) {
+  src <- fixture_corpus_path()
+  tmp <- withr::local_tempdir(.local_envir = parent.frame())
+  file.copy(list.files(src, full.names = TRUE), tmp, recursive = TRUE)
+
+  manifest_path <- file.path(tmp, "manifest.json")
+  m <- jsonlite::fromJSON(manifest_path, simplifyVector = FALSE)
+  m$schema_version <- as.integer(version)
+  writeLines(
+    jsonlite::toJSON(m, auto_unbox = TRUE, pretty = TRUE, null = "null"),
+    manifest_path
+  )
+
+  old_url <- Sys.getenv("USCOGDATA_URL", unset = NA)
+  uscogdata:::cog_close()
+  Sys.setenv(USCOGDATA_URL = paste0(tmp, "/"))
+  on.exit({
+    uscogdata:::cog_close()
+    if (is.na(old_url)) Sys.unsetenv("USCOGDATA_URL") else Sys.setenv(USCOGDATA_URL = old_url)
+  }, add = TRUE)
+  force(code)
+}

tests/testthat/test-explain.R

@@ -31,6 +31,45 @@ test_that("cog_explain errors on non-verb input", {
   expect_error(cog_explain(df), "provenance")
 })
 
+test_that("cog_explain prints basis + harmonization block", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", 2020L, "Corrections")
+  txt <- paste(c(
+    capture.output(cog_explain(r)),
+    capture.output(cog_explain(r), type = "message")
+  ), collapse = "\n")
+  expect_true(grepl("Basis: harmonized", txt))
+  expect_true(grepl("Harmonization", txt))
+  expect_true(grepl("Excluded 0 row", txt))
+})
+
+test_that("cog_explain prints a Recipe section for recipe = results", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", c(2011L, 2012L), recipe = "corrections_combined")
+  txt <- paste(c(
+    capture.output(cog_explain(r)),
+    capture.output(cog_explain(r), type = "message")
+  ), collapse = "\n")
+  expect_true(grepl("Recipe", txt))
+  expect_true(grepl("corrections_combined", txt))
+  expect_true(grepl("E04", txt))
+  expect_true(grepl("E05", txt))
+})
+
+test_that("cog_explain prints a Suggestions section when the provenance has one", {
+  skip_if_no_corpus()
+  r <- suppressMessages(
+    cog_spending("121011212191", c(2011L, 2012L), category = "Corrections")
+  )
+  txt <- paste(c(
+    capture.output(cog_explain(r)),
+    capture.output(cog_explain(r), type = "message")
+  ), collapse = "\n")
+  expect_true(grepl("Suggestions", txt))
+  expect_true(grepl("corrections_combined", txt))
+  expect_true(grepl("re-run with recipe", txt))
+})
+
 test_that("cog_explain prints denominator + popyear_range + counts", {
   skip_if_no_corpus()
   with_fixture_corpus({

tests/testthat/test-manifest.R

@@ -107,6 +107,49 @@ test_that("cog_manifest returns the active session's parsed manifest", {
     expect_true(m$schema_version >= 4L)
     yrs <- vapply(m$files$long_partitions, function(p) as.integer(p$year),
                   integer(1))
-    expect_setequal(yrs, c(2019L, 2020L))
+    expect_setequal(yrs, c(2011L, 2012L, 2019L, 2020L))
+  })
+})
+
+test_that(".validate_schema accepts schema_version 4, 5 and 6, rejects others", {
+  expect_silent(uscogdata:::.validate_schema(list(schema_version = 4L)))
+  expect_silent(uscogdata:::.validate_schema(list(schema_version = 5L)))
+  # v6 = FIPS geography harmonization (2026-07-22): _code -> _asof rename +
+  # cog_legacy_* columns (26 -> 28 cols). This package references none of the
+  # renamed columns and its geography comes from the xwalk, so v6 is accepted
+  # without behavioural change -- see .validate_schema()'s note.
+  expect_silent(uscogdata:::.validate_schema(list(schema_version = 6L)))
+  expect_error(
+    uscogdata:::.validate_schema(list(schema_version = 3L)),
+    "schema_version"
+  )
+  expect_error(
+    uscogdata:::.validate_schema(list(schema_version = 7L)),
+    "schema_version"
+  )
+})
+
+test_that("cog_open succeeds against a doctored schema_version 4 corpus (dual-accept)", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    con <- cog_open()
+    expect_true(DBI::dbIsValid(con))
+    expect_equal(as.integer(cog_manifest()$schema_version), 4L)
+
+    # Core (pre-Phase-R2) views must still register on a v4 corpus.
+    views <- DBI::dbGetQuery(con,
+      "SELECT table_name FROM information_schema.tables
+       WHERE table_schema = 'main' AND table_type = 'VIEW'"
+    )$table_name
+    expect_true(all(c("spending_annotated", "revenue_annotated") %in% views))
+
+    # Schema-v5-only harmonization views must NOT register on a v4 corpus:
+    # their parquet sources don't exist there and DuckDB's read_parquet()
+    # errors eagerly at CREATE VIEW time for a missing file/glob, so
+    # .register_views() gates these on manifest$schema_version >= 5.
+    expect_false(any(c(
+      "spending_long_harmonized", "spending_annotated_harmonized",
+      "harmonization_recipes", "harmonization_map", "series_breaks_pq"
+    ) %in% views))
   })
 })

tests/testthat/test-recipes.R

@@ -0,0 +1,312 @@
+# tests/testthat/test-recipes.R
+#
+# cog_recipes(), recipe = in cog_spending()/cog_revenue(), and the
+# recipe-component-driven signposting in prov$suggestions (Phase R2 /
+# Task 11, schema_version 5).
+
+test_that("cog_recipes lists the curated catalog including corrections_combined", {
+  skip_if_no_corpus()
+  r <- cog_recipes()
+  expect_s3_class(r, "tbl_df")
+  expect_equal(names(r), c("recipe_id", "label", "n_components", "year_min", "year_max"))
+  expect_equal(nrow(r), 24L)
+  expect_true("corrections_combined" %in% r$recipe_id)
+  expect_true("t19_selective_sales_wide" %in% r$recipe_id)
+  expect_true("ig_federal_b89_wide" %in% r$recipe_id)
+  expect_true("rents_royalties_u4_wide" %in% r$recipe_id)
+  expect_true("higher_ed_e18_wide" %in% r$recipe_id)
+  expect_true("cash_securities_z77_wide" %in% r$recipe_id)
+  # Superseded id from the pre-curation brief text must NOT be present.
+  expect_false("corrections_judicial_combined" %in% r$recipe_id)
+})
+
+test_that("cog_recipes(pattern=) filters by recipe_id or label", {
+  skip_if_no_corpus()
+  r <- cog_recipes("corrections")
+  expect_true(nrow(r) >= 1L)
+  expect_true(all(grepl("corrections", r$recipe_id, ignore.case = TRUE) |
+                    grepl("corrections", r$label, ignore.case = TRUE)))
+})
+
+test_that("cog_recipes requires schema_version >= 5", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    expect_error(cog_recipes(), class = "uscogdata_schema_unsupported")
+  })
+})
+
+# --- recipe = : generic join, no is_aggregate filter -----------------------
+
+test_that("recipe = 'corrections_combined' is continuous across the 2011->2012 seam", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", years = c(2011L, 2012L),
+                    recipe = "corrections_combined")
+  expect_equal(nrow(r), 2L)
+  expect_true(all(c("year", "canonical_govid", "gov_name", "spend_subtype",
+                    "category", "amt_nominal", "codes_included",
+                    "aggregate_fallback", "notes") %in% names(r)))
+  expect_equal(unique(r$spend_subtype), "recipe")
+  expect_equal(unique(r$category), "Corrections (functions 04+05 combined)")
+  expect_false(any(r$aggregate_fallback))
+
+  r2011 <- r$amt_nominal[r$year == 2011L]
+  r2012 <- r$amt_nominal[r$year == 2012L]
+  # 2011: E05 only exists as a wide-era AGGREGATE row (is_aggregate = TRUE)
+  # for Broward -- data-verified $216,088,000. Since .run_recipe() does NOT
+  # filter is_aggregate (amendment: the recipe join must not, because these
+  # families exist ONLY as aggregate rows in the wide era), the recipe
+  # correctly picks this up.
+  expect_equal(r2011, 216088000)
+  # 2012: modern E04 leaf ($213,056,000); Broward reports no E05 leaf that
+  # year, so the recipe total equals E04 alone -- still continuous with the
+  # 2011 aggregate, proving the wide-aggregate -> modern-leaf handoff.
+  expect_equal(r2012, 213056000)
+  expect_true(all(grepl("E04|E05", r$codes_included)))
+})
+
+test_that("recipe result carries a recipe provenance block with component rows", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", years = c(2011L, 2012L),
+                    recipe = "corrections_combined")
+  prov <- attr(r, "provenance")
+  expect_equal(prov$basis, "recipe")
+  expect_equal(prov$category, "Corrections (functions 04+05 combined)")
+  expect_type(prov$recipe, "list")
+  expect_equal(prov$recipe$recipe_id, "corrections_combined")
+  expect_equal(prov$recipe$label, "Corrections (functions 04+05 combined)")
+  expect_length(prov$recipe$components, 2L)
+  comp_codes <- vapply(prov$recipe$components, function(x) x$component_code, character(1))
+  expect_setequal(comp_codes, c("E04", "E05"))
+  # A recipe query resolves its own coverage; it should never also carry
+  # suggestions for itself.
+  expect_length(prov$suggestions, 0L)
+})
+
+test_that("recipe results report an unambiguous basis/harmonization, ignoring basis=", {
+  skip_if_no_corpus()
+  # A recipe query bypasses spending_annotated(_harmonized) entirely --
+  # .run_recipe() joins `long` directly -- so `basis` must never read
+  # "harmonized"/"raw" (which would describe a code path this query never
+  # took) regardless of what the caller passed for `basis`. Task 12
+  # consumes provenance verbatim, so this needs to be unambiguous.
+  r_default <- cog_spending("121011212191", years = c(2011L, 2012L),
+                            recipe = "corrections_combined")
+  r_raw <- cog_spending("121011212191", years = c(2011L, 2012L),
+                        recipe = "corrections_combined", basis = "raw")
+  r_harm <- cog_spending("121011212191", years = c(2011L, 2012L),
+                         recipe = "corrections_combined", basis = "harmonized")
+
+  for (r in list(r_default, r_raw, r_harm)) {
+    prov <- attr(r, "provenance")
+    expect_equal(prov$basis, "recipe")
+    expect_true(is.na(prov$basis_note))
+    expect_false(prov$harmonization$applied)
+    expect_equal(prov$harmonization$na_rows_excluded, 0L)
+    expect_match(prov$harmonization$note, "recipe", ignore.case = TRUE)
+  }
+
+  # basis= truly has zero effect on a recipe query's actual numbers.
+  expect_equal(r_raw$amt_nominal, r_harm$amt_nominal)
+  expect_equal(r_default$amt_nominal, r_raw$amt_nominal)
+})
+
+test_that("recipe = 't19_selective_sales_wide' sums the local T11/T14 legs when present", {
+  skip_if_no_corpus()
+  # Westminster City, CA (canonical_govid 082001211654): T11 = 0 in 2011,
+  # T11 = 568 (T14 = 0/absent) in 2012 -- a real, data-verified equality/
+  # inequality pair inside the amended fixture window (2011-2012), standing
+  # in for the brief's original 2004/2005 example (out of scope per the
+  # amended fixture years; the underlying local-tax-split boundary is
+  # nationally FY2005, but this government's own T11 reporting activates
+  # within our 2011-2012 window).
+  r <- cog_revenue("082001211654", years = c(2011L, 2012L),
+                   recipe = "t19_selective_sales_wide")
+
+  # Raw, single-code T19 total (not the "Other Taxes" category total, which
+  # would also sum in T11/T14/T21/T23/T27/T29/T53/T99 -- queried directly to
+  # isolate exactly the code the brief's equality/inequality check is about).
+  con <- uscogdata:::.ensure_session()
+  raw_t19 <- DBI::dbGetQuery(con, "
+    SELECT year, SUM(amt) * 1000.0 AS amt
+    FROM revenue_long
+    WHERE canonical_govid = '082001211654' AND item_code = 'T19'
+      AND year IN (2011, 2012)
+    GROUP BY year ORDER BY year
+  ")
+  raw_t19_2011 <- raw_t19$amt[raw_t19$year == 2011L]
+  raw_t19_2012 <- raw_t19$amt[raw_t19$year == 2012L]
+  expect_equal(raw_t19_2011, 2231000)
+  expect_equal(raw_t19_2012, 2365000)
+
+  recipe_2011 <- r$amt_nominal[r$year == 2011L]
+  recipe_2012 <- r$amt_nominal[r$year == 2012L]
+
+  expect_equal(recipe_2011, raw_t19_2011)       # equality: no local T11/T14 yet
+  expect_gt(recipe_2012, raw_t19_2012)          # inequality: local T11 joins in
+  expect_equal(recipe_2012, raw_t19_2012 + 568000)
+})
+
+test_that("recipe = and category = together aborts", {
+  skip_if_no_corpus()
+  expect_error(
+    cog_spending("121011212191", 2020L, category = "Corrections",
+                recipe = "corrections_combined"),
+    class = "uscogdata_recipe_category_conflict"
+  )
+})
+
+test_that("unknown recipe id aborts and lists valid ids", {
+  skip_if_no_corpus()
+  err <- tryCatch(
+    cog_spending("121011212191", 2020L, recipe = "does_not_exist"),
+    error = identity
+  )
+  expect_s3_class(err, "uscogdata_unknown_recipe")
+  expect_match(conditionMessage(err), "corrections_combined")
+})
+
+test_that("recipe = requires schema_version >= 5", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    expect_error(
+      cog_spending("121011212191", 2020L, recipe = "corrections_combined"),
+      class = "uscogdata_schema_unsupported"
+    )
+  })
+})
+
+# --- signposting -------------------------------------------------------
+#
+# Phase R3 / Task 19c: .build_suggestions() was narrowed from a whole-result
+# gap check (R2: does the ENTIRE category result have zero rows in a
+# requested year) to per-code gap detection (does a specific recipe
+# component -- itself a member of the requested category -- have zero rows
+# in a year the recipe's own generic join otherwise covers). See
+# R/suggestions.R's header comment and docs/phase_r_harmonization_review.md
+# § 0.3. The R2 test below ("...across the 2011->2012 gap") is unaffected
+# by the refinement (it already passed under both the coarse and per-code
+# rule). The next few pin cases the coarse rule specifically could NOT see.
+
+test_that("signposting suggests corrections_combined across the 2011->2012 gap", {
+  skip_if_no_corpus()
+  expect_message(
+    r <- cog_spending("121011212191", years = c(2011L, 2012L),
+                      category = "Corrections"),
+    "recipe"
+  )
+  prov <- attr(r, "provenance")
+  expect_true(length(prov$suggestions) >= 1L)
+  ids <- vapply(prov$suggestions, function(s) s$recipe_id, character(1))
+  expect_true("corrections_combined" %in% ids)
+  hit <- prov$suggestions[[which(ids == "corrections_combined")]]
+  expect_equal(hit$hint, "re-run with recipe = 'corrections_combined'")
+  expect_equal(hit$available_years, c(1967L, 2023L))
+})
+
+test_that("per-code gap does NOT fire when a code's only coverage is its own aggregate row (self-coverage is not \"other components\")", {
+  skip_if_no_corpus()
+  # Broward, FY2011 ONLY (isolating the 2011 half of the query above): E05,
+  # F05, and G05 each report SOLELY as a wide-era AGGREGATE row that year
+  # (216088, 1453, 270 respectively); E04/F04/G04 -- their modern-only
+  # siblings -- don't exist as codes at all before 2012, corpus-wide (zero
+  # rows for any government). Each component's own aggregate row would
+  # trivially satisfy a same-component "covered" check, but review-doc
+  # § 0.3's criterion is explicit that a gap must be covered by "OTHER
+  # components", not the gapped component's own aggregate form. With no
+  # OTHER component present for any of the three Corrections recipes in
+  # 2011, none of them should fire -- this is what the combined
+  # 2011-2012 test above actually relies on 2012 (E05 gapped, E04 -- a
+  # genuinely different component -- covers) to fire, not 2011.
+  r <- cog_spending("121011212191", years = 2011L, category = "Corrections")
+  prov <- attr(r, "provenance")
+  expect_length(prov$suggestions, 0L)
+})
+
+test_that("per-code gap fires even when a sibling code masks the whole-result check (Cleburne County, FY2012)", {
+  skip_if_no_corpus()
+  # Cleburne County, AL (canonical_govid 011029122489), FY2012: E04 ($854)
+  # and E05 ($1) both report ("operations" subtype), and G04 ($14,000,
+  # corrections_other_capital_combined's modern-only leg) also reports
+  # ("capital" subtype) -- so the WHOLE category result is non-empty for
+  # 2012 (2 rows) and the R2 whole-result check would never look further.
+  # But G05 -- G04's OWN recipe sibling, the 1967-2023 wide leg -- has
+  # ZERO rows at all that year: a genuine, per-code gap the recipe exists
+  # to bridge, invisible at the category-result grain because it's masked
+  # by G04's own data, let alone the unrelated E04/E05 pair.
+  r <- cog_spending("011029122489", years = 2012L, category = "Corrections")
+  expect_equal(nrow(r), 2L)  # operations + capital rows: a non-empty result
+
+  prov <- attr(r, "provenance")
+  ids <- vapply(prov$suggestions, function(s) s$recipe_id, character(1))
+  expect_true("corrections_other_capital_combined" %in% ids)
+  hit <- prov$suggestions[[which(ids == "corrections_other_capital_combined")]]
+  expect_equal(hit$hint, "re-run with recipe = 'corrections_other_capital_combined'")
+  expect_equal(hit$available_years, c(1967L, 2023L))
+
+  # corrections_combined must NOT fire: E04 AND E05 both have real 2012
+  # data for this government, so neither of ITS OWN components is gapped.
+  expect_false("corrections_combined" %in% ids)
+})
+
+test_that("per-code gap does not fire when no recipe component has any data at all (ordinary reporting variance, not a format-boundary gap)", {
+  skip_if_no_corpus()
+  # Same government/year as above: F04 and F05 (corrections_capital_combined)
+  # are BOTH completely absent -- Cleburne simply never reported capital
+  # corrections spending under that code family in 2012, wide-era or
+  # modern. The recipe's own generic join (aggregate-inclusive, either
+  # component) has nothing to offer either, so this must stay silent --
+  # the per-government `covered` guard the header comment describes is
+  # unchanged and still does this filtering.
+  r <- cog_spending("011029122489", years = 2012L, category = "Corrections")
+  prov <- attr(r, "provenance")
+  ids <- vapply(prov$suggestions, function(s) s$recipe_id, character(1))
+  expect_false("corrections_capital_combined" %in% ids)
+})
+
+test_that("per-code gap fires for Broward 2019-2020 even though the category result looks complete", {
+  skip_if_no_corpus()
+  # Broward reports E04 + G04 (modern leaf codes) in BOTH 2019 and 2020 but
+  # never reports E05 or G05 (their own recipe siblings) in either year --
+  # a real per-code gap in two of the three Corrections recipes, invisible
+  # under the R2 coarse check because the category *result* is non-empty
+  # both years (this replaces the old R2-era "full year coverage" test,
+  # whose premise -- that a non-empty result implies nothing to signpost --
+  # is exactly what this refinement narrows; see data-raw/
+  # measure_signposting_rate.R for the measured rate change this causes).
+  # corrections_capital_combined correctly stays silent: Broward reports
+  # neither F04 nor F05 in 2019 or 2020, so that recipe's own join has
+  # nothing to offer either (ordinary non-reporting, not a format-boundary
+  # gap) -- the per-government `covered` guard still does its job here too.
+  r <- cog_spending("121011212191", years = 2019:2020, category = "Corrections")
+  prov <- attr(r, "provenance")
+  ids <- vapply(prov$suggestions, function(s) s$recipe_id, character(1))
+  expect_true("corrections_combined" %in% ids)
+  expect_true("corrections_other_capital_combined" %in% ids)
+  expect_false("corrections_capital_combined" %in% ids)
+})
+
+test_that("no signposting when every recipe component genuinely has data (true full per-code coverage)", {
+  skip_if_no_corpus()
+  # Maricopa County, AZ (canonical_govid 041013160815): all six Corrections
+  # codes (E04, E05, F04, F05, G04, G05) report real, nonzero, non-aggregate
+  # amounts in BOTH 2019 and 2020 -- genuinely nothing for any recipe to
+  # fill, even at the finer per-code grain this refinement now checks.
+  r <- cog_spending("041013160815", years = 2019:2020, category = "Corrections")
+  prov <- attr(r, "provenance")
+  expect_length(prov$suggestions, 0L)
+})
+
+test_that("no signposting when category is NULL (unscoped query)", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", years = c(2011L, 2012L))
+  prov <- attr(r, "provenance")
+  expect_length(prov$suggestions, 0L)
+})
+
+test_that("no signposting under basis = 'raw'", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", years = c(2011L, 2012L),
+                    category = "Corrections", basis = "raw")
+  prov <- attr(r, "provenance")
+  expect_length(prov$suggestions, 0L)
+})

tests/testthat/test-revenue.R

@@ -36,3 +36,20 @@ test_that("cog_revenue result has provenance attribute", {
 test_that("cog_revenue rejects invalid inputs", {
   expect_error(cog_revenue(list(), 2020L), "character|data frame")
 })
+
+test_that("cog_revenue basis = 'harmonized' (default) matches 'raw' in this fixture window", {
+  skip_if_no_corpus()
+  r_raw  <- cog_revenue("121011212191", 2019:2020, basis = "raw")
+  r_harm <- cog_revenue("121011212191", 2019:2020, basis = "harmonized")
+  expect_equal(attr(r_raw, "provenance")$basis, "raw")
+  expect_equal(attr(r_harm, "provenance")$basis, "harmonized")
+  expect_equal(sum(r_raw$amt_nominal), sum(r_harm$amt_nominal))
+})
+
+test_that("cog_revenue provenance carries the harmonization block", {
+  skip_if_no_corpus()
+  r <- cog_revenue("121011212191", 2020L)
+  h <- attr(r, "provenance")$harmonization
+  expect_true(h$applied)
+  expect_true(h$na_rows_excluded >= 0L)
+})

tests/testthat/test-signposting-harness.R

@@ -0,0 +1,164 @@
+# tests/testthat/test-signposting-harness.R
+#
+# Pins the subset-relation REPORTING in data-raw/measure_signposting_rate.R.
+#
+# Phase R3 Task 19c narrowed signposting from a coarse whole-result gap
+# check to per-code gap detection. Those two checks are partly DISJOINT,
+# not nested: a query can fire under coarse and stay silent under per-code,
+# so the harness's `*_delta_pp` figures are NETS that can hide a coverage
+# loss. `.measure_subset_relation()` is what separates the two flows, and
+# `.measure_format_subset_report()` is what puts the loss in front of a
+# human. Both are load-bearing for the Checkpoint R3 ruling, so both are
+# pinned here: if the violation detection is deleted, inverted, or quietly
+# downgraded to a count with no identities, these tests fail.
+#
+# These tests do NOT assert that the violation set is empty -- it is
+# genuinely non-empty, and asserting otherwise would be pinning a bug as a
+# contract. They assert only that a real violation is DETECTED and NAMED.
+
+# The harness lives in data-raw/, which is .Rbuildignore'd, so it is absent
+# from an installed/checked tarball. Source it into an env parented on the
+# namespace so it resolves the package internals it calls (.build_verb_sql,
+# .build_suggestions) exactly as it does when run for real.
+harness_env <- function() {
+  path <- testthat::test_path("..", "..", "data-raw", "measure_signposting_rate.R")
+  skip_if_not(file.exists(path),
+              "data-raw/ is .Rbuildignore'd; harness not present in this tree")
+  env <- new.env(parent = asNamespace("uscogdata"))
+  source(path, local = env)
+  env
+}
+
+# A detail frame in exactly the shape .measure_one_query() emits, covering
+# all four quadrants of the coarse x percode cross-tab.
+fake_detail <- function() {
+  data.frame(
+    category         = c("Corrections", "Other Taxes", "Police", "Fire"),
+    category_type    = c("expenditure", "revenue", "expenditure", "expenditure"),
+    canonical_govid  = c("121011212191", "472155175824", "011029122489",
+                         "041013160815"),
+    n_result_rows    = c(0L, 1L, 2L, 6L),
+    coarse_gap_years = c("2011", "", "", ""),
+    fired_coarse     = c(TRUE, FALSE, TRUE, FALSE),
+    fired_percode    = c(FALSE, TRUE, TRUE, FALSE),
+    coarse_recipes   = c("corrections_combined", "", "police_combined", ""),
+    percode_recipes  = c("", "t29_license_wide", "police_combined", ""),
+    stringsAsFactors = FALSE
+  )
+}
+
+test_that(".measure_subset_relation() separates coverage LOST from coverage ADDED", {
+  e   <- harness_env()
+  rel <- e$.measure_subset_relation(fake_detail())
+
+  # Row 1 (coarse fired, per-code silent) is the violation; row 2 is the
+  # addition; row 3 agrees; row 4 is silent.
+  expect_false(rel$holds)
+  expect_equal(rel$n_violations, 1L)
+  expect_equal(rel$n_additions, 1L)
+  expect_equal(rel$n_coarse_fired, 2L)
+  expect_equal(rel$n_percode_fired, 2L)
+  expect_equal(rel$n_both, 1L)
+  expect_equal(rel$n_queries, 4L)
+
+  # The violation must be NAMED down to (category, government, year), not
+  # merely counted -- that is what makes it inspectable at Checkpoint R3.
+  expect_equal(rel$violations$category, "Corrections")
+  expect_equal(rel$violations$canonical_govid, "121011212191")
+  expect_equal(rel$violations$coarse_gap_years, "2011")
+  expect_equal(rel$violations$coarse_recipes, "corrections_combined")
+
+  # Inversion guard: an implementation that swapped the two directions
+  # would report the addition as a violation and vice versa.
+  expect_false("Other Taxes" %in% rel$violations$category)
+  expect_equal(rel$additions$category, "Other Taxes")
+  expect_false("Corrections" %in% rel$additions$category)
+
+  # Agreeing and silent queries belong to neither set.
+  expect_false("Police" %in% c(rel$violations$category, rel$additions$category))
+  expect_false("Fire" %in% c(rel$violations$category, rel$additions$category))
+})
+
+test_that(".measure_subset_relation() reports holds = TRUE only when nothing fires coarse-only", {
+  e <- harness_env()
+  # Drop the violating row: coarse is now genuinely a subset of per-code.
+  clean <- fake_detail()[-1L, , drop = FALSE]
+  rel   <- e$.measure_subset_relation(clean)
+
+  expect_true(rel$holds)
+  expect_equal(rel$n_violations, 0L)
+  expect_equal(nrow(rel$violations), 0L)
+  expect_equal(rel$n_additions, 1L)
+})
+
+test_that(".measure_subset_relation() validates its input rather than silently mis-reporting", {
+  e <- harness_env()
+  expect_error(e$.measure_subset_relation("not a data frame"), "must be a data frame")
+  expect_error(e$.measure_subset_relation(fake_detail()[, c("category", "canonical_govid")]),
+               "fired_coarse")
+  bad <- fake_detail()
+  bad$fired_percode[1] <- NA
+  expect_error(e$.measure_subset_relation(bad), "non-NA logicals")
+})
+
+test_that("the subset report NAMES a coarse-only firing as a violation", {
+  e   <- harness_env()
+  txt <- paste(e$.measure_format_subset_report(e$.measure_subset_relation(fake_detail())),
+               collapse = "\n")
+
+  # Stated plainly as a violation, not buried.
+  expect_match(txt, "VIOLATED")
+  expect_match(txt, "COVERAGE LOST")
+  expect_no_match(txt, "HOLDS")
+  # ...and the offending query named, so a human can go look at it.
+  expect_match(txt, "Corrections")
+  expect_match(txt, "121011212191")
+  expect_match(txt, "2011")
+  # ...and the delta explicitly flagged as a net of both directions.
+  expect_match(txt, "NET")
+})
+
+test_that("the subset report says HOLDS when coarse really is a subset", {
+  e   <- harness_env()
+  rel <- e$.measure_subset_relation(fake_detail()[-1L, , drop = FALSE])
+  txt <- paste(e$.measure_format_subset_report(rel), collapse = "\n")
+
+  expect_match(txt, "HOLDS")
+  expect_no_match(txt, "VIOLATED")
+  expect_no_match(txt, "COVERAGE LOST")
+})
+
+test_that("a REAL coarse-fires/per-code-silent query is measured and reported as a violation", {
+  skip_if_no_corpus()
+  # Broward County FY2011, Corrections: E05/F05/G05 report SOLELY as
+  # wide-era aggregate rows, which basis = "harmonized" excludes, so the
+  # whole category result is empty -- coarse's trigger. Their modern-only
+  # siblings E04/F04/G04 do not exist as codes at all before 2012, so no
+  # OTHER component can supply per-code's covering evidence and per-code
+  # is structurally unable to fire. This is the disjointness the harness
+  # exists to surface, measured end-to-end through the real git-loaded
+  # coarse arm and the live per-code arm (not a hand-built frame).
+  e   <- harness_env()
+  con <- uscogdata:::cog_open()
+  row <- e$.measure_one_query(
+    con,
+    coarse_env  = e$.measure_load_git_impl("b0df1ec"),
+    selfcov_env = e$.measure_load_git_impl("da72bf3"),
+    category = "Corrections", category_type = "expenditure",
+    govid = "121011212191", years = 2011L
+  )
+
+  expect_true(row$fired_coarse)
+  expect_false(row$fired_percode)
+  expect_equal(row$n_result_rows, 0L)
+  expect_equal(row$coarse_gap_years, "2011")
+
+  rel <- e$.measure_subset_relation(row)
+  expect_false(rel$holds)
+  expect_equal(rel$n_violations, 1L)
+  expect_equal(rel$violations$canonical_govid, "121011212191")
+
+  txt <- paste(e$.measure_format_subset_report(rel), collapse = "\n")
+  expect_match(txt, "VIOLATED")
+  expect_match(txt, "121011212191")
+})

tests/testthat/test-spending.R

@@ -189,3 +189,134 @@ test_that("provenance records per-year denominator metadata", {
     expect_equal(length(pc$popyear_range), 2L)
   })
 })
+
+# --- basis = "harmonized" / "raw" (Phase R2, schema v5) --------------------
+
+test_that("basis = 'raw' reproduces the pre-harmonization Broward Police totals", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", years = 2019:2020, category = "Police",
+                      basis = "raw")
+    # Regression pin captured against the schema v5 fixture (2026-07-18,
+    # pipeline_commit ece9b32) before basis = "harmonized" existed as a
+    # concept; these are the same totals the pre-Phase-R2 default query
+    # returned (spending_annotated is untouched by the harmonized views).
+    ops <- r$amt_nominal[r$year == 2019L & r$spend_subtype == "operations"]
+    cap <- r$amt_nominal[r$year == 2020L & r$spend_subtype == "capital"]
+    expect_equal(ops, 483560000)
+    expect_equal(cap, 26693000)
+    expect_equal(attr(r, "provenance")$basis, "raw")
+  })
+})
+
+test_that("basis = 'harmonized' (default) matches 'raw' when no harmonization rule applies", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    # Every `method = "collapse"` mapping in the curated harmonization_map
+    # ends by FY2004 for codes inside the spending/revenue flow-type
+    # prefixes (E/F/G/K, T/A/U/B/C/D); the one collapse extending to FY2011
+    # (L38/M38 -> L36/M36) is intergovernmental-transfer (L/M prefix) codes
+    # that were never part of spending_long/revenue_long to begin with. So
+    # for the fixture's 2011-2020 window, basis = "harmonized" is a
+    # data-verified no-op vs "raw" for in-scope codes -- this is the
+    # positive-control counterpart to the synthetic REPLACE-mechanism test
+    # in test-views.R, which proves the fold itself works when data exists.
+    r_raw  <- cog_spending("121011212191", c(2011L, 2012L, 2019L, 2020L),
+                           "Police", basis = "raw")
+    r_harm <- cog_spending("121011212191", c(2011L, 2012L, 2019L, 2020L),
+                           "Police", basis = "harmonized")
+    expect_equal(attr(r_harm, "provenance")$basis, "harmonized")
+    expect_equal(
+      r_harm$amt_nominal[order(r_harm$year, r_harm$spend_subtype)],
+      r_raw$amt_nominal[order(r_raw$year, r_raw$spend_subtype)]
+    )
+  })
+})
+
+test_that("basis defaults to 'harmonized' when not passed", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2020L, "Police")
+    expect_equal(attr(r, "provenance")$basis, "harmonized")
+  })
+})
+
+test_that("provenance carries basis + harmonization block with na_rows_excluded", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2011:2012, "Corrections")
+    prov <- attr(r, "provenance")
+    expect_equal(prov$basis, "harmonized")
+    expect_true(prov$harmonization$applied)
+    expect_true(prov$harmonization$na_rows_excluded >= 0L)
+    expect_true(prov$harmonization$na_amount_excluded >= 0)
+    # Data-verified for the v6 fixture (corpus 2026-07-22). The Task 18 map
+    # extension added E/F/G-prefix discontinued_na rulings the earlier pin's
+    # comment predated: E21/F21/G21 (Education NEC local, SB184-186,
+    # "trivial; explicit-NA, full wide-era window"). Broward's 2011 legacy
+    # partition zero-pads exactly those three codes, so this query now
+    # excludes 3 NA-harmonized rows -- all with amt = 0, hence the excluded
+    # AMOUNT stays exactly zero. (The other discontinued_na rulings -- S74,
+    # Z61, X04, X06, the debt-detail family, L24 -- remain outside the
+    # E/F/G/K prefixes.) See docs/phase_r_harmonization_review.md § 1.3/1.4
+    # and cog_pipeline data/harmonization_map.csv E21/F21/G21 rows.
+    expect_equal(prov$harmonization$na_rows_excluded, 3L)
+    expect_equal(prov$harmonization$na_amount_excluded, 0)
+  })
+})
+
+test_that("basis = 'raw' never populates the harmonization exclusion block", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2020L, "Corrections", basis = "raw")
+    h <- attr(r, "provenance")$harmonization
+    expect_false(h$applied)
+    expect_equal(h$na_rows_excluded, 0L)
+  })
+})
+
+test_that("v4 corpus: basis silently resolves to raw (default) with a provenance note", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    r <- cog_spending("121011212191", 2019L, "Police")
+    prov <- attr(r, "provenance")
+    expect_equal(prov$basis, "raw")
+    expect_match(prov$basis_note, "raw", fixed = TRUE)
+    expect_match(prov$basis_note, "schema_version", fixed = TRUE)
+    expect_false(prov$harmonization$applied)
+  })
+})
+
+test_that("v4 corpus: explicit basis = 'harmonized' aborts", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    expect_error(
+      cog_spending("121011212191", 2019L, "Police", basis = "harmonized"),
+      class = "uscogdata_basis_unsupported"
+    )
+  })
+})
+
+test_that("provenance$series_break_refs is a populated-when-applicable character vector", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2020L, "Corrections")
+    refs <- attr(r, "provenance")$series_break_refs
+    expect_type(refs, "character")
+    # No catalogued series_breaks_pq row falls inside this fixture's
+    # 2011/2012/2019/2020 window for the codes this query touches (E04/G04)
+    # -- data-verified; the mechanism itself is what's under test here, via
+    # a query-shaped unit test in test-views.R since the fixture has no
+    # positive case to pin against.
+    expect_equal(refs, character(0))
+  })
+})
+
+test_that("v4 corpus: explicit basis = 'raw' still works", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    r <- cog_spending("121011212191", 2019L, "Police", basis = "raw")
+    expect_equal(attr(r, "provenance")$basis, "raw")
+    expect_gt(nrow(r), 0L)
+  })
+})

tests/testthat/test-views.R

@@ -14,6 +14,144 @@ test_that("all expected views register on session open", {
   expect_true(all(expected %in% views$table_name))
 })
 
+test_that("inst/sql/22- and 23- harmonized views enforce every WHERE predicate (real SQL text, synthetic parquet)", {
+  # spending_long_harmonized / revenue_long_harmonized are three-predicate
+  # views:
+  #   SELECT * REPLACE (harmonized_code AS item_code)
+  #   FROM long
+  #   WHERE NOT is_aggregate
+  #     AND harmonized_code IS NOT NULL
+  #     AND LEFT(harmonized_code, 1) IN (<flow prefixes>)
+  # None of the curated harmonization_map's `collapse` rulings land inside
+  # the bundled fixture's 2011-2020 window for spending/revenue-prefixed
+  # codes (see the "basis = 'harmonized' (default) matches 'raw'" test in
+  # test-spending.R and docs/phase_r_harmonization_review.md § 0.2/§ 2), so
+  # there is no real fixture row that exercises a nonzero fold or a
+  # predicate-excluded row. Rather than re-implement the WHERE clause by
+  # hand against an in-memory VALUES table (which would only prove the SQL
+  # *pattern* works, not that the deployed inst/sql/22-/23- text actually
+  # applies it), this test reads the real SQL files off disk, substitutes
+  # {url} exactly as .register_views() does, and executes them -- plus
+  # their 10-long.sql dependency -- against a synthetic hive-partitioned
+  # parquet tree written to a temp dir. A regression in any predicate (e.g.
+  # `NOT is_aggregate` dropped, the prefix list changed, the NULL guard
+  # removed) would change which of the rows below survive.
+  #
+  # The synthetic parquet is written with DuckDB's own COPY ... TO (FORMAT
+  # PARQUET) rather than the arrow package: this package has no arrow
+  # dependency (CLAUDE.md "No arrow dependency -- DuckDB reads parquet
+  # natively"), and DuckDB can round-trip its own parquet writer/reader
+  # without adding one for tests either.
+  skip_if_no_corpus()
+
+  tmp <- withr::local_tempdir()
+  part_dir <- file.path(tmp, "data", "long", "year=2004")
+  dir.create(part_dir, recursive = TRUE)
+  part_path <- file.path(part_dir, "part-0.parquet")
+
+  write_con <- DBI::dbConnect(duckdb::duckdb())
+  on.exit(DBI::dbDisconnect(write_con, shutdown = TRUE), add = TRUE)
+  DBI::dbExecute(write_con, sprintf("
+    COPY (
+      SELECT * FROM (VALUES
+        -- Spending (E/F/G/K) rows, exercised against spending_long_harmonized:
+        ('spend-A', 'E36', 100,    false, 'E36'), -- control: passes every predicate as-is
+        ('spend-B', 'E38', 50,     false, 'E36'), -- collapse-fold: passes every predicate, renamed to E36
+        ('spend-C', 'E05', 999999, true,  'E05'), -- excluded ONLY by `NOT is_aggregate`
+        ('spend-D', 'E99', 888888, false, NULL),  -- excluded by `harmonized_code IS NOT NULL`
+        -- 'S74' is outside BOTH flow families (E/F/G/K spending and
+        -- T/A/U/B/C/D revenue -- it mirrors the real corpus's own
+        -- non-flow-type codes like S74/Z61), so it can only leak into
+        -- EITHER view via the E/F/G/K or T/A/U/B/C/D prefix filter, never
+        -- both at once -- a prefix drawn from the other view's own family
+        -- (e.g. a real T-code for the spending row) would incorrectly
+        -- leak into the other view's assertion below and not discriminate
+        -- the predicate under test.
+        ('spend-E', 'S74', 777777, false, 'S74'), -- excluded ONLY by the E/F/G/K prefix filter
+        -- Revenue (T/A/U/B/C/D) rows, exercised against revenue_long_harmonized:
+        ('rev-A',   'U11', 200,    false, 'U11'), -- control: passes every predicate as-is
+        ('rev-B',   'U10', 25,     false, 'U11'), -- collapse-fold: passes every predicate, renamed to U11
+        ('rev-C',   'T29', 555555, true,  'T29'), -- excluded ONLY by `NOT is_aggregate`
+        ('rev-D',   'T88', 444444, false, NULL),  -- excluded by `harmonized_code IS NOT NULL`
+        ('rev-E',   'Z61', 333333, false, 'Z61')  -- excluded ONLY by the T/A/U/B/C/D prefix filter
+      ) AS t(canonical_govid, item_code, amt, is_aggregate, harmonized_code)
+    ) TO %s (FORMAT PARQUET)
+  ", uscogdata:::.sql_lit_chr(part_path)))
+
+  sql_dir <- system.file("sql", package = "uscogdata")
+  .read_view_sql <- function(filename) {
+    txt <- paste(readLines(file.path(sql_dir, filename), warn = FALSE), collapse = "\n")
+    gsub("\\{url\\}", paste0(tmp, "/"), txt, fixed = FALSE)
+  }
+
+  con <- DBI::dbConnect(duckdb::duckdb())
+  on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+  DBI::dbExecute(con, .read_view_sql("10-long.sql"))
+  DBI::dbExecute(con, .read_view_sql("22-spending_long_harmonized.sql"))
+  DBI::dbExecute(con, .read_view_sql("23-revenue_long_harmonized.sql"))
+
+  spend <- DBI::dbGetQuery(con,
+    "SELECT item_code, SUM(amt) AS amt FROM spending_long_harmonized
+     GROUP BY item_code ORDER BY item_code"
+  )
+  # Exactly one surviving row: spend-C (aggregate), spend-D (NULL
+  # harmonized_code), and spend-E (wrong prefix family) must all be gone,
+  # and spend-A + spend-B must be folded together under E36.
+  expect_equal(nrow(spend), 1L)
+  expect_equal(spend$item_code, "E36")
+  expect_equal(spend$amt, 150)
+
+  rev <- DBI::dbGetQuery(con,
+    "SELECT item_code, SUM(amt) AS amt FROM revenue_long_harmonized
+     GROUP BY item_code ORDER BY item_code"
+  )
+  expect_equal(nrow(rev), 1L)
+  expect_equal(rev$item_code, "U11")
+  expect_equal(rev$amt, 225)
+})
+
+test_that(".build_series_break_refs matches fin_code + break_year window", {
+  # No series_breaks_pq row falls inside the bundled fixture's 2011-2020
+  # window (data-verified; see the "series_break_refs" test in
+  # test-spending.R), so this proves the matching logic itself against the
+  # live view + a synthetic year window that DOES hit a cataloged break
+  # (SB075, fin_code E62, break_year 2005).
+  skip_if_no_corpus()
+  con <- cog_open()
+  on.exit(cog_close())
+  refs <- uscogdata:::.build_series_break_refs(
+    con, codes_observed = c("E62", "E04"), years = c(2003L, 2006L),
+    schema_version = 5L
+  )
+  expect_true("SB075" %in% refs)
+  expect_true("SB071" %in% refs)
+
+  # Gated on schema_version >= 5 even when the codes/years would otherwise match.
+  refs_v4 <- uscogdata:::.build_series_break_refs(
+    con, codes_observed = c("E62"), years = c(2003L, 2006L), schema_version = 4L
+  )
+  expect_equal(refs_v4, character(0))
+})
+
+test_that("schema v5 harmonization views register when the corpus supports them", {
+  skip_if_no_corpus()
+  con <- cog_open()
+  on.exit(cog_close())
+  manifest <- uscogdata:::.uscogdata_env$manifest
+  skip_if(as.integer(manifest$schema_version) < 5L, "fixture is schema_version < 5")
+
+  views <- DBI::dbGetQuery(con,
+    "SELECT table_name FROM information_schema.tables
+     WHERE table_schema = 'main' AND table_type = 'VIEW'"
+  )
+  expected_v5 <- c(
+    "spending_long_harmonized", "revenue_long_harmonized",
+    "spending_annotated_harmonized", "revenue_annotated_harmonized",
+    "harmonization_map", "harmonization_recipes", "series_breaks_pq"
+  )
+  expect_true(all(expected_v5 %in% views$table_name))
+})
+
 test_that("spending_long filters to E/F/G/K prefixes and excludes aggregates", {
   skip_if_no_corpus()
   con <- cog_open()
@@ -69,13 +207,15 @@ test_that("gov_population_yearly exposes one row per (year, canonical_govid)", {
        WHERE canonical_govid = '121011212191'
        ORDER BY year"
     )
-    expect_setequal(df$year, c(2019L, 2020L))
+    expect_setequal(df$year, c(2011L, 2012L, 2019L, 2020L))
-    expect_equal(nrow(df), 2L)
+    expect_equal(nrow(df), 4L)
     expect_true(all(!is.na(df$population)))
-    # Hardcoded values are from the bundled fixture (regenerated 2026-07-11
+    # Hardcoded values are from the bundled fixture (regenerated 2026-07-18
-    # against cog_pipeline publish tree, pipeline_commit 1a00925, Phase P
+    # against cog_pipeline publish tree, pipeline_commit ece9b32, Phase R2
-    # schema_version 4). Update if the fixture is rebuilt against a
+    # schema_version 5, years 2011/2012/2019/2020). Update if the fixture is
-    # different source vintage.
+    # rebuilt against a different source vintage.
+    expect_equal(df$population[df$year == 2011L], 1759591L)
+    expect_equal(df$population[df$year == 2012L], 1819773L)
     expect_equal(df$population[df$year == 2019L], 1935878L)
     expect_equal(df$population[df$year == 2020L], 1952778L)
     # Uniqueness on (year, canonical_govid) across the whole view.