fix: exercise real harmonized-view SQL in tests; unambiguous recipe provenance

test-views.R's harmonized-view test previously ran a hand-rolled REPLACE
query with no WHERE clause, so a regression in any of
inst/sql/22-spending_long_harmonized.sql / 23-revenue_long_harmonized.sql's
three predicates (NOT is_aggregate, harmonized_code IS NOT NULL, the
E/F/G/K or T/A/U/B/C/D prefix filter) would go uncaught. Replaced it with a
test that 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 via
DuckDB's own COPY ... TO (FORMAT PARQUET) (no arrow dependency, matching
this package's existing convention). Ten rows are crafted so each predicate
is independently falsifiable by a specific row; manually broke each
predicate in turn to confirm the test fails exactly as expected, then
restored the SQL files (see the task report for the RED-phase transcript).

Also fixes a provenance ambiguity: a recipe= query bypasses
spending_annotated(_harmonized)/revenue_annotated(_harmonized) entirely
(.run_recipe() joins `long` directly), so basis= has no effect on it, but
provenance was still reporting basis = "harmonized"/"raw" (whatever the
argument resolved to) with harmonization$applied = FALSE alongside it --
misleading, since it looks like harmonization was evaluated and found
nothing to exclude rather than "not applicable here." Recipe results now
report basis = "recipe" with an inert harmonization block carrying an
explicit note, regardless of what basis= was passed.

.Rbuildignore

@@ -10,3 +10,9 @@
 ^\.gitignore$
 \.gitkeep$
 ^vignettes$
+^specs$
+^plans$
+^doc$
+^Meta$
+^\.gitea$
+^CLAUDE\.md$

DESCRIPTION

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

NAMESPACE

@@ -7,7 +7,9 @@ export(cog_explain)
 export(cog_find_peers)
 export(cog_geographic_rollup)
 export(cog_gov_search)
+export(cog_manifest)
 export(cog_mirror)
 export(cog_peer_compare)
+export(cog_recipes)
 export(cog_revenue)
 export(cog_spending)

NEWS.md

@@ -1,5 +1,84 @@
 # uscogdata 0.1.0 (development)
 
+## Breaking: corpus schema_version 4 (Phase P canonical ids)
+
+* The package now requires corpus `schema_version = 4` (`MinCorpusSchema` /
+  `MaxCorpusSchema` in `DESCRIPTION` are both `4`); older corpora built
+  against schema 3 are rejected by `cog_open()` with a clear version-mismatch
+  error. `canonical_govid` is now uniformly 12 characters across every
+  vintage the corpus covers (previously a mix of 9-char legacy ids and
+  12-char FIPS ids depending on source year) — **every hardcoded
+  `canonical_govid` literal from a pre-Phase-P corpus is now invalid** and
+  must be re-resolved via `cog_gov_search()` or the new `canonical_alias`
+  lookup table. `canonical_fips_xwalk` gains four columns
+  (`legacy_govs_id`, `census_geoid`, `id_source`; `confidence` is renamed to
+  `pop_confidence`) and a companion `canonical_alias` table ships in the
+  corpus for mapping legacy/alternate ids onto the current canonical
+  namespace. The bundled fixture corpus (`inst/extdata/fixture_corpus/`) has
+  been regenerated against the Phase P publish tree, now ships the full
+  `canonical_fips_xwalk` and `canonical_alias` master tables alongside the
+  2019-2020 long partitions, and is reproducible via
+  `data-raw/regenerate_fixture_corpus.R`.
+
+## Clearer errors when `USCOGDATA_URL` is unconfigured or returns non-JSON
+
+* `cog_open()` now aborts with the `uscogdata_url_not_configured` error
+  class when the resolved corpus URL still contains the placeholder
+  `REPLACE_WITH_SHARE_TOKEN` sentinel (or is empty). The message lists both
+  remediation paths (`Sys.setenv(USCOGDATA_URL = ...)` and
+  `options(uscogdata.url = ...)`) and points at the bundled fixture for
+  offline testing. Previously the package proceeded to fetch the placeholder
+  URL, cached the resulting HTML welcome page, and failed downstream with a
+  cryptic `jsonlite` lexical-error.
+* `.fetch_or_cache_manifest()` now parses the HTTP response body before
+  persisting it. Non-JSON responses (login pages, 404 HTML) raise
+  `uscogdata_invalid_manifest` with the URL, Content-Type, and underlying
+  parse error — and never write to the on-disk cache.
+* Manifest cache writes are now atomic (write to `manifest.json.tmp.<pid>`
+  in `cache_dir`, then `file.rename` over the target), so an interrupted
+  fetch cannot replace a previously-good cache.
+* Existing caches with non-JSON content (poisoned by the prior code path)
+  are silently refetched instead of returning a parse error to the caller.
+* Local `USCOGDATA_URL` paths whose `manifest.json` is not valid JSON now
+  surface the same `uscogdata_invalid_manifest` class with file context.
+
+## Per-capita denominators now use per-year Census F-33 population
+
+* `cog_spending()` and `cog_revenue()` previously divided all years' amounts
+  by a single ACS 2018-2022 estimate (`canonical_fips_xwalk.population_acs`),
+  producing biased per-capita values for time-series analysis. They now
+  divide by the F-33 `population` recorded on each gov-year via the new
+  `gov_population_yearly` view. Result tibbles gain a `pop_source` column
+  with values `"census_f33"` or `"unavailable"`. `notes` is updated to
+  concatenate multiple notes with `"; "`.
+
+## Peer cohorts can be set to a chosen year
+
+* `cog_find_peers()` adds a `year` argument (default: most recent year for
+  which the target has an observed population in `gov_population_yearly`).
+  The returned column previously named `population_acs` is now `population`
+  and reflects the cohort year's vintage. The cohort year is attached to the
+  returned tibble as `attr(x, "cohort_year")`.
+* `cog_peer_compare()` now stamps a `cohort_year` column on its result (read
+  from the peers tibble's attribute) and records `cohort_year` plus
+  `cohort_govids` in provenance. When the caller supplies a bare character
+  vector instead of a `cog_find_peers()` result, `cohort_year` is `NA`.
+
+## Rollups exclude govs missing population
+
+* `cog_geographic_rollup(per_capita = TRUE)` drops rows whose government has
+  `pop_source == "unavailable"` and records the dropped govids in
+  `provenance$rollup$excluded_govids`. This excludes special districts
+  (type 4) and school districts (type 5) from per-capita rollups by design.
+
+## New: vignette and provenance metadata
+
+* New vignette `population-denominators` covers the four population sources,
+  the type-4/5 coverage gap, the popyear quirk, and how to build moving-window
+  peer cohorts manually.
+* Provenance gains `transformations$per_capita$popyear_range` and
+  `pop_source_counts`. `cog_explain()` renders both.
+
 ## New features
 
 * `cog_gov_search()` gains a **basket mode**: passing vector `name`

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)) {
@@ -74,6 +116,16 @@ cog_explain <- function(result, format = c("print", "list")) {
   pc <- prov$transformations$per_capita
   if (isTRUE(pc$applied)) {
     cli::cli_text("Per-capita denominator: {pc$denominator_source}")
+    if (length(pc$popyear_range) == 2L) {
+      lo <- .expand_popyear(pc$popyear_range[1])
+      hi <- .expand_popyear(pc$popyear_range[2])
+      cli::cli_text("  popyear range: {lo}-{hi}")
+    }
+    if (!is.null(pc$pop_source_counts)) {
+      cli::cli_text(
+        "  pop_source counts: census_f33={pc$pop_source_counts$census_f33}, unavailable={pc$pop_source_counts$unavailable}"
+      )
+    }
   }
   infl <- prov$transformations$inflation
   if (isTRUE(infl$applied)) {
@@ -98,3 +150,37 @@ 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+.
+#' @noRd
+.expand_popyear <- function(yy) {
+  yy <- as.integer(yy)
+  if (length(yy) == 0L || is.na(yy)) return(NA_integer_)
+  if (yy >= 100L) return(yy)        # already 4-digit
+  if (yy < 70L)  return(2000L + yy)
+  1900L + yy
+}

R/manifest.R

@@ -1,5 +1,66 @@
 # R/manifest.R
 
+# Sentinel substring baked into the placeholder default URL. If we see this
+# in the resolved URL, the user hasn't configured USCOGDATA_URL yet.
+.PLACEHOLDER_TOKEN <- "REPLACE_WITH_SHARE_TOKEN"
+
+#' Abort with actionable guidance when the resolved corpus URL is still the
+#' placeholder shipped with the package (or any URL containing the sentinel).
+#' Called from `cog_open()` before any I/O so users see a clear message
+#' instead of a downstream JSON parse error.
+#' @noRd
+.check_url_configured <- function(url) {
+  if (!is.character(url) || length(url) != 1L || !nzchar(url)) {
+    cli::cli_abort(c(
+      "USCOGDATA_URL is not configured.",
+      i = "Set the corpus location via one of:",
+      "*" = "{.code Sys.setenv(USCOGDATA_URL = \"<url-or-local-path>/\")}",
+      "*" = "{.code options(uscogdata.url = \"<url-or-local-path>/\")}",
+      i = "For an offline smoke test, use the bundled fixture: {.code system.file(\"extdata/fixture_corpus\", package = \"uscogdata\")}."
+    ), class = "uscogdata_url_not_configured")
+  }
+  if (grepl(.PLACEHOLDER_TOKEN, url, fixed = TRUE)) {
+    sentinel <- .PLACEHOLDER_TOKEN
+    cli::cli_abort(c(
+      "USCOGDATA_URL is not configured (placeholder URL detected).",
+      x = "Current value contains the sentinel {.val {sentinel}}: {.url {url}}",
+      i = "Set the corpus location via one of:",
+      "*" = "{.code Sys.setenv(USCOGDATA_URL = \"<url-or-local-path>/\")}",
+      "*" = "{.code options(uscogdata.url = \"<url-or-local-path>/\")}",
+      i = "For an offline smoke test, use the bundled fixture: {.code system.file(\"extdata/fixture_corpus\", package = \"uscogdata\")}.",
+      i = "For the live Civilytics corpus, request the Nextcloud share URL from the package maintainer."
+    ), class = "uscogdata_url_not_configured")
+  }
+  invisible(url)
+}
+
+#' Try to parse a JSON file. Returns parsed object on success, NULL on
+#' any parse failure (so callers can decide whether to refetch).
+#' @noRd
+.try_parse_manifest_file <- function(path) {
+  tryCatch(
+    jsonlite::fromJSON(path, simplifyVector = FALSE),
+    error = function(e) NULL
+  )
+}
+
+#' Abort with a clear, classified error when a manifest payload (string or
+#' file) cannot be parsed as JSON. Surfaces the URL, content-type if known,
+#' and the underlying parse error.
+#' @noRd
+.abort_invalid_manifest <- function(source, content_type = NA_character_, parse_error = NULL) {
+  ct <- if (is.na(content_type) || !nzchar(content_type)) "<unknown>" else content_type
+  pmsg <- if (is.null(parse_error)) "" else conditionMessage(parse_error)
+  cli::cli_abort(c(
+    "Corpus manifest is not valid JSON.",
+    x = "Source: {source}",
+    i = "Content-Type: {ct}",
+    i = "Likely causes: USCOGDATA_URL points at a login page, a 404 HTML page, or the wrong share; or the corpus has not been published yet.",
+    i = "Set USCOGDATA_URL to a directory (local path or HTTPS) that serves manifest.json directly.",
+    if (nzchar(pmsg)) c(">" = "Parse error: {pmsg}") else NULL
+  ), class = "uscogdata_invalid_manifest")
+}
+
 #' Fetch manifest.json from URL (or read from a local fixture path),
 #' cache locally, validate TTL.
 #' @noRd
@@ -11,23 +72,54 @@
     if (!file.exists(local_manifest)) {
       cli::cli_abort("Local fixture has no manifest.json at {local_manifest}")
     }
-    return(jsonlite::fromJSON(local_manifest, simplifyVector = FALSE))
+    return(tryCatch(
+      jsonlite::fromJSON(local_manifest, simplifyVector = FALSE),
+      error = function(e) .abort_invalid_manifest(source = local_manifest, parse_error = e)
+    ))
   }
 
   cache_path <- file.path(cache_dir, "manifest.json")
   ttl <- as.integer(.cfg("manifest_ttl_secs"))
 
-  needs_fetch <- !file.exists(cache_path) ||
+  cache_fresh <- file.exists(cache_path) &&
-                 difftime(Sys.time(), file.info(cache_path)$mtime, units = "secs") > ttl
+    difftime(Sys.time(), file.info(cache_path)$mtime, units = "secs") <= ttl
 
-  if (needs_fetch) {
+  # Honor a fresh cache only if its contents still parse as JSON. A previous
-    resp <- httr2::request(paste0(url, "manifest.json")) |>
+  # version of this package could write HTML directly into the cache; treat
-      httr2::req_error(is_error = function(r) httr2::resp_status(r) >= 400) |>
+  # such poisoned caches as if they were missing so the next call recovers.
-      httr2::req_perform()
+  if (cache_fresh) {
-    writeLines(httr2::resp_body_string(resp), cache_path)
+    parsed <- .try_parse_manifest_file(cache_path)
+    if (!is.null(parsed)) return(parsed)
   }
 
-  jsonlite::fromJSON(cache_path, simplifyVector = FALSE)
+  resp <- httr2::request(paste0(url, "manifest.json")) |>
+    httr2::req_error(is_error = function(r) httr2::resp_status(r) >= 400) |>
+    httr2::req_perform()
+  body <- httr2::resp_body_string(resp)
+
+  # Parse BEFORE persisting. If the server returned HTML / a login page /
+  # any non-JSON body with a 2xx status, we must not write it to the cache.
+  parsed <- tryCatch(
+    jsonlite::fromJSON(body, simplifyVector = FALSE),
+    error = function(e) {
+      ct <- tryCatch(httr2::resp_content_type(resp), error = function(e2) NA_character_)
+      .abort_invalid_manifest(
+        source = paste0(url, "manifest.json"),
+        content_type = ct,
+        parse_error = e
+      )
+    }
+  )
+
+  # Atomic write: tmp file alongside cache_path (same filesystem -> no EXDEV)
+  # then rename. Ensures a partial write or interrupted process never
+  # replaces a previously-good cache.
+  if (!dir.exists(cache_dir)) dir.create(cache_dir, recursive = TRUE)
+  tmp <- paste0(cache_path, ".tmp.", Sys.getpid())
+  on.exit(if (file.exists(tmp)) unlink(tmp), add = TRUE)
+  writeLines(body, tmp)
+  file.rename(tmp, cache_path)
+  parsed
 }
 
 #' @noRd
@@ -36,11 +128,11 @@
 }
 
 #' @noRd
-.validate_schema <- function(manifest, expected_version) {
+.validate_schema <- function(manifest, supported = c(4L, 5L)) {
-  if (manifest$schema_version != expected_version) {
+  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."
     ))
   }
@@ -54,3 +146,18 @@
 }
 
 `%||%` <- function(a, b) if (is.null(a) || (length(a) == 1 && is.na(a))) b else a
+
+#' Return the parsed corpus manifest for the active session.
+#'
+#' Opens a session (connecting to the configured corpus) if none is active,
+#' then returns the manifest exactly as parsed from `manifest.json`. Useful
+#' for consumers that need the published year range (`years` block, schema
+#' v5+) or the partition list without issuing a data query.
+#'
+#' @return Named list: `schema_version`, `built_at`, `pipeline_commit`,
+#'   `data_vintage`, `scope`, `years` (schema v5+), `schema`, `files`.
+#' @export
+cog_manifest <- function() {
+  .ensure_session()
+  .uscogdata_env$manifest
+}

R/peers.R

@@ -2,31 +2,33 @@
 
 #' Find peer governments by similarity criteria
 #'
-#' Selects peer governments from `canonical_fips_xwalk` by combinations of
+#' Selects peer governments by combinations of government type, state, and
-#' government type, state, and population range. Peers are ordered by
+#' population range at a chosen `year`. Peers are ordered by `|log(pop_ratio)|`
-#' `|log(pop_ratio)|` ascending (closest to the target's population first).
+#' ascending (closest to the target's population first).
 #'
 #' @param target_govid Character scalar — `canonical_govid` of the target.
+#' @param year Integer scalar. Cohort vintage. When `NULL` (default), uses the
+#'   most recent year for which the target has an observed population in
+#'   `gov_population_yearly`.
 #' @param same_type If `TRUE` (default) restrict peers to the target's
 #'   `govs_type`.
 #' @param same_state If `TRUE` restrict peers to the target's `fips_state`.
 #'   Default `FALSE`.
 #' @param pop_range Length-2 numeric vector giving lower/upper bounds.
 #' @param is_ratio If `TRUE` (default) `pop_range` is multiplied by the
-#'   target's `population_acs` to produce absolute bounds. If `FALSE`,
+#'   target's population at `year` to produce absolute bounds. If `FALSE`,
 #'   `pop_range` is interpreted as absolute population counts.
-#' @param pop_year Reserved for future use (selecting ACS vintage). Currently
-#'   the corpus has a single snapshot so this argument has no effect.
 #' @param max_peers Integer cap on the number of peers returned.
 #' @return Tibble with columns `canonical_govid`, `gov_name`, `fips_state`,
-#'   `population_acs`, `pop_ratio`, `rank`.
+#'   `population`, `pop_ratio`, `rank`. The cohort year is attached as
+#'   `attr(x, "cohort_year")`.
 #' @export
 cog_find_peers <- function(target_govid,
+                           year = NULL,
                            same_type = TRUE,
                            same_state = FALSE,
                            pop_range = c(0.7, 1.3),
                            is_ratio = TRUE,
-                           pop_year = NULL,
                            max_peers = 10L) {
   if (!is.character(target_govid) || length(target_govid) != 1L) {
     cli::cli_abort("`target_govid` must be a length-1 character string.")
@@ -35,58 +37,96 @@ cog_find_peers <- function(target_govid,
       pop_range[1] >= pop_range[2]) {
     cli::cli_abort("`pop_range` must be a length-2 numeric with lo < hi.")
   }
+  if (!is.null(year) &&
+      (!(is.numeric(year) || is.integer(year)) || length(year) != 1L)) {
+    cli::cli_abort("`year` must be NULL or a length-1 integer.")
+  }
 
   con <- .ensure_session()
 
-  target_sql <- sprintf(
+  # Confirm target exists in the xwalk and pull govs_type / fips_state.
-    "SELECT canonical_govid, gov_name, govs_type, fips_state, population_acs
+  meta_sql <- sprintf(
+    "SELECT canonical_govid, gov_name, govs_type, fips_state
      FROM canonical_fips_xwalk
      WHERE canonical_govid = %s",
     .sql_lit_chr(target_govid)
   )
-  target <- DBI::dbGetQuery(con, target_sql)
+  meta <- DBI::dbGetQuery(con, meta_sql)
-  if (nrow(target) == 0L) {
+  if (nrow(meta) == 0L) {
     cli::cli_abort(c(
       "govid {target_govid} not found in corpus.",
       i = "v0.1 covers types 0-3 only (state/county/city/township); see vignette('coverage-scope')."
     ))
   }
-  if (is.na(target$population_acs) || target$population_acs <= 0) {
+
-    cli::cli_abort("Target {target_govid} has missing or non-positive population; cannot build pop_ratio band.")
+  cohort_year <- .resolve_cohort_year(con, target_govid, year)
+
+  pop_sql <- sprintf(
+    "SELECT population FROM gov_population_yearly
+     WHERE canonical_govid = %s AND year = %d",
+    .sql_lit_chr(target_govid), as.integer(cohort_year)
+  )
+  target_pop <- DBI::dbGetQuery(con, pop_sql)$population
+  if (length(target_pop) == 0L || is.na(target_pop) || target_pop <= 0) {
+    cli::cli_abort(c(
+      "Target {target_govid} has no observed population in {cohort_year}.",
+      i = "Use a year for which population is observed; see gov_population_yearly."
+    ))
   }
 
   if (isTRUE(is_ratio)) {
-    lo <- target$population_acs * pop_range[1]
+    lo <- target_pop * pop_range[1]
-    hi <- target$population_acs * pop_range[2]
+    hi <- target_pop * pop_range[2]
   } else {
     lo <- pop_range[1]; hi <- pop_range[2]
   }
 
   preds <- c(
-    sprintf("canonical_govid != %s", .sql_lit_chr(target_govid)),
+    sprintf("p.canonical_govid != %s", .sql_lit_chr(target_govid)),
-    sprintf("population_acs BETWEEN %.6f AND %.6f", lo, hi)
+    sprintf("p.year = %d", as.integer(cohort_year)),
+    sprintf("p.population BETWEEN %.6f AND %.6f", lo, hi)
   )
-  if (isTRUE(same_type))  preds <- c(preds, sprintf("govs_type = %d", target$govs_type))
+  if (isTRUE(same_type))  preds <- c(preds, sprintf("x.govs_type = %d", meta$govs_type))
-  if (isTRUE(same_state)) preds <- c(preds, sprintf("fips_state = %s", .sql_lit_chr(target$fips_state)))
+  if (isTRUE(same_state)) preds <- c(preds, sprintf("x.fips_state = %s", .sql_lit_chr(meta$fips_state)))
 
   peers_sql <- sprintf(
-    "SELECT canonical_govid, gov_name, fips_state, population_acs,
+    "SELECT p.canonical_govid, x.gov_name, x.fips_state, p.population,
-            population_acs / %.6f AS pop_ratio
+            p.population / %.6f AS pop_ratio
-     FROM canonical_fips_xwalk
+     FROM gov_population_yearly p
+     JOIN canonical_fips_xwalk x USING (canonical_govid)
      WHERE %s
-     ORDER BY ABS(LN(CAST(population_acs AS DOUBLE) / %.6f))
+     ORDER BY ABS(LN(CAST(p.population AS DOUBLE) / %.6f))
      LIMIT %d",
-    target$population_acs,
+    target_pop,
     paste(preds, collapse = " AND "),
-    target$population_acs,
+    target_pop,
     as.integer(max_peers)
   )
   peers <- tibble::as_tibble(DBI::dbGetQuery(con, peers_sql))
-  if (nrow(peers) > 0L) peers$rank <- seq_len(nrow(peers))
+  peers$rank <- if (nrow(peers) > 0L) seq_len(nrow(peers)) else integer(0)
-  else peers$rank <- integer(0)
+  attr(peers, "cohort_year") <- as.integer(cohort_year)
+  attr(peers, "pop_range")   <- as.numeric(pop_range)
+  attr(peers, "is_ratio")    <- isTRUE(is_ratio)
   peers
 }
 
+#' @noRd
+.resolve_cohort_year <- function(con, target_govid, year) {
+  if (!is.null(year)) return(as.integer(year))
+  sql <- sprintf(
+    "SELECT MAX(year) AS y FROM gov_population_yearly
+     WHERE canonical_govid = %s",
+    .sql_lit_chr(target_govid)
+  )
+  y <- DBI::dbGetQuery(con, sql)$y
+  if (length(y) == 0L || is.na(y)) {
+    cli::cli_abort(
+      "Target {target_govid} has no observed population in any year."
+    )
+  }
+  as.integer(y)
+}
+
 #' Compare a target government against a peer set
 #'
 #' Pulls spending for the target plus a peer set (either a
@@ -105,9 +145,12 @@ cog_find_peers <- function(target_govid,
 #' @param adjust_to_year Integer base year for CPI-U conversion or `NULL`.
 #' @return Tibble matching [cog_spending()]'s columns, plus a `role`
 #'   column taking values `"target"`, `"peer"`, `"summary_p25"`,
-#'   `"summary_p50"`, or `"summary_p75"`, and `target_rank` (target's rank
+#'   `"summary_p50"`, or `"summary_p75"`, `target_rank` (target's rank
-#'   among target+peers at `max(years)`, NA for other rows). Provenance
+#'   among target+peers at `max(years)`, NA for other rows), and
-#'   attribute reports `verb = "cog_peer_compare"` and `peer_count`.
+#'   `cohort_year` (the year used to build the peer cohort, read from
+#'   `attr(peers, "cohort_year")`; `NA` when `peers` was a bare character
+#'   vector). Provenance reports `verb = "cog_peer_compare"`, `peer_count`,
+#'   `cohort_year`, and `cohort_govids`.
 #' @export
 cog_peer_compare <- function(target_govid, peers, category, years,
                              per_capita = TRUE, adjust_to_year = NULL) {
@@ -115,6 +158,14 @@ cog_peer_compare <- function(target_govid, peers, category, years,
   if (!is.character(target_govid) || length(target_govid) != 1L) {
     cli::cli_abort("`target_govid` must be a length-1 character string.")
   }
+  cohort_year <- if (is.data.frame(peers)) {
+    ay <- attr(peers, "cohort_year")
+    if (is.null(ay)) NA_integer_ else as.integer(ay)
+  } else {
+    NA_integer_
+  }
+  pop_range <- if (is.data.frame(peers)) attr(peers, "pop_range") else NULL
+  is_ratio  <- if (is.data.frame(peers)) attr(peers, "is_ratio")  else NULL
   peer_govids <- if (is.data.frame(peers)) {
     as.character(peers$canonical_govid)
   } else {
@@ -132,11 +183,16 @@ cog_peer_compare <- function(target_govid, peers, category, years,
   out <- dplyr::bind_rows(r, summary_rows)
   rank_val <- .peer_target_rank(r, target_govid, years, value_col)
   out$target_rank <- ifelse(out$role == "target", rank_val, NA_integer_)
+  out$cohort_year <- cohort_year
 
   prov <- attr(r, "provenance") %||% list()
-  prov$verb       <- "cog_peer_compare"
+  prov$verb        <- "cog_peer_compare"
-  prov$call       <- paste(deparse(call), collapse = " ")
+  prov$call        <- paste(deparse(call), collapse = " ")
-  prov$peer_count <- length(peer_govids)
+  prov$peer_count  <- length(peer_govids)
+  prov$cohort_year <- cohort_year
+  prov$cohort_govids <- peer_govids
+  prov$pop_range   <- pop_range
+  prov$is_ratio    <- is_ratio
   prov$target     <- list(
     canonical_govid = target_govid,
     gov_name        = unique(r$gov_name[r$role == "target"])

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)),
@@ -61,9 +80,27 @@
       per_capita = list(
         applied = isTRUE(per_capita),
         denominator_source = if (isTRUE(per_capita)) {
-          "ACS 2018-2022 B01003_001 (population_acs from canonical_fips_xwalk)"
+          "Census F-33 population (per-year, from long.population)"
         } else {
           NA_character_
+        },
+        popyear_range = if (isTRUE(per_capita)) {
+          attr(result, ".popyear_range") %||% integer(0)
+        } else {
+          integer(0)
+        },
+        pop_source_counts = if (isTRUE(per_capita)) {
+          ps <- result[["pop_source"]]
+          if (is.null(ps) || length(ps) == 0L) {
+            list(census_f33 = 0L, unavailable = 0L)
+          } else {
+            list(
+              census_f33  = sum(ps == "census_f33", na.rm = TRUE),
+              unavailable = sum(ps == "unavailable", na.rm = TRUE)
+            )
+          }
+        } else {
+          NULL
         }
       ),
       inflation = list(
@@ -72,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

@@ -11,19 +11,23 @@
 #' @return Tibble with columns `year`, `canonical_govid`, `gov_name`,
 #'   `revenue_subtype`, `category`, `amt_nominal`, optional `amt_real`,
 #'   optional `amt_per_capita_nominal`, optional `amt_per_capita_real`,
-#'   `codes_included`, `aggregate_fallback`, `notes`.
+#'   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/rollup.R

@@ -8,28 +8,35 @@
 #' "place portraits" that compare a city to the surrounding county and
 #' containing state on one set of axes.
 #'
+#' When `per_capita = TRUE`, rows whose government has no observed
+#' population in that year (`pop_source == "unavailable"`) are dropped from
+#' the result. The dropped govids are recorded in
+#' `provenance$rollup$excluded_govids`. This excludes special districts
+#' (gov type 4) and school districts (gov type 5) from per-capita rollups
+#' by design — see `vignette('population-denominators')`.
+#'
 #' @param govids Named list with any non-empty subset of elements named
 #'   `state`, `county`, `city`. Each element is a character vector of
 #'   `canonical_govid` values. At least one layer required.
 #' @param category Single category name or character vector (passed through
 #'   to [cog_spending()]).
 #' @param years Integer vector of years.
-#' @param per_capita If `TRUE`, per-capita uses each layer's own population
+#' @param per_capita If `TRUE`, per-capita uses each gov's own per-year
-#'   from `canonical_fips_xwalk.population_acs`.
+#'   population from `gov_population_yearly`. Govs with missing population
+#'   are excluded from the result.
 #' @param adjust_to_year Integer base year for CPI-U conversion, or `NULL`.
 #' @return Tibble with columns `year`, `layer`, `canonical_govid`, `gov_name`,
 #'   `spend_subtype`, `category`, `amt_nominal`, optional `amt_real` /
-#'   `amt_per_capita_nominal` / `amt_per_capita_real`, `codes_included`,
+#'   `amt_per_capita_nominal` / `amt_per_capita_real`, optional `pop_source`,
-#'   `aggregate_fallback`, `scope_note`, `notes`. Carries a `provenance`
+#'   `codes_included`, `aggregate_fallback`, `scope_note`, `notes`. Carries a
-#'   attribute with `verb = "cog_geographic_rollup"` and `layers`.
+#'   `provenance` attribute with `verb = "cog_geographic_rollup"`, `layers`,
+#'   and `rollup$included_govids` / `rollup$excluded_govids`.
 #' @export
 cog_geographic_rollup <- function(govids, category, years,
                                   per_capita = FALSE, adjust_to_year = NULL) {
   call <- match.call()
   .validate_rollup_layers(govids)
 
-  # Accept character vector OR a data.frame with canonical_govid per layer,
-  # so cog_gov_search() output can be piped into one of the layer slots.
   govids <- lapply(govids, .coerce_govid_input, arg = "govids[[layer]]")
   if (any(lengths(govids) == 0L)) {
     cli::cli_abort("Each layer in `govids` must be non-empty after coercion.")
@@ -45,12 +52,25 @@ cog_geographic_rollup <- function(govids, category, years,
   r <- dplyr::left_join(r, layer_map, by = "canonical_govid",
                         relationship = "many-to-many")
   r$scope_note <- .rollup_scope_note(r$layer)
+
+  excluded <- character(0)
+  if (isTRUE(per_capita) && "pop_source" %in% names(r)) {
+    drop <- r$pop_source == "unavailable"
+    excluded <- unique(r$canonical_govid[drop])
+    r <- r[!drop, , drop = FALSE]
+  }
+  included <- unique(r$canonical_govid)
+
   r <- .reorder_rollup_cols(r)
 
   prov <- attr(r, "provenance")
   prov$verb   <- "cog_geographic_rollup"
   prov$call   <- paste(deparse(call), collapse = " ")
   prov$layers <- layer_names
+  prov$rollup <- list(
+    included_govids = included,
+    excluded_govids = excluded
+  )
   attr(r, "provenance") <- prov
 
   r

R/search.R

@@ -126,9 +126,10 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
     canonical_govid = character(0), gov_name = character(0),
     govs_type = integer(0), type_label = character(0),
     fips_state = character(0), fips_county = character(0),
-    fips_place = character(0), first_year = integer(0),
+    fips_place = character(0), legacy_govs_id = character(0),
-    last_year = integer(0), population_acs = integer(0),
+    first_year = integer(0), last_year = integer(0),
-    confidence = character(0)
+    census_geoid = character(0), population_acs = integer(0),
+    pop_confidence = character(0), id_source = character(0)
   )
 }
 

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

@@ -5,13 +5,14 @@
 #' @noRd
 cog_open <- function(url = .resolve_url(),
                      cache_dir = .resolve_cache_dir()) {
+  .check_url_configured(url)
   if (!dir.exists(cache_dir)) dir.create(cache_dir, recursive = TRUE)
 
   con <- DBI::dbConnect(duckdb::duckdb())
   DBI::dbExecute(con, "INSTALL httpfs; LOAD httpfs;")
 
   manifest <- .fetch_or_cache_manifest(url, cache_dir)
-  .validate_schema(manifest, expected_version = 3L)
+  .validate_schema(manifest, supported = c(4L, 5L))
   .validate_scope(manifest)
 
   .register_views(con, url, manifest)

R/spending.R

@@ -13,46 +13,101 @@
 #' @param category Character vector of category names (from
 #'   `summary_categories.category`), or `NULL` for all categories.
 #' @param per_capita If `TRUE`, adds `amt_per_capita_nominal` (and
-#'   `amt_per_capita_real` when `adjust_to_year` is set) using
+#'   `amt_per_capita_real` when `adjust_to_year` is set) using the per-year
-#'   `population_acs` from the canonical xwalk.
+#'   Census F-33 population from `gov_population_yearly`. Result also gains
+#'   a `pop_source` column with values `"census_f33"` or `"unavailable"`
+#'   (the latter for gov types 4/5 and any row whose population is missing
+#'   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`,
-#'   `codes_included`, `aggregate_fallback`, `notes`. Carries a `provenance`
+#'   optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
-#'   attribute matching `inst/schemas/provenance-v1.json`.
+#'   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)
 
-  sql <- .build_verb_sql(view, subtype_col, govid, years, category)
+  resolved <- .resolve_basis(basis, basis_explicit, manifest)
-  result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+
+  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)) {
@@ -61,27 +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, result, 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.")
   }
@@ -100,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)
@@ -143,18 +251,32 @@ cog_spending <- function(govid, years, category = NULL,
 .attach_per_capita <- function(result, con, govid) {
   if (nrow(result) == 0L) {
     result$amt_per_capita_nominal <- numeric(0)
+    result$pop_source <- character(0)
+    attr(result, ".popyear_range") <- integer(0)
     return(result)
   }
+  years_lit <- paste(unique(as.integer(result$year)), collapse = ",")
   sql <- sprintf(
-    "SELECT canonical_govid, population_acs
+    "SELECT canonical_govid, year, population, popyear
-     FROM canonical_fips_xwalk
+     FROM gov_population_yearly
-     WHERE canonical_govid IN (%s)",
+     WHERE canonical_govid IN (%s)
-    .sql_lit_chr(govid)
+       AND year IN (%s)",
+    .sql_lit_chr(govid), years_lit
   )
   pops <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
-  result <- dplyr::left_join(result, pops, by = "canonical_govid")
+  result <- dplyr::left_join(result, pops,
-  result$amt_per_capita_nominal <- result$amt_nominal / result$population_acs
+                             by = c("canonical_govid", "year"))
-  result$population_acs <- NULL
+  result$amt_per_capita_nominal <- result$amt_nominal / result$population
+  result$pop_source <- ifelse(is.na(result$population),
+                              "unavailable", "census_f33")
+  py <- result$popyear[!is.na(result$popyear)]
+  attr(result, ".popyear_range") <- if (length(py) > 0L) {
+    as.integer(c(min(py), max(py)))
+  } else {
+    integer(0)
+  }
+  result$population <- NULL
+  result$popyear <- NULL
   result
 }
 
@@ -176,10 +298,30 @@ cog_spending <- function(govid, years, category = NULL,
 
 #' @noRd
 .notes_column <- function(result) {
-  if (nrow(result) == 0L) return(character(0))
+  n <- nrow(result)
-  ifelse(
+  if (n == 0L) return(character(0))
-    isTRUE(result$aggregate_fallback) | result$aggregate_fallback %in% TRUE,
+  parts <- vector("list", 2L)
-    "Aggregate fallback applied; see cog_explain()",
+  agg <- result[["aggregate_fallback"]]
-    ""
+  parts[[1]] <- if (!is.null(agg)) {
-  )
+    ifelse(agg %in% TRUE,
+           "Aggregate fallback applied; see cog_explain()",
+           NA_character_)
+  } else {
+    rep(NA_character_, n)
+  }
+  ps <- result[["pop_source"]]
+  parts[[2]] <- if (!is.null(ps)) {
+    ifelse(ps == "unavailable",
+           "No population denominator available for this gov type",
+           NA_character_)
+  } else {
+    rep(NA_character_, n)
+  }
+  out <- character(n)
+  for (i in seq_len(n)) {
+    pieces <- vapply(parts, `[[`, character(1), i)
+    pieces <- pieces[!is.na(pieces)]
+    out[i] <- if (length(pieces) == 0L) "" else paste(pieces, collapse = "; ")
+  }
+  out
 }

R/suggestions.R

@@ -0,0 +1,119 @@
+# R/suggestions.R
+# Recipe-component-driven signposting: when a basis = "harmonized" query for
+# a category comes back with a coverage gap in some requested years (the
+# result has no rows at all in that year) that a harmonization recipe would
+# actually fill 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: signposting only runs when the caller
+# supplied a `category` (an un-scoped, all-categories query has no single
+# coverage question to answer) and only flags a recipe when the ACTUAL
+# result has zero rows in a requested year AND the candidate recipe's own
+# generic join (same join .run_recipe() uses, including its wide-era
+# aggregate rows) produces at least one row for this government in that
+# year. Checking presence per-government (not corpus-wide) avoids false
+# positives from ordinary reporting variance -- most governments don't use
+# every sibling code in a multi-code category every year, and that is not
+# a format-boundary gap worth signposting.
+
+#' Build the `prov$suggestions` list for a (non-recipe) basis = "harmonized"
+#' verb call: recipes whose generic join would fill a real gap in `result`.
+#'
+#' @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 result The verb's already-computed result tibble (post basis
+#'   query, pre per_capita/adjust_to_year).
+#' @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, result, basis) {
+  if (!identical(basis, "harmonized") || is.null(category)) return(list())
+
+  candidates <- DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT recipe_id FROM harmonization_recipes
+     WHERE component_code IN (
+       SELECT DISTINCT item_code FROM summary_categories WHERE category IN (%s)
+     )",
+    .sql_lit_chr(category)
+  ))$recipe_id
+  if (length(candidates) == 0L) return(list())
+
+  result_years <- if (is.null(result) || nrow(result) == 0L) {
+    integer(0)
+  } else {
+    unique(as.integer(result$year))
+  }
+  gap_years <- setdiff(as.integer(years), result_years)
+  if (length(gap_years) == 0L) return(list())
+
+  meta <- 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, year) pairs the recipe's own generic join actually
+  # covers for this government, restricted to the gap years -- the same
+  # join .run_recipe() uses (component year_min/year_max + gov_type_scope,
+  # no is_aggregate filter), just checking existence instead of summing.
+  covered <- DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT r.recipe_id, 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),
+    paste(gap_years, collapse = ",")
+  ))
+
+  suggestions <- list()
+  for (rid in candidates) {
+    if (!rid %in% covered$recipe_id) 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)

data-raw/regenerate_fixture_corpus.R

@@ -0,0 +1,234 @@
+# data-raw/regenerate_fixture_corpus.R
+#
+# Regenerate inst/extdata/fixture_corpus/ from a cog_pipeline publish tree.
+#
+# What this does:
+#   1. Copies each requested year's long partition as-is (byte-for-byte)
+#      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,
+#      summary_categories.parquet, harmonization_map.parquet,
+#      harmonization_recipes.parquet, and series_breaks.parquet metadata
+#      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/.
+#   4. Hand-builds manifest.json for just the files the fixture ships,
+#      following the shape of the previous fixture manifest but with
+#      schema_version bumped to whatever the source manifest reports, and
+#      freshly computed sha256 / row_count / size_bytes for every fixture
+#      file (never copied from the source manifest, since paths and byte
+#      layout can differ subtly between a full corpus and a fixture).
+#
+# This is never a manual job: run it whenever cog_pipeline publishes a new
+# corpus vintage that the fixture should track.
+#
+# Usage (from the uscogdata package root):
+#   Rscript data-raw/regenerate_fixture_corpus.R
+#   Rscript data-raw/regenerate_fixture_corpus.R /path/to/publish_cache
+#
+# Or from R:
+#   source("data-raw/regenerate_fixture_corpus.R")
+#   regenerate_fixture_corpus(publish_cache_dir = "/path/to/publish_cache")
+
+regenerate_fixture_corpus <- function(
+    publish_cache_dir = file.path(
+      "..", "cog_pipeline", "_targets", "publish_cache"
+    ),
+    fixture_dir = file.path("inst", "extdata", "fixture_corpus"),
+    fixture_years = c(2011L, 2012L, 2019L, 2020L)) {
+  stopifnot(
+    requireNamespace("digest", quietly = TRUE),
+    requireNamespace("jsonlite", quietly = TRUE),
+    requireNamespace("duckdb", quietly = TRUE),
+    requireNamespace("DBI", quietly = TRUE)
+  )
+
+  publish_cache_dir <- normalizePath(publish_cache_dir, mustWork = TRUE)
+  if (!dir.exists(fixture_dir)) dir.create(fixture_dir, recursive = TRUE)
+
+  source_manifest <- jsonlite::fromJSON(
+    file.path(publish_cache_dir, "manifest.json"),
+    simplifyVector = TRUE
+  )
+
+  .copy_long_partitions(publish_cache_dir, fixture_dir, fixture_years)
+  .copy_metadata_parquets(publish_cache_dir, fixture_dir)
+  .copy_docs(publish_cache_dir, fixture_dir)
+
+  manifest <- .build_fixture_manifest(
+    fixture_dir, source_manifest, fixture_years
+  )
+  manifest_path <- file.path(fixture_dir, "manifest.json")
+  writeLines(
+    jsonlite::toJSON(manifest, auto_unbox = TRUE, pretty = TRUE, null = "null"),
+    manifest_path
+  )
+
+  size_bytes <- sum(file.info(
+    list.files(fixture_dir, recursive = TRUE, full.names = TRUE)
+  )$size)
+  message(sprintf(
+    "Fixture corpus regenerated at %s (%.2f MB total).",
+    fixture_dir, size_bytes / 1024^2
+  ))
+  invisible(manifest)
+}
+
+# Copy each requested year's partition directory (just the parquet file
+# inside it) from the publish tree into the fixture, as-is.
+#' @noRd
+.copy_long_partitions <- function(publish_cache_dir, fixture_dir, years) {
+  for (yr in years) {
+    part_rel <- file.path("data", "long", sprintf("year=%d", yr), "part-0.parquet")
+    src <- file.path(publish_cache_dir, part_rel)
+    dst <- file.path(fixture_dir, part_rel)
+    if (!file.exists(src)) {
+      stop(sprintf("Source partition missing: %s", src))
+    }
+    dir.create(dirname(dst), recursive = TRUE, showWarnings = FALSE)
+    ok <- file.copy(src, dst, overwrite = TRUE)
+    if (!ok) stop(sprintf("Failed to copy %s -> %s", src, dst))
+  }
+  invisible(NULL)
+}
+
+# Copy the full (not year-scoped) canonical_fips_xwalk, canonical_alias,
+# 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",
+    "harmonization_map.parquet",
+    "harmonization_recipes.parquet",
+    "series_breaks.parquet"
+  )
+  for (f in files) {
+    src <- file.path(publish_cache_dir, "data", f)
+    dst <- file.path(fixture_dir, "data", f)
+    if (!file.exists(src)) {
+      stop(sprintf("Source metadata file missing: %s", src))
+    }
+    dir.create(dirname(dst), recursive = TRUE, showWarnings = FALSE)
+    ok <- file.copy(src, dst, overwrite = TRUE)
+    if (!ok) stop(sprintf("Failed to copy %s -> %s", src, dst))
+  }
+  invisible(NULL)
+}
+
+# Resync the four reference docs shipped alongside the fixture.
+#' @noRd
+.copy_docs <- function(publish_cache_dir, fixture_dir) {
+  docs <- c(
+    "data_dictionary.md", "reader-specification.md",
+    "README.md", "series_breaks.md"
+  )
+  dst_dir <- file.path(fixture_dir, "docs")
+  dir.create(dst_dir, recursive = TRUE, showWarnings = FALSE)
+  for (f in docs) {
+    src <- file.path(publish_cache_dir, "docs", f)
+    if (!file.exists(src)) {
+      stop(sprintf("Source doc missing: %s", src))
+    }
+    ok <- file.copy(src, file.path(dst_dir, f), overwrite = TRUE)
+    if (!ok) stop(sprintf("Failed to copy doc %s", f))
+  }
+  invisible(NULL)
+}
+
+# Count rows in a parquet file via an ephemeral DuckDB connection.
+#' @noRd
+.parquet_row_count <- function(path) {
+  con <- DBI::dbConnect(duckdb::duckdb())
+  on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+  DBI::dbGetQuery(con, sprintf(
+    "SELECT COUNT(*) AS n FROM read_parquet(%s)",
+    .sql_quote(path)
+  ))$n
+}
+
+#' @noRd
+.sql_quote <- function(x) paste0("'", gsub("'", "''", x), "'")
+
+# Hand-build manifest.json following the shape of the previous fixture
+# manifest: schema_version / built_at / pipeline_commit / fixture_note /
+# data_vintage / scope / schema / files.long_partitions / files.metadata /
+# series_breaks_ref / reader_spec_ref. Every sha256 / row_count / size_bytes
+# is freshly computed against the files actually written into fixture_dir.
+#' @noRd
+.build_fixture_manifest <- function(fixture_dir, source_manifest, years) {
+  long_partitions <- lapply(years, function(yr) {
+    rel <- file.path("data", "long", sprintf("year=%d", yr), "part-0.parquet")
+    path <- file.path(fixture_dir, rel)
+    list(
+      year       = as.integer(yr),
+      path       = gsub("\\\\", "/", rel),
+      sha256     = digest::digest(path, algo = "sha256", file = TRUE),
+      row_count  = as.integer(.parquet_row_count(path)),
+      size_bytes = as.integer(file.info(path)$size)
+    )
+  })
+
+  metadata_files <- c(
+    "canonical_alias.parquet",
+    "canonical_fips_xwalk.parquet",
+    "summary_categories.parquet",
+    "harmonization_map.parquet",
+    "harmonization_recipes.parquet",
+    "series_breaks.parquet"
+  )
+  metadata <- lapply(metadata_files, function(f) {
+    rel <- file.path("data", f)
+    path <- file.path(fixture_dir, rel)
+    list(
+      path        = gsub("\\\\", "/", rel),
+      sha256      = digest::digest(path, algo = "sha256", file = TRUE),
+      description = f
+    )
+  })
+
+  list(
+    schema_version = as.integer(source_manifest$schema_version),
+    built_at       = format(Sys.time(), "%Y-%m-%dT%H:%M:%SZ", tz = "UTC"),
+    pipeline_commit = source_manifest$pipeline_commit,
+    fixture_note = paste(
+      "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    = source_manifest$data_vintage,
+    scope           = source_manifest$scope,
+    schema          = source_manifest$schema,
+    files = list(
+      long_partitions = long_partitions,
+      metadata        = metadata
+    ),
+    series_breaks_ref = source_manifest$series_breaks_ref,
+    reader_spec_ref   = source_manifest$reader_spec_ref
+  )
+}
+
+if (identical(environment(), globalenv()) && sys.nframe() == 0L) {
+  args <- commandArgs(trailingOnly = TRUE)
+  if (length(args) >= 1L) {
+    regenerate_fixture_corpus(publish_cache_dir = args[[1]])
+  } else {
+    regenerate_fixture_corpus()
+  }
+}

inst/extdata/fixture_corpus/manifest.json

@@ -1,83 +1,84 @@
 {
-  "schema_version": 3,
+  "schema_version": 5,
-  "built_at": "2026-04-27T16:43:46Z",
+  "built_at": "2026-07-19T03:08:41Z",
-  "pipeline_commit": "899af37",
+  "pipeline_commit": "ece9b32",
-  "fixture_note": "Two-year (2019-2020) fixture for uscogdata tests. Full corpus available via USCOGDATA_URL.",
+  "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",
     "cpi_vintage": "FRED CPIAUCSL",
     "acs_vintage": "ACS 2018-2022 5-year"
   },
   "scope": {
-    "gov_types_included": [
+    "gov_types_included": [0, 1, 2, 3],
-      0,
+    "gov_types_excluded": [4, 5],
-      1,
-      2,
-      3
-    ],
-    "gov_types_excluded": [
-      4,
-      5
-    ],
     "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": 26,
-    "long_columns": [
+    "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", "harmonized_code", "survey_weight"],
-      "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"
-    ],
     "data_dictionary": "docs/data_dictionary.md"
   },
   "files": {
     "long_partitions": [
+      {
+        "year": 2011,
+        "path": "data/long/year=2011/part-0.parquet",
+        "sha256": "76c2153ef0a94c3551a24751aa08225fc937fafb579f10fd0750848d781dc471",
+        "row_count": 3037606,
+        "size_bytes": 4985667
+      },
+      {
+        "year": 2012,
+        "path": "data/long/year=2012/part-0.parquet",
+        "sha256": "a9bb10d04887490376a4b030a03a4ca1e0f1f9fbca5f7d99ea7ca7434c42b2bc",
+        "row_count": 1163338,
+        "size_bytes": 5900191
+      },
       {
         "year": 2019,
         "path": "data/long/year=2019/part-0.parquet",
-        "sha256": "e1c9f426c6d7d3c51d06b3a652473b987b304836619c213f019cee4887714daa",
+        "sha256": "475481fcfbb030f96cae457cbda824512a2da6f97f4c6d12a75130979e4331f9",
         "row_count": 318139,
-        "size_bytes": 1424231
+        "size_bytes": 1717345
       },
       {
         "year": 2020,
         "path": "data/long/year=2020/part-0.parquet",
-        "sha256": "9b795853a848e8c955c80261b96b79630fc77394dcfb1a1ca288e2cd634053a3",
+        "sha256": "7cdf3eb1a73c34e6a3d22befdf7ccdd0b44b1e6b1cf4491159a38a5a24a1b005",
         "row_count": 317500,
-        "size_bytes": 1427150
+        "size_bytes": 1720794
       }
     ],
     "metadata": [
+      {
+        "path": "data/canonical_alias.parquet",
+        "sha256": "fa27ce4b59e286ace05fbcb9f59a910b8e483fd6c58ddf01875a210b966809cb",
+        "description": "canonical_alias.parquet"
+      },
       {
         "path": "data/canonical_fips_xwalk.parquet",
-        "sha256": "86e53e04a35f6f90bb74bb1a273e053392afa782d6f518e3e3da9c976d47f7af",
+        "sha256": "b6e2c4cb748141f6f2e324d6704c13830d2a68ad57bbb73d04ac54b881d1285b",
         "description": "canonical_fips_xwalk.parquet"
       },
       {
         "path": "data/summary_categories.parquet",
-        "sha256": "60045e22bc2723318fa2cb73f8e5038250dc54d24b3447c6750dfe29035335b8",
+        "sha256": "8e6fcd4dd9bb4723841a67233b19388c9762dfc23b4479501183cebf7ea3c1b5",
         "description": "summary_categories.parquet"
+      },
+      {
+        "path": "data/harmonization_map.parquet",
+        "sha256": "52b4f3f94e65231aa869445946df7a0cfebf8fef1940f800e08287b53d09ab42",
+        "description": "harmonization_map.parquet"
+      },
+      {
+        "path": "data/harmonization_recipes.parquet",
+        "sha256": "1133e9a0b02f8f34f5f936e55c5ecd596bb8a55d8425dcce76767f0f3203581c",
+        "description": "harmonization_recipes.parquet"
+      },
+      {
+        "path": "data/series_breaks.parquet",
+        "sha256": "049bd7365dae14e357f6e57765dc5a069440bab46facce237d7ba2ec94b0b113",
+        "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/32-gov_population_yearly.sql

@@ -0,0 +1,8 @@
+CREATE OR REPLACE VIEW gov_population_yearly AS
+SELECT DISTINCT
+  year,
+  canonical_govid,
+  population,
+  popyear
+FROM long
+WHERE population IS NOT NULL;

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_find_peers.Rd

@@ -6,17 +6,21 @@
 \usage{
 cog_find_peers(
   target_govid,
+  year = NULL,
   same_type = TRUE,
   same_state = FALSE,
   pop_range = c(0.7, 1.3),
   is_ratio = TRUE,
-  pop_year = NULL,
   max_peers = 10L
 )
 }
 \arguments{
 \item{target_govid}{Character scalar — `canonical_govid` of the target.}
 
+\item{year}{Integer scalar. Cohort vintage. When `NULL` (default), uses the
+most recent year for which the target has an observed population in
+`gov_population_yearly`.}
+
 \item{same_type}{If `TRUE` (default) restrict peers to the target's
 `govs_type`.}
 
@@ -26,20 +30,18 @@ Default `FALSE`.}
 \item{pop_range}{Length-2 numeric vector giving lower/upper bounds.}
 
 \item{is_ratio}{If `TRUE` (default) `pop_range` is multiplied by the
-target's `population_acs` to produce absolute bounds. If `FALSE`,
+target's population at `year` to produce absolute bounds. If `FALSE`,
 `pop_range` is interpreted as absolute population counts.}
 
-\item{pop_year}{Reserved for future use (selecting ACS vintage). Currently
-the corpus has a single snapshot so this argument has no effect.}
-
 \item{max_peers}{Integer cap on the number of peers returned.}
 }
 \value{
 Tibble with columns `canonical_govid`, `gov_name`, `fips_state`,
-  `population_acs`, `pop_ratio`, `rank`.
+  `population`, `pop_ratio`, `rank`. The cohort year is attached as
+  `attr(x, "cohort_year")`.
 }
 \description{
-Selects peer governments from `canonical_fips_xwalk` by combinations of
+Selects peer governments by combinations of government type, state, and
-government type, state, and population range. Peers are ordered by
+population range at a chosen `year`. Peers are ordered by `|log(pop_ratio)|`
-`|log(pop_ratio)|` ascending (closest to the target's population first).
+ascending (closest to the target's population first).
 }

man/cog_geographic_rollup.Rd

@@ -22,17 +22,19 @@ to [cog_spending()]).}
 
 \item{years}{Integer vector of years.}
 
-\item{per_capita}{If `TRUE`, per-capita uses each layer's own population
+\item{per_capita}{If `TRUE`, per-capita uses each gov's own per-year
-from `canonical_fips_xwalk.population_acs`.}
+population from `gov_population_yearly`. Govs with missing population
+are excluded from the result.}
 
 \item{adjust_to_year}{Integer base year for CPI-U conversion, or `NULL`.}
 }
 \value{
 Tibble with columns `year`, `layer`, `canonical_govid`, `gov_name`,
   `spend_subtype`, `category`, `amt_nominal`, optional `amt_real` /
-  `amt_per_capita_nominal` / `amt_per_capita_real`, `codes_included`,
+  `amt_per_capita_nominal` / `amt_per_capita_real`, optional `pop_source`,
-  `aggregate_fallback`, `scope_note`, `notes`. Carries a `provenance`
+  `codes_included`, `aggregate_fallback`, `scope_note`, `notes`. Carries a
-  attribute with `verb = "cog_geographic_rollup"` and `layers`.
+  `provenance` attribute with `verb = "cog_geographic_rollup"`, `layers`,
+  and `rollup$included_govids` / `rollup$excluded_govids`.
 }
 \description{
 Wraps [cog_spending()], tags each row with its layer, and attaches a
@@ -41,3 +43,11 @@ human-readable `scope_note` documenting geographic-scope caveats (e.g.
 "place portraits" that compare a city to the surrounding county and
 containing state on one set of axes.
 }
+\details{
+When `per_capita = TRUE`, rows whose government has no observed
+population in that year (`pop_source == "unavailable"`) are dropped from
+the result. The dropped govids are recorded in
+`provenance$rollup$excluded_govids`. This excludes special districts
+(gov type 4) and school districts (gov type 5) from per-capita rollups
+by design — see `vignette('population-denominators')`.
+}

man/cog_manifest.Rd

@@ -0,0 +1,18 @@
+% Generated by roxygen2: do not edit by hand
+% Please edit documentation in R/manifest.R
+\name{cog_manifest}
+\alias{cog_manifest}
+\title{Return the parsed corpus manifest for the active session.}
+\usage{
+cog_manifest()
+}
+\value{
+Named list: `schema_version`, `built_at`, `pipeline_commit`,
+  `data_vintage`, `scope`, `years` (schema v5+), `schema`, `files`.
+}
+\description{
+Opens a session (connecting to the configured corpus) if none is active,
+then returns the manifest exactly as parsed from `manifest.json`. Useful
+for consumers that need the published year range (`years` block, schema
+v5+) or the partition list without issuing a data query.
+}

man/cog_peer_compare.Rd

@@ -31,9 +31,12 @@ population.}
 \value{
 Tibble matching [cog_spending()]'s columns, plus a `role`
   column taking values `"target"`, `"peer"`, `"summary_p25"`,
-  `"summary_p50"`, or `"summary_p75"`, and `target_rank` (target's rank
+  `"summary_p50"`, or `"summary_p75"`, `target_rank` (target's rank
-  among target+peers at `max(years)`, NA for other rows). Provenance
+  among target+peers at `max(years)`, NA for other rows), and
-  attribute reports `verb = "cog_peer_compare"` and `peer_count`.
+  `cohort_year` (the year used to build the peer cohort, read from
+  `attr(peers, "cohort_year")`; `NA` when `peers` was a bare character
+  vector). Provenance reports `verb = "cog_peer_compare"`, `peer_count`,
+  `cohort_year`, and `cohort_govids`.
 }
 \description{
 Pulls spending for the target plus a peer set (either a

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{
@@ -21,17 +23,44 @@ cog_revenue(
 `summary_categories.category`), or `NULL` for all categories.}
 
 \item{per_capita}{If `TRUE`, adds `amt_per_capita_nominal` (and
-`amt_per_capita_real` when `adjust_to_year` is set) using
+`amt_per_capita_real` when `adjust_to_year` is set) using the per-year
-`population_acs` from the canonical xwalk.}
+Census F-33 population from `gov_population_yearly`. Result also gains
+a `pop_source` column with values `"census_f33"` or `"unavailable"`
+(the latter for gov types 4/5 and any row whose population is missing
+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`,
   `revenue_subtype`, `category`, `amt_nominal`, optional `amt_real`,
   optional `amt_per_capita_nominal`, optional `amt_per_capita_real`,
-  `codes_included`, `aggregate_fallback`, `notes`.
+  optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
 }
 \description{
 Mirror of [cog_spending()] for revenue categories. One row per

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{
@@ -21,18 +23,45 @@ cog_spending(
 `summary_categories.category`), or `NULL` for all categories.}
 
 \item{per_capita}{If `TRUE`, adds `amt_per_capita_nominal` (and
-`amt_per_capita_real` when `adjust_to_year` is set) using
+`amt_per_capita_real` when `adjust_to_year` is set) using the per-year
-`population_acs` from the canonical xwalk.}
+Census F-33 population from `gov_population_yearly`. Result also gains
+a `pop_source` column with values `"census_f33"` or `"unavailable"`
+(the latter for gov types 4/5 and any row whose population is missing
+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`,
   `spend_subtype`, `category`, `amt_nominal`, optional `amt_real`,
   optional `amt_per_capita_nominal`, optional `amt_per_capita_real`,
-  `codes_included`, `aggregate_fallback`, `notes`. Carries a `provenance`
+  optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
-  attribute matching `inst/schemas/provenance-v1.json`.
+  Carries a `provenance` attribute matching `inst/schemas/provenance-v1.json`.
 }
 \description{
 One row per `(year, canonical_govid, spend_subtype, category)`. Amounts are

specs/2026-04-29-per-year-population-denominator-design.md

@@ -0,0 +1,285 @@
+# Per-year population denominators in uscogdata
+
+**Date:** 2026-04-29
+**Status:** Design — pending implementation
+**Scope:** uscogdata 0.1 (pre-release; no version bump)
+**Related:** cog_pipeline (data dictionary updates)
+
+## Problem
+
+`uscogdata::cog_spending(per_capita = TRUE)` and `cog_revenue(per_capita = TRUE)`
+currently divide every year's nominal amount by a single static population value
+— `canonical_fips_xwalk.population_acs`, the ACS 2018-2022 5-year estimate.
+
+For a 24-year corpus (2000–2023) this introduces a systematic bias proportional
+to each government's population change over that span. Fast-growing places have
+their early-year per-capita numbers understated; shrinking places have theirs
+overstated. The bias commonly exceeds 20% and can exceed 50% for cities like
+Detroit. Provenance currently advertises this denominator explicitly, so the
+error is visible to careful users — but the default behavior produces wrong
+numbers.
+
+`cog_geographic_rollup()` has the same bug. `cog_find_peers()` /
+`cog_peer_compare()` use the same static value to define peer cohorts, which
+is defensible for matching but is no longer necessary now that per-year
+population is available.
+
+## Background — population sources
+
+| Source | What it is | Where it lives |
+|---|---|---|
+| **Census F-33 `population`** | Population value Census uses on each COG row to compute its own per-capita tables. Almost always a Population Estimates Program (PEP) estimate; sometimes lagged a year for fiscal-year alignment, recorded in `popyear` | `long.population`, `long.popyear` (per row) |
+| **PEP** (raw) | Census Bureau's official annual intercensal estimates. Distinct from F-33 because F-33 sometimes uses a lagged vintage | Not in corpus; available via tidycensus |
+| **ACS 5-year** | American Community Survey 5-year rolling average. Different methodology, includes margin of error, only available 2005-2009 onward | `canonical_fips_xwalk.population_acs` (one fixed vintage) |
+| **Decennial** | Actual count, every 10 years | Not in corpus |
+
+F-33 `population` is the right default: it's what Census itself uses, so per-
+capita results published by uscogdata reconcile with Census's own published
+tables.
+
+## Approach
+
+Use the per-row `population` already present in `long`, joined on
+`(canonical_govid, year)`. No new external data dependency. Coverage:
+
+- **Types 0–3** (state, county, city, township): observed every year by design
+- **Types 4–5** (special districts, schools): always NA — masked in
+  `cog_pipeline/R/read_modern.R` because the F-33 schema does not carry a
+  population value for these gov types
+
+Type-4 and type-5 govs return `NA` per-capita with a `pop_source = "unavailable"`
+flag and a note. No silent substitution.
+
+The architecture leaves the door open for future denominators (PEP, ACS,
+decennial) by surfacing `pop_source` as a first-class result column. Adding a
+new source later is a join change, not an API change.
+
+## Detailed design
+
+### New view: `gov_population_yearly`
+
+```sql
+-- inst/sql/32-gov_population_yearly.sql
+CREATE OR REPLACE VIEW gov_population_yearly AS
+SELECT DISTINCT
+  year,
+  canonical_govid,
+  population,
+  popyear
+FROM long
+WHERE population IS NOT NULL;
+```
+
+`SELECT DISTINCT` collapses the metadata column duplicated across each gov-year's
+item rows. A test asserts `(year, canonical_govid)` is unique to catch any
+future source-data divergence.
+
+### `cog_spending()` and `cog_revenue()`
+
+`.attach_per_capita()` (in `R/spending.R`) is rewritten to:
+
+1. Query `gov_population_yearly` for the requested govids and years.
+2. `LEFT JOIN` on `(canonical_govid, year)` so missing rows produce NA.
+3. Compute `amt_per_capita_nominal = amt_nominal / population`. NA when
+   population is NA.
+4. Drop `population` from the returned tibble (keep `pop_source` instead).
+
+Result tibble gains one new column when `per_capita = TRUE`:
+
+- `pop_source`: `"census_f33"` when a denominator was found, `"unavailable"`
+  when NA.
+
+`notes` is extended: when `pop_source == "unavailable"`, append
+`"No population denominator available for this gov type"`. The `notes` column
+is updated to concatenate multiple notes with `"; "` (it currently holds at
+most one).
+
+`amt_per_capita_real` is NA whenever `amt_per_capita_nominal` is NA.
+
+### `cog_geographic_rollup()`
+
+The current implementation does **not** sum amounts within a layer — it returns
+one row per `(year, canonical_govid, subtype, category)` tagged with its
+layer, intended for side-by-side "place portrait" comparisons (a city, the
+county containing it, the state containing both). That semantics is preserved.
+
+The only behavior change in this work is per-row exclusion when `per_capita = TRUE`:
+
+1. After `cog_spending()` returns with the per-row per-year denominator from
+   Task 3, drop rows where `pop_source == "unavailable"` so the result never
+   contains NA per-capita rows.
+2. Record the dropped `canonical_govid`s in `provenance$rollup$excluded_govids`
+   and the kept ones in `provenance$rollup$included_govids`.
+
+Documentation states explicitly: *Per-capita rollups include only governments
+observed in both the finance and population panels for the given year. Special
+districts and school districts (gov types 4 and 5) are therefore excluded from
+per-capita rollups by design.*
+
+Provenance gains:
+
+- `rollup.included_govids` — `canonical_govid`s present in the result
+- `rollup.excluded_govids` — `canonical_govid`s dropped for missing pop
+
+### `cog_find_peers()`
+
+Signature: `cog_find_peers(target_govid, year = NULL, pop_range = c(0.5, 2), ...)`
+
+- `year` is a single integer. When `NULL`, defaults to the most recent year
+  present in `gov_population_yearly` for the target.
+- Looks up target's `population` at `year`. Errors if NA, with a message
+  listing nearby years where target *is* observed.
+- Filters candidates by `gov_population_yearly.population` at the same `year`,
+  within `pop_range[1] * target_pop` and `pop_range[2] * target_pop`.
+- Orders by `|log(pop_ratio)|` ascending.
+
+Returned columns: `canonical_govid`, `gov_name`, `govs_type`, `fips_state`,
+`population`, `pop_ratio`, `rank`. The column previously named `population_acs`
+is renamed to `population`.
+
+The cohort year is attached as a tibble attribute: `attr(x, "cohort_year")`.
+
+### `cog_peer_compare()`
+
+Existing signature unchanged:
+`cog_peer_compare(target_govid, peers, category, years, per_capita = TRUE, adjust_to_year = NULL)`.
+The caller supplies `peers` (either a `cog_find_peers()` result tibble or a
+character vector of `canonical_govid`). The cohort year is implicit in
+whichever year the caller used to call `cog_find_peers()`.
+
+Behavior changes:
+
+- When `peers` is a tibble carrying `attr(peers, "cohort_year")`,
+  `cog_peer_compare()` reads it and stamps every result row with a constant
+  `cohort_year` column.
+- When `peers` is a bare character vector, `cohort_year` in the result is `NA`.
+- Provenance gets `cohort_year` (scalar or NA) and the cohort govids list.
+
+Users who want time-varying cohorts call `cog_find_peers()` per year and
+stitch the `cog_peer_compare()` results themselves — documented in the
+vignette with a worked example.
+
+### Provenance updates
+
+`provenance$transformations$per_capita` becomes:
+
+```r
+list(
+  applied = TRUE,
+  denominator_source = "Census F-33 population (per-year, from long.population)",
+  popyear_range = c(<min>, <max>),
+  pop_source_counts = list(census_f33 = N1, unavailable = N2)
+)
+```
+
+For peer compare results, additional provenance:
+
+```r
+list(
+  cohort_year = <int>,
+  cohort_govids = <character>,
+  pop_range = c(<lo>, <hi>)
+)
+```
+
+For rollup results, additional provenance:
+
+```r
+list(
+  rollup = list(
+    included_govids = <character>,
+    excluded_govids = <character>
+  )
+)
+```
+
+`R/explain.R` is updated to render the new fields.
+
+### Documentation
+
+**New vignette** `vignettes/population-denominators.Rmd`:
+
+1. The four population sources explained
+2. Why F-33 is the default — and how it reconciles with Census's own per-capita
+   tables
+3. The `popyear` quirk: Census sometimes uses a lagged estimate for fiscal-year
+   alignment. Recorded in provenance, not in the result.
+4. Worked example showing the bias from the old static-ACS approach versus
+   per-year F-33 (e.g., Detroit 2003 vs. 2023)
+5. Worked example of a rolling-cohort peer comparison built by looping
+   `cog_peer_compare()` per year
+6. Future direction: `pop_source` is structured so PEP, ACS time-series, or
+   decennial denominators can be added later without API changes
+
+**`cog_pipeline/docs/data_dictionary.md`** entry for `long.population` and
+`long.popyear`: definition, source (F-33 fixed-width files, byte ranges),
+type-4/5 masking rule, relationship to PEP.
+
+### Tests
+
+- `gov_population_yearly` returns one row per `(year, canonical_govid)` (uniqueness)
+- `cog_spending(per_capita = TRUE)` returns different denominators for
+  different years for a known gov in the fixture (use any gov whose population
+  changes between 2019 and 2020)
+- Type-4 and type-5 govids in the fixture return `pop_source = "unavailable"`
+  and `NA` per-capita with the expected note
+- `cog_geographic_rollup(per_capita = TRUE)` excludes missing-pop govs and
+  records them in provenance
+- `cog_find_peers()` defaults `year` to the most recent year for a target
+  with known population history
+- `cog_find_peers()` errors with a helpful message when target has no observed
+  population in the requested year
+- `cog_peer_compare()` defaults `cohort_year` and produces a result with a
+  constant `cohort_year` column
+- Provenance carries `denominator_source`, `popyear_range`, and
+  `pop_source_counts`
+- Regression test against a fixed govid+year showing the new per-capita value
+  differs from the old (static-ACS) by exactly the ratio of `population_acs`
+  to `long.population` for that gov-year
+
+### Migration
+
+Pre-release; no version bump. `NEWS.md` Unreleased entry:
+
+> **Per-capita denominators now use per-year Census F-33 population.**
+> Previously, `cog_spending()` and `cog_revenue()` divided all years' amounts
+> by a single ACS 2018-2022 population, producing biased per-capita values
+> for time-series. They now divide by the F-33 `population` recorded for each
+> gov-year. Type-4 (special districts) and type-5 (school districts) govs
+> return `NA` per-capita with `pop_source = "unavailable"`.
+>
+> **Peer matching now uses per-year population.** `cog_find_peers()` gains a
+> `year` argument (defaults to most recent observed year). `cog_peer_compare()`
+> gains `cohort_year`. Cohorts are still fixed for a single peer-compare call;
+> users wanting moving cohorts loop themselves.
+>
+> **Rollups exclude govs with missing population.** `cog_geographic_rollup()`
+> per-capita totals include only govs where both the finance variable and
+> population are observed in that year; excluded govids are recorded in
+> provenance.
+>
+> Returned column `population_acs` from `cog_find_peers()` is renamed to
+> `population` and reflects the cohort-year vintage.
+
+### File impact
+
+| File | Change |
+|---|---|
+| `inst/sql/32-gov_population_yearly.sql` | New |
+| `R/spending.R` (`.attach_per_capita`, `.notes_column`) | Per-year join, `pop_source`, multi-note concat |
+| `R/peers.R` (`cog_find_peers`, `cog_peer_compare`) | `year` / `cohort_year` args, query new view, column rename |
+| `R/rollup.R` | Skip-with-record for missing-pop govs |
+| `R/provenance.R` | New denominator/cohort/rollup fields |
+| `R/explain.R` | Render new fields |
+| `vignettes/population-denominators.Rmd` | New |
+| `tests/testthat/` | Per-year denominator, type-4/5, rollup exclusion, peer cohort, provenance |
+| `cog_pipeline/docs/data_dictionary.md` | Document `long.population`, `long.popyear`, masking |
+| `NEWS.md` | Unreleased entry |
+
+## Out of scope
+
+- PEP/ACS/decennial denominators — architected for, not implemented
+- `per_pupil` denominator using `long.enrollment` for type-5 — deferred
+- Covering-county fallback for type-4 — deliberately not done
+- Backfilling population for type-4/5 from any external source
+- Changes to `cog_explorer` callers — separate follow-up, after this lands

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-adjust.R

@@ -41,3 +41,10 @@ test_that(".inflate preserves NA amounts", {
   expect_true(is.na(result[2]))
   expect_false(any(is.na(result[c(1, 3)])))
 })
+
+test_that("bundled CPI covers the full 1967+ corpus era through this year", {
+  cpi <- .cpi_table()
+  expect_lte(min(cpi$year), 1967L)
+  expect_gte(max(cpi$year), as.integer(format(Sys.Date(), "%Y")))
+  expect_false(any(is.na(cpi$cpi)))
+})

tests/testthat/test-explain.R

@@ -1,6 +1,6 @@
 test_that("cog_explain prints verb header and target", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections")
+  r <- cog_spending("121011212191", 2020L, "Corrections")
   # cli writes to stderr; capture both stdout and message streams.
   txt <- paste(c(
     capture.output(cog_explain(r)),
@@ -8,19 +8,19 @@ test_that("cog_explain prints verb header and target", {
   ), collapse = "\n")
   expect_true(grepl("cog_spending", txt))
   expect_true(grepl("Corrections", txt))
-  expect_true(grepl("101006006", txt))
+  expect_true(grepl("121011212191", txt))
 })
 
 test_that("cog_explain format='list' returns structured provenance", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections")
+  r <- cog_spending("121011212191", 2020L, "Corrections")
   prov <- cog_explain(r, format = "list")
   expect_identical(prov, attr(r, "provenance"))
 })
 
 test_that("cog_explain returns result invisibly for chaining", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections")
+  r <- cog_spending("121011212191", 2020L, "Corrections")
   res <- withVisible(cog_explain(r))
   expect_false(res$visible)
   expect_identical(res$value, r)
@@ -30,3 +30,60 @@ test_that("cog_explain errors on non-verb input", {
   df <- tibble::tibble(a = 1)
   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({
+    r <- cog_spending("121011212191", years = 2019:2020,
+                      category = "Police", per_capita = TRUE)
+    out <- paste(c(
+      capture.output(cog_explain(r)),
+      capture.output(cog_explain(r), type = "message")
+    ), collapse = "\n")
+    expect_true(grepl("Census F-33", out))
+    expect_true(grepl("popyear", out, ignore.case = TRUE))
+    expect_true(grepl("census_f33", out))
+    # popyear_range should render as 4-digit calendar years, not raw 2-digit
+    expect_true(grepl("2019-2020", out))
+    expect_false(grepl("popyear range: 19-20", out, fixed = TRUE))
+  })
+})

tests/testthat/test-manifest.R

@@ -0,0 +1,150 @@
+# tests/testthat/test-manifest.R
+#
+# Tests for the guards on .fetch_or_cache_manifest() and cog_open() that
+# protect users from silent failures when USCOGDATA_URL is misconfigured
+# or returns non-JSON content.
+
+test_that("cog_open aborts with actionable error when URL is the placeholder default", {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+
+  placeholder <- "https://cloud.civilytics.org/s/REPLACE_WITH_SHARE_TOKEN/download/"
+  withr::with_envvar(c(USCOGDATA_URL = placeholder), {
+    expect_error(
+      uscogdata:::cog_open(),
+      class = "uscogdata_url_not_configured"
+    )
+  })
+})
+
+test_that("placeholder guard fires for any URL containing the sentinel token", {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+
+  # Sentinel detection should be substring-based — covers any host that still
+  # has REPLACE_WITH_SHARE_TOKEN baked in (default or partial user edit).
+  withr::with_envvar(c(USCOGDATA_URL = "https://other.example/s/REPLACE_WITH_SHARE_TOKEN/x/"), {
+    expect_error(
+      uscogdata:::cog_open(),
+      class = "uscogdata_url_not_configured"
+    )
+  })
+})
+
+test_that("placeholder guard error names both env var and option as remediation", {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+
+  placeholder <- "https://cloud.civilytics.org/s/REPLACE_WITH_SHARE_TOKEN/download/"
+  withr::with_envvar(c(USCOGDATA_URL = placeholder), {
+    msg <- tryCatch(uscogdata:::cog_open(), error = conditionMessage)
+    expect_match(msg, "USCOGDATA_URL", fixed = TRUE)
+    expect_match(msg, "uscogdata.url", fixed = TRUE)
+  })
+})
+
+test_that("local manifest containing HTML produces uscogdata_invalid_manifest, not raw parse error", {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+
+  tmp <- withr::local_tempdir()
+  writeLines(
+    c("<html>", "  <head><title>Welcome to our server</title></head>", "</html>"),
+    file.path(tmp, "manifest.json")
+  )
+
+  withr::with_envvar(c(USCOGDATA_URL = paste0(tmp, "/")), {
+    err <- expect_error(
+      uscogdata:::cog_open(),
+      class = "uscogdata_invalid_manifest"
+    )
+    expect_match(conditionMessage(err), "manifest", ignore.case = TRUE)
+  })
+})
+
+test_that("remote manifest fetch does not poison cache when response is HTML", {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+
+  tmp_cache <- withr::local_tempdir()
+  cache_path <- file.path(tmp_cache, "manifest.json")
+
+  # Pretend the cache already exists with stale-but-fresh-by-mtime HTML
+  # (simulating a previous poisoned write from the old behavior). When the
+  # fetcher sees invalid JSON in the cache, it must refetch rather than
+  # silently returning a parse error to the caller.
+  writeLines("<html>poisoned</html>", cache_path)
+  Sys.setFileTime(cache_path, Sys.time())  # ensure within TTL
+
+  # We don't have a live HTTP fixture here, so the refetch will fail at the
+  # network layer — but the failure should NOT be a jsonlite parse error on
+  # the cached HTML; it should be a network-level httr2 error. The cache
+  # file itself must remain untouched (no atomic-write half-states).
+  withr::with_envvar(
+    c(
+      USCOGDATA_URL = "https://invalid.localhost.uscogdata.test/",
+      USCOGDATA_CACHE_DIR = tmp_cache
+    ),
+    {
+      err <- tryCatch(uscogdata:::cog_open(), error = identity)
+      expect_s3_class(err, "error")
+      # Must not be a JSON lexical error on HTML.
+      expect_false(grepl("lexical error", conditionMessage(err), fixed = TRUE))
+    }
+  )
+
+  # Atomic write contract: no stray tmp files left behind in cache_dir.
+  expect_length(
+    list.files(tmp_cache, pattern = "manifest\\.json\\.tmp"),
+    0L
+  )
+})
+
+test_that("cog_manifest returns the active session's parsed manifest", {
+  with_fixture_corpus({
+    m <- cog_manifest()
+    expect_type(m, "list")
+    expect_true(m$schema_version >= 4L)
+    yrs <- vapply(m$files$long_partitions, function(p) as.integer(p$year),
+                  integer(1))
+    expect_setequal(yrs, c(2011L, 2012L, 2019L, 2020L))
+  })
+})
+
+test_that(".validate_schema accepts schema_version 4 and 5, rejects others", {
+  expect_silent(uscogdata:::.validate_schema(list(schema_version = 4L)))
+  expect_silent(uscogdata:::.validate_schema(list(schema_version = 5L)))
+  expect_error(
+    uscogdata:::.validate_schema(list(schema_version = 3L)),
+    "schema_version"
+  )
+  expect_error(
+    uscogdata:::.validate_schema(list(schema_version = 6L)),
+    "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-mirror.R

@@ -61,7 +61,7 @@ test_that("cog_mirror reads back via a fresh session against the mirror", {
   cog_close()
   options(uscogdata.url = paste0(normalizePath(tmp), "/"))
 
-  r <- cog_spending("101006006", 2020L, "Corrections")
+  r <- cog_spending("121011212191", 2020L, "Corrections")
   expect_gt(nrow(r), 0L)
-  expect_equal(unique(r$canonical_govid), "101006006")
+  expect_equal(unique(r$canonical_govid), "121011212191")
 })

tests/testthat/test-peers.R

@@ -1,29 +1,29 @@
 test_that("cog_find_peers returns same-type peers in the default pop band", {
   skip_if_no_corpus()
-  peers <- cog_find_peers("101006006")                # Broward County
+  peers <- cog_find_peers("121011212191")                # Broward County
   expect_s3_class(peers, "tbl_df")
   expected_cols <- c("canonical_govid", "gov_name", "fips_state",
-                     "population_acs", "pop_ratio", "rank")
+                     "population", "pop_ratio", "rank")
   expect_true(all(expected_cols %in% names(peers)))
   expect_true(all(peers$pop_ratio >= 0.7 & peers$pop_ratio <= 1.3))
-  expect_false("101006006" %in% peers$canonical_govid)
+  expect_false("121011212191" %in% peers$canonical_govid)
   expect_equal(peers$rank, seq_len(nrow(peers)))
 })
 
 test_that("cog_find_peers respects same_state restriction", {
   skip_if_no_corpus()
-  peers <- cog_find_peers("101006006", same_state = TRUE,
+  peers <- cog_find_peers("121011212191", same_state = TRUE,
                           pop_range = c(0.1, 10))
   expect_true(all(peers$fips_state == "12"))
 })
 
 test_that("cog_find_peers absolute pop range works", {
   skip_if_no_corpus()
-  peers <- cog_find_peers("101006006",
+  peers <- cog_find_peers("121011212191",
                           pop_range = c(1.5e6, 2.5e6),
                           is_ratio = FALSE, max_peers = 20L)
-  expect_true(all(peers$population_acs >= 1.5e6 &
+  expect_true(all(peers$population >= 1.5e6 &
-                  peers$population_acs <= 2.5e6))
+                  peers$population <= 2.5e6))
 })
 
 test_that("cog_find_peers errors cleanly on unknown govid", {
@@ -33,8 +33,8 @@ test_that("cog_find_peers errors cleanly on unknown govid", {
 
 test_that("cog_peer_compare accepts a cog_find_peers result directly", {
   skip_if_no_corpus()
-  peers <- cog_find_peers("101006006", max_peers = 4L)
+  peers <- cog_find_peers("121011212191", max_peers = 4L)
-  r <- cog_peer_compare("101006006", peers, "Police", years = 2020L)
+  r <- cog_peer_compare("121011212191", peers, "Police", years = 2020L)
   expect_s3_class(r, "tbl_df")
   expect_true("role" %in% names(r))
   expect_setequal(
@@ -47,8 +47,8 @@ test_that("cog_peer_compare accepts a cog_find_peers result directly", {
 test_that("cog_peer_compare accepts a character vector of govids", {
   skip_if_no_corpus()
   r <- cog_peer_compare(
-    "101006006",
+    "121011212191",
-    peers = c("441015015", "441220220"),     # Bexar, Tarrant
+    peers = c("481029175853", "481439135072"),     # Bexar, Tarrant
     category = "Police", years = 2020L
   )
   expect_true("peer" %in% r$role)
@@ -58,8 +58,8 @@ test_that("cog_peer_compare accepts a character vector of govids", {
 test_that("cog_peer_compare summary rows use real per-capita when requested", {
   skip_if_no_corpus()
   r <- cog_peer_compare(
-    "101006006",
+    "121011212191",
-    peers = c("441015015", "441220220", "231082082"),
+    peers = c("481029175853", "481439135072", "261163166615"),
     category = "Police", years = 2019:2020,
     per_capita = TRUE, adjust_to_year = 2022L
   )
@@ -72,8 +72,8 @@ test_that("cog_peer_compare summary rows use real per-capita when requested", {
 
 test_that("cog_peer_compare provenance reports the outer verb + peer count", {
   skip_if_no_corpus()
-  r <- cog_peer_compare("101006006",
+  r <- cog_peer_compare("121011212191",
-                        peers = c("441015015", "441220220"),
+                        peers = c("481029175853", "481439135072"),
                         category = "Police", years = 2020L)
   prov <- attr(r, "provenance")
   expect_equal(prov$verb, "cog_peer_compare")
@@ -82,9 +82,57 @@ test_that("cog_peer_compare provenance reports the outer verb + peer count", {
 
 test_that("cog_peer_compare handles zero peers gracefully", {
   skip_if_no_corpus()
-  r <- cog_peer_compare("101006006",
+  r <- cog_peer_compare("121011212191",
                         peers = character(0),
                         category = "Police", years = 2020L)
   expect_true(all(r$role == "target"))
   expect_equal(sum(grepl("^summary_", r$role)), 0L)
 })
+
+test_that("cog_find_peers defaults `year` to most recent observed year for target", {
+  skip_if_no_corpus()
+  peers <- cog_find_peers("121011212191")
+  expect_equal(attr(peers, "cohort_year"), 2020L)
+  # Returned column is now `population`, not `population_acs`
+  expect_true("population" %in% names(peers))
+  expect_false("population_acs" %in% names(peers))
+})
+
+test_that("cog_find_peers honors an explicit `year`", {
+  skip_if_no_corpus()
+  peers <- cog_find_peers("121011212191", year = 2019L)
+  expect_equal(attr(peers, "cohort_year"), 2019L)
+})
+
+test_that("cog_find_peers errors when target has no observed pop in `year`", {
+  skip_if_no_corpus()
+  expect_error(
+    cog_find_peers("121011212191", year = 1999L),
+    "no observed population"
+  )
+})
+
+test_that("cog_peer_compare stamps cohort_year from peers attribute", {
+  skip_if_no_corpus()
+  peers <- cog_find_peers("121011212191", year = 2019L, max_peers = 4L,
+                          pop_range = c(0.5, 1.5))
+  r <- cog_peer_compare("121011212191", peers, "Police", years = 2020L)
+  expect_true("cohort_year" %in% names(r))
+  expect_true(all(r$cohort_year == 2019L))
+  prov <- attr(r, "provenance")
+  expect_equal(prov$cohort_year, 2019L)
+  expect_equal(length(prov$cohort_govids), nrow(peers))
+  # pop_range and is_ratio propagate from cog_find_peers attrs
+  expect_equal(prov$pop_range, c(0.5, 1.5))
+  expect_true(prov$is_ratio)
+})
+
+test_that("cog_peer_compare cohort_year is NA for bare character peers", {
+  skip_if_no_corpus()
+  r <- cog_peer_compare(
+    "121011212191",
+    peers = c("481029175853", "481439135072"),
+    category = "Police", years = 2020L
+  )
+  expect_true(all(is.na(r$cohort_year)))
+})

tests/testthat/test-recipes.R

@@ -0,0 +1,216 @@
+# 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 -------------------------------------------------------
+
+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("no signposting when the result already has full year coverage", {
+  skip_if_no_corpus()
+  r <- cog_spending("121011212191", 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

@@ -1,24 +1,24 @@
 test_that("cog_revenue returns expected shape for Broward Property Tax 2020", {
   skip_if_no_corpus()
-  r <- cog_revenue("101006006", years = 2020L, category = "Property Tax")
+  r <- cog_revenue("121011212191", years = 2020L, category = "Property Tax")
   expect_s3_class(r, "tbl_df")
   expected_cols <- c("year", "canonical_govid", "gov_name", "revenue_subtype",
                      "category", "amt_nominal", "codes_included",
                      "aggregate_fallback", "notes")
   expect_true(all(expected_cols %in% names(r)))
-  expect_equal(unique(r$canonical_govid), "101006006")
+  expect_equal(unique(r$canonical_govid), "121011212191")
   expect_equal(unique(r$year), 2020L)
 })
 
 test_that("cog_revenue with no category filter returns multiple categories", {
   skip_if_no_corpus()
-  r <- cog_revenue("101006006", years = 2020L)
+  r <- cog_revenue("121011212191", years = 2020L)
   expect_gt(length(unique(r$category)), 1L)
 })
 
 test_that("cog_revenue with per_capita + adjust_to_year adds all columns", {
   skip_if_no_corpus()
-  r <- cog_revenue("101006006", 2020L,
+  r <- cog_revenue("121011212191", 2020L,
                    per_capita = TRUE, adjust_to_year = 2022L)
   expect_true(all(c("amt_nominal", "amt_real",
                     "amt_per_capita_nominal", "amt_per_capita_real") %in%
@@ -27,7 +27,7 @@ test_that("cog_revenue with per_capita + adjust_to_year adds all columns", {
 
 test_that("cog_revenue result has provenance attribute", {
   skip_if_no_corpus()
-  r <- cog_revenue("101006006", 2020L)
+  r <- cog_revenue("121011212191", 2020L)
   prov <- attr(r, "provenance")
   expect_equal(prov$verb, "cog_revenue")
   expect_true(grepl("revenue_annotated", prov$sql_query))
@@ -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-rollup.R

@@ -2,9 +2,9 @@ test_that("cog_geographic_rollup aggregates state + county + city layers", {
   skip_if_no_corpus()
   r <- cog_geographic_rollup(
     govids = list(
-      state  = "100000000",               # Florida state govt
+      state  = "120000226351",               # Florida state govt
-      county = "101006006",               # Broward County
+      county = "121011212191",               # Broward County
-      city   = "102006004"                # Fort Lauderdale City
+      city   = "122011161585"                # Fort Lauderdale City
     ),
     category = "Police",
     years = 2019:2020
@@ -23,7 +23,7 @@ test_that("cog_geographic_rollup aggregates state + county + city layers", {
 test_that("cog_geographic_rollup respects per_capita + adjust_to_year", {
   skip_if_no_corpus()
   r <- cog_geographic_rollup(
-    govids = list(county = "101006006", city = "102006004"),
+    govids = list(county = "121011212191", city = "122011161585"),
     category = "Police",
     years = 2020L,
     per_capita = TRUE,
@@ -43,8 +43,8 @@ test_that("cog_geographic_rollup respects per_capita + adjust_to_year", {
 test_that("cog_geographic_rollup scope_notes describe each layer", {
   skip_if_no_corpus()
   r <- cog_geographic_rollup(
-    govids = list(state = "100000000", county = "101006006",
+    govids = list(state = "120000226351", county = "121011212191",
-                  city = "102006004"),
+                  city = "122011161585"),
     category = "Police", years = 2020L
   )
   state_notes <- unique(r$scope_note[r$layer == "state"])
@@ -58,7 +58,7 @@ test_that("cog_geographic_rollup scope_notes describe each layer", {
 test_that("cog_geographic_rollup single-layer call works", {
   skip_if_no_corpus()
   r <- cog_geographic_rollup(
-    govids = list(county = c("101006006")),
+    govids = list(county = c("121011212191")),
     category = "Corrections",
     years = 2020L
   )
@@ -69,7 +69,7 @@ test_that("cog_geographic_rollup single-layer call works", {
 test_that("cog_geographic_rollup provenance reports the outer verb", {
   skip_if_no_corpus()
   r <- cog_geographic_rollup(
-    govids = list(state = "100000000", county = "101006006"),
+    govids = list(state = "120000226351", county = "121011212191"),
     category = "Police", years = 2020L
   )
   prov <- attr(r, "provenance")
@@ -80,7 +80,7 @@ test_that("cog_geographic_rollup provenance reports the outer verb", {
 
 test_that("cog_geographic_rollup accepts data.frames per layer", {
   skip_if_no_corpus()
-  fl_state  <- cog_gov_search("^FLORIDA STATE GOVT$", type = "state")
+  fl_state  <- cog_gov_search("^FLORIDA$", type = "state")
   broward   <- cog_gov_search("^BROWARD COUNTY$", state = "FL", type = "county")
   r <- cog_geographic_rollup(
     govids = list(state = fl_state, county = broward),
@@ -92,9 +92,47 @@ test_that("cog_geographic_rollup accepts data.frames per layer", {
 
 test_that("cog_geographic_rollup rejects invalid inputs", {
   expect_error(cog_geographic_rollup(list(), "Police", 2020L), "length")
-  expect_error(cog_geographic_rollup(c("101006006"), "Police", 2020L), "list")
+  expect_error(cog_geographic_rollup(c("121011212191"), "Police", 2020L), "list")
   expect_error(
-    cog_geographic_rollup(list(planet = "100000000"), "Police", 2020L),
+    cog_geographic_rollup(list(planet = "120000226351"), "Police", 2020L),
     "state|county|city"
   )
 })
+
+test_that("cog_geographic_rollup per-capita uses summed per-year populations", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_geographic_rollup(
+      govids   = list(state  = "010000226085",
+                      county = "121011212191"),
+      category = "Police",
+      years    = 2019:2020,
+      per_capita = TRUE
+    )
+    state_ops <- r[r$layer == "state" & r$spend_subtype == "operations", ]
+    county_ops <- r[r$layer == "county" & r$spend_subtype == "operations", ]
+    state_implied <- state_ops$amt_nominal / state_ops$amt_per_capita_nominal
+    county_implied <- county_ops$amt_nominal /
+      county_ops$amt_per_capita_nominal
+    # Per-year, per-layer denominator is the layer's own per-year population
+    expect_equal(state_implied[state_ops$year == 2019], 4874747, tolerance = 1)
+    expect_equal(state_implied[state_ops$year == 2020], 4903185, tolerance = 1)
+    expect_equal(county_implied[county_ops$year == 2019], 1935878, tolerance = 1)
+  })
+})
+
+test_that("cog_geographic_rollup records included/excluded govids in provenance", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_geographic_rollup(
+      govids   = list(county = "121011212191"),
+      category = "Police",
+      years    = 2019:2020,
+      per_capita = TRUE
+    )
+    prov <- attr(r, "provenance")
+    expect_true("rollup" %in% names(prov))
+    expect_true("121011212191" %in% prov$rollup$included_govids)
+    expect_true(is.character(prov$rollup$excluded_govids))
+  })
+})

tests/testthat/test-search.R

@@ -146,7 +146,7 @@ test_that(".resolve_basket_row exact match returns one row", {
   expect_equal(out$match_method, "exact")
   expect_equal(out$n_candidates, 1L)
   expect_equal(nrow(out$row), 1L)
-  expect_equal(out$row$canonical_govid, "101006006")
+  expect_equal(out$row$canonical_govid, "121011212191")
   expect_equal(out$row$gov_name, "BROWARD COUNTY")
 })
 
@@ -157,7 +157,7 @@ test_that(".resolve_basket_row exact match is case-insensitive", {
   )
   expect_equal(out$status, "resolved")
   expect_equal(out$match_method, "exact")
-  expect_equal(out$row$canonical_govid, "101006006")
+  expect_equal(out$row$canonical_govid, "121011212191")
 })
 
 test_that(".resolve_basket_row exact match honors per-row type", {
@@ -166,7 +166,7 @@ test_that(".resolve_basket_row exact match honors per-row type", {
     name = "SAN DIEGO CITY", state = "CA", type = "city", con = con
   )
   expect_equal(out$status, "resolved")
-  expect_equal(out$row$canonical_govid, "052037010")
+  expect_equal(out$row$canonical_govid, "062073207598")
 })
 
 test_that(".resolve_basket_row substring fallback resolves single match", {
@@ -177,7 +177,7 @@ test_that(".resolve_basket_row substring fallback resolves single match", {
   expect_equal(out$status, "resolved")
   expect_equal(out$match_method, "substring")
   expect_equal(out$n_candidates, 1L)
-  expect_equal(out$row$canonical_govid, "101006006")
+  expect_equal(out$row$canonical_govid, "121011212191")
 })
 
 test_that(".resolve_basket_row no_match returns 0-row tibble", {
@@ -207,15 +207,19 @@ test_that(".resolve_basket_row treats empty/whitespace name as no_match", {
 
 test_that(".resolve_basket_row largest_pop within single type", {
   # FL Miami substring matches 10 cities (all govs_type = 2), largest pop
-  # is MIAMI CITY at 443665.
+  # is MIAMI CITY at 443665. Under Phase P canonical naming, MIAMI-DADE
+  # COUNTY (govs_type = 1) also contains "Miami", so `type = "city"` pins
+  # the match set to a single type (as the query docs promise it will for
+  # per-row `type`), keeping this test's original intent: multiple
+  # same-type name matches resolve to the largest-population row.
   con <- uscogdata:::.ensure_session()
   out <- uscogdata:::.resolve_basket_row(
-    name = "Miami", state = "FL", type = NA_character_, con = con
+    name = "Miami", state = "FL", type = "city", con = con
   )
   expect_equal(out$status, "largest_pop")
   expect_equal(out$match_method, "substring")
   expect_gte(out$n_candidates, 2L)
-  expect_equal(out$row$canonical_govid, "102013013")
+  expect_equal(out$row$canonical_govid, "122086194757")
   expect_equal(out$row$gov_name, "MIAMI CITY")
 })
 
@@ -241,7 +245,7 @@ test_that(".resolve_basket_row resolves with type override on ambiguous case", {
   )
   expect_equal(out$status, "resolved")
   expect_equal(out$match_method, "substring")
-  expect_equal(out$row$canonical_govid, "052037010")
+  expect_equal(out$row$canonical_govid, "062073207598")
 })
 
 # ---- basket mode public surface ----
@@ -254,7 +258,7 @@ test_that("cog_gov_search basket mode resolves clean inputs in input order", {
   )
   expect_s3_class(basket, "tbl_df")
   expect_equal(nrow(basket), 3L)
-  expect_equal(basket$canonical_govid, c("101006006", "052037010", "442227001"))
+  expect_equal(basket$canonical_govid, c("121011212191", "062073207598", "482453176394"))
   expect_equal(basket$gov_name, c("BROWARD COUNTY", "SAN DIEGO CITY", "AUSTIN CITY"))
 })
 
@@ -284,7 +288,7 @@ test_that("cog_gov_search basket mode skips ambiguous and no_match rows", {
   ))
   # Broward resolves; San Diego ambiguous; Notarealplace no_match.
   expect_equal(nrow(basket), 1L)
-  expect_equal(basket$canonical_govid, "101006006")
+  expect_equal(basket$canonical_govid, "121011212191")
   res <- attr(basket, "resolution")
   expect_equal(nrow(res), 3L)
   expect_equal(res$status, c("resolved", "ambiguous", "no_match"))
@@ -306,20 +310,24 @@ test_that("cog_gov_search basket mode recycles single state", {
     state = "CA"
   )
   expect_equal(nrow(basket), 2L)
-  expect_equal(basket$canonical_govid, c("052037010", "052001009"))
+  expect_equal(basket$canonical_govid, c("062073207598", "062001123093"))
 })
 
 test_that("cog_gov_search basket mode within-type largest_pop records candidates", {
   skip_if_no_corpus()
+  # `type = "city"` for the Miami row pins the match set to govs_type = 2;
+  # under Phase P canonical naming MIAMI-DADE COUNTY also contains "Miami"
+  # and would otherwise make this an ambiguous (cross-type) match.
   basket <- suppressMessages(cog_gov_search(
     name  = c("Miami",  "OAKLAND CITY"),
-    state = c("FL",     "CA")
+    state = c("FL",     "CA"),
+    type  = c("city",   NA)
   ))
   expect_equal(nrow(basket), 2L)
   res <- attr(basket, "resolution")
   miami_row <- res[res$query_name == "Miami", ]
   expect_equal(miami_row$status, "largest_pop")
-  expect_equal(miami_row$canonical_govid, "102013013")
+  expect_equal(miami_row$canonical_govid, "122086194757")
   expect_gte(miami_row$n_candidates, 2L)
   expect_gte(nrow(miami_row$candidates[[1]]), 2L)
 })
@@ -396,7 +404,7 @@ test_that("cog_gov_search basket mode skips per-row excluded type without aborti
   ))
   # Broward should resolve; the special_district row should be no_match.
   expect_equal(nrow(basket), 1L)
-  expect_equal(basket$canonical_govid, "101006006")
+  expect_equal(basket$canonical_govid, "121011212191")
   res <- attr(basket, "resolution")
   expect_equal(res$status, c("resolved", "no_match"))
   # query_type should record what the user passed for the excluded-type row

tests/testthat/test-spending.R

@@ -1,12 +1,12 @@
 test_that("cog_spending returns expected shape for Broward Corrections 2020", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", years = 2020L, category = "Corrections")
+  r <- cog_spending("121011212191", years = 2020L, category = "Corrections")
   expect_s3_class(r, "tbl_df")
   expected_cols <- c("year", "canonical_govid", "gov_name", "spend_subtype",
                      "category", "amt_nominal", "codes_included",
                      "aggregate_fallback", "notes")
   expect_true(all(expected_cols %in% names(r)))
-  expect_equal(unique(r$canonical_govid), "101006006")
+  expect_equal(unique(r$canonical_govid), "121011212191")
   expect_equal(unique(r$year), 2020L)
   expect_equal(unique(r$category), "Corrections")
   expect_true(all(r$spend_subtype %in% c("operations", "capital")))
@@ -15,7 +15,7 @@ test_that("cog_spending returns expected shape for Broward Corrections 2020", {
 
 test_that("cog_spending vectorised years + categories", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2019:2020,
+  r <- cog_spending("121011212191", 2019:2020,
                     category = c("Corrections", "Police"))
   expect_true(all(r$year %in% 2019:2020))
   expect_true(all(r$category %in% c("Corrections", "Police")))
@@ -24,7 +24,7 @@ test_that("cog_spending vectorised years + categories", {
 
 test_that("cog_spending with per_capita adds per-capita nominal column", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections", per_capita = TRUE)
+  r <- cog_spending("121011212191", 2020L, "Corrections", per_capita = TRUE)
   expect_true("amt_per_capita_nominal" %in% names(r))
   expect_false("amt_real" %in% names(r))
   expect_false("amt_per_capita_real" %in% names(r))
@@ -34,7 +34,7 @@ test_that("cog_spending with per_capita adds per-capita nominal column", {
 
 test_that("cog_spending with adjust_to_year adds real column", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2019:2020, "Corrections",
+  r <- cog_spending("121011212191", 2019:2020, "Corrections",
                     adjust_to_year = 2022L)
   expect_true("amt_real" %in% names(r))
   r2019 <- dplyr::filter(r, year == 2019L)
@@ -43,7 +43,7 @@ test_that("cog_spending with adjust_to_year adds real column", {
 
 test_that("cog_spending with per_capita + adjust_to_year adds all columns", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections",
+  r <- cog_spending("121011212191", 2020L, "Corrections",
                     per_capita = TRUE, adjust_to_year = 2022L)
   expect_true(all(c("amt_nominal", "amt_real",
                     "amt_per_capita_nominal", "amt_per_capita_real") %in%
@@ -68,16 +68,16 @@ test_that("cog_spending for unknown govid returns empty tibble + informs", {
 test_that("cog_spending records found + missing govids in provenance", {
   skip_if_no_corpus()
   suppressMessages(
-    r <- cog_spending(c("101006006", "XXXINVALID"), 2020L, "Corrections")
+    r <- cog_spending(c("121011212191", "XXXINVALID"), 2020L, "Corrections")
   )
   prov <- attr(r, "provenance")
-  expect_equal(sort(prov$scope$govids_found), "101006006")
+  expect_equal(sort(prov$scope$govids_found), "121011212191")
   expect_equal(sort(prov$scope$govids_missing), "XXXINVALID")
 })
 
 test_that("cog_spending result has provenance attribute matching schema", {
   skip_if_no_corpus()
-  r <- cog_spending("101006006", 2020L, "Corrections")
+  r <- cog_spending("121011212191", 2020L, "Corrections")
   prov <- attr(r, "provenance")
   expect_type(prov, "list")
   expect_equal(prov$verb, "cog_spending")
@@ -93,7 +93,7 @@ test_that("cog_spending result has provenance attribute matching schema", {
 
 test_that("cog_spending rejects invalid inputs", {
   expect_error(cog_spending(list(), 2020L), "character|data frame")
-  expect_error(cog_spending("101006006", "2020"), "years")
+  expect_error(cog_spending("121011212191", "2020"), "years")
 })
 
 test_that("cog_spending accepts a cog_gov_search result directly", {
@@ -101,12 +101,12 @@ test_that("cog_spending accepts a cog_gov_search result directly", {
   picks <- cog_gov_search("^BROWARD COUNTY$", state = "FL", type = "county")
   expect_gt(nrow(picks), 0L)
   r <- cog_spending(picks, 2020L, "Corrections")
-  expect_equal(unique(r$canonical_govid), "101006006")
+  expect_equal(unique(r$canonical_govid), "121011212191")
 })
 
 test_that("cog_spending accepts a cog_find_peers result directly", {
   skip_if_no_corpus()
-  peers <- cog_find_peers("101006006", max_peers = 3L)
+  peers <- cog_find_peers("121011212191", max_peers = 3L)
   r <- cog_spending(peers, 2020L, "Police")
   expect_setequal(unique(r$canonical_govid),
                   sort(peers$canonical_govid))
@@ -128,3 +128,190 @@ test_that("cog_spending accepts a basket-mode cog_gov_search result", {
   expect_s3_class(spending, "tbl_df")
   expect_setequal(unique(spending$canonical_govid), basket$canonical_govid)
 })
+
+test_that("per_capita denominator is the per-year F-33 population", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", years = 2019:2020,
+                      category = "Police", per_capita = TRUE)
+    r_ops <- r[r$spend_subtype == "operations", ]
+    # Implied denominator from amt_nominal / amt_per_capita_nominal
+    implied_pop <- r_ops$amt_nominal / r_ops$amt_per_capita_nominal
+    names(implied_pop) <- r_ops$year
+    # Use absolute tolerance: within 1 person of per-year F-33 values.
+    # Hardcoded values are Broward County's per-year Census F-33 population
+    # from the bundled fixture (regenerated 2026-07-11 against cog_pipeline
+    # publish tree, pipeline_commit 1a00925, Phase P schema_version 4).
+    # 1,940,907 is the static ACS 2018-2022 5-year value the legacy
+    # implementation would use; we assert it is NOT what we get.
+    expect_true(abs(implied_pop[["2019"]] - 1935878) < 1)
+    expect_true(abs(implied_pop[["2020"]] - 1952778) < 1)
+    expect_false(all(abs(implied_pop - 1940907) < 1))
+  })
+})
+
+test_that("pop_source = 'census_f33' does not produce unavailable-pop note", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", years = 2019L,
+                      category = "Police", per_capita = TRUE)
+    expect_true(all(r$pop_source == "census_f33"))
+    expect_true(all(is.na(r$notes) | r$notes == "" |
+                    !grepl("No population denominator", r$notes)))
+  })
+})
+
+test_that("aggregate fallback + unavailable pop produce concatenated notes", {
+  # Unit-level test of .notes_column with a synthetic data frame so we don't
+  # depend on having a type-4/5 gov in the fixture.
+  result <- tibble::tibble(
+    aggregate_fallback = c(FALSE, TRUE,  TRUE),
+    pop_source         = c("census_f33", "census_f33", "unavailable")
+  )
+  notes <- uscogdata:::.notes_column(result)
+  expect_equal(notes[1], "")
+  expect_equal(notes[2], "Aggregate fallback applied; see cog_explain()")
+  expect_equal(notes[3],
+               "Aggregate fallback applied; see cog_explain(); No population denominator available for this gov type")
+})
+
+test_that("provenance records per-year denominator metadata", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", years = 2019:2020,
+                      category = "Police", per_capita = TRUE)
+    pc <- attr(r, "provenance")$transformations$per_capita
+    expect_true(pc$applied)
+    expect_match(pc$denominator_source, "Census F-33", fixed = FALSE)
+    expect_match(pc$denominator_source, "per-year", fixed = TRUE)
+    expect_equal(pc$pop_source_counts$census_f33, nrow(r))
+    expect_equal(pc$pop_source_counts$unavailable, 0L)
+    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 this fixture: none of the discontinued_na rulings
+    # (S74, Z61, X04, X06, the debt-detail family, L24) fall inside the
+    # E/F/G/K spending prefixes, so the exclusion count is exactly zero for
+    # every year in the bundled window -- see
+    # docs/phase_r_harmonization_review.md § 1.3/1.4.
+    expect_equal(prov$harmonization$na_rows_excluded, 0L)
+    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()
@@ -57,3 +195,36 @@ test_that("spending_annotated carries category + xwalk columns", {
     expect_true(nm %in% names(row), info = paste("missing column:", nm))
   }
 })
+
+test_that("gov_population_yearly exposes one row per (year, canonical_govid)", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    con <- uscogdata:::.ensure_session()
+    df <- DBI::dbGetQuery(
+      con,
+      "SELECT year, canonical_govid, population, popyear
+       FROM gov_population_yearly
+       WHERE canonical_govid = '121011212191'
+       ORDER BY year"
+    )
+    expect_setequal(df$year, c(2011L, 2012L, 2019L, 2020L))
+    expect_equal(nrow(df), 4L)
+    expect_true(all(!is.na(df$population)))
+    # Hardcoded values are from the bundled fixture (regenerated 2026-07-18
+    # against cog_pipeline publish tree, pipeline_commit ece9b32, Phase R2
+    # schema_version 5, years 2011/2012/2019/2020). Update if the fixture is
+    # 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.
+    dup <- DBI::dbGetQuery(
+      con,
+      "SELECT year, canonical_govid, COUNT(*) AS n
+       FROM gov_population_yearly
+       GROUP BY year, canonical_govid HAVING n > 1"
+    )
+    expect_equal(nrow(dup), 0L)
+  })
+})

vignettes/population-denominators.Rmd

@@ -0,0 +1,62 @@
+---
+title: "Population denominators"
+output: rmarkdown::html_vignette
+vignette: >
+  %\VignetteIndexEntry{Population denominators}
+  %\VignetteEngine{knitr::rmarkdown}
+  %\VignetteEncoding{UTF-8}
+---
+
+```{r setup, include = FALSE}
+knitr::opts_chunk$set(eval = FALSE, collapse = TRUE, comment = "#>")
+```
+
+# Why per-year population matters
+
+Per-capita finance numbers divide each year's spending or revenue by a population denominator. The choice of denominator is a research decision, not an implementation detail: a 24-year corpus paired with a single 5-year ACS estimate produces biased per-capita values whose magnitude scales with each government's population change.
+
+`uscogdata` defaults to the **Census F-33 population value Census itself uses to compute its published per-capita tables.** That value is recorded on every COG row as `population`, with `popyear` indicating the vintage. For a city that grew from 200,000 to 300,000 between 2000 and 2023, this default reproduces the per-capita value Census published. A static ACS denominator would have understated 2000 per-capita by ~33%.
+
+# The four population sources
+
+| Source | What it is | Default in uscogdata? |
+|---|---|---|
+| Census F-33 `population` | Population value Census used on each COG row to compute its published per-capita tables. Almost always a Population Estimates Program (PEP) estimate; sometimes lagged a year for fiscal-year alignment, recorded in `popyear`. | **Yes — default for `cog_spending(per_capita = TRUE)` etc.** |
+| PEP (raw) | Census Bureau's official annual intercensal estimates, distinct from F-33 because F-33 sometimes uses a lagged vintage. | No (not in corpus) |
+| ACS 5-year | American Community Survey 5-year rolling average. Different methodology, has margin of error, only available 2005-2009 onward. | Used by `cog_find_peers()` historically; replaced in 0.1 by per-year F-33. Still available in `canonical_fips_xwalk.population_acs` for non-time-series uses. |
+| Decennial count | Actual count, every 10 years. | No (not in corpus) |
+
+The F-33 denominator is preferred because it's the same value Census uses internally — so `uscogdata` per-capita numbers reconcile with Census's own published tables.
+
+# Coverage
+
+F-33 `population` is observed for gov types 0–3 (state, county, city, township). Gov types 4 (special districts) and 5 (school districts) have `population` masked to NA in the F-33 schema. uscogdata returns:
+
+- `pop_source = "census_f33"` and a numeric `amt_per_capita_*` for types 0–3.
+- `pop_source = "unavailable"` and `NA` per-capita for types 4–5, with a corresponding entry in `notes`.
+
+`cog_geographic_rollup(per_capita = TRUE)` excludes unavailable-pop rows from the result; the dropped govids are listed in `provenance\$rollup\$excluded_govids`.
+
+# The popyear quirk
+
+Census sometimes uses a population estimate from one year prior to the fiscal year being reported (e.g., FY2018 paired with a 2017 PEP estimate) so the denominator is available before the fiscal year closes. `popyear` records which vintage was paired; `cog_spending()` returns the popyear range in `provenance\$transformations\$per_capita\$popyear_range` rather than as a per-row column.
+
+# Time-varying peer cohorts
+
+`cog_find_peers(target, year = Y)` builds a cohort matched on each candidate's population at year `Y`. The cohort is fixed once chosen; `cog_peer_compare()` then runs that cohort across whatever `years` you ask for. To run a moving-window comparison, build cohorts year-by-year and stitch the results:
+
+```r
+years <- 2010:2023
+out <- purrr::map_dfr(years, function(y) {
+  peers <- cog_find_peers("261163166615", year = y, max_peers = 10L)
+  cog_peer_compare("261163166615", peers,
+                   category = "Police", years = y,
+                   per_capita = TRUE)
+})
+```
+
+Each row in `out` has `cohort_year == year`, so a faceted plot shows cohort drift directly.
+
+# Future direction
+
+`pop_source` is a column on the result, not a fixed value, so adding a new denominator (PEP from tidycensus, decennial counts, ACS time-series) is a join change rather than an API change. A future release may add `cog_spending(..., pop_source = "pep")` for users who need a single externally-audited series.