feat: limit/offset on cog_gov_search() and cog_balances() (#57 )

verbs were left materializing everything and slicing in R -- the pattern behind the 2026-08-06 production incident. cog_gov_search() had no LIMIT at all, so an unfiltered call returns the entire 40,336-row crosswalk. Extracted the #39 machinery into R/pagination.R first (.validate_pagination(), .paginate_sql(), .take_pagination_total()) rather than growing a third inline copy: three definitions of what total_rows means is three places for it to drift. Conflict refusals stay at the call sites because each verb's conflict set differs. .verb_spendrev() now uses the shared helpers and is unchanged in behaviour. The empty-page fallback query is now passed as a thunk, so the unpaginated SQL is only BUILT when an offset actually lands past the end instead of on every paged call. Two things #57 did not anticipate: - cog_gov_search()'s ORDER BY was not a total order. population_acs DESC NULLS LAST leaves ties -- and the whole NULL block -- in scan order, so two requests can order them differently and a paged sweep duplicates one row while dropping another. Added canonical_govid as tiebreaker. Unpaginated output changes only in the relative order of already-tied rows. - Basket mode returns one resolved row per requested name plus a sidecar covering all of them, so a page of it is not a page of anything the caller asked for. Refused with uscogdata_basket_pagination_conflict rather than silently ignoring the arguments. Both default to NULL, so cog-api adopts them behind its existing formals() probe with no lockstep deploy. Suite: 1067 passed, 0 failed, 0 warnings (2 pre-existing live-corpus skips).
Merge pull request 'feat: cog_open() honours a DuckDB thread and memory budget (#60 )' (#62 ) from feat/duckdb-threads-60 into main
2026-08-10 19:27:11 -04:00 · 2026-08-10 19:23:14 -04:00 · 2026-08-10 18:51:56 -04:00 · 2026-08-09 14:31:53 -04:00 · 2026-08-09 14:15:21 -04:00 · 2026-08-09 09:08:41 -04:00
28 changed files with 1450 additions and 134 deletions
@@ -1,7 +1,7 @@
 Package: uscogdata
 Type: Package
 Title: Curated Reader for the Civilytics US Census of Governments Finance Corpus
-Version: 0.3.0
+Version: 0.4.0
 Authors@R: c(
    person(c("Jared", "E."), "Knowles",
           email = "jared@civilytics.com",
@@ -1,3 +1,108 @@
+# uscogdata 0.4.0
+
+## DuckDB's resource budget is configurable
+
+`USCOGDATA_DUCKDB_THREADS` and `USCOGDATA_DUCKDB_MEMORY_LIMIT` (with matching
+`options(uscogdata.duckdb_threads = )` / `options(uscogdata.duckdb_memory_limit = )`
+spellings) cap the DuckDB connection the package opens. Both follow the same
+env-var > option > default precedence as `USCOGDATA_URL`.
+
+Unset, **no pragma is issued at all** and DuckDB's own defaults apply exactly as
+before -- every visible core. That is right for one interactive session on a
+dedicated machine and wrong for a server: where several readers share a host, each
+otherwise claims the whole machine and they contend. Capping measured ~5% on a
+single-government all-years query (502 ms at 2 threads vs 475 ms uncapped on 16
+cores), which is cheap enough that a server should always cap.
+
+This replaces a workaround in which a consumer reached into the package namespace
+at boot -- `getFromNamespace(".ensure_session", "uscogdata")()` followed by a manual
+`SET threads` -- depending both on a private name and on the session already being
+open.
+
+## `cog_gov_search()` and `cog_balances()` gain `limit`/`offset`
+
+Pagination arrived on `cog_spending()`/`cog_revenue()` in 0.3.0; the other two
+verbs were left materializing everything and slicing in R. Both now take
+`limit`/`offset` with the same semantics: `NULL` default, the page applied in
+SQL behind a deterministic `ORDER BY`, and the unpaginated count returned as a
+`total_rows` attribute computed by `COUNT(*) OVER()` in the same scan rather
+than a second query.
+
+`cog_gov_search()` had no `LIMIT` at all, which made it the one verb that
+returns the entire 40,336-row crosswalk when called with no filter.
+
+Two refusals rather than silent surprises:
+
+* `cog_balances(recipe = , limit = )` aborts with class
+  `uscogdata_recipe_pagination_conflict` -- a recipe's result comes from a
+  separate query that pagination is not wired into.
+* `cog_gov_search()` in basket mode (`length(name) > 1`) aborts with class
+  `uscogdata_basket_pagination_conflict`. Basket mode returns one resolved row
+  per requested name with a sidecar covering all of them; a page of that is not
+  a page of anything the caller asked for.
+
+## Cohorts can be named by predicate, not just by id
+
+`cog_spending()`, `cog_revenue()` and `cog_balances()` gain optional `state`
+and `type` arguments. Both default to `NULL`, so every existing call behaves
+exactly as before.
+
+Passing them expresses the cohort as a subquery against `canonical_fips_xwalk`
+inside each statement, instead of round-tripping the ids through R and
+rendering them back into a literal `IN` list:
+
+```r
+# before: resolve 20,106 ids in R, then embed them in every statement
+ids <- cog_gov_search(NULL, state = "CA", type = "city")$canonical_govid
+cog_spending(ids, years = 2022)
+
+# now: the cohort never leaves the database
+cog_spending(years = 2022, state = "CA", type = "city")
+```
+
+Measured against the production corpus, same FY2022 aggregate over the
+20,106-government `type = "city"` cohort:
+
+| cohort expressed as | time |
+|---|---:|
+| `IN (20,106 literals)` | 449 ms |
+| join against a temp cohort table | 99 ms |
+| predicate on `canonical_fips_xwalk` | **94 ms** |
+| no cohort filter at all (the floor) | 88 ms |
+
+**4.8x, within 7% of the floor.** The rendered `IN` list was 301,591
+characters and was re-parsed in 5-8 separate statements per call, so the cost
+was paid repeatedly; the predicate's size is constant in the cohort.
+
+`state` and `type` use the same vocabulary and the same internal coercion as
+`cog_gov_search()` -- `state` is a postal abbreviation (`"WI"`) even though the
+crosswalk column holds a FIPS code (`"55"`).
+
+Supplying `govid` **and** `state`/`type` intersects them: the governments in
+`govid` that also match the predicate. Naming no cohort at all now aborts with
+class `uscogdata_no_cohort` rather than R's "argument is missing" error.
+
+When the cohort is named by predicate there is no id list to report, so
+`provenance$scope$govids_found`/`govids_missing` are empty and
+`provenance$scope$cohort` carries `state`, `type` and `n_governments` instead.
+A `govid`-named cohort's provenance is unchanged.
+
+## Fixes
+
+* `cog_gov_search()` now orders by `population_acs DESC NULLS LAST,
+  canonical_govid`. **`population_acs` alone is not a total order** -- ties, and
+  the entire `NULLS LAST` block, came back in whatever order the scan produced.
+  That was invisible while every call returned the full result set, but it makes
+  a paged sweep unsound: two requests can order tied rows differently, so a row
+  is duplicated on one page and missing from the next. Unpaginated results are
+  unchanged except for the relative order of rows that were already tied.
+
+* An unknown `state` abbreviation now aborts with "Unknown state abbreviation"
+  (class `uscogdata_unknown_state`) instead of base R's "subscript out of
+  bounds". `.state_abbrev_to_fips` is a named character vector, so `[[` on an
+  absent name threw before the curated message could be reached -- making that
+  message unreachable dead code in every verb that takes a `state`.
+
 # uscogdata 0.3.0

 First public release.
@@ -18,7 +18,9 @@
 #' comparable to a GAAP fund balance from an ACFR.
 #'
 #' @param govid Canonical govid(s): a character vector, or a data frame with a
-#'   `canonical_govid` column (e.g. from [cog_gov_search()]).
+#'   `canonical_govid` column (e.g. from [cog_gov_search()]). `NULL` to name
+#'   the cohort by `state`/`type` instead.
+#' @inheritParams cog_spending
 #' @param years Integer vector of fiscal years.
 #' @param category Optional character vector of categories to keep. One of
 #'   `"Fund Balances"`, `"Insurance Trust Balances"`,
@@ -45,6 +47,12 @@
 #' @param recipe Optional harmonization recipe id (see [cog_recipes()]).
 #'   `"cash_securities_z77_wide"` and `"cash_securities_z78_wide"` bridge the
 #'   wide era to the modern one.
+#' @param limit Maximum number of result rows to return, pushed into the SQL
+#'   rather than applied after materializing every row. `NULL` (default)
+#'   returns everything. Cannot be combined with `recipe` -- see `offset` and
+#'   `total_rows`.
+#' @param offset Rows to skip before `limit` starts counting (0-based).
+#'   Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.
 #'
 #' @return Tibble with columns `year`, `canonical_govid`, `gov_name`,
 #'   `balance_subtype`, `category`, `amt_nominal`, `codes_included`,
@@ -62,16 +70,22 @@
 #'   and `truncated` (the observed subtypes whose coverage falls short of the
 #'   requested years). `expenditure_concept`/`revenue_concept` are `NA` --
 #'   holdings are a stock, not a flow, so neither concept vocabulary applies.
+#'
+#'   When `limit` is set, also carries a `total_rows` attribute: the full
+#'   unpaginated row count, computed by the same query (`COUNT(*) OVER()`)
+#'   rather than a second scan.
 #' @export
-cog_balances <- function(govid, years, category = NULL,
+cog_balances <- function(govid = NULL, years, category = NULL,
                         per_capita = FALSE, adjust_to_year = NULL,
-                         basis = c("harmonized", "raw"), recipe = NULL) {
+                         basis = c("harmonized", "raw"), recipe = NULL,
+                         state = NULL, type = NULL,
+                         limit = NULL, offset = NULL) {
  call <- match.call()
  basis <- match.arg(basis, c("harmonized", "raw"))
  # Coerce FIRST, validate second: .validate_verb_inputs() asserts
  # is.character(govid), and a data-frame govid (cog_gov_search() output) has
  # not been unwrapped yet at this point.
-  govid <- .coerce_govid_input(govid)
+  govid <- if (is.null(govid)) NULL else .coerce_govid_input(govid)
  # The money verbs' validator, reused rather than re-implemented (R/spending.R).
  # It covers the exact superset cog_balances() needs -- including the
  # recipe/category mutual-exclusivity guard -- so a second local copy would
@@ -88,9 +102,27 @@ cog_balances <- function(govid, years, category = NULL,
  # validator's own doc comment for the incident that made that matter.
  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year,
                        recipe)
+
+  # Same semantics as the money verbs (R/pagination.R). Only the `recipe`
+  # conflict applies here: cog_balances() has no `complete` argument, and a
+  # recipe's result comes from .run_recipe()'s own query, which pagination is
+  # not wired into.
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
+  if (!is.null(limit) && !is.null(recipe)) {
+    cli::cli_abort(c(
+      "`limit`/`offset` cannot be combined with `recipe`.",
+      "i" = "A recipe's result comes from a separate query (`.run_recipe()`) that pagination is not wired into yet.",
+      "*" = "Drop `limit`/`offset`, or drop `recipe`."
+    ), class = "uscogdata_recipe_pagination_conflict")
+  }
+
  years <- as.integer(years)
  if (!is.null(adjust_to_year)) adjust_to_year <- as.integer(adjust_to_year)

+  cohort <- .make_cohort(govid, state, type)
+
  con <- .ensure_session()
  .require_balance_support(con)
  scope <- .check_govids_in_scope(govid)
@@ -103,13 +135,14 @@ cog_balances <- function(govid, years, category = NULL,
  manifest <- .uscogdata_env$manifest
  recipe_block <- NULL
  category_for_prov <- category
+  total_rows <- NULL  # set below only when limit is non-NULL (non-recipe path)

  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)
+    result <- .run_recipe(con, recipe, cohort, years)
    sql <- attr(result, "sql_query")
    result <- .shape_recipe_result(result, "balance_subtype", recipe_label)
    recipe_block <- list(
@@ -119,15 +152,25 @@ cog_balances <- function(govid, years, category = NULL,
    category_for_prov <- recipe_label
  } else {
    sql <- .build_verb_sql("balance_annotated", "balance_subtype",
-                           govid, years, category,
-                           ig_view = NULL, subtype_scope = NULL)
+                           cohort, years, category,
+                           ig_view = NULL, subtype_scope = NULL,
+                           limit = limit, offset = offset)
    result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
+    if (!is.null(limit)) {
+      paged <- .take_pagination_total(result, con, function() {
+        .build_verb_sql("balance_annotated", "balance_subtype",
+                        cohort, years, category,
+                        ig_view = NULL, subtype_scope = NULL)
+      })
+      result     <- paged$result
+      total_rows <- paged$total_rows
+    }
  }

  # Order matters (matches .verb_spendrev()): per-capita first, so
  # .attach_real_dollars() deflates the nominal per-capita column into
  # amt_per_capita_real rather than needing amt_per_capita_nominal recomputed.
-  if (isTRUE(per_capita)) result <- .attach_per_capita(result, con, govid)
+  if (isTRUE(per_capita)) result <- .attach_per_capita(result, con)
  if (!is.null(adjust_to_year)) {
    result <- .attach_real_dollars(result, adjust_to_year, per_capita)
  }
@@ -145,6 +188,7 @@ cog_balances <- function(govid, years, category = NULL,
  )
  prov$scope$govids_found   <- scope$found
  prov$scope$govids_missing <- scope$missing
+  prov$scope$cohort <- .cohort_provenance(con, cohort)

  prov$balance_caveats <- .balance_caveats(
    con, prov$codes_summed$observed, years
@@ -152,6 +196,7 @@ cog_balances <- function(govid, years, category = NULL,
  .emit_balance_caveats(prov$balance_caveats)

  attr(result, "provenance") <- prov
+  if (!is.null(limit)) attr(result, "total_rows") <- total_rows
  result
 }

@@ -54,7 +54,7 @@
 #' expenditure_concept = "total": ig_long_harmonized COALESCEs rather than
 #' drops NULL-harmonized rows, so harmonization never excludes an IG row.
 #' @noRd
-.build_harmonization_block <- function(con, govid, years, resolved,
+.build_harmonization_block <- function(con, cohort, years, resolved,
                                       subtype_col, subtype_scope) {
  if (!identical(resolved$basis, "harmonized")) {
    return(list(
@@ -68,12 +68,12 @@
  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)
+     WHERE %s AND year IN (%s)
       AND NOT is_aggregate AND harmonized_code IS NULL
       AND item_code IN (
         SELECT item_code FROM summary_categories WHERE %s IN (%s)
       )",
-    .sql_lit_chr(govid), paste(as.integer(years), collapse = ","),
+    .cohort_sql(cohort), paste(as.integer(years), collapse = ","),
    subtype_col, .sql_lit_chr(subtype_scope)
  )
  na <- DBI::dbGetQuery(con, sql)
@@ -0,0 +1,127 @@
+# How a verb names the set of governments it queries.
+#
+# Historically there was one way: a `govid` character vector, rendered by
+# .sql_lit_chr() into a quoted IN list. That is fine for a handful of
+# governments and pathological for a fleet. Measured against the production
+# corpus, the same FY2022 aggregate over the 20,106-government `type = "city"`
+# cohort:
+#
+#   cohort expressed as                            time
+#   IN (20,106 literals)                           449 ms
+#   join against a temp cohort table                99 ms
+#   predicate on canonical_fips_xwalk               94 ms
+#   no cohort filter at all (the floor)             88 ms
+#
+# 4.8x, and within 7% of the no-filter floor. The rendered IN list is 301,591
+# characters and .verb_spendrev() embeds it in 5-8 separate statements per
+# call, so the parse-and-plan cost is paid over and over (uscogdata#58).
+#
+# A cohort therefore has two independent halves, and a query can carry either
+# or both:
+#
+#   ids        an explicit canonical_govid vector      -> literal IN list
+#   predicate  state/type over canonical_fips_xwalk    -> IN (SELECT ...)
+#
+# Both together is an INTERSECTION -- "these ids, narrowed to that state/type"
+# -- never a precedence rule where one silently wins.
+
+#' Build the internal cohort object shared by every query verb.
+#'
+#' `state` and `type` are coerced with the SAME helpers `cog_gov_search()`
+#' uses. That is load-bearing, not tidiness: the public argument is a postal
+#' abbreviation (`"WI"`) while `canonical_fips_xwalk.fips_state` holds a FIPS
+#' code (`"55"`), and `type` is a label (`"city"`) against an integer
+#' `govs_type`. A predicate written against the raw parameter matches nothing
+#' and returns an empty result indistinguishable from "this government
+#' reported nothing" -- cog-api hit exactly that trap optimizing this path.
+#' One definition of the translation, not two.
+#'
+#' @param govid Already-coerced character vector of canonical_govids, or NULL.
+#' @param state Postal abbreviation or FIPS code, or NULL.
+#' @param type Type label or integer code, or NULL.
+#' @noRd
+.make_cohort <- function(govid = NULL, state = NULL, type = NULL) {
+  if (is.null(govid) && is.null(state) && is.null(type)) {
+    cli::cli_abort(c(
+      "A cohort must be named.",
+      "*" = "Pass {.arg govid} for specific governments, or {.arg state}/{.arg type} for every government matching a predicate.",
+      "i" = "Passing both intersects them: the governments in {.arg govid} that also match {.arg state}/{.arg type}."
+    ), class = "uscogdata_no_cohort")
+  }
+  structure(
+    list(
+      ids        = govid,
+      state      = state,
+      type       = type,
+      state_fips = if (is.null(state)) NULL else .coerce_state_to_fips(state),
+      type_int   = if (is.null(type)) NULL else .coerce_type(type)
+    ),
+    class = "uscogdata_cohort"
+  )
+}
+
+#' Is any part of this cohort expressed as an xwalk predicate?
+#' @noRd
+.cohort_by_predicate <- function(cohort) {
+  !is.null(cohort$state_fips) || !is.null(cohort$type_int)
+}
+
+#' Render the cohort as a SQL boolean expression over `col`.
+#'
+#' `col` may be qualified (`"l.canonical_govid"`, `"x.canonical_govid"`) --
+#' several call sites join the xwalk under an alias. The subquery's own
+#' projected column stays unqualified: it selects from canonical_fips_xwalk,
+#' not from the outer relation.
+#' @noRd
+.cohort_sql <- function(cohort, col = "canonical_govid") {
+  preds <- character(0)
+
+  if (!is.null(cohort$ids)) {
+    preds <- c(preds, sprintf("%s IN (%s)", col, .sql_lit_chr(cohort$ids)))
+  }
+
+  if (.cohort_by_predicate(cohort)) {
+    xwalk_preds <- character(0)
+    if (!is.null(cohort$state_fips)) {
+      xwalk_preds <- c(xwalk_preds,
+                       sprintf("fips_state = %s", .sql_lit_chr(cohort$state_fips)))
+    }
+    if (!is.null(cohort$type_int)) {
+      xwalk_preds <- c(xwalk_preds, sprintf("govs_type = %d", cohort$type_int))
+    }
+    preds <- c(preds, sprintf(
+      "%s IN (SELECT canonical_govid FROM canonical_fips_xwalk WHERE %s)",
+      col, paste(xwalk_preds, collapse = " AND ")
+    ))
+  }
+
+  paste(preds, collapse = " AND ")
+}
+
+#' How many governments the cohort covers.
+#'
+#' One COUNT against the crosswalk, used only to populate the provenance
+#' `scope$cohort` block. Deliberately a count rather than the id list: a
+#' fleet-scale cohort would otherwise put 20,000 ids into every response body,
+#' which is the cost this issue exists to remove.
+#' @noRd
+.cohort_count <- function(con, cohort) {
+  sql <- sprintf(
+    "SELECT COUNT(*) AS n FROM canonical_fips_xwalk WHERE %s",
+    .cohort_sql(cohort)
+  )
+  as.integer(DBI::dbGetQuery(con, sql)$n[[1]])
+}
+
+#' The provenance `scope$cohort` block for a predicate cohort, or NULL when
+#' the cohort was named by id alone (in which case `govids_found`/
+#' `govids_missing` already describe it exactly).
+#' @noRd
+.cohort_provenance <- function(con, cohort) {
+  if (!.cohort_by_predicate(cohort)) return(NULL)
+  list(
+    state         = if (is.null(cohort$state)) NA_character_ else as.character(cohort$state),
+    type          = if (is.null(cohort$type)) NA_character_ else as.character(cohort$type),
+    n_governments = .cohort_count(con, cohort)
+  )
+}
@@ -57,7 +57,7 @@
 #' aggregate rows. Without it the grid would offer cells the verb structurally
 #' never returns, so every one of them would fill as a phantom $0.
 #' @noRd
-.completion_grid_sql <- function(subtype_col, govid, years, category,
+.completion_grid_sql <- function(subtype_col, cohort, years, category,
                                 subtype_scope) {
  category_pred <- if (is.null(category)) {
    ""
@@ -76,13 +76,13 @@
     JOIN canonical_fips_xwalk x ON x.govs_type = cs.type
     JOIN summary_categories c   ON c.item_code = cs.item_code
     JOIN representation r       ON r.year = cs.year
-     WHERE x.canonical_govid IN (%2$s)
+     WHERE %2$s
       AND cs.year IN (%3$s)
       AND NOT cs.is_aggregate
       AND c.category IS NOT NULL
       AND c.%1$s IN (%4$s)
       %5$s",
-    subtype_col, .sql_lit_chr(govid),
+    subtype_col, .cohort_sql(cohort, "x.canonical_govid"),
    paste(as.integer(years), collapse = ","),
    .sql_lit_chr(subtype_scope), category_pred
  )
@@ -94,10 +94,10 @@
 #' provenance block. Reported rows are passed through untouched -- filling
 #' must never alter or drop what the corpus actually published.
 #' @noRd
-.complete_result <- function(result, con, subtype_col, govid, years, category,
+.complete_result <- function(result, con, subtype_col, cohort, years, category,
                             subtype_scope) {
  grid <- tibble::as_tibble(DBI::dbGetQuery(
-    con, .completion_grid_sql(subtype_col, govid, years, category, subtype_scope)
+    con, .completion_grid_sql(subtype_col, cohort, years, category, subtype_scope)
  ))

  result$value_source <- rep("reported", nrow(result))
@@ -18,7 +18,15 @@
  # Nextcloud share or a local copy made by cog_mirror().
  url               = "https://huggingface.co/datasets/civilytics/us-cog-finance/resolve/main/",
  cache_dir         = NULL,
-  manifest_ttl_secs = 3600L
+  manifest_ttl_secs = 3600L,
+  # NULL means "emit no pragma", which leaves DuckDB's own defaults intact:
+  # every visible core, and 80% of RAM. That is right for one interactive
+  # session on a dedicated machine and wrong for a server, where several
+  # readers share a box and each would otherwise claim all of it. See
+  # .resolve_duckdb_threads() for why this is a supported option rather than
+  # something a consumer reaches into the namespace to set.
+  duckdb_threads      = NULL,
+  duckdb_memory_limit = NULL
 )

 #' Resolve a config value: env var > option > default
@@ -61,3 +69,54 @@
  v <- .cfg("cache_dir")
  if (is.null(v)) tools::R_user_dir("uscogdata", "cache") else v
 }
+
+#' Resolve the DuckDB thread cap, or NULL to leave DuckDB's default alone.
+#'
+#' `cog_open()` used to connect with a bare `dbConnect()` and set no `threads`
+#' pragma, so DuckDB claimed every core it could see. cog-api works around that
+#' by reaching into this namespace at boot --
+#' `getFromNamespace(".ensure_session", "uscogdata")()` followed by a manual
+#' `SET threads` -- which depends on a private name and on the session already
+#' being open. Making it a resolved option removes the reason to do that.
+#'
+#' `.cfg()` returns an environment variable as CHARACTER, so this coerces
+#' rather than trusting the type: `USCOGDATA_DUCKDB_THREADS=4` arrives as "4",
+#' and `sprintf("SET threads TO %d", "4")` would abort inside the connection
+#' path with an error about the pragma rather than about the setting.
+#' @noRd
+.resolve_duckdb_threads <- function() {
+  v <- .cfg("duckdb_threads")
+  if (is.null(v) || (is.character(v) && !nzchar(v))) return(NULL)
+  n <- suppressWarnings(as.integer(v))
+  if (length(n) != 1L || is.na(n) || n < 1L) {
+    cli::cli_abort(c(
+      "{.envvar USCOGDATA_DUCKDB_THREADS} must be a single positive integer.",
+      x = "Got {.val {v}}.",
+      i = "Unset it (or {.code options(uscogdata.duckdb_threads = NULL)}) to use DuckDB's default of every visible core."
+    ), class = "uscogdata_invalid_duckdb_threads")
+  }
+  n
+}
+
+#' Resolve the DuckDB memory limit, or NULL to leave DuckDB's default alone.
+#'
+#' The value is a DuckDB size string (`"4GB"`, `"512MB"`). Only its SHAPE is
+#' checked here -- DuckDB owns the unit vocabulary, and re-implementing that
+#' parse would be a second definition free to drift from the engine's. An
+#' unrecognised unit therefore surfaces as DuckDB's own error at `SET` time,
+#' which names the setting correctly; the check here exists to reject the
+#' inputs that would otherwise reach the connection as a SQL fragment.
+#' @noRd
+.resolve_duckdb_memory_limit <- function() {
+  v <- .cfg("duckdb_memory_limit")
+  if (is.null(v) || (is.character(v) && !nzchar(v))) return(NULL)
+  if (length(v) != 1L || !is.character(v) ||
+      !grepl("^[0-9]+(\\.[0-9]+)?\\s*[A-Za-z]{0,3}$", v)) {
+    cli::cli_abort(c(
+      "{.envvar USCOGDATA_DUCKDB_MEMORY_LIMIT} must be a single DuckDB size string.",
+      x = "Got {.val {v}}.",
+      i = "Examples: {.val 4GB}, {.val 512MB}, {.val 1.5GB}."
+    ), class = "uscogdata_invalid_duckdb_memory_limit")
+  }
+  trimws(v)
+}
@@ -0,0 +1,81 @@
+# R/pagination.R
+#
+# Shared limit/offset machinery. #39 established the semantics inside
+# .verb_spendrev(); #57 extends them to cog_gov_search() and cog_balances(),
+# which is what made a single definition worth having: three inline copies of
+# "coerce, refuse, unwrap the count" would be three places for the meaning of
+# `total_rows` to drift.
+#
+# The SQL side stays in .build_verb_sql() (R/spending.R) -- it already wraps
+# the aggregate in an outer SELECT so COUNT(*) OVER() sees the post-GROUP-BY
+# row count rather than the pre-aggregation one, and that is the subtle part
+# worth not duplicating either.
+
+#' Coerce and check a limit/offset pair.
+#'
+#' Returns the coerced pair, or NULL for `limit` when no page was requested.
+#' `offset` defaults to 0 whenever `limit` is set, so a caller can supply just
+#' `limit` and get the first page.
+#'
+#' Conflicts with other arguments are deliberately NOT checked here: they
+#' differ per verb (`complete`/`recipe` for the money verbs, basket mode for
+#' `cog_gov_search()`, `recipe` alone for `cog_balances()`), and a shared
+#' function taking a list of conflict flags would be harder to read than the
+#' three explicit refusals at the call sites.
+#' @noRd
+.validate_pagination <- function(limit, offset) {
+  if (is.null(limit)) {
+    return(list(limit = NULL, offset = NULL))
+  }
+  limit <- as.integer(limit)
+  if (length(limit) != 1L || is.na(limit) || limit < 0L) {
+    cli::cli_abort("`limit` must be a single non-negative integer.",
+                   class = "uscogdata_invalid_pagination")
+  }
+  offset <- if (is.null(offset)) 0L else as.integer(offset)
+  if (length(offset) != 1L || is.na(offset) || offset < 0L) {
+    cli::cli_abort("`offset` must be a single non-negative integer.",
+                   class = "uscogdata_invalid_pagination")
+  }
+  list(limit = limit, offset = offset)
+}
+
+#' Wrap a query so one page comes back carrying the unpaginated total.
+#'
+#' `COUNT(*) OVER()` rides along as an ordinary column, so the caller gets the
+#' true total from the SAME scan instead of a second round trip. The outer
+#' `SELECT *` matters: appending LIMIT/OFFSET directly to a grouped query would
+#' have the window function count pre-aggregation rows.
+#' @noRd
+.paginate_sql <- function(base_sql, limit, offset) {
+  if (is.null(limit)) return(base_sql)
+  sprintf(
+    "SELECT *, COUNT(*) OVER() AS pagination_total_rows
+     FROM (%s) AS _paged
+     LIMIT %d OFFSET %d",
+    base_sql, limit, offset
+  )
+}
+
+#' Strip the count column back out and report the unpaginated total.
+#'
+#' Returns `list(result = , total_rows = )`.
+#'
+#' An empty page -- an offset past the end -- carries no row to read the window
+#' function off, so that one case falls back to a second, unpaginated
+#' `COUNT(*)` rather than reporting a wrong zero. `unpaged_sql` is passed as a
+#' function so the fallback query is only BUILT when it is actually needed;
+#' every caller's unpaginated SQL is otherwise constructed on every paged call
+#' and thrown away.
+#' @noRd
+.take_pagination_total <- function(result, con, unpaged_sql) {
+  if (nrow(result) > 0L) {
+    total <- result$pagination_total_rows[[1]]
+    result$pagination_total_rows <- NULL
+    return(list(result = result, total_rows = as.integer(total)))
+  }
+  count_sql <- sprintf("SELECT COUNT(*) AS n FROM (%s) AS _uncounted",
+                       if (is.function(unpaged_sql)) unpaged_sql() else unpaged_sql)
+  list(result = result,
+       total_rows = as.integer(DBI::dbGetQuery(con, count_sql)$n[[1]]))
+}
@@ -113,7 +113,7 @@ cog_recipes <- function(pattern = NULL) {
 #' `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) {
+.run_recipe <- function(con, recipe_id, cohort, years) {
  sql <- sprintf(
    "SELECT l.year, l.canonical_govid,
            COALESCE(x.gov_name, l.gov_name) AS gov_name,
@@ -128,11 +128,11 @@ cog_recipes <- function(pattern = NULL) {
           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 %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),
+    .sql_lit_chr(recipe_id), .cohort_sql(cohort, "l.canonical_govid"),
    paste(as.integer(years), collapse = ",")
  )
  result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
@@ -50,11 +50,12 @@
 #'   optional `pop_source`, `codes_included`, `aggregate_fallback`, `notes`,
 #'   and `value_source` when `complete = TRUE`.
 #' @export
-cog_revenue <- function(govid, years, category = NULL,
+cog_revenue <- function(govid = NULL, years, category = NULL,
                        per_capita = FALSE, adjust_to_year = NULL,
                        basis = c("harmonized", "raw"), recipe = NULL,
                        revenue_concept = c("general", "total"),
-                        complete = FALSE, limit = NULL, offset = NULL) {
+                        complete = FALSE, limit = NULL, offset = NULL,
+                        state = NULL, type = NULL) {
  # flow_prefixes no longer classifies rows (crosswalk revenue_subtype
  # membership does -- General Revenue, i.e. everything except
  # insurance_trust) -- it only scopes the recipe-suggestion machinery to
@@ -75,6 +76,8 @@ cog_revenue <- function(govid, years, category = NULL,
    revenue_concept = revenue_concept,
    complete       = complete,
    limit          = limit,
-    offset         = offset
+    offset         = offset,
+    state          = state,
+    type           = type
  )
 }
@@ -44,10 +44,22 @@
 #'   in basket mode (recycles from length 1). Excluded types `4`/`5` (or
 #'   `"special_district"` / `"school_district"`) trigger an explanatory
 #'   message and an empty result.
+#' @param limit Maximum number of rows to return, applied in SQL. `NULL`
+#'   (default) returns every match -- which, with no other filter, is the
+#'   entire crosswalk. Utility mode only: pagination has no meaning in basket
+#'   mode, where the result is one resolved row per requested name in input
+#'   order, and is refused there with class
+#'   `uscogdata_basket_pagination_conflict`.
+#' @param offset Rows to skip before `limit` starts counting (0-based).
+#'   Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.
 #' @return A tibble of `canonical_fips_xwalk` rows. In utility mode, all
-#'   matches sorted by `population_acs` desc. In basket mode, resolved
-#'   rows in input order, with `attr(., "resolution")` set to the
-#'   sidecar tibble.
+#'   matches sorted by `population_acs` desc, ties broken by
+#'   `canonical_govid`. In basket mode, resolved rows in input order, with
+#'   `attr(., "resolution")` set to the sidecar tibble.
+#'
+#'   When `limit` is set, carries a `total_rows` attribute: the full
+#'   unpaginated match count, computed by the same query (`COUNT(*) OVER()`)
+#'   rather than a second scan.
 #' @seealso [cog_basket_resolution()], [cog_basket_unresolved()],
 #'   [cog_spending()], [cog_revenue()].
 #' @examples
@@ -82,7 +94,12 @@
 #' )
 #' }
 #' @export
-cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
+cog_gov_search <- function(name = NULL, state = NULL, type = NULL,
+                           limit = NULL, offset = NULL) {
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
+
  if (!is.null(type) && length(type) == 1L && .is_excluded_type(type)) {
    cli::cli_inform(c(
      i = "v0.1 covers gov_types 0-3 (state/county/city/township) only.",
@@ -94,6 +111,18 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
  con <- .ensure_session()

  if (length(name) > 1L) {
+    # Basket mode returns one resolved row per requested name, in input order,
+    # with a resolution sidecar describing how each was matched. A page of that
+    # is not a page of anything the caller asked for -- the sidecar would still
+    # describe every name -- so refuse rather than silently ignoring the
+    # arguments. Same shape as the recipe/complete refusals in .verb_spendrev().
+    if (!is.null(limit)) {
+      cli::cli_abort(c(
+        "`limit`/`offset` cannot be combined with basket mode.",
+        "i" = "Basket mode ({.code length(name) > 1}) returns one resolved row per requested name, in input order, with a resolution sidecar covering all of them.",
+        "*" = "Drop `limit`/`offset`, or search one name at a time."
+      ), class = "uscogdata_basket_pagination_conflict")
+    }
    return(.resolve_basket(name = name, state = state, type = type, con = con))
  }

@@ -123,12 +152,26 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
  }

  where <- if (length(preds) == 0L) "" else paste("WHERE", paste(preds, collapse = " AND "))
-  sql <- paste(
+  # canonical_govid breaks ties. population_acs alone is NOT a total order --
+  # governments sharing a population, and the whole NULLS LAST block, came back
+  # in whatever order the scan produced. That was invisible while every call
+  # returned the full result set, but it makes a paged sweep unsound: two
+  # requests can order the tied rows differently, so a row is duplicated on one
+  # page and missing from the next. Any pagination has to sit on a total order.
+  base_sql <- paste(
    "SELECT * FROM canonical_fips_xwalk",
    where,
-    "ORDER BY population_acs DESC NULLS LAST"
+    "ORDER BY population_acs DESC NULLS LAST, canonical_govid"
  )
-  tibble::as_tibble(DBI::dbGetQuery(con, sql))
+  result <- tibble::as_tibble(
+    DBI::dbGetQuery(con, .paginate_sql(base_sql, limit, offset))
+  )
+  if (is.null(limit)) return(result)
+
+  paged <- .take_pagination_total(result, con, base_sql)
+  out <- paged$result
+  attr(out, "total_rows") <- paged$total_rows
+  out
 }

 #' @noRd
@@ -184,11 +227,18 @@ cog_gov_search <- function(name = NULL, state = NULL, type = NULL) {
  if (!is.character(state) || length(state) != 1L) {
    cli::cli_abort("`state` must be a 2-letter USPS abbrev or a FIPS integer.")
  }
-  fips <- .state_abbrev_to_fips[[toupper(state)]]
-  if (is.null(fips)) {
-    cli::cli_abort("Unknown state abbreviation: {state}.")
+  # Membership tested before the lookup, not after: `.state_abbrev_to_fips` is
+  # a named CHARACTER vector, and `[[` on a name it does not carry throws
+  # base R's "subscript out of bounds" rather than returning NULL -- which
+  # made the curated message below unreachable dead code. Reported as a bare
+  # subscript error, `cog_gov_search(state = "ZZ")` gave no hint that the
+  # argument wants a postal abbreviation.
+  key <- toupper(state)
+  if (!key %in% names(.state_abbrev_to_fips)) {
+    cli::cli_abort("Unknown state abbreviation: {state}.",
+                   class = "uscogdata_unknown_state")
  }
-  fips
+  .state_abbrev_to_fips[[key]]
 }

 # USPS state / territory abbreviation -> 2-digit FIPS code.
@@ -2,13 +2,23 @@

 #' Internal: open session, register views, cache manifest.
 #' Not exported. Called lazily by verbs via .ensure_session().
+#'
+#' `threads` and `memory_limit` default to the resolved configuration and are
+#' applied as pragmas on the new connection. When both resolve to NULL -- which
+#' is the case unless the operator sets one -- NO pragma is issued at all, so an
+#' unconfigured session connects exactly as it did before this argument existed.
 #' @noRd
 cog_open <- function(url = .resolve_url(),
-                     cache_dir = .resolve_cache_dir()) {
+                     cache_dir = .resolve_cache_dir(),
+                     threads = .resolve_duckdb_threads(),
+                     memory_limit = .resolve_duckdb_memory_limit()) {
  .check_url_configured(url)
  if (!dir.exists(cache_dir)) dir.create(cache_dir, recursive = TRUE)

  con <- DBI::dbConnect(duckdb::duckdb())
+  # Before anything else touches the connection: httpfs reads the corpus, and
+  # a remote read should already be bound by whatever budget the operator set.
+  .apply_duckdb_limits(con, threads, memory_limit)
  DBI::dbExecute(con, "INSTALL httpfs; LOAD httpfs;")

  manifest <- .fetch_or_cache_manifest(url, cache_dir)
@@ -25,6 +35,26 @@ cog_open <- function(url = .resolve_url(),
  invisible(con)
 }

+#' Apply the operator's DuckDB resource budget to a fresh connection.
+#'
+#' Split out from cog_open() so the "unset changes nothing" property is one
+#' readable branch rather than two conditionals buried in the connection path.
+#' Both settings are session-scoped in DuckDB, so this must run per connection;
+#' cog_close() discards the connection and the next cog_open() re-resolves,
+#' which is what makes a changed option take effect on the next session.
+#' @noRd
+.apply_duckdb_limits <- function(con, threads, memory_limit) {
+  if (!is.null(threads)) {
+    DBI::dbExecute(con, sprintf("SET threads TO %d", threads))
+  }
+  if (!is.null(memory_limit)) {
+    # Quoted as a string literal: DuckDB's memory_limit takes '4GB', not 4GB.
+    DBI::dbExecute(con, sprintf("SET memory_limit TO %s",
+                                .sql_lit_chr(memory_limit)))
+  }
+  invisible(con)
+}
+
 #' @noRd
 .ensure_session <- function() {
  if (is.null(.uscogdata_env$con) ||
@@ -68,7 +68,9 @@
 #' millions/billions). The conversion is recorded in the provenance attribute
 #' under `transformations$units_conversion`.
 #'
-#' @param govid Character vector of `canonical_govid` values.
+#' @param govid Character vector of `canonical_govid` values, or `NULL` to name
+#'   the cohort by `state`/`type` instead. One of `govid`, `state`, or `type`
+#'   is required.
 #' @param years Integer vector of years.
 #' @param category Character vector of category names (from
 #'   `summary_categories.category`), or `NULL` for all categories broken out
@@ -185,6 +187,29 @@
 #'   `recipe` and with `complete = TRUE` -- see `offset` and `total_rows`.
 #' @param offset Rows to skip before `limit` starts counting (0-based).
 #'   Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.
+#' @param state,type Name the cohort by predicate instead of by id: `state` is
+#'   a 2-letter USPS abbreviation (or a FIPS code) and `type` is one of
+#'   `"state"`, `"county"`, `"city"`, `"township"` (or the integer `0:3`) --
+#'   the same vocabulary, and the same internal coercion, as
+#'   [cog_gov_search()]. Both default to `NULL`.
+#'
+#'   The cohort is then expressed as a subquery against `canonical_fips_xwalk`
+#'   inside each statement rather than round-tripped through R as a literal id
+#'   list. For a fleet-scale cohort that is the difference between a
+#'   301,591-character `IN` list re-parsed in 5--8 statements per call and a
+#'   constant-size predicate: measured at **94 ms versus 449 ms** for the same
+#'   FY2022 aggregate over the 20,106-government `type = "city"` cohort, within
+#'   7% of the no-filter floor.
+#'
+#'   Supplying `govid` **and** `state`/`type` INTERSECTS them -- the
+#'   governments in `govid` that also match the predicate -- rather than one
+#'   silently taking precedence. Naming no cohort at all (`govid`, `state` and
+#'   `type` all `NULL`) aborts with class `uscogdata_no_cohort`.
+#'
+#'   When the cohort is named by predicate, `provenance$scope$govids_found`
+#'   and `govids_missing` are empty -- there is no id list to report against --
+#'   and `provenance$scope$cohort` carries `state`, `type` and
+#'   `n_governments` instead. A `govid`-named cohort reports exactly as before.
 #' @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`,
@@ -198,11 +223,12 @@
 #'   round trip -- so a caller walking pages never has to ask "how many are
 #'   there" separately.
 #' @export
-cog_spending <- function(govid, years, category = NULL,
+cog_spending <- function(govid = NULL, years, category = NULL,
                         per_capita = FALSE, adjust_to_year = NULL,
                         basis = c("harmonized", "raw"), recipe = NULL,
                         expenditure_concept = c("primary", "direct", "total"),
-                         complete = FALSE, limit = NULL, offset = NULL) {
+                         complete = FALSE, limit = NULL, offset = NULL,
+                         state = NULL, type = NULL) {
  # flow_prefixes no longer classifies rows (crosswalk subtype membership
  # does, per expenditure_concept) -- it only scopes the recipe-suggestion
  # machinery to this verb's recipe families (see R/suggestions.R; the
@@ -223,7 +249,9 @@ cog_spending <- function(govid, years, category = NULL,
    expenditure_concept = expenditure_concept,
    complete      = complete,
    limit         = limit,
-    offset        = offset
+    offset        = offset,
+    state         = state,
+    type          = type
  )
 }

@@ -251,7 +279,8 @@ cog_spending <- function(govid, years, category = NULL,
                           basis = c("harmonized", "raw"), recipe = NULL,
                           expenditure_concept = c("primary", "direct", "total"),
                           revenue_concept = c("general", "total"),
-                           complete = FALSE, limit = NULL, offset = NULL) {
+                           complete = FALSE, limit = NULL, offset = NULL,
+                           state = NULL, type = NULL) {
  basis_explicit <- length(basis) == 1L
  basis <- match.arg(basis, c("harmonized", "raw"))
  # match.arg() itself throws a base `simpleError`, not an rlang-classed
@@ -291,7 +320,10 @@ cog_spending <- function(govid, years, category = NULL,
    .revenue_concept_subtypes(revenue_concept)
  }

-  govid <- .coerce_govid_input(govid, arg = "govid")
+  # NULL `govid` means "the cohort is named by predicate"; anything else is
+  # coerced and validated exactly as before, so an empty or wrong-typed vector
+  # still fails with its original message rather than being read as absent.
+  govid <- if (is.null(govid)) NULL else .coerce_govid_input(govid, arg = "govid")
  # allow_all_categories = TRUE: cog_spending()/cog_revenue() are the two
  # verbs the reserved pseudo-category is defined for. cog_balances() shares
  # this validator but leaves the argument at its FALSE default, so it
@@ -300,6 +332,10 @@ cog_spending <- function(govid, years, category = NULL,
  .validate_verb_inputs(govid, years, category, per_capita, adjust_to_year,
                        recipe, allow_all_categories = TRUE)

+  # Built after validation so the argument-shape errors above keep firing
+  # first, and before .ensure_session() so a bad state/type costs no I/O.
+  cohort <- .make_cohort(govid, state, type)
+
  # Recognize the reserved pseudo-category. Detected after type validation so a
  # non-character `category` still fails with the ordinary type error.
  all_categories <- !is.null(category) && .ALL_CATEGORIES %in% category
@@ -362,17 +398,10 @@ cog_spending <- function(govid, years, category = NULL,
  # up front rather than silently ignored: complete = TRUE fills a grid over
  # the FULL requested (year, category) space, and a recipe's result comes
  # from .run_recipe()'s own query, which this function does not touch.
+  paging <- .validate_pagination(limit, offset)
+  limit  <- paging$limit
+  offset <- paging$offset
  if (!is.null(limit)) {
-    limit <- as.integer(limit)
-    if (length(limit) != 1L || is.na(limit) || limit < 0L) {
-      cli::cli_abort("`limit` must be a single non-negative integer.",
-                     class = "uscogdata_invalid_pagination")
-    }
-    offset <- if (is.null(offset)) 0L else as.integer(offset)
-    if (length(offset) != 1L || is.na(offset) || offset < 0L) {
-      cli::cli_abort("`offset` must be a single non-negative integer.",
-                     class = "uscogdata_invalid_pagination")
-    }
    if (complete) {
      cli::cli_abort(c(
        "`limit`/`offset` cannot be combined with `complete = TRUE`.",
@@ -407,7 +436,7 @@ cog_spending <- function(govid, years, category = NULL,
    .validate_recipe_id(con, recipe)
    comps <- .recipe_components(con, recipe)
    recipe_label <- comps$label[[1]]
-    result <- .run_recipe(con, recipe, govid, years)
+    result <- .run_recipe(con, recipe, cohort, years)
    sql <- attr(result, "sql_query")
    result <- .shape_recipe_result(result, subtype_col, recipe_label)
    recipe_block <- list(
@@ -423,31 +452,21 @@ cog_spending <- function(govid, years, category = NULL,
    } else {
      NULL
    }
-    sql <- .build_verb_sql(view, subtype_col, govid, years,
+    sql <- .build_verb_sql(view, subtype_col, cohort, years,
                           if (all_categories) NULL else category,
                           ig_view, subtype_scope,
                           all_categories = all_categories,
                           limit = limit, offset = offset)
    result <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
    if (!is.null(limit)) {
-      # COUNT(*) OVER() rides along as an ordinary column so the total comes
-      # from the same scan when this page has any rows -- see
-      # .build_verb_sql(). An empty page (offset past the end) carries no
-      # such row to read it from, so that one case falls back to a second,
-      # unpaginated COUNT(*) query rather than reporting a wrong zero.
-      if (nrow(result) > 0L) {
-        total_rows <- result$pagination_total_rows[[1]]
-        result$pagination_total_rows <- NULL
-      } else {
-        count_sql <- sprintf(
-          "SELECT COUNT(*) AS n FROM (%s) AS _uncounted",
-          .build_verb_sql(view, subtype_col, govid, years,
-                          if (all_categories) NULL else category,
-                          ig_view, subtype_scope,
-                          all_categories = all_categories)
-        )
-        total_rows <- as.integer(DBI::dbGetQuery(con, count_sql)$n[[1]])
-      }
+      paged <- .take_pagination_total(result, con, function() {
+        .build_verb_sql(view, subtype_col, cohort, years,
+                        if (all_categories) NULL else category,
+                        ig_view, subtype_scope,
+                        all_categories = all_categories)
+      })
+      result     <- paged$result
+      total_rows <- paged$total_rows
    }
  }

@@ -457,13 +476,13 @@ cog_spending <- function(govid, years, category = NULL,
  # spurious 0.
  completion <- list(applied = FALSE, rows_filled = 0L, absence_means = list())
  if (complete) {
-    result <- .complete_result(result, con, subtype_col, govid, years,
+    result <- .complete_result(result, con, subtype_col, cohort, years,
                               category, subtype_scope)
    completion <- attr(result, ".completion")
    attr(result, ".completion") <- NULL
  }

-  if (per_capita) result <- .attach_per_capita(result, con, govid)
+  if (per_capita) result <- .attach_per_capita(result, con)
  if (!is.null(adjust_to_year)) {
    result <- .attach_real_dollars(result, adjust_to_year, per_capita)
  }
@@ -491,7 +510,7 @@ cog_spending <- function(govid, years, category = NULL,
    basis_for_prov <- resolved$basis
    basis_note_for_prov <- resolved$note
    harmonization <- .build_harmonization_block(
-      con, govid, years, resolved, subtype_col, subtype_scope
+      con, cohort, years, resolved, subtype_col, subtype_scope
    )
    # C1(a): gap detection must run against the Direct leg alone. `result`
    # can also carry UNION'd intergovernmental rows (expenditure_concept =
@@ -506,7 +525,7 @@ cog_spending <- function(govid, years, category = NULL,
    } else {
      result
    }
-    suggestions <- .build_suggestions(con, govid, years, category,
+    suggestions <- .build_suggestions(con, cohort, years, category,
                                       direct_leg_result,
                                       resolved$basis, flow_prefixes,
                                       .select_long_view(view_base, resolved$basis),
@@ -611,6 +630,12 @@ cog_spending <- function(govid, years, category = NULL,
  )
  prov$scope$govids_found   <- scope$found
  prov$scope$govids_missing <- scope$missing
+  # A predicate-named cohort has no id list to report found/missing against
+  # (both stay empty), so it describes itself instead. Deliberately a COUNT
+  # rather than the resolved ids: enumerating them would put 20,000 govids in
+  # every fleet-scale response body, which is the cost this path exists to
+  # remove. NULL for a govid-named cohort, so that output is untouched.
+  prov$scope$cohort <- .cohort_provenance(con, cohort)
  attr(result, "provenance") <- prov
  attr(result, ".popyear_range") <- NULL
  # Attached here, after every downstream transform (per_capita/real-dollar
@@ -639,7 +664,11 @@ cog_spending <- function(govid, years, category = NULL,
 .validate_verb_inputs <- function(govid, years, category,
                                  per_capita, adjust_to_year, recipe = NULL,
                                  allow_all_categories = FALSE) {
-  if (!is.character(govid) || length(govid) == 0L) {
+  # NULL is allowed only because the caller has already established that the
+  # cohort is named some other way (`state`/`type`); .make_cohort() is what
+  # refuses a call that names no cohort at all. A supplied-but-empty `govid`
+  # still fails here, exactly as before.
+  if (!is.null(govid) && (!is.character(govid) || length(govid) == 0L)) {
    cli::cli_abort("`govid` must be a non-empty character vector.")
  }
  if (!(is.integer(years) || is.numeric(years)) || length(years) == 0L) {
@@ -740,10 +769,10 @@ cog_spending <- function(govid, years, category = NULL,
 }

 #' @noRd
-.build_verb_sql <- function(view, subtype_col, govid, years, category,
+.build_verb_sql <- function(view, subtype_col, cohort, years, category,
                            ig_view = NULL, subtype_scope = NULL,
                            all_categories = FALSE, limit = NULL, offset = NULL) {
-  govid_lit <- .sql_lit_chr(govid)
+  cohort_pred <- .cohort_sql(cohort)
  years_lit <- paste(as.integer(years), collapse = ",")
  # In all-categories mode there is no category filter: the sum is defined by
  # the concept's SUBTYPE allowlist (subtype_pred below), which is the real
@@ -813,13 +842,13 @@ cog_spending <- function(govid, years, category = NULL,
       string_agg(DISTINCT item_code, ',' ORDER BY item_code) AS codes_included,
       bool_or(is_aggregate) AS aggregate_fallback
     FROM %2$s
-     WHERE canonical_govid IN (%3$s)
+     WHERE %3$s
       AND year IN (%4$s)
       %5$s
       %6$s
     GROUP BY year, canonical_govid, gov_name, xwalk_gov_name, %1$s%8$s
     ORDER BY year, canonical_govid, %1$s%8$s",
-    subtype_col, source_expr, govid_lit, years_lit, category_pred, subtype_pred,
+    subtype_col, source_expr, cohort_pred, years_lit, category_pred, subtype_pred,
    category_select, category_group
  )

@@ -827,26 +856,21 @@ cog_spending <- function(govid, years, category = NULL,
  # matching row across the network only to slice and discard most of it
  # afterward (the pattern behind the 2026-08-06 production incident: a
  # 193,105-row/194-page sweep re-ran the full query and re-listified every
-  # row on EVERY page). COUNT(*) OVER() rides along as an ordinary column so
-  # the caller gets the true total from this same scan -- see the call site
-  # in .verb_spendrev(), which reads it off row 1 and strips it back out.
-  # The outer SELECT * wrapping (rather than appending LIMIT/OFFSET directly
-  # to base_sql) is what makes COUNT(*) OVER() see the post-GROUP-BY row
-  # count, not the pre-aggregation one.
-  if (is.null(limit)) {
-    base_sql
-  } else {
-    sprintf(
-      "SELECT *, COUNT(*) OVER() AS pagination_total_rows
-       FROM (%s) AS _paged
-       LIMIT %d OFFSET %d",
-      base_sql, limit, offset
-    )
-  }
+  # row on EVERY page). See .paginate_sql() in R/pagination.R for why the
+  # wrapping is an outer SELECT rather than a bare LIMIT on base_sql.
+  .paginate_sql(base_sql, limit, offset)
 }

+#' Join population onto a result and derive the per-capita columns.
+#'
+#' The population lookup is keyed on the govids PRESENT IN `result`, not on the
+#' cohort that produced it. Those are the only ones the LEFT JOIN below can
+#' match, so the joined output is identical either way -- but it means this
+#' works unchanged for a cohort named by predicate (where no id list exists in
+#' R at all), and on a paginated call it looks up one page's governments
+#' instead of the whole fleet's.
 #' @noRd
-.attach_per_capita <- function(result, con, govid) {
+.attach_per_capita <- function(result, con) {
  if (nrow(result) == 0L) {
    result$amt_per_capita_nominal <- numeric(0)
    result$pop_source <- character(0)
@@ -859,7 +883,7 @@ cog_spending <- function(govid, years, category = NULL,
     FROM gov_population_yearly
     WHERE canonical_govid IN (%s)
       AND year IN (%s)",
-    .sql_lit_chr(govid), years_lit
+    .sql_lit_chr(unique(result$canonical_govid)), years_lit
  )
  pops <- tibble::as_tibble(DBI::dbGetQuery(con, sql))
  result <- dplyr::left_join(result, pops,
@@ -42,8 +42,8 @@
 #' 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 cohort The verb's cohort object (see `.make_cohort()`), naming the
+#'   governments by id, by state/type predicate, or both.
 #' @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).
@@ -85,7 +85,7 @@
 #'   ig_recipe_id, trigger, suppressed_amount, suppressed_years,
 #'   suppressed_codes)`, possibly empty.
 #' @noRd
-.build_suggestions <- function(con, govid, years, category, result, basis,
+.build_suggestions <- function(con, cohort, years, category, result, basis,
                                flow_prefixes, long_view,
                                all_categories = FALSE,
                                subtype_col = NULL, subtype_scope = NULL) {
@@ -160,7 +160,7 @@
  # violation would kill signposting, the exact failure class uscogdata#9
  # exists to prevent. Owner's call: keep this simple; a batch-aware
  # optimization, if one is worth building, is a separate issue.
-  supp <- .suppressed_components(con, candidates, govid, years, long_view, flow_prefixes)
+  supp <- .suppressed_components(con, candidates, cohort, years, long_view, flow_prefixes)

  if (length(gap_years) == 0L && nrow(supp) == 0L) return(list())

@@ -188,9 +188,9 @@
             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 %s
         AND l.year IN (%s)",
-      .sql_lit_chr(candidates), .sql_lit_chr(govid),
+      .sql_lit_chr(candidates), .cohort_sql(cohort, "l.canonical_govid"),
      paste(gap_years, collapse = ",")
    ))
  }
@@ -50,7 +50,9 @@
 #'
 #' @param con Active DuckDB connection.
 #' @param candidates Character vector of recipe ids to measure.
-#' @param govid Character vector of canonical_govid values.
+#' @param cohort The verb's cohort object (see `.make_cohort()`), rendered
+#'   into the govid predicate on both the outer scan and the restated
+#'   NOT EXISTS filter.
 #' @param years Integer vector of requested years.
 #' @param long_view Name of the verb's long view, from `.select_long_view()`.
 #' @param flow_prefixes The calling verb's own flow-type prefixes (see
@@ -60,7 +62,7 @@
 #'   dollars), `suppressed_codes` (comma-joined, sorted). Zero rows when
 #'   nothing is suppressed.
 #' @noRd
-.suppressed_components <- function(con, candidates, govid, years, long_view,
+.suppressed_components <- function(con, candidates, cohort, years, long_view,
                                    flow_prefixes) {
  empty <- tibble::tibble(
    recipe_id = character(0), year = numeric(0),
@@ -93,7 +95,7 @@
           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 (%1$s)
-       AND l.canonical_govid IN (%2$s)
+       AND %2$s
       AND l.year IN (%3$s)
       AND l.amt <> 0
       AND LEFT(r.component_code, 1) IN (%5$s)
@@ -103,13 +105,13 @@
           AND v.year = l.year
           AND v.item_code = l.item_code
           AND v.year IN (%3$s)              -- restated: enables partition pruning (I3a)
-           AND v.canonical_govid IN (%2$s)   -- restated: pushes the govid filter (I3a)
+           AND %6$s   -- restated: pushes the cohort filter (I3a)
       )
     GROUP BY 1, 2
     ORDER BY 1, 2",
-    .sql_lit_chr(candidates), .sql_lit_chr(govid),
+    .sql_lit_chr(candidates), .cohort_sql(cohort, "l.canonical_govid"),
    paste(as.integer(years), collapse = ","), long_view,
-    .sql_lit_chr(flow_prefixes)
+    .sql_lit_chr(flow_prefixes), .cohort_sql(cohort, "v.canonical_govid")
  )
  tibble::as_tibble(DBI::dbGetQuery(con, sql))
 }
@@ -122,9 +122,25 @@
 #' fallback -- correct for the local temp corpora the direct-execution tests
 #' build.
 #' @noRd
+#' `fixed = TRUE` is load-bearing, not a style choice.
+#'
+#' In regex mode, `gsub()` interprets backslashes in the REPLACEMENT string as
+#' escape sequences and silently drops them. A Windows corpus path is full of
+#' them, so `C:\Users\RUNNER\AppData\...` was substituted in as
+#' `C:UsersRUNNERAppData...` and every DuckDB read failed with "No files found
+#' that match the pattern". `fixed = TRUE` treats pattern and replacement as
+#' literal text, which is what a filesystem path needs.
+#'
+#' This is why the package could not read a LOCAL corpus on Windows at all --
+#' including the test fixture, hence the entire suite, and any `cog_mirror()`
+#' copy. Remote https URLs were unaffected, having no backslashes, which is
+#' part of why it stayed hidden: the bug predates the `{long_files}` token and
+#' lived in the original `{url}` substitution, unnoticed because nothing ever
+#' ran on Windows until the mirror's check matrix existed.
+#' @noRd
 .render_view_sql <- function(sql, url, manifest = list()) {
-  sql <- gsub("\\{long_files\\}", .long_files_sql(url, manifest), sql, fixed = FALSE)
-  gsub("\\{url\\}", url, sql, fixed = FALSE)
+  sql <- gsub("{long_files}", .long_files_sql(url, manifest), sql, fixed = TRUE)
+  gsub("{url}", url, sql, fixed = TRUE)
 }

 #' Register DuckDB views from inst/sql/ SQL files
@@ -110,6 +110,17 @@ After that, nothing in your analysis touches an external service.
 - `USCOGDATA_URL` — corpus root: an HTTPS URL or a local path, **trailing slash required**
 - `USCOGDATA_CACHE_DIR` — where the manifest is cached (default: user cache dir)
 - `USCOGDATA_MANIFEST_TTL_SECS` — manifest re-fetch interval (default 3600)
+- `USCOGDATA_DUCKDB_THREADS` — cap DuckDB's thread count (default: every visible core)
+- `USCOGDATA_DUCKDB_MEMORY_LIMIT` — cap DuckDB's memory, e.g. `"4GB"` (default: DuckDB's own)
+
+Each also has an `options()` spelling — `uscogdata.url`, `uscogdata.duckdb_threads`,
+and so on — and the environment variable wins where both are set.
+
+The two DuckDB caps exist for **servers**, not laptops. Unset, DuckDB claims every
+core it can see, which is right for one interactive session on your own machine and
+wrong when several readers share a box: each claims the whole machine and they fight.
+Capping costs roughly 5% on a single query and is worth it anywhere the process is
+sharing hardware.

 ## Amounts are in full US dollars

@@ -5,18 +5,23 @@
 \title{Cash and security holdings for one or more governments}
 \usage{
 cog_balances(
-  govid,
+  govid = NULL,
  years,
  category = NULL,
  per_capita = FALSE,
  adjust_to_year = NULL,
  basis = c("harmonized", "raw"),
-  recipe = NULL
+  recipe = NULL,
+  state = NULL,
+  type = NULL,
+  limit = NULL,
+  offset = NULL
 )
 }
 \arguments{
 \item{govid}{Canonical govid(s): a character vector, or a data frame with a
-`canonical_govid` column (e.g. from [cog_gov_search()]).}
+`canonical_govid` column (e.g. from [cog_gov_search()]). `NULL` to name
+the cohort by `state`/`type` instead.}

 \item{years}{Integer vector of fiscal years.}

@@ -49,6 +54,38 @@ and raw space are identical for holdings. Reported in
 \item{recipe}{Optional harmonization recipe id (see [cog_recipes()]).
 `"cash_securities_z77_wide"` and `"cash_securities_z78_wide"` bridge the
 wide era to the modern one.}
+
+\item{state, type}{Name the cohort by predicate instead of by id: `state` is
+  a 2-letter USPS abbreviation (or a FIPS code) and `type` is one of
+  `"state"`, `"county"`, `"city"`, `"township"` (or the integer `0:3`) --
+  the same vocabulary, and the same internal coercion, as
+  [cog_gov_search()]. Both default to `NULL`.
+
+  The cohort is then expressed as a subquery against `canonical_fips_xwalk`
+  inside each statement rather than round-tripped through R as a literal id
+  list. For a fleet-scale cohort that is the difference between a
+  301,591-character `IN` list re-parsed in 5--8 statements per call and a
+  constant-size predicate: measured at **94 ms versus 449 ms** for the same
+  FY2022 aggregate over the 20,106-government `type = "city"` cohort, within
+  7% of the no-filter floor.
+
+  Supplying `govid` **and** `state`/`type` INTERSECTS them -- the
+  governments in `govid` that also match the predicate -- rather than one
+  silently taking precedence. Naming no cohort at all (`govid`, `state` and
+  `type` all `NULL`) aborts with class `uscogdata_no_cohort`.
+
+  When the cohort is named by predicate, `provenance$scope$govids_found`
+  and `govids_missing` are empty -- there is no id list to report against --
+  and `provenance$scope$cohort` carries `state`, `type` and
+  `n_governments` instead. A `govid`-named cohort reports exactly as before.}
+
+\item{limit}{Maximum number of result rows to return, pushed into the SQL
+rather than applied after materializing every row. `NULL` (default)
+returns everything. Cannot be combined with `recipe` -- see `offset` and
+`total_rows`.}
+
+\item{offset}{Rows to skip before `limit` starts counting (0-based).
+Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.}
 }
 \value{
 Tibble with columns `year`, `canonical_govid`, `gov_name`,
@@ -67,6 +104,10 @@ Tibble with columns `year`, `canonical_govid`, `gov_name`,
  and `truncated` (the observed subtypes whose coverage falls short of the
  requested years). `expenditure_concept`/`revenue_concept` are `NA` --
  holdings are a stock, not a flow, so neither concept vocabulary applies.
+
+  When `limit` is set, also carries a `total_rows` attribute: the full
+  unpaginated row count, computed by the same query (`COUNT(*) OVER()`)
+  rather than a second scan.
 }
 \description{
 Returns Census cash-and-security holdings (`category_type = "balance"`):
@@ -4,7 +4,13 @@
 \alias{cog_gov_search}
 \title{Search for governments by name, state, and/or type}
 \usage{
-cog_gov_search(name = NULL, state = NULL, type = NULL)
+cog_gov_search(
+  name = NULL,
+  state = NULL,
+  type = NULL,
+  limit = NULL,
+  offset = NULL
+)
 }
 \arguments{
 \item{name}{Character vector of place name(s). Length 1 = utility mode;
@@ -19,12 +25,26 @@ all entries; otherwise must match `length(name)`.}
 in basket mode (recycles from length 1). Excluded types `4`/`5` (or
 `"special_district"` / `"school_district"`) trigger an explanatory
 message and an empty result.}
+
+\item{limit}{Maximum number of rows to return, applied in SQL. `NULL`
+(default) returns every match -- which, with no other filter, is the
+entire crosswalk. Utility mode only: pagination has no meaning in basket
+mode, where the result is one resolved row per requested name in input
+order, and is refused there with class
+`uscogdata_basket_pagination_conflict`.}
+
+\item{offset}{Rows to skip before `limit` starts counting (0-based).
+Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.}
 }
 \value{
 A tibble of `canonical_fips_xwalk` rows. In utility mode, all
-  matches sorted by `population_acs` desc. In basket mode, resolved
-  rows in input order, with `attr(., "resolution")` set to the
-  sidecar tibble.
+  matches sorted by `population_acs` desc, ties broken by
+  `canonical_govid`. In basket mode, resolved rows in input order, with
+  `attr(., "resolution")` set to the sidecar tibble.
+
+  When `limit` is set, carries a `total_rows` attribute: the full
+  unpaginated match count, computed by the same query (`COUNT(*) OVER()`)
+  rather than a second scan.
 }
 \description{
 Resolves human-readable place names into rows of `canonical_fips_xwalk`,
@@ -5,7 +5,7 @@
 \title{Summarized revenue by category}
 \usage{
 cog_revenue(
-  govid,
+  govid = NULL,
  years,
  category = NULL,
  per_capita = FALSE,
@@ -15,11 +15,15 @@ cog_revenue(
  revenue_concept = c("general", "total"),
  complete = FALSE,
  limit = NULL,
-  offset = NULL
+  offset = NULL,
+  state = NULL,
+  type = NULL
 )
 }
 \arguments{
-\item{govid}{Character vector of `canonical_govid` values.}
+\item{govid}{Character vector of `canonical_govid` values, or `NULL` to name
+the cohort by `state`/`type` instead. One of `govid`, `state`, or `type`
+is required.}

 \item{years}{Integer vector of years.}

@@ -126,6 +130,30 @@ exactly as before this parameter existed. Mutually exclusive with

 \item{offset}{Rows to skip before `limit` starts counting (0-based).
 Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.}
+
+\item{state, type}{Name the cohort by predicate instead of by id: `state` is
+  a 2-letter USPS abbreviation (or a FIPS code) and `type` is one of
+  `"state"`, `"county"`, `"city"`, `"township"` (or the integer `0:3`) --
+  the same vocabulary, and the same internal coercion, as
+  [cog_gov_search()]. Both default to `NULL`.
+
+  The cohort is then expressed as a subquery against `canonical_fips_xwalk`
+  inside each statement rather than round-tripped through R as a literal id
+  list. For a fleet-scale cohort that is the difference between a
+  301,591-character `IN` list re-parsed in 5--8 statements per call and a
+  constant-size predicate: measured at **94 ms versus 449 ms** for the same
+  FY2022 aggregate over the 20,106-government `type = "city"` cohort, within
+  7% of the no-filter floor.
+
+  Supplying `govid` **and** `state`/`type` INTERSECTS them -- the
+  governments in `govid` that also match the predicate -- rather than one
+  silently taking precedence. Naming no cohort at all (`govid`, `state` and
+  `type` all `NULL`) aborts with class `uscogdata_no_cohort`.
+
+  When the cohort is named by predicate, `provenance$scope$govids_found`
+  and `govids_missing` are empty -- there is no id list to report against --
+  and `provenance$scope$cohort` carries `state`, `type` and
+  `n_governments` instead. A `govid`-named cohort reports exactly as before.}
 }
 \value{
 Tibble with columns `year`, `canonical_govid`, `gov_name`,
@@ -5,7 +5,7 @@
 \title{Summarized spending by category}
 \usage{
 cog_spending(
-  govid,
+  govid = NULL,
  years,
  category = NULL,
  per_capita = FALSE,
@@ -15,11 +15,15 @@ cog_spending(
  expenditure_concept = c("primary", "direct", "total"),
  complete = FALSE,
  limit = NULL,
-  offset = NULL
+  offset = NULL,
+  state = NULL,
+  type = NULL
 )
 }
 \arguments{
-\item{govid}{Character vector of `canonical_govid` values.}
+\item{govid}{Character vector of `canonical_govid` values, or `NULL` to name
+the cohort by `state`/`type` instead. One of `govid`, `state`, or `type`
+is required.}

 \item{years}{Integer vector of years.}

@@ -146,6 +150,30 @@ exactly as before this parameter existed. Mutually exclusive with

 \item{offset}{Rows to skip before `limit` starts counting (0-based).
 Ignored if `limit` is `NULL`; defaults to `0L` when `limit` is set.}
+
+\item{state, type}{Name the cohort by predicate instead of by id: `state` is
+  a 2-letter USPS abbreviation (or a FIPS code) and `type` is one of
+  `"state"`, `"county"`, `"city"`, `"township"` (or the integer `0:3`) --
+  the same vocabulary, and the same internal coercion, as
+  [cog_gov_search()]. Both default to `NULL`.
+
+  The cohort is then expressed as a subquery against `canonical_fips_xwalk`
+  inside each statement rather than round-tripped through R as a literal id
+  list. For a fleet-scale cohort that is the difference between a
+  301,591-character `IN` list re-parsed in 5--8 statements per call and a
+  constant-size predicate: measured at **94 ms versus 449 ms** for the same
+  FY2022 aggregate over the 20,106-government `type = "city"` cohort, within
+  7% of the no-filter floor.
+
+  Supplying `govid` **and** `state`/`type` INTERSECTS them -- the
+  governments in `govid` that also match the predicate -- rather than one
+  silently taking precedence. Naming no cohort at all (`govid`, `state` and
+  `type` all `NULL`) aborts with class `uscogdata_no_cohort`.
+
+  When the cohort is named by predicate, `provenance$scope$govids_found`
+  and `govids_missing` are empty -- there is no id list to report against --
+  and `provenance$scope$cohort` carries `state`, `type` and
+  `n_governments` instead. A `govid`-named cohort reports exactly as before.}
 }
 \value{
 Tibble with columns `year`, `canonical_govid`, `gov_name`,
@@ -4,7 +4,7 @@ test_that(".build_verb_sql emits a literal category and no category filter in al
  sql <- uscogdata:::.build_verb_sql(
    view         = "spending_annotated",
    subtype_col  = "spend_subtype",
-    govid        = "552025209777",
+    cohort = uscogdata:::.make_cohort("552025209777"),
    years        = 2019L,
    category     = NULL,
    subtype_scope = c("operations", "capital"),
@@ -24,7 +24,7 @@ test_that(".build_verb_sql emits a literal category and no category filter in al
 test_that(".build_verb_sql is unchanged when all_categories is FALSE", {
  args <- list(
    view = "spending_annotated", subtype_col = "spend_subtype",
-    govid = "552025209777", years = 2019L, category = NULL,
+    cohort = uscogdata:::.make_cohort("552025209777"), years = 2019L, category = NULL,
    subtype_scope = c("operations", "capital")
  )
  old <- do.call(uscogdata:::.build_verb_sql, args)
@@ -244,7 +244,7 @@ test_that('"All Categories" candidate scoping is symmetric with .build_verb_sql(
  con <- uscogdata:::.ensure_session()

  none <- uscogdata:::.build_suggestions(
-    con, govid = "010000226085", years = 2011L,
+    con, cohort = uscogdata:::.make_cohort("010000226085"), years = 2011L,
    category = "All Categories", result = NULL, basis = "harmonized",
    flow_prefixes = c("E", "F", "G"),
    long_view = "spending_long_harmonized",
@@ -255,7 +255,7 @@ test_that('"All Categories" candidate scoping is symmetric with .build_verb_sql(
  expect_length(none, 0L)

  scoped <- uscogdata:::.build_suggestions(
-    con, govid = "010000226085", years = 2011L,
+    con, cohort = uscogdata:::.make_cohort("010000226085"), years = 2011L,
    category = "All Categories", result = NULL, basis = "harmonized",
    flow_prefixes = c("E", "F", "G"),
    long_view = "spending_long_harmonized",
@@ -0,0 +1,218 @@
+# The money verbs accepting a cohort by predicate (uscogdata#58).
+#
+# The load-bearing property is EQUIVALENCE: naming the same set of governments
+# by id and by state/type must return the same rows. Everything else here is
+# about the ways that equivalence could silently break -- the postal/FIPS
+# translation, the intersection rule, and provenance no longer having an id
+# list to describe.
+#
+# Fixture cohorts used: RI (fips 44) cities = 8 governments, DE (fips 10)
+# counties = 3. Small on purpose; the size of the win is measured against the
+# production corpus, not here.
+
+# --- equivalence -----------------------------------------------------------
+
+test_that("a predicate cohort returns exactly what the same ids return", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "RI", type = "city")$canonical_govid
+    expect_gt(length(ids), 1L)
+
+    by_id   <- cog_spending(govid = ids, years = 2019)
+    by_pred <- cog_spending(years = 2019, state = "RI", type = "city")
+
+    # Compare the data itself, ignoring the provenance attribute -- which is
+    # SUPPOSED to differ (see the scope tests below).
+    expect_equal(
+      as.data.frame(by_id[order(by_id$canonical_govid, by_id$category), ]),
+      as.data.frame(by_pred[order(by_pred$canonical_govid, by_pred$category), ]),
+      ignore_attr = TRUE
+    )
+    expect_gt(nrow(by_pred), 0L)
+  })
+})
+
+test_that("equivalence holds for cog_revenue()", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "DE", type = "county")$canonical_govid
+    by_id   <- cog_revenue(govid = ids, years = 2019)
+    by_pred <- cog_revenue(years = 2019, state = "DE", type = "county")
+    expect_equal(nrow(by_id), nrow(by_pred))
+    expect_equal(sum(by_id$amt_nominal), sum(by_pred$amt_nominal))
+  })
+})
+
+test_that("equivalence holds for cog_balances()", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "DE", type = "county")$canonical_govid
+    by_id   <- cog_balances(govid = ids, years = 2019)
+    by_pred <- cog_balances(years = 2019, state = "DE", type = "county")
+    expect_equal(nrow(by_id), nrow(by_pred))
+    expect_equal(sum(by_id$amt_nominal), sum(by_pred$amt_nominal))
+  })
+})
+
+test_that("equivalence survives per_capita, adjust_to_year and pagination", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "RI", type = "city")$canonical_govid
+
+    # per_capita now keys its population lookup on the rows in the result
+    # rather than on the requested cohort; these must stay identical.
+    by_id   <- cog_spending(govid = ids, years = 2019, per_capita = TRUE,
+                            adjust_to_year = 2020)
+    by_pred <- cog_spending(years = 2019, state = "RI", type = "city",
+                            per_capita = TRUE, adjust_to_year = 2020)
+    expect_equal(by_id$amt_per_capita_nominal, by_pred$amt_per_capita_nominal)
+    expect_equal(by_id$amt_per_capita_real, by_pred$amt_per_capita_real)
+    expect_equal(by_id$pop_source, by_pred$pop_source)
+
+    paged_id   <- cog_spending(govid = ids, years = 2019, limit = 5, offset = 5)
+    paged_pred <- cog_spending(years = 2019, state = "RI", type = "city",
+                               limit = 5, offset = 5)
+    expect_equal(as.data.frame(paged_id), as.data.frame(paged_pred),
+                 ignore_attr = TRUE)
+    expect_identical(attr(paged_id, "total_rows"), attr(paged_pred, "total_rows"))
+  })
+})
+
+test_that("a state-only predicate spans every type in that state", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "DE", type = NULL)$canonical_govid
+    by_id   <- cog_spending(govid = ids, years = 2019)
+    by_pred <- cog_spending(years = 2019, state = "DE")
+    expect_equal(nrow(by_id), nrow(by_pred))
+  })
+})
+
+# --- the postal/FIPS trap --------------------------------------------------
+
+test_that("a postal abbreviation resolves to rows, not to silence", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    # canonical_fips_xwalk.fips_state holds "44", not "RI". A predicate built
+    # from the raw parameter matches nothing and returns an empty result that
+    # reads as "these governments reported nothing" -- the exact trap cog-api
+    # hit. A zero-row result here is the regression.
+    r <- cog_spending(years = 2019, state = "RI", type = "city")
+    expect_gt(nrow(r), 0L)
+
+    # And the FIPS form is accepted as the same cohort.
+    expect_equal(nrow(cog_spending(years = 2019, state = "44", type = "city")),
+                 nrow(r))
+  })
+})
+
+test_that("an unknown state abbreviation aborts with a message that names the problem", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    # Regression: `.state_abbrev_to_fips` is a named character vector, so `[[`
+    # on an absent name threw base R's "subscript out of bounds" and the
+    # curated message was unreachable.
+    expect_error(cog_spending(years = 2019, state = "ZZ"),
+                 class = "uscogdata_unknown_state")
+    expect_error(cog_spending(years = 2019, state = "ZZ"),
+                 "Unknown state abbreviation")
+  })
+})
+
+# --- naming the cohort -----------------------------------------------------
+
+test_that("naming no cohort at all is refused", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    expect_error(cog_spending(years = 2019), class = "uscogdata_no_cohort")
+    expect_error(cog_revenue(years = 2019), class = "uscogdata_no_cohort")
+    expect_error(cog_balances(years = 2019), class = "uscogdata_no_cohort")
+  })
+})
+
+test_that("an empty govid vector still fails as it always did", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    # Supplied-but-empty is a caller error, not "cohort named some other way".
+    expect_error(cog_spending(character(0), 2019),
+                 "must be a non-empty character vector")
+  })
+})
+
+test_that("govid and state/type together intersect", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    cities   <- cog_gov_search(NULL, state = "RI", type = "city")$canonical_govid
+    counties <- cog_gov_search(NULL, state = "DE", type = "county")$canonical_govid
+
+    # The documented rule: the governments in `govid` that ALSO match the
+    # predicate -- never one silently taking precedence over the other.
+    both <- cog_spending(govid = c(cities, counties), years = 2019,
+                         state = "RI", type = "city")
+    only <- cog_spending(govid = cities, years = 2019)
+    expect_equal(as.data.frame(both), as.data.frame(only), ignore_attr = TRUE)
+
+    # A disjoint intersection is empty, not "whichever one won".
+    none <- cog_spending(govid = counties, years = 2019,
+                         state = "RI", type = "city")
+    expect_equal(nrow(none), 0L)
+  })
+})
+
+# --- provenance ------------------------------------------------------------
+
+test_that("a govid cohort's provenance scope is untouched", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "DE", type = "county")$canonical_govid
+    prov <- attr(cog_spending(govid = ids, years = 2019), "provenance")
+    expect_setequal(prov$scope$govids_found, ids)
+    expect_length(prov$scope$govids_missing, 0L)
+    # No cohort block: govids_found already describes this cohort exactly.
+    expect_null(prov$scope$cohort)
+  })
+})
+
+test_that("a predicate cohort describes itself instead of listing ids", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    prov <- attr(cog_spending(years = 2019, state = "DE", type = "county"),
+                 "provenance")
+    # Deliberately NOT the resolved id list: a fleet-scale cohort would put
+    # 20,000 govids into every response body.
+    expect_length(prov$scope$govids_found, 0L)
+    expect_length(prov$scope$govids_missing, 0L)
+    expect_identical(prov$scope$cohort$state, "DE")
+    expect_identical(prov$scope$cohort$type, "county")
+    expect_identical(prov$scope$cohort$n_governments, 3L)
+  })
+})
+
+test_that("the cohort block counts the intersection, not the predicate alone", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    counties <- cog_gov_search(NULL, state = "DE", type = "county")$canonical_govid
+    prov <- attr(
+      cog_spending(govid = counties[1], years = 2019, state = "DE", type = "county"),
+      "provenance"
+    )
+    expect_identical(prov$scope$cohort$n_governments, 1L)
+  })
+})
+
+# --- the SQL actually changed ----------------------------------------------
+
+test_that("a predicate cohort never renders the ids into the query", {
+  skip_if_no_corpus()
+  with_fixture_corpus({
+    ids <- cog_gov_search(NULL, state = "RI", type = "city")$canonical_govid
+    prov <- attr(cog_spending(years = 2019, state = "RI", type = "city"),
+                 "provenance")
+    sql <- prov$sql_query %||% prov$sql
+    skip_if(is.null(sql), "provenance carries no SQL for this verb")
+    # The whole point: cohort size does not enter the SQL string.
+    for (id in ids) expect_false(grepl(id, sql, fixed = TRUE))
+    expect_match(sql, "SELECT canonical_govid FROM canonical_fips_xwalk",
+                 fixed = TRUE)
+  })
+})
@@ -0,0 +1,83 @@
+# A cohort is how every verb names the set of governments it queries. It can be
+# named by explicit id, by a predicate over canonical_fips_xwalk, or by both
+# (intersection). These tests cover the SQL construction itself -- pure string
+# building, no corpus needed -- because that is where the postal/FIPS trap and
+# the 40k-literal blowup both live.
+
+test_that(".make_cohort() keeps an explicit id vector as ids", {
+  ch <- .make_cohort(govid = c("550000227544", "060000000001"))
+  expect_identical(ch$ids, c("550000227544", "060000000001"))
+  expect_null(ch$state_fips)
+  expect_null(ch$type_int)
+  expect_false(.cohort_by_predicate(ch))
+})
+
+test_that(".make_cohort() translates a postal abbreviation to FIPS", {
+  # The trap this whole issue exists to avoid: canonical_fips_xwalk.fips_state
+  # holds "55", not "WI". A predicate written against the raw parameter matches
+  # nothing and returns an empty result that reads as "reported nothing".
+  ch <- .make_cohort(state = "WI")
+  expect_identical(ch$state_fips, "55")
+  expect_true(.cohort_by_predicate(ch))
+})
+
+test_that(".make_cohort() translates a type label to its integer code", {
+  expect_identical(.make_cohort(type = "city")$type_int, 2L)
+  expect_identical(.make_cohort(type = "state")$type_int, 0L)
+  expect_identical(.make_cohort(type = 1)$type_int, 1L)
+})
+
+test_that(".make_cohort() reuses the search verb's coercers for invalid input", {
+  expect_error(.make_cohort(state = "ZZ"), "Unknown state abbreviation")
+  expect_error(.make_cohort(type = "special_district"), "Unknown type")
+  # Out-of-scope types (4 = special district, 5 = school district) are refused
+  # by the numeric branch, with the v0.1-scope message.
+  expect_error(.make_cohort(type = 4), "type must be 0, 1, 2, or 3")
+})
+
+test_that(".make_cohort() rejects naming no cohort at all", {
+  expect_error(.make_cohort(), class = "uscogdata_no_cohort")
+})
+
+test_that(".cohort_sql() renders an id cohort as a literal IN list", {
+  sql <- .cohort_sql(.make_cohort(govid = c("a", "b")))
+  expect_identical(sql, "canonical_govid IN ('a','b')")
+})
+
+test_that(".cohort_sql() renders a predicate cohort as an xwalk subquery", {
+  # The point of the issue: the cohort never becomes a literal list, so its
+  # size does not enter the SQL string at all.
+  sql <- .cohort_sql(.make_cohort(state = "WI", type = "city"))
+  expect_match(sql, "SELECT canonical_govid FROM canonical_fips_xwalk", fixed = TRUE)
+  expect_match(sql, "fips_state = '55'", fixed = TRUE)
+  expect_match(sql, "govs_type = 2", fixed = TRUE)
+  expect_false(grepl("'WI'", sql, fixed = TRUE))
+})
+
+test_that(".cohort_sql() renders ids and a predicate as an intersection", {
+  sql <- .cohort_sql(.make_cohort(govid = c("a", "b"), type = "city"))
+  expect_match(sql, "canonical_govid IN ('a','b')", fixed = TRUE)
+  expect_match(sql, "AND canonical_govid IN (SELECT", fixed = TRUE)
+})
+
+test_that(".cohort_sql() honours a column alias", {
+  # Several call sites join the xwalk under an alias (`l.`, `x.`, `v.`), so the
+  # predicate has to be able to name the qualified column.
+  sql <- .cohort_sql(.make_cohort(state = "WI"), col = "l.canonical_govid")
+  expect_match(sql, "l.canonical_govid IN (SELECT", fixed = TRUE)
+  # The subquery's own column stays unqualified -- it selects from the xwalk,
+  # not from the outer relation.
+  expect_match(sql, "SELECT canonical_govid FROM", fixed = TRUE)
+})
+
+test_that(".cohort_sql() escapes quotes in ids", {
+  sql <- .cohort_sql(.make_cohort(govid = "o'brien"))
+  expect_match(sql, "'o''brien'", fixed = TRUE)
+})
+
+test_that("a predicate cohort's SQL does not grow with cohort size", {
+  # The regression this guards: 20,106 ids rendered to a 301,591-character
+  # IN list, embedded in 5-8 statements per call.
+  wide <- .cohort_sql(.make_cohort(state = "CA", type = "city"))
+  expect_lt(nchar(wide), 200L)
+})
@@ -0,0 +1,134 @@
+# tests/testthat/test-duckdb-limits.R
+#
+# uscogdata#60. cog_open() used to connect with a bare dbConnect() and set no
+# resource pragmas, so DuckDB claimed every visible core. That is right for one
+# interactive session on a dedicated machine and wrong for a server: cog-api
+# runs two replicas on an 8-core host budgeted 4, and without a cap each
+# replica independently claims all 8 and they fight.
+#
+# The consumer-side workaround this replaces reached into the namespace at
+# boot -- getFromNamespace(".ensure_session", "uscogdata")() followed by a
+# manual SET threads -- which depends on a private name AND on the session
+# already being open.
+#
+# The load-bearing property is the NEGATIVE one: unset must emit no pragma at
+# all, so an unconfigured session is byte-identical to pre-#60 behaviour.
+
+# Open a session under a given configuration and read a DuckDB setting back.
+# Each call closes first, because both settings are session-scoped: an
+# already-open connection would be reused by .ensure_session() and report the
+# PREVIOUS test's value, which is exactly the false pass to avoid here.
+setting_under <- function(setting, envvars = character(0), opts = list()) {
+  uscogdata:::cog_close()
+  on.exit(uscogdata:::cog_close(), add = TRUE)
+  withr::with_envvar(envvars, {
+    withr::with_options(opts, {
+      con <- uscogdata:::cog_open()
+      DBI::dbGetQuery(
+        con, sprintf("SELECT current_setting('%s') AS v", setting)
+      )$v[[1]]
+    })
+  })
+}
+
+test_that("USCOGDATA_DUCKDB_THREADS caps the connection's thread count", {
+  skip_if_no_corpus()
+  expect_equal(
+    as.integer(setting_under("threads", c(USCOGDATA_DUCKDB_THREADS = "2"))),
+    2L
+  )
+})
+
+test_that("the option spelling works, and the env var beats it", {
+  skip_if_no_corpus()
+  expect_equal(
+    as.integer(setting_under("threads",
+                             c(USCOGDATA_DUCKDB_THREADS = NA),
+                             list(uscogdata.duckdb_threads = 3L))),
+    3L
+  )
+  # Same precedence .cfg() gives every other setting: env var > option.
+  expect_equal(
+    as.integer(setting_under("threads",
+                             c(USCOGDATA_DUCKDB_THREADS = "1"),
+                             list(uscogdata.duckdb_threads = 3L))),
+    1L
+  )
+})
+
+test_that("unset leaves DuckDB's own default in place", {
+  skip_if_no_corpus()
+  # Not asserting a specific number -- the default is core-count-dependent and
+  # a literal would fail on a different machine. The claim is that NO pragma
+  # was issued, so the session sees whatever DuckDB would have chosen on its
+  # own. Compared against a plain connection opened the pre-#60 way.
+  unset <- setting_under("threads",
+                         c(USCOGDATA_DUCKDB_THREADS = NA),
+                         list(uscogdata.duckdb_threads = NULL))
+  bare <- local({
+    con <- DBI::dbConnect(duckdb::duckdb())
+    on.exit(DBI::dbDisconnect(con, shutdown = TRUE), add = TRUE)
+    DBI::dbGetQuery(con, "SELECT current_setting('threads') AS v")$v[[1]]
+  })
+  expect_equal(as.integer(unset), as.integer(bare))
+})
+
+test_that("USCOGDATA_DUCKDB_MEMORY_LIMIT is applied", {
+  skip_if_no_corpus()
+  v <- setting_under("memory_limit", c(USCOGDATA_DUCKDB_MEMORY_LIMIT = "2GB"))
+  # DuckDB does not echo back the string it was given: it stores bytes and
+  # reports BINARY units, so "2GB" (2e9 bytes) comes back as "1.8 GiB". Assert
+  # the magnitude it actually means rather than the spelling this package sent
+  # -- matching on "2" passes for the wrong reason and fails on the right one.
+  expect_match(as.character(v), "GiB", fixed = TRUE)
+  # DuckDB also truncates the display to one decimal ("1.8 GiB" for 1.863), so
+  # the tolerance covers rounding, not slack in the setting itself.
+  gib <- as.numeric(sub("\\s*GiB$", "", as.character(v)))
+  expect_equal(gib, 2e9 / 1024^3, tolerance = 0.05)
+})
+
+# --- Validation -------------------------------------------------------------
+# .cfg() returns an env var as CHARACTER. Without coercion here,
+# sprintf("SET threads TO %d", "4") aborts inside the connection path with an
+# error about the pragma rather than about the setting the operator got wrong.
+
+test_that(".resolve_duckdb_threads coerces a character env var to integer", {
+  withr::local_envvar(USCOGDATA_DUCKDB_THREADS = "4")
+  expect_identical(uscogdata:::.resolve_duckdb_threads(), 4L)
+})
+
+test_that(".resolve_duckdb_threads returns NULL when unset or empty", {
+  withr::local_options(uscogdata.duckdb_threads = NULL)
+  withr::local_envvar(USCOGDATA_DUCKDB_THREADS = NA)
+  expect_null(uscogdata:::.resolve_duckdb_threads())
+
+  withr::local_envvar(USCOGDATA_DUCKDB_THREADS = "")
+  expect_null(uscogdata:::.resolve_duckdb_threads())
+})
+
+test_that(".resolve_duckdb_threads rejects values that are not positive integers", {
+  for (bad in c("0", "-1", "two", "1.5.2")) {
+    withr::local_envvar(USCOGDATA_DUCKDB_THREADS = bad)
+    expect_error(uscogdata:::.resolve_duckdb_threads(),
+                 class = "uscogdata_invalid_duckdb_threads")
+  }
+})
+
+test_that(".resolve_duckdb_memory_limit accepts size strings and rejects junk", {
+  withr::local_envvar(USCOGDATA_DUCKDB_MEMORY_LIMIT = "4GB")
+  expect_identical(uscogdata:::.resolve_duckdb_memory_limit(), "4GB")
+
+  withr::local_envvar(USCOGDATA_DUCKDB_MEMORY_LIMIT = "1.5GB")
+  expect_identical(uscogdata:::.resolve_duckdb_memory_limit(), "1.5GB")
+
+  # A SQL fragment must not reach the connection as one.
+  withr::local_envvar(USCOGDATA_DUCKDB_MEMORY_LIMIT = "4GB'; DROP TABLE x; --")
+  expect_error(uscogdata:::.resolve_duckdb_memory_limit(),
+               class = "uscogdata_invalid_duckdb_memory_limit")
+})
+
+test_that(".apply_duckdb_limits issues no statement when both are NULL", {
+  # The negative property, asserted directly rather than inferred: a connection
+  # that would ERROR on any statement proves none was sent.
+  expect_silent(uscogdata:::.apply_duckdb_limits(NULL, NULL, NULL))
+})
@@ -62,3 +62,36 @@ test_that("registered `long` view reads through the enumerated list", {
    expect_true(all(c(2011, 2012, 2019, 2020) %in% yrs))
  })
 })
+
+test_that("a Windows-style corpus path survives token substitution", {
+  # gsub() in regex mode treats backslashes in the REPLACEMENT as escape
+  # sequences and silently drops them, so a Windows path went in as
+  # C:\Users\RUNNER\... and came out as C:UsersRUNNER..., after which every
+  # DuckDB read failed with "No files found that match the pattern".
+  #
+  # That made a LOCAL corpus unreadable on Windows -- the bundled fixture
+  # included, so the whole suite failed there -- while remote https URLs
+  # worked fine, having no backslashes. It went unnoticed for the life of the
+  # package because nothing ever ran on Windows.
+  #
+  # Reproducible on any platform: this is string handling, not a filesystem
+  # behaviour, so it does not need a Windows runner to catch.
+  win <- "C:\\Users\\RUNNER~1\\AppData\\Local\\Temp\\Rtmp123/"
+
+  out <- uscogdata:::.render_view_sql(
+    "FROM read_parquet('{url}data/summary_categories.parquet')", win
+  )
+  expect_true(grepl("C:\\Users\\RUNNER~1\\AppData", out, fixed = TRUE))
+  expect_false(grepl("C:Users", out, fixed = TRUE))
+
+  # The same must hold through the {long_files} path, which embeds the url
+  # once per enumerated partition.
+  manifest <- list(files = list(long_partitions = list(
+    list(year = 2011L, path = "data/long/year=2011/part-0.parquet")
+  )))
+  out2 <- uscogdata:::.render_view_sql(
+    "FROM read_parquet({long_files}, hive_partitioning = true)", win, manifest
+  )
+  expect_true(grepl("C:\\Users\\RUNNER~1\\AppData", out2, fixed = TRUE))
+  expect_false(grepl("C:Users", out2, fixed = TRUE))
+})
@@ -251,7 +251,7 @@ test_that(".suppressed_components measures the E67/E68 dollars Public Welfare dr
  s <- uscogdata:::.suppressed_components(
    con,
    candidates = c("welfare_cash_e67_wide", "welfare_cash_e68_wide"),
-    govid = "061037123085", years = 2011L,
+    cohort = uscogdata:::.make_cohort("061037123085"), years = 2011L,
    long_view = "spending_long_harmonized",
    flow_prefixes = c("E", "F", "G"))

@@ -269,7 +269,7 @@ test_that(".suppressed_components finds nothing in a modern year", {
  s <- uscogdata:::.suppressed_components(
    con,
    candidates = c("welfare_cash_e67_wide", "welfare_cash_e68_wide"),
-    govid = "061037123085", years = 2019L,
+    cohort = uscogdata:::.make_cohort("061037123085"), years = 2019L,
    long_view = "spending_long_harmonized",
    flow_prefixes = c("E", "F", "G"))
  expect_equal(nrow(s), 0L)
@@ -280,7 +280,7 @@ test_that(".suppressed_components rejects a long_view outside the allowlist", {
  con <- uscogdata:::.ensure_session()
  expect_error(
    uscogdata:::.suppressed_components(
-      con, candidates = "welfare_cash_e67_wide", govid = "061037123085",
+      con, candidates = "welfare_cash_e67_wide", cohort = uscogdata:::.make_cohort("061037123085"),
      years = 2011L, long_view = "long; DROP TABLE x",
      flow_prefixes = c("E", "F", "G")),
    class = "uscogdata_internal_error")
@@ -298,7 +298,7 @@ test_that(".suppressed_components never measures a component from the other flow
  s <- uscogdata:::.suppressed_components(
    con,
    candidates = c("welfare_cash_e67_wide", "welfare_cash_e68_wide"),
-    govid = "061037123085", years = 2011L,
+    cohort = uscogdata:::.make_cohort("061037123085"), years = 2011L,
    long_view = "revenue_long_harmonized",
    flow_prefixes = c("T", "A", "U", "B", "C", "D"))
  expect_equal(nrow(s), 0L)
@@ -0,0 +1,178 @@
+# tests/testthat/test-search-balances-pagination.R
+#
+# uscogdata#57. cog_spending()/cog_revenue() gained limit/offset in #39;
+# cog_gov_search() and cog_balances() did not, so every consumer of those two
+# was back to materialize-then-slice -- the exact pattern that wedged the
+# production API for hours on 2026-08-06.
+#
+# cog_gov_search() was also the one verb with no LIMIT at all, so an
+# unfiltered call returns the entire 40,336-row crosswalk by accident.
+
+# --- cog_gov_search() -------------------------------------------------------
+
+test_that("cog_gov_search() limit returns the first page of the unpaginated result", {
+  skip_if_no_corpus()
+  full <- cog_gov_search(state = "WI", type = "city")
+  skip_if(nrow(full) < 12L, "fixture has too few WI cities to page")
+
+  page <- cog_gov_search(state = "WI", type = "city", limit = 5L)
+  expect_equal(nrow(page), 5L)
+  expect_equal(page$canonical_govid, full$canonical_govid[1:5])
+})
+
+test_that("cog_gov_search() offset skips ahead without gaps or overlap", {
+  skip_if_no_corpus()
+  full <- cog_gov_search(state = "WI", type = "city")
+  skip_if(nrow(full) < 12L, "fixture has too few WI cities to page")
+
+  p1 <- cog_gov_search(state = "WI", type = "city", limit = 5L)
+  p2 <- cog_gov_search(state = "WI", type = "city", limit = 5L, offset = 5L)
+  expect_equal(p2$canonical_govid, full$canonical_govid[6:10])
+  expect_length(intersect(p1$canonical_govid, p2$canonical_govid), 0L)
+})
+
+test_that("walking every page reconstructs the unpaginated search exactly", {
+  skip_if_no_corpus()
+  full <- cog_gov_search(state = "WI", type = "city")
+  n <- nrow(full)
+  limit <- 7L
+  pages <- list()
+  offset <- 0L
+  repeat {
+    p <- cog_gov_search(state = "WI", type = "city", limit = limit, offset = offset)
+    if (nrow(p) == 0L) break
+    pages[[length(pages) + 1L]] <- p
+    offset <- offset + limit
+    if (offset > n + limit) stop("test runaway: paging did not terminate")
+  }
+  walked <- dplyr::bind_rows(pages)
+  expect_equal(nrow(walked), n)
+  expect_equal(walked$canonical_govid, full$canonical_govid)
+})
+
+test_that("cog_gov_search() total_rows reports the full unpaginated count", {
+  skip_if_no_corpus()
+  full <- cog_gov_search(state = "WI", type = "city")
+  page <- cog_gov_search(state = "WI", type = "city", limit = 3L)
+  expect_equal(attr(page, "total_rows"), nrow(full))
+})
+
+test_that("cog_gov_search() offset past the end reports the true total, not zero", {
+  skip_if_no_corpus()
+  full <- cog_gov_search(state = "WI", type = "city")
+  # No row survives to carry COUNT(*) OVER(), so this is the branch that has
+  # to fall back to a second count rather than reporting 0 rows out of 0.
+  page <- cog_gov_search(state = "WI", type = "city",
+                         limit = 5L, offset = nrow(full) + 50L)
+  expect_equal(nrow(page), 0L)
+  expect_equal(attr(page, "total_rows"), nrow(full))
+})
+
+test_that("cog_gov_search() bounds an otherwise-unfiltered crosswalk sweep", {
+  skip_if_no_corpus()
+  # The reason this verb needed a limit most: with no filter it returns the
+  # whole crosswalk.
+  page <- cog_gov_search(limit = 10L)
+  expect_equal(nrow(page), 10L)
+  expect_gt(attr(page, "total_rows"), 10L)
+})
+
+test_that("cog_gov_search() orders by a total order, not population alone", {
+  skip_if_no_corpus()
+  # population_acs is not unique -- NA in particular repeats across many rows
+  # -- so paging on it alone can duplicate a row on one page and drop it from
+  # the next. The tiebreaker is what makes the sequence reproducible.
+  full <- cog_gov_search(state = "WI")
+  skip_if(nrow(full) < 5L, "fixture has too few WI governments")
+  expect_equal(cog_gov_search(state = "WI")$canonical_govid,
+               full$canonical_govid)
+
+  ties <- full[is.na(full$population_acs), ]
+  skip_if(nrow(ties) < 2L, "no tied rows in the fixture to order")
+  expect_false(is.unsorted(ties$canonical_govid))
+})
+
+test_that("cog_gov_search() refuses pagination in basket mode", {
+  skip_if_no_corpus()
+  expect_error(
+    cog_gov_search(name = c("MADISON CITY", "MILWAUKEE CITY"),
+                   state = c("WI", "WI"), limit = 1L),
+    class = "uscogdata_basket_pagination_conflict"
+  )
+})
+
+test_that("cog_gov_search() rejects a malformed limit or offset", {
+  skip_if_no_corpus()
+  expect_error(cog_gov_search(state = "WI", limit = -1L),
+               class = "uscogdata_invalid_pagination")
+  expect_error(cog_gov_search(state = "WI", limit = 5L, offset = -1L),
+               class = "uscogdata_invalid_pagination")
+})
+
+# --- cog_balances() ---------------------------------------------------------
+
+test_that("cog_balances() limit/offset walk the unpaginated result exactly", {
+  skip_if_no_corpus()
+  full <- cog_balances(years = 2019:2020, state = "WI", type = "city")
+  skip_if(nrow(full) < 6L, "fixture has too few WI city balance rows to page")
+
+  key <- c("year", "canonical_govid", "balance_subtype", "amt_nominal")
+  p1 <- cog_balances(years = 2019:2020, state = "WI", type = "city", limit = 3L)
+  p2 <- cog_balances(years = 2019:2020, state = "WI", type = "city",
+                     limit = 3L, offset = 3L)
+
+  expect_equal(nrow(p1), 3L)
+  expect_equal(p1[key], full[1:3, key], ignore_attr = TRUE)
+  expect_equal(p2[key], full[4:6, key], ignore_attr = TRUE)
+  # The window-function column is an implementation detail and must not reach
+  # the caller's data frame.
+  expect_false("pagination_total_rows" %in% names(p1))
+})
+
+test_that("cog_balances() total_rows reports the full unpaginated count", {
+  skip_if_no_corpus()
+  full <- cog_balances(years = 2019:2020, state = "WI", type = "city")
+  page <- cog_balances(years = 2019:2020, state = "WI", type = "city", limit = 2L)
+  expect_equal(attr(page, "total_rows"), nrow(full))
+})
+
+test_that("cog_balances() offset past the end reports the true total", {
+  skip_if_no_corpus()
+  full <- cog_balances(years = 2019:2020, state = "WI", type = "city")
+  page <- cog_balances(years = 2019:2020, state = "WI", type = "city",
+                       limit = 5L, offset = nrow(full) + 50L)
+  expect_equal(nrow(page), 0L)
+  expect_equal(attr(page, "total_rows"), nrow(full))
+})
+
+test_that("cog_balances() refuses pagination alongside a recipe", {
+  skip_if_no_corpus()
+  expect_error(
+    cog_balances(years = 2011, state = "WI", type = "city",
+                 recipe = "cash_securities_z77_wide", limit = 5L),
+    class = "uscogdata_recipe_pagination_conflict"
+  )
+})
+
+test_that("cog_balances() rejects a malformed limit or offset", {
+  skip_if_no_corpus()
+  expect_error(cog_balances(years = 2019, state = "WI", type = "city", limit = -1L),
+               class = "uscogdata_invalid_pagination")
+  expect_error(cog_balances(years = 2019, state = "WI", type = "city",
+                            limit = 5L, offset = -1L),
+               class = "uscogdata_invalid_pagination")
+})
+
+# --- Unchanged without the arguments ----------------------------------------
+
+test_that("both verbs are unchanged when limit is not supplied", {
+  skip_if_no_corpus()
+  # The adoption contract for cog-api: NULL default, so a formals() probe can
+  # feature-detect without any call site changing behaviour.
+  s <- cog_gov_search(state = "WI", type = "city")
+  b <- cog_balances(years = 2019, state = "WI", type = "city")
+  expect_null(attr(s, "total_rows"))
+  expect_null(attr(b, "total_rows"))
+  expect_true(all(c("limit", "offset") %in% names(formals(cog_gov_search))))
+  expect_true(all(c("limit", "offset") %in% names(formals(cog_balances))))
+})
Author	SHA1	Message	Date
jared	2bb9d41d72	feat: limit/offset on cog_gov_search() and cog_balances() (#57 ) R-CMD-check / check (pull_request) Successful in 4m22s Details R-CMD-check / check (push) Successful in 4m28s Details verbs were left materializing everything and slicing in R -- the pattern behind the 2026-08-06 production incident. cog_gov_search() had no LIMIT at all, so an unfiltered call returns the entire 40,336-row crosswalk. Extracted the #39 machinery into R/pagination.R first (.validate_pagination(), .paginate_sql(), .take_pagination_total()) rather than growing a third inline copy: three definitions of what total_rows means is three places for it to drift. Conflict refusals stay at the call sites because each verb's conflict set differs. .verb_spendrev() now uses the shared helpers and is unchanged in behaviour. The empty-page fallback query is now passed as a thunk, so the unpaginated SQL is only BUILT when an offset actually lands past the end instead of on every paged call. Two things #57 did not anticipate: - cog_gov_search()'s ORDER BY was not a total order. population_acs DESC NULLS LAST leaves ties -- and the whole NULL block -- in scan order, so two requests can order them differently and a paged sweep duplicates one row while dropping another. Added canonical_govid as tiebreaker. Unpaginated output changes only in the relative order of already-tied rows. - Basket mode returns one resolved row per requested name plus a sidecar covering all of them, so a page of it is not a page of anything the caller asked for. Refused with uscogdata_basket_pagination_conflict rather than silently ignoring the arguments. Both default to NULL, so cog-api adopts them behind its existing formals() probe with no lockstep deploy. Suite: 1067 passed, 0 failed, 0 warnings (2 pre-existing live-corpus skips).	2026-08-10 19:27:11 -04:00
jared	700ae93c9c	Merge pull request 'feat: cog_open() honours a DuckDB thread and memory budget (#60 )' (#62 ) from feat/duckdb-threads-60 into main Mirror to GitHub / mirror (push) Successful in 10s Details R-CMD-check / check (push) Successful in 4m6s Details Reviewed-on: #62	2026-08-10 19:23:14 -04:00
jared	f4ab9b6d90	feat: cog_open() honours a DuckDB thread and memory budget (#60 ) R-CMD-check / check (push) Successful in 4m11s Details R-CMD-check / check (pull_request) Successful in 4m11s Details cog_open() connected with a bare dbConnect() and set no resource pragmas, so DuckDB claimed every visible core. Right for one interactive session on a dedicated machine; wrong for a server, where cog-api runs two replicas on an 8-core host budgeted 4 and each replica independently claims all 8. USCOGDATA_DUCKDB_THREADS and USCOGDATA_DUCKDB_MEMORY_LIMIT now resolve through .cfg() -- inheriting the env var > option > default precedence USCOGDATA_URL already had -- and are applied as pragmas when the connection is created. Unset issues NO pragma, so an unconfigured session is byte-identical to before. That negative property is asserted directly against a connection opened the pre-change way rather than against a hardcoded core count. .cfg() returns an env var as character, so both resolvers coerce and validate rather than trusting the type: sprintf("SET threads TO %d", "4") would otherwise abort inside the connection path with an error naming the pragma instead of the setting the operator got wrong. Replaces cog-api's getFromNamespace(".ensure_session", "uscogdata") workaround, which depended on a private name and on the session already being open.	2026-08-10 18:51:56 -04:00
jared	9508b98676	Merge pull request 'feat: cohort predicates instead of a 40k-id IN list (#58 )' (#61 ) from feat/cohort-predicates-58 into main Mirror to GitHub / mirror (push) Successful in 9s Details R-CMD-check / check (push) Successful in 3m18s Details Reviewed-on: #61	2026-08-09 14:31:53 -04:00
jared	0fbae00e27	feat: name a cohort by state/type predicate instead of a 40k-id IN list R-CMD-check / check (push) Successful in 4m29s Details R-CMD-check / check (pull_request) Successful in 4m19s Details cog_spending(), cog_revenue() and cog_balances() gain optional state/type arguments. Both default to NULL, so every existing govid-based call is unchanged. The verbs took a cohort only as a govid vector, which .sql_lit_chr() rendered into a quoted IN list and .verb_spendrev() embedded into 5-8 separate statements per call: the scope check, the main aggregate, the per-capita join, the harmonization block, and the suggestion and suppression queries. For type = "city" that list is 301,589 characters, parsed and planned from scratch every time it appears. Passing state/type instead expresses the cohort as a subquery against canonical_fips_xwalk, so its size never enters the SQL string at all. Measured on the production corpus, same FY2022 aggregate over the 20,106-government city cohort, DUCKDB_THREADS=2, median of 5: IN (20,106 literals) -- 0.3.0 432 ms join against a temp cohort table 132 ms predicate on canonical_fips_xwalk 102 ms no cohort filter at all (the floor) 105 ms The predicate reaches the no-filter floor: the cohort restriction is now free. End to end through cog_spending(category = "Police"), 1080 ms -> 271 ms, 3.99x -- larger than the single-query saving, because the repetition across statements is what actually cost. Design decisions, both made explicitly rather than left implicit: - govid AND state/type INTERSECT. "These ids, narrowed to that state/type" is a real query, and an error here could never be relaxed later without breaking callers. - A predicate cohort has no id list to report, so provenance$scope$govids_found/govids_missing stay empty and a new scope$cohort block carries state, type and n_governments. Resolving the ids just to report them would put 20,000 govids in every fleet-scale response body -- the cost this change removes. A govid-named cohort's provenance is untouched. state/type are coerced with .coerce_state_to_fips()/.coerce_type(), the same helpers cog_gov_search() uses. That is load-bearing: the argument is a postal abbreviation ("WI") while fips_state holds a FIPS code ("55"), and a predicate on the raw parameter matches nothing and returns an empty result indistinguishable from "reported nothing". cog-api hit exactly this trap optimizing the same path. .attach_per_capita() now keys its population lookup on the govids present in the result rather than the requested cohort. Those are the only ones its LEFT JOIN can match, so the output is identical -- but it needs no id list, and on a paginated call it looks up one page instead of the fleet. Fixes uscogdata#58.	2026-08-09 14:15:21 -04:00
jared	698812a25c	Merge pull request 'fix: local corpus paths were unreadable on Windows (backslashes eaten)' (#55 ) from fix/windows-backslash-paths into main Mirror to GitHub / mirror (push) Failing after 8s Details R-CMD-check / check (push) Successful in 3m34s Details	2026-08-09 09:08:41 -04:00
jared	d74ecdd4a5	Merge pull request 'ci: mirror main and tags to the GitHub mirror' (#54 ) from ci/mirror-to-github into main R-CMD-check / check (push) Successful in 3m29s Details Mirror to GitHub / mirror (push) Failing after 11s Details	2026-08-09 09:04:28 -04:00
jared	72b2cc3a26	fix: local corpus paths were unreadable on Windows (backslashes eaten) R-CMD-check / check (push) Successful in 3m35s Details R-CMD-check / check (pull_request) Successful in 3m49s Details gsub() in regex mode treats backslashes in the REPLACEMENT string as escape sequences and silently drops them. A Windows corpus path is full of them, so C:\Users\RUNNER\AppData\... was substituted into the view SQL as C:UsersRUNNERAppData... and every DuckDB read failed with 'No files found that match the pattern'. Effect: uscogdata could not read a LOCAL corpus on Windows at all -- the bundled fixture included, so the entire test suite failed there, and any cog_mirror() copy was unusable. Remote https URLs were unaffected, having no backslashes, which is part of why it stayed hidden. The bug predates the {long_files} token; it lived in the original {url} substitution since that code was written. Nothing ever ran on Windows until the GitHub mirror's check matrix existed, which found it on its first run: 14 of 15 Windows failures were this, the 15th a downstream consequence of view registration failing. fixed = TRUE treats pattern and replacement as literal text. The regression test reproduces on any platform -- it is string handling, not filesystem behaviour, so it needs no Windows runner.	2026-08-09 09:03:44 -04:00