fix(#3): normalize the corpus URL's trailing slash at resolution

Investigating "gov search doesn't work" (#3) turned up two separate things.

THE REPORTED SYMPTOM IS ALREADY FIXED.
#3 reported `cog_gov_search("Orange")` dying in jsonlite with
`lexical error: invalid char in json text. <html> <head>`. That was fixed the
same day the issue was filed, by 8743472 "fix(manifest): actionable errors when
USCOGDATA_URL is unset or returns non-JSON" (issue filed 2026-05-27 11:30;
commit 2026-05-27). The issue was simply never closed. Verified now: injecting
an HTML manifest.json raises a typed `uscogdata_invalid_manifest` condition
naming the likely causes, with the raw parse error demoted to a footnote, and
`cog_gov_search("Orange")` returns 62 rows against the live corpus.

THE ROOT CAUSE OF THAT HTML WAS STILL LIVE -- and is what this commit fixes.

Every consumer builds locations by CONCATENATION:
  manifest.R:95   paste0(url, "manifest.json")
  mirror.R:48,125 paste0(url, e$path)
  views.R         the parquet glob
and mirror.R:104 documents the invariant outright ('url ends in "/"'). The
error messages tell users to set `"<url-or-local-path>/"`. But `.resolve_url()`
was a bare `.cfg("url")` passthrough -- the invariant was assumed everywhere and
enforced nowhere.

So a URL entered without the slash failed silently and misleadingly:
  HTTPS -> ".../downloadmanifest.json"; the host answers with an HTML 404 page,
           which lands in the JSON parser as EXACTLY the #3 symptom -- and the
           guard then blames "login page / 404 / wrong share" when the real
           cause was one missing character.
  local -> ".../corpusdata/long/**/*.parquet" and a DuckDB "No files found".

Reproduced both: pointing USCOGDATA_URL at the bundled fixture without a
trailing slash gave
  No files found that match ".../fixture_corpusdata/long/**/*.parquet"

Normalizing once at resolution fixes every consumer at the same time, rather
than having each call site re-derive the invariant. An empty setting passes
through untouched so manifest.R's "not configured" guard still fires instead of
the value degrading into a bare "/" filesystem root.

RED->GREEN: 4 tests added, 2 failed first (append-missing-slash, local-path
normalization); the already-correct cases (slash present, empty setting) passed
throughout and pin them against regression. Same fixture path that produced the
DuckDB error above now returns 62 rows.

Suite: FAIL 0 | WARN 0 | SKIP 0 | PASS 471 (was 463; +8 = the new tests).

.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
-MaxCorpusSchema: 3
+MinCorpusSchema: 4
+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/config.R

@@ -21,7 +21,30 @@
   .uscogdata_defaults[[key]]
 }
 
-.resolve_url <- function() .cfg("url")
+#' Resolve the corpus URL, guaranteeing the trailing slash the package assumes.
+#'
+#' Every consumer builds locations by CONCATENATION -- `paste0(url,
+#' "manifest.json")` in manifest.R, `paste0(url, e$path)` in mirror.R, and the
+#' parquet glob in views.R -- and mirror.R:104 documents the invariant outright
+#' ('url ends in "/"'). Nothing enforced it, so a URL entered without the slash
+#' failed silently and misleadingly:
+#'
+#'   HTTPS -> ".../downloadmanifest.json"; the host answers with an HTML 404
+#'            page, which lands in the JSON parser as the lexical error
+#'            reported in issue #3 -- pointing the user at "login page / wrong
+#'            share" when the real cause was one missing character.
+#'   local -> ".../corpusdata/long/**/*.parquet" and a DuckDB "No files found".
+#'
+#' Normalizing here fixes every consumer at once, rather than each call site
+#' re-deriving the same invariant. An empty setting is passed through
+#' untouched so manifest.R's "not configured" guard still fires instead of the
+#' value degrading into a bare "/" filesystem root.
+#' @noRd
+.resolve_url <- function() {
+  url <- .cfg("url")
+  if (is.null(url) || !nzchar(url) || grepl("/$", url)) return(url)
+  paste0(url, "/")
+}
 
 .resolve_cache_dir <- function() {
   v <- .cfg("cache_dir")

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) ||
-                 difftime(Sys.time(), file.info(cache_path)$mtime, units = "secs") > ttl
+  cache_fresh <- file.exists(cache_path) &&
+    difftime(Sys.time(), file.info(cache_path)$mtime, units = "secs") <= ttl
+
+  # Honor a fresh cache only if its contents still parse as JSON. A previous
+  # version of this package could write HTML directly into the cache; treat
+  # such poisoned caches as if they were missing so the next call recovers.
+  if (cache_fresh) {
+    parsed <- .try_parse_manifest_file(cache_path)
+    if (!is.null(parsed)) return(parsed)
+  }
 
-  if (needs_fetch) {
   resp <- httr2::request(paste0(url, "manifest.json")) |>
     httr2::req_error(is_error = function(r) httr2::resp_status(r) >= 400) |>
     httr2::req_perform()
-    writeLines(httr2::resp_body_string(resp), cache_path)
-  }
+  body <- httr2::resp_body_string(resp)
 
-  jsonlite::fromJSON(cache_path, simplifyVector = FALSE)
+  # 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,21 @@
 }
 
 #' @noRd
-.validate_schema <- function(manifest, expected_version) {
-  if (manifest$schema_version != expected_version) {
+#' Schema v6 (FIPS geography harmonization, 2026-07-22) is accepted alongside
+#' 4/5. v6 renamed the long table's fips_state_code/fips_county_code to
+#' fips_state_asof/fips_county_asof and added cog_legacy_state/
+#' cog_legacy_county (26 -> 28 cols); this package references NONE of those
+#' columns, so no code change was needed. NOTE the SILENT semantic change for
+#' any consumer of the raw long table: long fips_state/fips_county are now
+#' PRESENT/harmonized geography (current county identity carried back to every
+#' year, matching canonical_fips_xwalk) rather than as-of-year; as-of-year
+#' moved to the *_asof columns. This package's own geography always came from
+#' the xwalk (already present-based), so behaviour is unchanged.
+.validate_schema <- function(manifest, supported = c(4L, 5L, 6L)) {
+  if (!manifest$schema_version %in% supported) {
     cli::cli_abort(c(
       "Corpus schema version mismatch.",
-      x = "Package expects schema_version = {expected_version}; corpus has {manifest$schema_version}.",
+      x = "Package supports schema_version in {paste(supported, collapse = ', ')}; corpus has {manifest$schema_version}.",
       i = "Update uscogdata (install.packages or pak::pkg_install) or re-publish corpus."
     ))
   }
@@ -54,3 +156,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
-#' government type, state, and population range. Peers are ordered by
-#' `|log(pop_ratio)|` ascending (closest to the target's population first).
+#' Selects peer governments by combinations of government type, state, and
+#' population range at a chosen `year`. Peers are ordered by `|log(pop_ratio)|`
+#' 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(
-    "SELECT canonical_govid, gov_name, govs_type, fips_state, population_acs
+  # Confirm target exists in the xwalk and pull govs_type / fips_state.
+  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)
-  if (nrow(target) == 0L) {
+  meta <- DBI::dbGetQuery(con, meta_sql)
+  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]
-    hi <- target$population_acs * pop_range[2]
+    lo <- target_pop * pop_range[1]
+    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("population_acs BETWEEN %.6f AND %.6f", lo, hi)
+    sprintf("p.canonical_govid != %s", .sql_lit_chr(target_govid)),
+    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_state)) preds <- c(preds, sprintf("fips_state = %s", .sql_lit_chr(target$fips_state)))
+  if (isTRUE(same_type))  preds <- c(preds, sprintf("x.govs_type = %d", meta$govs_type))
+  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,
-            population_acs / %.6f AS pop_ratio
-     FROM canonical_fips_xwalk
+    "SELECT p.canonical_govid, x.gov_name, x.fips_state, p.population,
+            p.population / %.6f AS pop_ratio
+     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))
-  else peers$rank <- integer(0)
+  peers$rank <- if (nrow(peers) > 0L) seq_len(nrow(peers)) else 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
-#'   among target+peers at `max(years)`, NA for other rows). Provenance
-#'   attribute reports `verb = "cog_peer_compare"` and `peer_count`.
+#'   `"summary_p50"`, or `"summary_p75"`, `target_rank` (target's rank
+#'   among target+peers at `max(years)`, NA for other rows), and
+#'   `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$call        <- paste(deparse(call), collapse = " ")
   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
-#'   from `canonical_fips_xwalk.population_acs`.
+#' @param per_capita If `TRUE`, per-capita uses each gov's own per-year
+#'   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`,
-#'   `aggregate_fallback`, `scope_note`, `notes`. Carries a `provenance`
-#'   attribute with `verb = "cog_geographic_rollup"` and `layers`.
+#'   `amt_per_capita_nominal` / `amt_per_capita_real`, optional `pop_source`,
+#'   `codes_included`, `aggregate_fallback`, `scope_note`, `notes`. Carries a
+#'   `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),
-    last_year = integer(0), population_acs = integer(0),
-    confidence = character(0)
+    fips_place = character(0), legacy_govs_id = character(0),
+    first_year = integer(0), last_year = integer(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, 6L))
   .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
-#'   `population_acs` from the canonical xwalk.
+#'   `amt_per_capita_real` when `adjust_to_year` is set) using the per-year
+#'   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`
-#'   attribute matching `inst/schemas/provenance-v1.json`.
+#'   optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
+#'   Carries a `provenance` attribute matching `inst/schemas/provenance-v1.json`.
 #' @export
 cog_spending <- function(govid, years, category = NULL,
-                         per_capita = FALSE, adjust_to_year = NULL) {
+                         per_capita = FALSE, adjust_to_year = NULL,
+                         basis = c("harmonized", "raw"), recipe = NULL) {
   .verb_spendrev(
     verb          = "cog_spending",
-    view          = "spending_annotated",
+    view_base     = "spending_annotated",
     subtype_col   = "spend_subtype",
+    flow_prefixes = c("E", "F", "G", "K"),
     call          = match.call(),
     govid         = govid,
     years         = years,
     category      = category,
     per_capita    = per_capita,
-    adjust_to_year = adjust_to_year
+    adjust_to_year = adjust_to_year,
+    basis         = basis,
+    recipe        = recipe
   )
 }
 
 #' @noRd
-.verb_spendrev <- function(verb, view, subtype_col, call,
+.verb_spendrev <- function(verb, view_base, subtype_col, flow_prefixes, call,
                            govid, years, category,
-                           per_capita, adjust_to_year) {
+                           per_capita, adjust_to_year,
+                           basis = c("harmonized", "raw"), recipe = NULL) {
+  basis_explicit <- length(basis) == 1L
+  basis <- match.arg(basis, c("harmonized", "raw"))
+
   govid <- .coerce_govid_input(govid, arg = "govid")
-  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year)
+  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year,
+                        recipe)
 
   years   <- as.integer(years)
   if (!is.null(adjust_to_year)) adjust_to_year <- as.integer(adjust_to_year)
 
   con <- .ensure_session()
+  manifest <- .uscogdata_env$manifest
   scope <- .check_govids_in_scope(govid)
 
+  resolved <- .resolve_basis(basis, basis_explicit, manifest)
+
+  recipe_block <- NULL
+  category_for_prov <- category
+  if (!is.null(recipe)) {
+    .require_schema_v5(con, manifest, "recipe =")
+    .validate_recipe_id(con, recipe)
+    comps <- .recipe_components(con, recipe)
+    recipe_label <- comps$label[[1]]
+    result <- .run_recipe(con, recipe, govid, years)
+    sql <- attr(result, "sql_query")
+    result <- .shape_recipe_result(result, subtype_col, recipe_label)
+    recipe_block <- list(
+      recipe_id = recipe, label = recipe_label,
+      components = .df_to_row_list(comps)
+    )
+    category_for_prov <- recipe_label
+  } else {
+    view <- .select_view(view_base, resolved$basis)
     sql <- .build_verb_sql(view, subtype_col, govid, years, category)
     result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  }
 
   if (per_capita) result <- .attach_per_capita(result, con, govid)
   if (!is.null(adjust_to_year)) {
@@ -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
-     FROM canonical_fips_xwalk
-     WHERE canonical_govid IN (%s)",
-    .sql_lit_chr(govid)
+    "SELECT canonical_govid, year, population, popyear
+     FROM gov_population_yearly
+     WHERE canonical_govid IN (%s)
+       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$amt_per_capita_nominal <- result$amt_nominal / result$population_acs
-  result$population_acs <- NULL
+  result <- dplyr::left_join(result, pops,
+                             by = c("canonical_govid", "year"))
+  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))
-  ifelse(
-    isTRUE(result$aggregate_fallback) | result$aggregate_fallback %in% TRUE,
+  n <- nrow(result)
+  if (n == 0L) return(character(0))
+  parts <- vector("list", 2L)
+  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,97 @@
 {
-  "schema_version": 3,
-  "built_at": "2026-04-27T16:43:46Z",
-  "pipeline_commit": "899af37",
-  "fixture_note": "Two-year (2019-2020) fixture for uscogdata tests. Full corpus available via USCOGDATA_URL.",
+  "schema_version": 6,
+  "built_at": "2026-07-23T16:14:30Z",
+  "pipeline_commit": "4f992a0",
+  "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",
+    "source_vintages": {
+      "2012": "10162019",
+      "2013": "10162019",
+      "2014": "10162019",
+      "2015": "10162019",
+      "2016": "10162019",
+      "2017": "06102021",
+      "2018": "06102021",
+      "2019": "06102021",
+      "2020": "06122023",
+      "2021": "06122023",
+      "2022": "06052025",
+      "2023": "06052025"
+    },
+    "registry_rows": 148,
     "acs_vintage": "ACS 2018-2022 5-year"
   },
   "scope": {
-    "gov_types_included": [
-      0,
-      1,
-      2,
-      3
-    ],
-    "gov_types_excluded": [
-      4,
-      5
-    ],
+    "gov_types_included": [0, 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_columns": [
-      "fips_state",
-      "type",
-      "fips_county",
-      "govid",
-      "gov_blank",
-      "gov_name",
-      "county_name",
-      "fips_state_code",
-      "fips_county_code",
-      "fips_place_code",
-      "population",
-      "popyear",
-      "enrollment",
-      "enrollyear",
-      "function_code",
-      "sch_level_code",
-      "fiscal_year_end",
-      "srvy_year",
-      "item_code",
-      "amt",
-      "srv_data",
-      "impute_flag",
-      "is_aggregate",
-      "canonical_govid"
-    ],
+    "long_column_count": 28,
+    "long_columns": ["fips_state", "type", "fips_county", "govid", "gov_blank", "gov_name", "county_name", "fips_state_asof", "fips_county_asof", "cog_legacy_state", "cog_legacy_county", "fips_place_code", "population", "popyear", "enrollment", "enrollyear", "function_code", "sch_level_code", "fiscal_year_end", "srvy_year", "item_code", "amt", "srv_data", "impute_flag", "is_aggregate", "canonical_govid", "harmonized_code", "survey_weight"],
     "data_dictionary": "docs/data_dictionary.md"
   },
   "files": {
     "long_partitions": [
+      {
+        "year": 2011,
+        "path": "data/long/year=2011/part-0.parquet",
+        "sha256": "84302ab364dc9fc3b3fbbc3c3f8b826e3508b4d73ff7c42d094d3863cd1e37b5",
+        "row_count": 2864212,
+        "size_bytes": 3845911
+      },
+      {
+        "year": 2012,
+        "path": "data/long/year=2012/part-0.parquet",
+        "sha256": "b82ac82d5e35f844b26c887445601f3748438c52c998ba4e403b025941a6f170",
+        "row_count": 1163338,
+        "size_bytes": 5929917
+      },
       {
         "year": 2019,
         "path": "data/long/year=2019/part-0.parquet",
-        "sha256": "e1c9f426c6d7d3c51d06b3a652473b987b304836619c213f019cee4887714daa",
+        "sha256": "5cbd4726dcc7d0dab5c2a05a64702e979533ae119ed0587073cd31c089e0d737",
         "row_count": 318139,
-        "size_bytes": 1424231
+        "size_bytes": 1719548
       },
       {
         "year": 2020,
         "path": "data/long/year=2020/part-0.parquet",
-        "sha256": "9b795853a848e8c955c80261b96b79630fc77394dcfb1a1ca288e2cd634053a3",
+        "sha256": "ee548fec80bf1beda844fe03916ac145f10dd34c45968407cc330ec260935f00",
         "row_count": 317500,
-        "size_bytes": 1427150
+        "size_bytes": 1722918
       }
     ],
     "metadata": [
+      {
+        "path": "data/canonical_alias.parquet",
+        "sha256": "3f617051c23a99bea322889857f7106df0c92954564afeec181df7083ee6698e",
+        "description": "canonical_alias.parquet"
+      },
       {
         "path": "data/canonical_fips_xwalk.parquet",
-        "sha256": "86e53e04a35f6f90bb74bb1a273e053392afa782d6f518e3e3da9c976d47f7af",
+        "sha256": "f98742f941269dacf8f7de5c273aa4dd4e75017a5bb70c054da35852a95a8d46",
         "description": "canonical_fips_xwalk.parquet"
       },
       {
         "path": "data/summary_categories.parquet",
-        "sha256": "60045e22bc2723318fa2cb73f8e5038250dc54d24b3447c6750dfe29035335b8",
+        "sha256": "8e6fcd4dd9bb4723841a67233b19388c9762dfc23b4479501183cebf7ea3c1b5",
         "description": "summary_categories.parquet"
+      },
+      {
+        "path": "data/harmonization_map.parquet",
+        "sha256": "4cf32d0f817079ba4f28dc0ce65450d3247ebbf08d94c0c26c0d02af597bf812",
+        "description": "harmonization_map.parquet"
+      },
+      {
+        "path": "data/harmonization_recipes.parquet",
+        "sha256": "1133e9a0b02f8f34f5f936e55c5ecd596bb8a55d8425dcce76767f0f3203581c",
+        "description": "harmonization_recipes.parquet"
+      },
+      {
+        "path": "data/series_breaks.parquet",
+        "sha256": "b0b6794b6887a4f300079adfa10029c2a77109faa4952fbff1c5a270793cc02b",
+        "description": "series_breaks.parquet"
       }
     ]
   },

inst/schemas/provenance-v1.json

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

inst/sql/22-spending_long_harmonized.sql

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

inst/sql/23-revenue_long_harmonized.sql

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

inst/sql/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
-government type, state, and population range. Peers are ordered by
-`|log(pop_ratio)|` ascending (closest to the target's population first).
+Selects peer governments by combinations of government type, state, and
+population range at a chosen `year`. Peers are ordered by `|log(pop_ratio)|`
+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
-from `canonical_fips_xwalk.population_acs`.}
+\item{per_capita}{If `TRUE`, per-capita uses each gov's own per-year
+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`,
-  `aggregate_fallback`, `scope_note`, `notes`. Carries a `provenance`
-  attribute with `verb = "cog_geographic_rollup"` and `layers`.
+  `amt_per_capita_nominal` / `amt_per_capita_real`, optional `pop_source`,
+  `codes_included`, `aggregate_fallback`, `scope_note`, `notes`. Carries a
+  `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
-  among target+peers at `max(years)`, NA for other rows). Provenance
-  attribute reports `verb = "cog_peer_compare"` and `peer_count`.
+  `"summary_p50"`, or `"summary_p75"`, `target_rank` (target's rank
+  among target+peers at `max(years)`, NA for other rows), and
+  `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
-`population_acs` from the canonical xwalk.}
+`amt_per_capita_real` when `adjust_to_year` is set) using the per-year
+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
-`population_acs` from the canonical xwalk.}
+`amt_per_capita_real` when `adjust_to_year` is set) using the per-year
+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`
-  attribute matching `inst/schemas/provenance-v1.json`.
+  optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`.
+  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-config.R

@@ -26,3 +26,41 @@ test_that(".resolve_cache_dir falls back to R_user_dir", {
     })
   })
 })
+
+# ---------------------------------------------------------------------------
+# Trailing-slash normalization (uscogdata #3 follow-up).
+#
+# EVERY consumer builds paths by concatenation: paste0(url, "manifest.json")
+# (manifest.R), paste0(url, e$path) (mirror.R), and the parquet glob in
+# views.R. mirror.R:104 even comments 'url ends in "/"' -- an assumption the
+# package documents and relies on but never enforced.
+#
+# A URL missing its trailing slash therefore fails SILENTLY and confusingly:
+#   HTTPS -> ".../downloadmanifest.json" -> the host answers with an HTML 404
+#            page -> the jsonlite lexical error that issue #3 reported;
+#   local -> ".../corpusdata/long/**/*.parquet" -> DuckDB "No files found".
+# Neither message points at the real cause. Normalize once, at resolution.
+# ---------------------------------------------------------------------------
+
+test_that(".resolve_url appends a missing trailing slash", {
+  withr::local_envvar(USCOGDATA_URL = "https://example.org/s/TOKEN/download")
+  expect_equal(.resolve_url(), "https://example.org/s/TOKEN/download/")
+})
+
+test_that(".resolve_url leaves an existing trailing slash alone", {
+  withr::local_envvar(USCOGDATA_URL = "https://example.org/s/TOKEN/download/")
+  expect_equal(.resolve_url(), "https://example.org/s/TOKEN/download/")
+})
+
+test_that(".resolve_url normalizes a local path without a trailing slash", {
+  withr::local_envvar(USCOGDATA_URL = "/tmp/corpus")
+  expect_equal(.resolve_url(), "/tmp/corpus/")
+})
+
+test_that(".resolve_url does not invent a slash for an empty setting", {
+  # An unset/empty URL must stay empty so the "not configured" guard in
+  # manifest.R still fires, rather than degrading into a bare "/" root.
+  withr::local_envvar(USCOGDATA_URL = "")
+  withr::local_options(uscogdata.url = "")
+  expect_equal(.resolve_url(), "")
+})

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

tests/testthat/test-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 &
-                  peers$population_acs <= 2.5e6))
+  expect_true(all(peers$population >= 1.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)
-  r <- cog_peer_compare("101006006", peers, "Police", years = 2020L)
+  peers <- cog_find_peers("121011212191", max_peers = 4L)
+  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",
-    peers = c("441015015", "441220220"),     # Bexar, Tarrant
+    "121011212191",
+    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",
-    peers = c("441015015", "441220220", "231082082"),
+    "121011212191",
+    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",
-                        peers = c("441015015", "441220220"),
+  r <- cog_peer_compare("121011212191",
+                        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
-      county = "101006006",               # Broward County
-      city   = "102006004"                # Fort Lauderdale City
+      state  = "120000226351",               # Florida state govt
+      county = "121011212191",               # Broward County
+      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",
-                  city = "102006004"),
+    govids = list(state = "120000226351", county = "121011212191",
+                  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,195 @@ 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 the v6 fixture (corpus 2026-07-22). The Task 18 map
+    # extension added E/F/G-prefix discontinued_na rulings the earlier pin's
+    # comment predated: E21/F21/G21 (Education NEC local, SB184-186,
+    # "trivial; explicit-NA, full wide-era window"). Broward's 2011 legacy
+    # partition zero-pads exactly those three codes, so this query now
+    # excludes 3 NA-harmonized rows -- all with amt = 0, hence the excluded
+    # AMOUNT stays exactly zero. (The other discontinued_na rulings -- S74,
+    # Z61, X04, X06, the debt-detail family, L24 -- remain outside the
+    # E/F/G/K prefixes.) See docs/phase_r_harmonization_review.md § 1.3/1.4
+    # and cog_pipeline data/harmonization_map.csv E21/F21/G21 rows.
+    expect_equal(prov$harmonization$na_rows_excluded, 3L)
+    expect_equal(prov$harmonization$na_amount_excluded, 0)
+  })
+})
+
+test_that("basis = 'raw' never populates the harmonization exclusion block", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2020L, "Corrections", basis = "raw")
+    h <- attr(r, "provenance")$harmonization
+    expect_false(h$applied)
+    expect_equal(h$na_rows_excluded, 0L)
+  })
+})
+
+test_that("v4 corpus: basis silently resolves to raw (default) with a provenance note", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    r <- cog_spending("121011212191", 2019L, "Police")
+    prov <- attr(r, "provenance")
+    expect_equal(prov$basis, "raw")
+    expect_match(prov$basis_note, "raw", fixed = TRUE)
+    expect_match(prov$basis_note, "schema_version", fixed = TRUE)
+    expect_false(prov$harmonization$applied)
+  })
+})
+
+test_that("v4 corpus: explicit basis = 'harmonized' aborts", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    expect_error(
+      cog_spending("121011212191", 2019L, "Police", basis = "harmonized"),
+      class = "uscogdata_basis_unsupported"
+    )
+  })
+})
+
+test_that("provenance$series_break_refs is a populated-when-applicable character vector", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    r <- cog_spending("121011212191", 2020L, "Corrections")
+    refs <- attr(r, "provenance")$series_break_refs
+    expect_type(refs, "character")
+    # No catalogued series_breaks_pq row falls inside this fixture's
+    # 2011/2012/2019/2020 window for the codes this query touches (E04/G04)
+    # -- data-verified; the mechanism itself is what's under test here, via
+    # a query-shaped unit test in test-views.R since the fixture has no
+    # positive case to pin against.
+    expect_equal(refs, character(0))
+  })
+})
+
+test_that("v4 corpus: explicit basis = 'raw' still works", {
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    r <- cog_spending("121011212191", 2019L, "Police", basis = "raw")
+    expect_equal(attr(r, "provenance")$basis, "raw")
+    expect_gt(nrow(r), 0L)
+  })
+})

tests/testthat/test-views.R

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