Merge pull request 'feat(coverage): n_units_collected separates sampling from real zeros (#36)' (#71) from fix/coverage-collected-36 into main

provenance$coverage's n_units_reporting is category-conditional: it
counts governments with rows for the SPECIFIC requested category, which
conflates two different things -- a government never collected that
year (sampling), and one collected but genuinely spending nothing in
that category (a real zero). FY2012 Georgia Police is the motivating
case from the issue: a complete census year reads as a 69% "response
rate" because most of the gap is cities that contract policing to the
county sheriff, not non-response.

Adds a second counter, n_units_collected: how many of the caller's
expected cohort appear in the corpus that year for ANY category.
n_units_collected / n_units_expected is the true collection rate;
n_units_reporting / n_units_collected is category participation among
collected units. cog_geographic_rollup() and cog_peer_compare() both
carry it; cog_explain() prints it alongside n_units_reporting.

Two real bugs caught and fixed while finishing this (both against the
already-written, previously-uncommitted draft):

- .coverage_table()'s candidate list for the collection query was
  derived from the category-filtered result rows, not the caller's
  full expected cohort. A government with zero rows in the requested
  category across every requested year never appears in that result,
  so it was silently excluded from n_units_collected too -- collapsing
  the new counter back to the old, broken one for exactly the
  governments it exists to count. Fixed by threading an explicit
  `expected_ids` (all_govids / peer_govids) through instead.
- The collection query hardcoded long_view = "spending_long_harmonized",
  which does not exist on a corpus with schema_version < 5 (R/basis.R
  resolves basis = "raw" there; R/views.R only registers the
  harmonized views on v5+). cog_geographic_rollup()/cog_peer_compare()
  would hard-error on a corpus vintage the package otherwise explicitly
  supports. Fixed by deriving long_view from the basis cog_spending()
  actually resolved (prov$basis) via the existing .select_long_view()
  helper, matching how every other basis-aware query in the package
  already does this.

Also: cog_explain()'s general "complete census only in years ending in
2 or 7" footnote was gated on the OLD counter's absence, making it
permanently unreachable now that both callers always supply the new
one -- ungated it, since the explanation is orthogonal to which
counter set is present. Dropped a dead conditional branch, fixed two
stale roxygen blocks in R/peers.R/R/rollup.R still describing the old
two-counter model, fixed the same staleness in README.md, and switched
two `uscogdata:::` self-references to the package's own convention of
calling internal helpers unqualified.

1101 tests pass (2 skipped live-corpus), including new direct
regression tests for both bugs above (one exercising a government
collected-but-absent from a category result, one running the full
rollup/peer-compare path against a doctored schema_version 4 corpus).

Reviewed by an independent code-reviewer pass (1 HIGH, 1 MEDIUM, 3 LOW
-- all addressed above).

Closes #36.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.Rbuildignore

@@ -3,6 +3,7 @@
 ^\.Rproj\.user$
 ^_pkgdown\.yml$
 ^docs$
+^pm$
 ^Meta$
 ^doc$
 ^pkgdown$

.gitea/workflows/mirror-github.yml

@@ -39,5 +39,17 @@ jobs:
             echo "PAT_GH is unset -- add it under Settings > Actions > Secrets." >&2
             exit 1
           fi
+          # On a tag-triggered run, checkout materializes refs/tags/<tag> as a
+          # LIGHTWEIGHT tag at the commit SHA -- the annotated tag object Gitea
+          # holds is never fetched. Mirroring that strips the annotation, and the
+          # NEXT run on main (which does fetch the real object) is then rejected
+          # with "already exists" trying to correct it, because git will not
+          # clobber an existing tag. That is why v0.4.0 failed to mirror.
+          #
+          # Re-fetch canonical tag objects from Gitea first. --force here rewrites
+          # LOCAL tag refs only; it is not a force push and does not weaken the
+          # non-force guarantee on main documented above.
+          git fetch --tags --force origin
+
           git push "https://x-access-token:${PAT_GH}@github.com/civilytics/uscogdata.git" \
             HEAD:refs/heads/main --tags

.github/workflows/pr-welcome.yaml

@@ -0,0 +1,64 @@
+# Explain the mirror contribution flow on every incoming pull request.
+#
+# This repository is a MIRROR. A PR opened here is landed on the canonical Gitea
+# repository and syncs back; because the merge preserves the contributor's
+# commits at their original SHAs, GitHub marks the PR "Merged" on its own as
+# soon as the mirror syncs -- with nobody visibly clicking Merge.
+#
+# Without this comment, that reads as a rejection: the contributor sees their PR
+# close with no review, no merge button pressed, and no explanation. It is
+# actually the successful outcome. Say so up front, before it happens.
+#
+# WHY pull_request_target AND NOT pull_request:
+# a `pull_request` run from a fork gets a read-only token, so it cannot post a
+# comment -- which is exactly the case this workflow exists to serve.
+# `pull_request_target` runs in the context of the BASE repo and gets a writable
+# token. That is only safe because this job never checks out or executes the
+# contributor's code; it posts a fixed string. Do not add a checkout of
+# `github.event.pull_request.head.sha` here -- that combination is the standard
+# pull_request_target privilege-escalation hole.
+name: Explain the mirror flow
+
+on:
+  pull_request_target:
+    types: [opened]
+
+permissions:
+  pull-requests: write
+
+jobs:
+  comment:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Post the contribution-flow explainer
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const body = [
+              "Thanks for this — and one thing worth knowing before it happens.",
+              "",
+              "**This repository is a mirror.** Development happens on Gitea at",
+              "`gitea.civilytics.org/Civilytics/uscogdata`. Your pull request will be fetched",
+              "from here, landed there, and synced back.",
+              "",
+              "Because that merge preserves your commits at their original SHAs, **GitHub will",
+              "mark this pull request \"Merged\" on its own** — without anyone visibly clicking",
+              "the Merge button, and possibly without a review comment on this page first.",
+              "",
+              "> If your pull request closes as \"Merged\" and nobody appears to have merged it,",
+              "> that is the normal, successful outcome — not a rejection.",
+              "",
+              "If it is *not* going to be merged, you will get an actual reply saying so.",
+              "",
+              "Substantial contributions get a `ctb` entry in `DESCRIPTION`, which surfaces in",
+              "`citation(\"uscogdata\")`. There is no CLA and no DCO sign-off.",
+              "",
+              "Full details: [CONTRIBUTING.md](https://github.com/civilytics/uscogdata/blob/main/CONTRIBUTING.md).",
+            ].join("\n");
+
+            await github.rest.issues.createComment({
+              owner: context.repo.owner,
+              repo: context.repo.repo,
+              issue_number: context.payload.pull_request.number,
+              body,
+            });

.gitignore

@@ -4,6 +4,10 @@
 .Ruserdata
 *.Rproj
 inst/doc
+# pkgdown output. Compass used to keep its files in docs/pm/ and
+# docs/decisions/, which forced this to be written as children with two
+# re-includes -- git cannot re-include anything beneath an excluded directory.
+# Compass lives in pm/ now, so the whole directory can be excluded again.
 docs/
 /doc/
 /Meta/
@@ -12,3 +16,6 @@ docs/
 
 # SDD working artifacts (ledger, briefs, review packages) — plans/ stays tracked
 .superpowers/sdd/
+.compass-cache/
+# roborev snapshots
+/.roborev/

.roborev.toml

@@ -0,0 +1,60 @@
+# roborev configuration, initialised by compass.
+# Reviews are queued to a background daemon -- they never block a commit.
+
+post_commit_review = 'commit'
+excluded_commit_patterns = ['WIP', 'chore:', 'chore(', 'docs:', 'Merge ']
+
+review_guidelines = '''
+# --- compass:begin (generated -- edit the sources, not this) ---
+- Prefer returning new values to mutating arguments in place. A function that edits
+  its caller's object is a bug waiting for a second caller.
+- Validate at system boundaries -- user input, API responses, file contents, config.
+  Fail fast with a message naming the field and the file.
+- Never swallow an error. Handle it or let it propagate; a bare catch that continues
+  is worse than a crash.
+- No hardcoded secrets, tokens, or credentials, and no secrets in log output or error
+  messages.
+- Parameterise every query. String-built SQL is a defect even when the input looks safe.
+- Keep functions under roughly 50 lines and files under roughly 400. Flag nesting
+  deeper than four levels.
+- No magic numbers or hardcoded paths -- name them as constants or read them from config.
+- New behaviour needs a test. A bug fix needs a test that fails without the fix.
+- Prose a person reads -- an issue title or body, a journal entry, a decision record,
+  the narrative on the status board -- names the action or the thing, not the shape of
+  the machinery. Flag "gate", "seam", "surface area", "load-bearing", "first-class",
+  "primitive", "blast radius". A project's own defined vocabulary is not the target.
+- Use the native pipe `|>`, not magrittr `%>%`.
+- snake_case for objects and functions; UPPER_SNAKE for constants. Never use `.` as a
+  word separator in a function name -- it collides with S3 dispatch.
+- Validate arguments at the top of exported functions with `stopifnot()` or an explicit
+  check, and say which argument was wrong.
+- Never `setDT()`, `set()`, or otherwise modify by reference a data.table the caller
+  still owns. `as.data.table()` copies; use it.
+- Prefer `vapply()` to `sapply()` -- `sapply()` silently returns a list when the type
+  varies, which turns a type error into a downstream mystery.
+- Use `seq_len(n)` / `seq_along(x)`, never `1:n`, which iterates backwards when n is 0.
+- Compare strings with `==` only after checking for NA; use `identical()` for scalars
+  where NA would be wrong.
+- Do not call `library()` inside package or module files; attach packages in scripts and
+  test helpers only.
+- Namespace-qualify calls into other packages (`stats::sd`) in code that is sourced.
+- Every exported function needs roxygen with `@param` for each argument (type, meaning,
+  and why the default is what it is) and `@return`. Add `@examples` for exported API.
+- Declare dependencies in DESCRIPTION. Prefer base R or an existing dependency over
+  adding a new one; a package with zero hard deps is worth keeping that way.
+- Signal errors with `stop()` carrying a condition class, so callers can catch the kind
+  rather than matching on message text.
+- Keep internals internal. Export only what a user needs; an accidentally exported
+  helper becomes an API you have to keep.
+- Tests use testthat edition 3. Each test is self-sufficient -- no reliance on state
+  left by an earlier test or on a fixture built elsewhere in the file.
+- Prefer duplication in tests over a helper that hides what is being asserted.
+- Every verb calls .ensure_session() first, then queries via DBI::dbGetQuery().
+- A verb's return value is always a tbl_df carrying a provenance attribute.
+- govid inputs always go through .coerce_govid_input(); it accepts a character vector or a data frame.
+- SQL has two layers: view definitions are numbered .sql files in inst/sql/ registered by .register_views(); query construction is inline sprintf() in R. Add a view as a file; build a query in R.
+- No arrow dependency -- DuckDB reads parquet natively.
+- withr is Suggests-only and must appear in tests alone.
+- Tests must pass offline against the bundled fixture; tests/testthat/setup.R sets USCOGDATA_URL for that.
+# --- compass:end ---
+'''

CONTRIBUTING.md

@@ -101,5 +101,13 @@ but "usually" is not a release gate.
    variables set. This is the only check that catches a
    corpus-unreachable defect, and its absence is why 0.3.0 needed fixing.
 8. Bump `Version` and add a `NEWS.md` section.
-9. Tag, then update the r-universe registry pin at
-   `github.com/civilytics/civilytics.r-universe.dev`.
+9. Tag on **Gitea** (`git tag -a vX.Y.Z && git push origin vX.Y.Z`). The mirror
+   workflow carries tags to GitHub on its own — confirm the tag appears at
+   `github.com/civilytics/uscogdata/tags` before continuing.
+10. Update the r-universe registry pin at
+    `github.com/civilytics/civilytics.r-universe.dev` — edit `packages.json`'s
+    `branch` to the new tag. **r-universe will not pick up a release until this
+    is edited**: the pin is a tag, deliberately, so a mid-refactor `main` is
+    never published as a release. `"branch": "*release"` would track releases
+    automatically, but it needs a GitHub *Release* object and the mirror pushes
+    tags only — so it would silently never update.

DESCRIPTION

@@ -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",

NEWS.md

@@ -1,3 +1,132 @@
+# uscogdata 0.4.0
+
+## 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.
+
+## `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.
+
+## 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.
+
+## Documentation: the corpus-access table is re-measured and honest
+
+The README's "two ways to read the corpus" table carried figures taken before
+the corpus was re-chunked into row groups (cog_pipeline#93, published
+2026-08-09) and reported the mirrored column as "local speed" with no number at
+all. Re-measured 2026-08-10 against the published corpus (`pipeline_commit
+3d28ddd`), fresh R session per arm:
+
+* **A local mirror is roughly 60-80x faster.** A one-off question costs ~12 s
+  end to end remotely against ~0.15 s mirrored. That is the largest single
+  difference available to a user and it is now stated outright rather than left
+  as "local speed".
+* **Opening the session is the largest remote cost** (~7.5 s -- manifest fetch
+  plus 23 view registrations over HTTPS), larger than any individual query, and
+  it lands on the first query rather than on `library(uscogdata)`. The old table
+  did not account for it anywhere.
+* **The remote cost is round-trips, not scanning.** A repeat query over
+  already-touched partitions is ~1.5 s against ~4 s cold, and a full-history
+  query costs ~7 s whether it runs first or last.
+* The corpus size is **~201 MB**, not 190.6 MB -- row-group chunking added ~3.4%
+  and the old figure was ambiguous between MB and MiB besides.
+* Documented that a burst of remote queries can be rate-limited by the host
+  (`HTTP 429`), which is another reason to mirror for real work.
+
+## 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.

R/balances.R

@@ -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
 }
 

R/basis.R

@@ -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)

R/cohort.R

@@ -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)
+  )
+}

R/complete.R

@@ -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))

R/config.R

@@ -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)
+}

R/coverage.R

@@ -84,22 +84,92 @@
 #' `n_units_reporting = 0`, which is precisely the disclosure a silently
 #' missing year fails to make.
 #'
-#' `n_units_reporting` describes the result the caller actually received, so
-#' under `coverage = "consistent"` it reports the balanced count. `is_census_year`
-#' is a statement about the SURVEY CALENDAR, never a claim of completeness:
-#' FY1967 is a census year in which only 97 of Wisconsin's 608 cities report.
-#' `n_units_reporting` is the number that tells the truth.
+#' Three counters are returned, each answering a different question:
+#'
+#'   * `n_units_expected` -- the universe the caller named (govids passed in,
+#'     or peers for cog_peer_compare). "How many governments did you ask
+#'     about?"
+#'   * `n_units_collected` -- how many of those appear in the corpus at all
+#'     that year, in ANY category. This is a statement about survey collection,
+#'     independent of what was asked for: "of the governments you named, how
+#'     many did Census actually collect data from this year?" It separates
+#'     sampling (not collected) from real zeros (collected but spends nothing
+#'     in your category).
+#'   * `n_units_reporting` -- how many of those appear with rows for the
+#'     SPECIFIC category you requested. This is always <= n_units_collected:
+#'     a government can be collected but have no rows for "Police" because it
+#'     contracts policing to the county sheriff, not because it wasn't
+#'     surveyed.
+#'
+#' `n_units_reporting` therefore conflates two very different things: a unit
+#' that was not collected (sampling) and a unit that was collected but spends
+#' nothing in that category. The ratio n_units_collected / n_units_expected is
+#' the true collection rate; n_units_reporting / n_units_collected measures
+#' category participation among collected units.
+#'
+#' `is_census_year` is a statement about the SURVEY CALENDAR, never a claim of
+#' completeness: FY1967 is a census year in which only 97 of Wisconsin's 608
+#' cities report. The counters are what tell the truth.
+#'
+#' @param con Active DuckDB connection (used to look up n_units_collected).
+#' @param long_view The verb's own long view, used for the collection query;
+#'   NULL skips the lookup and leaves n_units_collected as NA_integer_.
+#' @param expected_ids The full EXPECTED cohort (govids the caller named),
+#'   used as the candidate list for the collection query. Required alongside
+#'   `con`/`long_view` for a correct count -- see the note below on why it
+#'   must not be derived from `result`/`rows`. `NULL`, or non-`NULL` but
+#'   empty after dropping `NA`/`""` entries, skips the lookup and leaves
+#'   n_units_collected as NA_integer_.
 #' @noRd
 .coverage_table <- function(result, years, n_expected,
-                            id_col = "canonical_govid", rows = NULL) {
+                            id_col = "canonical_govid", rows = NULL,
+                            con = NULL, long_view = NULL,
+                            expected_ids = NULL) {
   years <- sort(unique(as.integer(years)))
   src <- if (is.null(rows)) result else rows
   reporting <- vapply(years, function(y) {
     ids <- src[[id_col]][as.integer(src$year) == y]
     length(unique(ids[!is.na(ids)]))
   }, integer(1))
+
+  # n_units_collected: count EXPECTED cohort members present in the corpus
+  # for ANY category that year, not just the requested one. This separates
+  # sampling (not collected at all) from real zeros (collected but no rows
+  # for this category). Only computed when a connection, long_view, AND
+  # expected_ids are all provided; otherwise NA_integer_.
+  #
+  # The candidate list MUST be expected_ids, not derived from `result`/
+  # `rows`: a government with zero rows in the requested category across
+  # EVERY requested year never appears in `result` at all, so deriving
+  # candidates from it would silently exclude exactly the "collected but
+  # real zero" governments this counter exists to count -- collapsing
+  # n_units_collected back to n_units_reporting for precisely the case #36
+  # was filed over.
+  if (!is.null(con) && !is.null(long_view) && length(expected_ids) > 0L) {
+    cohort_chr <- .sql_lit_chr(unique(expected_ids[!is.na(expected_ids) &
+                                                       nzchar(expected_ids)]))
+    years_lit <- paste(years, collapse = ",")
+    collected_q <- sprintf(
+      "SELECT year, COUNT(DISTINCT canonical_govid) AS n
+       FROM %s
+       WHERE canonical_govid IN (%s)
+         AND year IN (%s)
+       GROUP BY year",
+      long_view, cohort_chr, years_lit
+    )
+    collected_df <- DBI::dbGetQuery(con, collected_q)
+    collected_map <- setNames(collected_df$n, as.integer(collected_df$year))
+    collected <- vapply(years, function(y) {
+      val <- collected_map[as.character(y)]
+      if (is.na(val)) 0L else as.integer(val)
+    }, integer(1))
+  } else {
+    collected <- rep(NA_integer_, length(years))
+  }
+
   tibble::tibble(
     year              = years,
+    n_units_collected = collected,
     n_units_reporting = as.integer(reporting),
     n_units_expected  = rep(as.integer(n_expected), length(years)),
     is_census_year    = .is_census_year(years)

R/explain.R

@@ -166,12 +166,30 @@ cog_explain <- function(result, format = c("print", "list")) {
     cli::cli_h2("Reporting coverage")
     cli::cli_text("Mode: {prov$coverage_mode %||% 'all'}")
     cov <- prov$coverage
-    cli::cli_ul(sprintf(
-      "%d: %d of %d units reporting (%.0f%%) -- %s year",
-      cov$year, cov$n_units_reporting, cov$n_units_expected,
-      100 * cov$n_units_reporting / pmax(cov$n_units_expected, 1L),
-      ifelse(cov$is_census_year, "census", "sample")
-    ))
+    has_collected <- "n_units_collected" %in% names(cov)
+    if (has_collected) {
+      # Three counters: collected separates sampling from real zeros;
+      # reporting is category-conditional and never a response rate.
+      cli::cli_ul(sprintf(
+        "%d: %d of %d units collected, %d reporting in this category -- %s year",
+        cov$year,
+        cov$n_units_collected,
+        cov$n_units_expected,
+        cov$n_units_reporting,
+        ifelse(cov$is_census_year, "census", "sample")
+      ))
+    } else {
+      cli::cli_ul(sprintf(
+        "%d: %d of %d units reporting (%.0f%%) -- %s year",
+        cov$year, cov$n_units_reporting, cov$n_units_expected,
+        100 * cov$n_units_reporting / pmax(cov$n_units_expected, 1L),
+        ifelse(cov$is_census_year, "census", "sample")
+      ))
+    }
+    # Explains what the per-row "-- sample year" tag means, regardless of
+    # which branch above rendered it -- not gated on has_collected, which
+    # would make this permanently unreachable now that both real callers
+    # (cog_geographic_rollup(), cog_peer_compare()) always supply it.
     if (any(!cov$is_census_year)) {
       cli::cli_text(
         "Note: the Census of Governments is a complete census only in years ending in 2 or 7; every other year is a sample."

R/pagination.R

@@ -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]]))
+}

R/peers.R

@@ -195,11 +195,13 @@ cog_find_peers <- function(target_govid,
 #'     a balanced panel.
 #'
 #'   Regardless of mode, `provenance$coverage` always carries per-year
-#'   `n_units_reporting`, `n_units_expected` and `is_census_year`, and
-#'   `provenance$coverage_mode` records the mode. `is_census_year` is a
-#'   statement about the **survey calendar**, never a claim of completeness:
-#'   FY1967 is a census year in which only 97 of Wisconsin's 608 cities
-#'   report. `n_units_reporting` is the number that tells the truth.
+#'   `n_units_expected`, `n_units_collected`, `n_units_reporting` and
+#'   `is_census_year`, and `provenance$coverage_mode` records the mode.
+#'   `is_census_year` is a statement about the **survey calendar**, never a
+#'   claim of completeness: FY1967 is a census year in which only 97 of
+#'   Wisconsin's 608 cities report. `n_units_reporting` is
+#'   category-conditional and is not a response rate on its own -- see
+#'   "Reading `coverage`" below for what each counter answers.
 #'
 #'   The comparison target is exempt from `"consistent"` balancing -- it is the
 #'   subject of the comparison, not a member of the cohort -- and the
@@ -241,14 +243,23 @@ cog_find_peers <- function(target_govid,
 #'     summarise(p50 = quantile(total, 0.5, na.rm = TRUE))
 #'   ```
 #' @section Reading `coverage`:
-#' `provenance$coverage` reports `n_units_reporting` against
-#' `n_units_expected` per year. **`n_units_reporting` is category-conditional:
-#' it counts cohort members with rows for the category you asked for, not
-#' cohort members collected that year.** A government that was surveyed and
-#' genuinely spends nothing in that category is indistinguishable here from one
-#' that was never surveyed.
+#' `provenance$coverage` carries three per-year counters:
+#'
+#'   * `n_units_expected` -- how many governments you asked about.
+#'   * `n_units_collected` -- how many of those appear in the corpus at all
+#'     that year (in ANY category), separating sampling from real zeros.
+#'   * `n_units_reporting` -- how many have rows for the SPECIFIC category you
+#'     requested. This is always <= n_units_collected: a government can be
+#'     collected but have no rows for "Police" because it contracts policing
+#'     to the county sheriff, not because it wasn't surveyed.
+#'
+#' **`n_units_reporting` is category-conditional** and therefore **not a
+#' response rate**: `n_units_reporting / n_units_expected` conflates sampling
+#' (never collected) with real zeros (collected but spends nothing in your
+#' category). Use `n_units_collected / n_units_expected` for the true
+#' collection rate, and `n_units_reporting / n_units_collected` for category
+#' participation among collected units.
 #'
-#' The ratio is therefore **not a response rate** and must not be used as one.
 #' In FY2022 — a complete census year — Georgia reports 393 of 567 cities for
 #' `category = "Police"`; the 174-city gap is overwhelmingly cities that
 #' contract policing to the county sheriff, not non-response.
@@ -325,10 +336,20 @@ cog_peer_compare <- function(target_govid, peers, category, years,
   # Counted over PEER rows only, against the cohort size: "3 of your 15 peers
   # reported in FY2019". Including the target would inflate every count by one
   # and make a cohort that has entirely stopped reporting look non-empty.
+  # n_units_collected is looked up against the spending long view matching
+  # whatever basis cog_spending() actually resolved above (prov$basis) --
+  # NOT hardcoded to spending_long_harmonized, which does not exist on a
+  # corpus with schema_version < 5 (R/basis.R resolves basis = "raw" there,
+  # and only *_long, not *_long_harmonized, is registered; see R/views.R).
+  # The con comes from .ensure_session() already called inside cog_spending().
+  con <- .ensure_session()
   prov$coverage_mode <- coverage
   prov$coverage <- .coverage_table(
     out, years, length(peer_govids),
-    rows = r[r$role == "peer", , drop = FALSE]
+    rows = r[r$role == "peer", , drop = FALSE],
+    con = con,
+    long_view = .select_long_view("spending_annotated", prov$basis),
+    expected_ids = peer_govids
   )
   attr(out, "provenance") <- prov
   out

R/recipes.R

@@ -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))

R/revenue.R

@@ -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
   )
 }

R/rollup.R

@@ -49,11 +49,13 @@
 #'     a balanced panel.
 #'
 #'   Regardless of mode, `provenance$coverage` always carries per-year
-#'   `n_units_reporting`, `n_units_expected` and `is_census_year`, and
-#'   `provenance$coverage_mode` records the mode. `is_census_year` is a
-#'   statement about the **survey calendar**, never a claim of completeness:
-#'   FY1967 is a census year in which only 97 of Wisconsin's 608 cities
-#'   report. `n_units_reporting` is the number that tells the truth.
+#'   `n_units_expected`, `n_units_collected`, `n_units_reporting` and
+#'   `is_census_year`, and `provenance$coverage_mode` records the mode.
+#'   `is_census_year` is a statement about the **survey calendar**, never a
+#'   claim of completeness: FY1967 is a census year in which only 97 of
+#'   Wisconsin's 608 cities report. `n_units_reporting` is
+#'   category-conditional and is not a response rate on its own -- see
+#'   "Reading `coverage`" below for what each counter answers.
 #' @return Tibble with columns `year`, `layer`, `canonical_govid`, `gov_name`,
 #'   `spend_subtype`, `category`, `amt_nominal`, optional `amt_real` /
 #'   `amt_per_capita_nominal` / `amt_per_capita_real`, optional `pop_source`,
@@ -61,14 +63,23 @@
 #'   `provenance` attribute with `verb = "cog_geographic_rollup"`, `layers`,
 #'   and `rollup$included_govids` / `rollup$excluded_govids`.
 #' @section Reading `coverage`:
-#' `provenance$coverage` reports `n_units_reporting` against
-#' `n_units_expected` per year. **`n_units_reporting` is category-conditional:
-#' it counts governments with rows for the category you asked for, not
-#' governments collected that year.** A government that was surveyed and
-#' genuinely spends nothing in that category is indistinguishable here from one
-#' that was never surveyed.
+#' `provenance$coverage` carries three per-year counters:
+#'
+#'   * `n_units_expected` -- how many governments you asked about.
+#'   * `n_units_collected` -- how many of those appear in the corpus at all
+#'     that year (in ANY category), separating sampling from real zeros.
+#'   * `n_units_reporting` -- how many have rows for the SPECIFIC category you
+#'     requested. This is always <= n_units_collected: a government can be
+#'     collected but have no rows for "Police" because it contracts policing
+#'     to the county sheriff, not because it wasn't surveyed.
+#'
+#' **`n_units_reporting` is category-conditional** and therefore **not a
+#' response rate**: `n_units_reporting / n_units_expected` conflates sampling
+#' (never collected) with real zeros (collected but spends nothing in your
+#' category). Use `n_units_collected / n_units_expected` for the true
+#' collection rate, and `n_units_reporting / n_units_collected` for category
+#' participation among collected units.
 #'
-#' The ratio is therefore **not a response rate** and must not be used as one.
 #' In FY2022 — a complete census year — Georgia reports 393 of 567 cities for
 #' `category = "Police"`; the 174-city gap is overwhelmingly cities that
 #' contract policing to the county sheriff, not non-response.
@@ -137,8 +148,20 @@ cog_geographic_rollup <- function(govids, category, years,
   # n_units_expected is the universe the CALLER named -- the govids passed in
   # -- not the national universe. That is what makes the ratio meaningful:
   # "597 of the 608 Wisconsin cities you asked about reported in FY2012".
+  # n_units_collected is looked up against the spending long view matching
+  # whatever basis cog_spending() actually resolved above (prov$basis) --
+  # NOT hardcoded to spending_long_harmonized, which does not exist on a
+  # corpus with schema_version < 5 (R/basis.R resolves basis = "raw" there,
+  # and only *_long, not *_long_harmonized, is registered; see R/views.R).
+  # The con comes from .ensure_session() already called inside cog_spending().
+  con <- .ensure_session()
   prov$coverage_mode <- coverage
-  prov$coverage <- .coverage_table(r, years, length(unique(all_govids)))
+  prov$coverage <- .coverage_table(
+    r, years, length(unique(all_govids)),
+    con = con,
+    long_view = .select_long_view("spending_annotated", prov$basis),
+    expected_ids = all_govids
+  )
   attr(r, "provenance") <- prov
 
   r

R/search.R

@@ -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.

R/session.R

@@ -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) ||

R/spending.R

@@ -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,

R/suggestions.R

@@ -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).
@@ -84,8 +84,23 @@
 #' @return List of `list(recipe_id, label, available_years, hint,
 #'   ig_recipe_id, trigger, suppressed_amount, suppressed_years,
 #'   suppressed_codes)`, possibly empty.
+#'
+#' Decomposed (Issue #33) into three extracted helpers to stay within the
+#' project's "functions under 50 lines" convention:
+#' \itemize{
+#'   \item `.query_candidate_recipes()` -- candidate recipe lookup by
+#'     category/subtype scope + `category_type` filter (#34) + M/L exclusion.
+#'   \item `.query_recipe_meta()` -- metadata (label, year spans).
+#'   \item `.query_covered_years()` -- Path 1 gap-year coverage via the
+#'     recipe's own generic join.
+#' }
+#' The for-loop that merges covered-years + suppressed-components into
+#' suggestion objects stays inline here because it interleaves
+#' empty_hit/supp_hit precedence with field assembly. Likewise kept inline:
+#' the M/L-exclusion design-comment block and the final
+#' `.attach_ig_counterparts()` call.
 #' @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) {
@@ -111,28 +126,16 @@
   # by `category` (`.ALL_CATEGORIES` is never a row in
   # `summary_categories.category`, so a category-keyed sub-select always
   # came back empty here). The M/L exclusion below is unchanged either way.
-  candidate_scope_sql <- if (isTRUE(all_categories)) {
-    sprintf(
-      "SELECT DISTINCT item_code FROM summary_categories WHERE %s IN (%s)",
-      subtype_col, .sql_lit_chr(subtype_scope)
-    )
-  } else {
-    sprintf(
-      "SELECT DISTINCT item_code FROM summary_categories WHERE category IN (%s)",
-      .sql_lit_chr(category)
-    )
-  }
-  candidates <- DBI::dbGetQuery(con, sprintf(
-    "SELECT DISTINCT recipe_id FROM harmonization_recipes
-     WHERE component_code IN (
-       %s
-     )
-     AND recipe_id NOT IN (
-       SELECT DISTINCT recipe_id FROM harmonization_recipes
-       WHERE LEFT(component_code, 1) IN ('M', 'L')
-     )",
-    candidate_scope_sql
-  ))$recipe_id
+  #
+  # Issue #34: scope the candidate query by `category_type` ('expenditure'
+  # vs 'revenue') to prevent cross-flow-family leakage -- e.g.
+  # `cog_revenue(category = "Corrections")` must not surface
+  # expenditure-only recipes (E04/E05) merely because they share the same
+  # category name in summary_categories. The type is derived from
+  # flow_prefixes: E/F/G -> 'expenditure', anything else -> 'revenue'.
+  candidates <- .query_candidate_recipes(con, category, flow_prefixes,
+                                          all_categories, subtype_col,
+                                          subtype_scope)
   if (length(candidates) == 0L) return(list())
 
   result_years <- if (is.null(result) || nrow(result) == 0L) {
@@ -160,40 +163,15 @@
   # 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())
 
-  meta <- tibble::as_tibble(DBI::dbGetQuery(con, sprintf(
-    "SELECT recipe_id, any_value(label) AS label,
-            MIN(year_min) AS year_min, MAX(year_max) AS year_max
-     FROM harmonization_recipes
-     WHERE recipe_id IN (%s)
-     GROUP BY recipe_id",
-    .sql_lit_chr(candidates)
-  )))
+  meta <- .query_recipe_meta(con, candidates)
 
   # Path 1 (unchanged): (recipe, year) pairs the recipe's own generic join
   # covers for this government, restricted to the gap years.
-  covered <- if (length(gap_years) == 0L) {
-    data.frame(recipe_id = character(0), year = integer(0))
-  } else {
-    DBI::dbGetQuery(con, sprintf(
-      "SELECT DISTINCT r.recipe_id, l.year
-       FROM long l
-       JOIN harmonization_recipes r
-         ON l.item_code = r.component_code
-        AND l.year BETWEEN r.year_min AND r.year_max
-        AND (r.gov_type_scope = 'all'
-             OR (r.gov_type_scope = 'state' AND l.type = 0)
-             OR (r.gov_type_scope = 'local' AND l.type BETWEEN 1 AND 3))
-       WHERE r.recipe_id IN (%s)
-         AND l.canonical_govid IN (%s)
-         AND l.year IN (%s)",
-      .sql_lit_chr(candidates), .sql_lit_chr(govid),
-      paste(gap_years, collapse = ",")
-    ))
-  }
+  covered <- .query_covered_years(con, candidates, cohort, gap_years)
 
   suggestions <- list()
   for (rid in candidates) {
@@ -227,6 +205,146 @@
   .attach_ig_counterparts(con, suggestions, flow_prefixes)
 }
 
+#' Query candidate harmonization recipe IDs for a coverage-gap suggestion.
+#'
+#' Selects recipes whose component codes fall within the requested scope
+#' (category or subtype allowlist), excluding any recipe that is ITSELF an
+#' intergovernmental (M/L) recipe -- i.e. every one of its own component
+#' codes is M/L-prefixed. Without this exclusion, a category whose
+#' summary_categories rows span both a Direct family (e.g. E04/E05,
+#' "Corrections") and its M/L counterpart (M04/M05) makes the M/L recipe
+#' itself a raw top-level candidate for a plain `cog_spending()` call --
+#' following that hint would silently return intergovernmental dollars
+#' under `expenditure_concept = "direct"` provenance.
+#'
+#' In all-categories mode (`all_categories = TRUE`) the inner sub-select is
+#' scoped by `subtype_col`/`subtype_scope` -- the same allowlist
+#' `.build_verb_sql()` applies as a WHERE predicate to make the summed
+#' result a *concept* (see R/spending.R), not by `category`.
+#' `.ALL_CATEGORIES` ("All Categories") is never itself a row in
+#' `summary_categories.category`, so a category-keyed sub-select always
+#' returns zero candidates and silently disables signposting.
+#'
+#' Scope is also by `category_type` ('expenditure' vs 'revenue', Issue #34)
+#' to prevent cross-flow-family leakage: `cog_revenue(category =
+#' "Corrections")` must not surface expenditure-only recipes (E04/E05)
+#' merely because they share the same category name in summary_categories.
+#' The type is derived from flow_prefixes: E/F/G -> 'expenditure', anything
+#' else -> 'revenue'.
+#'
+#' @param con Active DuckDB connection.
+#' @param category Category name, or `NULL`.
+#' @param flow_prefixes The calling verb's own flow-type prefixes (see
+#'   `.build_suggestions()`). Used to derive `category_type` (#34).
+#' @param all_categories `TRUE` when the caller used `.ALL_CATEGORIES`.
+#' @param subtype_col Name of the summary_categories subtype column to
+#'   scope by when `all_categories = TRUE`; ignored otherwise.
+#' @param subtype_scope Character vector of subtype values to scope by
+#'   when `all_categories = TRUE`; ignored otherwise.
+#' @return Character vector of recipe IDs (possibly empty).
+#' @noRd
+.query_candidate_recipes <- function(con, category, flow_prefixes,
+                                      all_categories = FALSE,
+                                      subtype_col = NULL,
+                                      subtype_scope = NULL) {
+  # Issue #34: derive category_type from flow_prefixes to prevent
+  # cross-flow-family leakage -- e.g. cog_revenue(category = "Corrections")
+  # must not surface expenditure-only recipes merely because they share the
+  # same category name in summary_categories.
+  category_type <- if (all(flow_prefixes %in% c("E", "F", "G"))) {
+    "expenditure"
+  } else {
+    "revenue"
+  }
+
+  candidate_scope_sql <- if (isTRUE(all_categories)) {
+    sprintf(
+      "SELECT DISTINCT item_code FROM summary_categories
+       WHERE %s IN (%s) AND category_type = '%s'",
+      subtype_col, .sql_lit_chr(subtype_scope), category_type
+    )
+  } else {
+    sprintf(
+      "SELECT DISTINCT item_code FROM summary_categories
+       WHERE category IN (%s) AND category_type = '%s'",
+      .sql_lit_chr(category), category_type
+    )
+  }
+
+  DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT recipe_id FROM harmonization_recipes
+     WHERE component_code IN (
+       %s
+     )
+     AND recipe_id NOT IN (
+       SELECT DISTINCT recipe_id FROM harmonization_recipes
+       WHERE LEFT(component_code, 1) IN ('M', 'L')
+     )",
+    candidate_scope_sql
+  ))$recipe_id
+}
+
+#' Query gap-year coverage: which (recipe, year) pairs the recipe's own
+#' generic join covers for this government, restricted to `gap_years`.
+#'
+#' This is Path 1 of a suggestion (unchanged): it finds recipes whose
+#' component codes' generic join produces at least one row for this
+#' government in each gap year -- i.e. the category returned nothing in
+#' that year but a recipe would fill it.
+#'
+#' @param con Active DuckDB connection.
+#' @param candidates Character vector of recipe IDs to check coverage for.
+#' @param cohort The verb's cohort object (see `.make_cohort()`), rendered
+#'   into the govid predicate on the joined `long` scan via `.cohort_sql()`.
+#' @param gap_years Integer vector of requested years absent from the
+#'   result.
+#' @return Data frame with columns `recipe_id` (character) and `year`
+#'   (integer). Returns an empty data frame (`recipe_id = character(0)`,
+#'   `year = integer(0)`) when `gap_years` or `candidates` is empty, so
+#'   callers can safely reference `$recipe_id`.
+#' @noRd
+.query_covered_years <- function(con, candidates, cohort, gap_years) {
+  if (length(gap_years) == 0L || length(candidates) == 0L) {
+    return(data.frame(recipe_id = character(0), year = integer(0)))
+  }
+  res <- DBI::dbGetQuery(con, sprintf(
+    "SELECT DISTINCT r.recipe_id, l.year
+     FROM long l
+     JOIN harmonization_recipes r
+       ON l.item_code = r.component_code
+      AND l.year BETWEEN r.year_min AND r.year_max
+      AND (r.gov_type_scope = 'all'
+           OR (r.gov_type_scope = 'state' AND l.type = 0)
+           OR (r.gov_type_scope = 'local' AND l.type BETWEEN 1 AND 3))
+     WHERE r.recipe_id IN (%s)
+       AND %s
+       AND l.year IN (%s)",
+    .sql_lit_chr(candidates), .cohort_sql(cohort, "l.canonical_govid"),
+    paste(gap_years, collapse = ",")
+  ))
+  res$year <- as.integer(res$year)
+  res
+}
+
+#' Query recipe metadata: labels and year spans for a set of candidate
+#' recipes.
+#'
+#' @param con Active DuckDB connection.
+#' @param candidates Character vector of recipe IDs to look up.
+#' @return Tibble with columns `recipe_id`, `label`, `year_min` (int), and
+#'   `year_max` (int).
+#' @noRd
+.query_recipe_meta <- function(con, candidates) {
+  tibble::as_tibble(DBI::dbGetQuery(con, sprintf(
+    "SELECT recipe_id, any_value(label) AS label,
+            MIN(year_min) AS year_min, MAX(year_max) AS year_max
+     FROM harmonization_recipes
+     WHERE recipe_id IN (%s)
+     GROUP BY recipe_id",
+    .sql_lit_chr(candidates)
+  )))
+}
+
 #' Attach `ig_recipe_id` to each suggestion: the intergovernmental-expenditure
 #' recipe (an M-to-local or L-to-state recipe) whose component codes cover
 #' exactly the same set of function suffixes as the firing recipe's own
@@ -272,7 +390,7 @@
 #'      `R/basis.R`). This blocks a recipe surfaced through a mis-scoped
 #'      category from ever reaching the M/L search, e.g. `cog_spending()`'s
 #'      flow_prefixes are `c("E","F","G")`, which `ig_federal_b47_wide`'s own
-#'      `"B"` is not part of.
+#'      "B" is not part of.
 #'   2. `own_prefix %in% c("E","F","G")`: M/L only ever pairs with the
 #'      DIRECT-expenditure family, never with revenue (`cog_revenue()`'s
 #'      flow_prefixes already fold B/C/D in as ordinary revenue -- there is
@@ -280,10 +398,6 @@
 #'      adds one for spending) and never with ANOTHER M/L recipe (without
 #'      this check, `ige_local_m47_wide` would wrongly match sibling
 #'      `ige_state_l47_wide` on their shared {"47","94"} suffix set).
-#'      Condition 1 alone does not catch this: under `cog_revenue()`,
-#'      `ig_federal_b47_wide`'s own `"B"` IS inside revenue's own
-#'      `flow_prefixes`, so only this second, family-specific check blocks
-#'      the search.
 #' @noRd
 .attach_ig_counterparts <- function(con, suggestions, flow_prefixes) {
   if (length(suggestions) == 0L) return(suggestions)

R/suppression.R

@@ -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))
 }

R/views.R

@@ -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

README.md

@@ -1,6 +1,7 @@
 # uscogdata
 
 <!-- badges: start -->
+[![R-CMD-check](https://github.com/civilytics/uscogdata/actions/workflows/R-CMD-check.yaml/badge.svg)](https://github.com/civilytics/uscogdata/actions/workflows/R-CMD-check.yaml)
 [![r-universe](https://civilytics.r-universe.dev/badges/uscogdata)](https://civilytics.r-universe.dev/uscogdata)
 [![License: MIT](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE.md)
 <!-- badges: end -->
@@ -18,7 +19,7 @@ carries provenance describing what was converted, what was aggregated, and
 which known series breaks intersect your query.
 
 **Scope:** government types 0–3 (state, county, municipality, township).
-56 fiscal years, 46,148,034 rows, 190.6 MB. There is no source data for FY1968
+56 fiscal years, 46,148,034 rows, ~201 MB. There is no source data for FY1968
 or FY1969. Special districts (type 4) and school districts (type 5) are
 excluded pending validation.
 
@@ -85,14 +86,38 @@ cog_explain(spend)
 
 | | Remote (default) | Mirrored |
 |---|---|---|
-| Setup | none | `cog_mirror(dest)`, 190.6 MB once |
-| Disk used | **0 MB** — HTTP range requests only | 190.6 MB |
-| Per query | ~4 s (one government, one year)<br>~6 s (one government, 23 years) | local speed |
+| Setup | none | `cog_mirror(dest)`, ~201 MB once |
+| Disk used | **0 MB** — HTTP range requests only | ~201 MB |
+| Opening a session | ~7.5 s | ~0.1 s |
+| One government, one year | ~4 s | ~0.05 s |
+| One government, full history | ~7 s | ~0.1 s |
+| Later queries, same session | ~1.5 s | ~0.05 s |
 | Good for | trying it out, teaching, one-off questions | repeated analysis, offline work, reproducibility |
 
+**A local mirror is roughly 60–80x faster, and it is one function call.** That is
+by far the largest difference any of these settings makes. If you are going to
+ask more than a handful of questions, mirror first.
+
+Measured 2026-08-10 on a 16-core Linux workstation against the published corpus
+(schema v7, `pipeline_commit 3d28ddd`), fresh R session per arm. A one-off
+question costs about **12 seconds end to end remotely and 0.15 seconds
+mirrored**, session setup included.
+
+Two things the per-query rows hide:
+
+- **Opening the session is the single largest remote cost** — larger than any
+  one query. It fetches the manifest and registers 23 SQL views over HTTPS, and
+  it lands on your first query, not on `library(uscogdata)`.
+- **The cost is network round-trips, not scanning.** A repeat query against
+  partitions this session has already touched is ~1.5 s rather than ~4 s, and a
+  full-history query costs ~7 s whether it runs first or last. What you are
+  paying for is reaching each of the 56 yearly files over HTTPS the first time.
+
 Nothing is written to disk in remote mode: DuckDB fetches the parquet footer,
 works out which row groups it needs, and reads only those. Nothing is cached
-between sessions either, so every query goes back to the network.
+between sessions either, so every query goes back to the network — and a session
+that issues many remote queries in quick succession can be rate-limited by the
+host (`HTTP Error: ... 429`). Both are further reasons to mirror for real work.
 
 The default points at a public HuggingFace mirror of the corpus. If you would
 rather not depend on a third party — for reproducibility, for an air-gapped
@@ -110,6 +135,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
 
@@ -191,7 +227,8 @@ A statewide total resting on a fifth of the universe looks exactly like one
 resting on all of it, so every multi-government result now says which it is:
 
 ```r
-attr(rollup, "provenance")$coverage   # per-year n_units_reporting, is_census_year
+attr(rollup, "provenance")$coverage
+# per-year n_units_expected, n_units_collected, n_units_reporting, is_census_year
 ```
 
 `cog_geographic_rollup()`, `cog_peer_compare()` and `cog_find_peers()` take a
@@ -199,8 +236,14 @@ attr(rollup, "provenance")$coverage   # per-year n_units_reporting, is_census_ye
 `"consistent"` (only units reporting in every requested year, a balanced
 panel).
 
-`n_units_reporting` is **category-conditional**, and it is not a response rate. A government that was surveyed and genuinely spends
-nothing in the requested category is indistinguishable from one never surveyed.
+`n_units_reporting` is **category-conditional**: it counts governments with
+rows for the *specific* category you asked for, so a government that was
+surveyed and genuinely spends nothing in that category is indistinguishable
+from one never surveyed — it is not a response rate on its own.
+`n_units_collected` is the number that separates them: governments present in
+the corpus that year for *any* category. `n_units_collected / n_units_expected`
+is the true collection rate; `n_units_reporting / n_units_collected` is
+category participation among collected units.
 
 ### Absent cells mean two different things
 

man/cog_balances.Rd

@@ -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"`):

man/cog_geographic_rollup.Rd

@@ -55,11 +55,13 @@ Direct spending); `"primary"` and `"direct"` combine safely.}
     a balanced panel.
 
   Regardless of mode, `provenance$coverage` always carries per-year
-  `n_units_reporting`, `n_units_expected` and `is_census_year`, and
-  `provenance$coverage_mode` records the mode. `is_census_year` is a
-  statement about the **survey calendar**, never a claim of completeness:
-  FY1967 is a census year in which only 97 of Wisconsin's 608 cities
-  report. `n_units_reporting` is the number that tells the truth.}
+  `n_units_expected`, `n_units_collected`, `n_units_reporting` and
+  `is_census_year`, and `provenance$coverage_mode` records the mode.
+  `is_census_year` is a statement about the **survey calendar**, never a
+  claim of completeness: FY1967 is a census year in which only 97 of
+  Wisconsin's 608 cities report. `n_units_reporting` is
+  category-conditional and is not a response rate on its own -- see
+  "Reading `coverage`" below for what each counter answers.}
 }
 \value{
 Tibble with columns `year`, `layer`, `canonical_govid`, `gov_name`,
@@ -86,14 +88,23 @@ by design — see `vignette('population-denominators')`.
 }
 \section{Reading `coverage`}{
 
-`provenance$coverage` reports `n_units_reporting` against
-`n_units_expected` per year. **`n_units_reporting` is category-conditional:
-it counts governments with rows for the category you asked for, not
-governments collected that year.** A government that was surveyed and
-genuinely spends nothing in that category is indistinguishable here from one
-that was never surveyed.
+`provenance$coverage` carries three per-year counters:
+
+  * `n_units_expected` -- how many governments you asked about.
+  * `n_units_collected` -- how many of those appear in the corpus at all
+    that year (in ANY category), separating sampling from real zeros.
+  * `n_units_reporting` -- how many have rows for the SPECIFIC category you
+    requested. This is always <= n_units_collected: a government can be
+    collected but have no rows for "Police" because it contracts policing
+    to the county sheriff, not because it wasn't surveyed.
+
+**`n_units_reporting` is category-conditional** and therefore **not a
+response rate**: `n_units_reporting / n_units_expected` conflates sampling
+(never collected) with real zeros (collected but spends nothing in your
+category). Use `n_units_collected / n_units_expected` for the true
+collection rate, and `n_units_reporting / n_units_collected` for category
+participation among collected units.
 
-The ratio is therefore **not a response rate** and must not be used as one.
 In FY2022 — a complete census year — Georgia reports 393 of 567 cities for
 `category = "Police"`; the 174-city gap is overwhelmingly cities that
 contract policing to the county sheriff, not non-response.

man/cog_gov_search.Rd

@@ -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`,

man/cog_peer_compare.Rd

@@ -50,11 +50,13 @@ safely.}
     a balanced panel.
 
   Regardless of mode, `provenance$coverage` always carries per-year
-  `n_units_reporting`, `n_units_expected` and `is_census_year`, and
-  `provenance$coverage_mode` records the mode. `is_census_year` is a
-  statement about the **survey calendar**, never a claim of completeness:
-  FY1967 is a census year in which only 97 of Wisconsin's 608 cities
-  report. `n_units_reporting` is the number that tells the truth.
+  `n_units_expected`, `n_units_collected`, `n_units_reporting` and
+  `is_census_year`, and `provenance$coverage_mode` records the mode.
+  `is_census_year` is a statement about the **survey calendar**, never a
+  claim of completeness: FY1967 is a census year in which only 97 of
+  Wisconsin's 608 cities report. `n_units_reporting` is
+  category-conditional and is not a response rate on its own -- see
+  "Reading `coverage`" below for what each counter answers.
 
   The comparison target is exempt from `"consistent"` balancing -- it is the
   subject of the comparison, not a member of the cohort -- and the
@@ -109,14 +111,23 @@ them.
 }
 \section{Reading `coverage`}{
 
-`provenance$coverage` reports `n_units_reporting` against
-`n_units_expected` per year. **`n_units_reporting` is category-conditional:
-it counts cohort members with rows for the category you asked for, not
-cohort members collected that year.** A government that was surveyed and
-genuinely spends nothing in that category is indistinguishable here from one
-that was never surveyed.
+`provenance$coverage` carries three per-year counters:
+
+  * `n_units_expected` -- how many governments you asked about.
+  * `n_units_collected` -- how many of those appear in the corpus at all
+    that year (in ANY category), separating sampling from real zeros.
+  * `n_units_reporting` -- how many have rows for the SPECIFIC category you
+    requested. This is always <= n_units_collected: a government can be
+    collected but have no rows for "Police" because it contracts policing
+    to the county sheriff, not because it wasn't surveyed.
+
+**`n_units_reporting` is category-conditional** and therefore **not a
+response rate**: `n_units_reporting / n_units_expected` conflates sampling
+(never collected) with real zeros (collected but spends nothing in your
+category). Use `n_units_collected / n_units_expected` for the true
+collection rate, and `n_units_reporting / n_units_collected` for category
+participation among collected units.
 
-The ratio is therefore **not a response rate** and must not be used as one.
 In FY2022 — a complete census year — Georgia reports 393 of 567 cities for
 `category = "Police"`; the 174-city gap is overwhelmingly cities that
 contract policing to the county sheriff, not non-response.

man/cog_revenue.Rd

@@ -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`,

man/cog_spending.Rd

@@ -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`,

pm/JOURNAL.md

@@ -0,0 +1,14 @@
+# Project journal
+
+Append-only, newest first. **Entries are never edited** — the value of this file is
+that it records what was believed at the time, including the parts that turned out
+wrong. Where things stand *today* is in `STATUS.md`, which is generated.
+
+Four lines per entry. The analysis belongs in the issue or the decision record; this
+file carries the reasoning and the pointers.
+
+- **Why** — the driver. The one line git cannot reconstruct later.
+- **Obligates** — issues this change created elsewhere. Numbers, not prose.
+- **Refs** — commits, issues, decision records.
+
+---

pm/STATUS.md

@@ -0,0 +1,67 @@
+# Project status
+
+> Between the compass markers is generated. Edit the sources, not this.
+
+<!-- compass:begin -->
+<!-- compass:board -->
+
+## Where this stands
+
+uscogdata is at 0.4.0 and its public surface is settled: the query verbs, the cohort
+predicates added in this release, and the provenance contract every verb returns.
+
+The six open issues split cleanly. Two are API work carried out of the #9 review pass
+and deliberately deferred there rather than fixed in that branch. Three concern the
+corpus layer, and the largest of them, partition-level caching, was named the single
+highest-leverage change on the remote path before being deferred. One, the
+data-correction intake (#52), is a decision rather than a task: it was parked during
+the 0.3.0 design, and the API announcement waits on it, because without it the corpus
+cannot make the "traceable and correctable" claim that most distinguishes it from
+Census's own files.
+
+Nothing here is blocked on anything else, so the ordering is a judgement about value
+rather than a dependency graph.
+
+Compass's own files moved out of `docs/` this session. They were sitting inside
+pkgdown's output directory, and `pkgdown::clean_site()` deletes every top-level entry
+there except `CNAME` and `dev` — asked directly, it listed `docs/pm` and
+`docs/decisions` among the 28 it would remove, with the guard that would have stopped
+it satisfied by `docs/pkgdown.yml`. They are in `pm/` now. Nothing was lost: the
+journal had no entries and there were no decision records yet, which made this the
+cheapest moment to move. The `.gitignore` workaround that re-included two children of
+an excluded `docs/` is gone with it.
+
+## Ready to work on next
+
+- **#34** cog_revenue() offers expenditure recipes as suggestions: scope the candidate query by category_type · `ws/api` — nothing is blocking it; something is currently wrong
+- **#36** n_units_reporting is category-conditional and cannot be read as a response rate · `ws/corpus` — nothing is blocking it; owed work from an earlier change
+- **#2** Extend population data to be households as an alternate spending denominator · `ws/corpus` — nothing is blocking it
+- **#33** Decompose .build_suggestions() (106 lines) into named helpers · `ws/api` — nothing is blocking it
+- **#52** Release 11/11: design the data-correction intake (deferred; gates the API announcement) · `ws/corpus` — nothing is blocking it
+- **#64** Partition-level caching: R/cache.R is still a stub, and the remote path pays for it every session · `ws/corpus` — nothing is blocking it
+
+## Workstreams
+
+| Stream | Commits since | Open | Debt | Owes docs |
+|---|---|---|---|---|
+| Query verbs and results | 77 | 2 | 0 | no |
+| Corpus, mirror, provenance | 39 | 4 | 1 | no |
+| Vignettes and guides | 34 | 0 | 0 | **yes** |
+
+## CI
+
+![R-CMD-check](https://gitea.civilytics.org/Civilytics/uscogdata/actions/workflows/ci.yml/badge.svg?branch=main)
+![Mirror to GitHub](https://gitea.civilytics.org/Civilytics/uscogdata/actions/workflows/mirror-github.yml/badge.svg?branch=main)
+
+<details>
+<summary>Dependency graph and detail</summary>
+
+_Nothing blocks anything else, so there is no graph to draw._
+
+- Marker: `none` (no journal entry yet)
+- Commits since: 165
+- Open issues: 6
+
+</details>
+
+<!-- compass:end -->

pm/compass.toml

@@ -0,0 +1,44 @@
+[project]
+name  = "uscogdata"
+forge = "Civilytics/uscogdata"
+
+# Three strands that go stale independently: what the verbs return, what the
+# corpus is and how it is mounted, and how both are explained to a reader.
+
+[[workstream]]
+id    = "api"
+title = "Query verbs and results"
+paths = [
+  "R/revenue.R", "R/spending.R", "R/balances.R", "R/peers.R", "R/search.R",
+  "R/categories.R", "R/recipes.R", "R/rollup.R", "R/explain.R", "R/basket.R",
+  "R/suggestions.R", "R/suppression.R", "R/complete.R", "R/cohort.R",
+  "R/basis.R", "R/adjust.R", "R/pagination.R",
+]
+docs  = ["vignettes/*.Rmd", "README.md"]
+
+[[workstream]]
+id    = "corpus"
+title = "Corpus, mirror, provenance"
+paths = [
+  "R/manifest.R", "R/mirror.R", "R/cache.R", "R/session.R", "R/provenance.R",
+  "R/coverage.R", "R/config.R", "R/views.R", "R/series_breaks.R",
+  "R/balance_caveats.R", "R/zzz.R", "data-raw/**", "inst/sql/**",
+]
+docs  = ["vignettes/*.Rmd", "NEWS.md"]
+
+[[workstream]]
+id    = "docs"
+title = "Vignettes and guides"
+paths = ["vignettes/**", "README.md", "_pkgdown.yml", "NEWS.md"]
+docs  = []
+
+[roborev]
+project_guidelines = [
+  "Every verb calls .ensure_session() first, then queries via DBI::dbGetQuery().",
+  "A verb's return value is always a tbl_df carrying a provenance attribute.",
+  "govid inputs always go through .coerce_govid_input(); it accepts a character vector or a data frame.",
+  "SQL has two layers: view definitions are numbered .sql files in inst/sql/ registered by .register_views(); query construction is inline sprintf() in R. Add a view as a file; build a query in R.",
+  "No arrow dependency -- DuckDB reads parquet natively.",
+  "withr is Suggests-only and must appear in tests alone.",
+  "Tests must pass offline against the bundled fixture; tests/testthat/setup.R sets USCOGDATA_URL for that.",
+]

pm/decisions/README.md

@@ -0,0 +1,12 @@
+# Decisions
+
+One file per decision, numbered and immutable. A decision that changes is superseded
+by a new record, never edited in place — the old reasoning is the point.
+
+The table below is **generated** by `compass:decide`. Do not hand-edit it.
+
+<!-- compass:begin decisions -->
+| # | Date | Decision | Status |
+|---|---|---|---|
+| — | — | *No decisions recorded yet.* | — |
+<!-- compass:end decisions -->

tests/testthat/test-all-categories.R

@@ -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",

tests/testthat/test-cohort-verbs.R

@@ -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)
+  })
+})

tests/testthat/test-cohort.R

@@ -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)
+})

tests/testthat/test-coverage-disclosure.R

@@ -48,6 +48,13 @@ test_that("multi-government aggregates disclose reporting coverage on every resu
   expect_equal(cov$n_units_reporting, c(152L, 597L, 112L, 114L))
   expect_equal(cov$is_census_year, c(FALSE, TRUE, FALSE, FALSE))
 
+  # uscogdata#36: with category = NULL (no category scope), "reported at
+  # all" and "collected" are the same question, so n_units_collected must
+  # equal n_units_reporting exactly here. This case alone cannot catch a
+  # regression in HOW n_units_collected is computed, though: see the
+  # category-scoped test below for that.
+  expect_equal(cov$n_units_collected, cov$n_units_reporting)
+
   # Cross-check against the raw partitions, scoped to the SAME universe the
   # rollup was given -- the 608 govids above. Scoping instead on the long
   # table's own `type`/`fips_state` asks a different question and answers 595:
@@ -80,6 +87,8 @@ test_that("multi-government aggregates disclose reporting coverage on every resu
   expect_equal(cov_peers$n_units_expected, rep(15L, 3L))
   expect_equal(cov_peers$n_units_reporting, c(15L, 3L, 3L))
   expect_equal(cov_peers$is_census_year, c(TRUE, FALSE, FALSE))
+  # uscogdata#36: same identity as the rollup case above, category = NULL.
+  expect_equal(cov_peers$n_units_collected, cov_peers$n_units_reporting)
 
   # -- the three coverage modes --------------------------------------------
   expect_equal(attr(cog_peer_compare(target_govid = chilton, peers = peers,
@@ -101,3 +110,102 @@ test_that("multi-government aggregates disclose reporting coverage on every resu
                                        coverage = "census")
   expect_equal(sort(unique(census_only$year)), 2012)
 })
+
+test_that("n_units_collected separates sampling from real zeros, category-scoped (uscogdata#36)", {
+  # The motivating case from the issue: Wisconsin cities, category = "Police".
+  # FY2012 is a complete census year -- collection is not partial -- yet a
+  # category-conditional n_units_reporting alone reads like a sampling gap.
+  # n_units_collected must diverge from n_units_reporting here, unlike the
+  # category = NULL cases above, because most of the FY2012 gap is cities
+  # that contract policing to the county sheriff (collected, real zero), not
+  # cities Census never surveyed.
+  wi <- cog_gov_search(name = NULL, state = "WI", type = "city")
+  roll <- suppressMessages(cog_geographic_rollup(
+    govids = list(city = wi$canonical_govid), category = "Police",
+    years = c(2011L, 2012L, 2019L, 2020L)))
+  cov <- wt_coverage(roll)
+
+  expect_equal(cov$n_units_expected, rep(608L, 4L))
+  expect_equal(cov$n_units_collected, c(152L, 597L, 112L, 114L))
+  expect_equal(cov$n_units_reporting, c(152L, 485L, 109L, 111L))
+
+  # The pair the issue actually wants: collected/expected is the true
+  # collection rate (98% in the FY2012 census year, matching the raw
+  # cross-check above); reporting/collected is category participation among
+  # collected units (81% -- most of the gap is real, not sampling).
+  expect_equal(round(cov$n_units_collected[cov$year == 2012] /
+                        cov$n_units_expected[cov$year == 2012], 2), 0.98)
+  expect_equal(round(cov$n_units_reporting[cov$year == 2012] /
+                        cov$n_units_collected[cov$year == 2012], 2), 0.81)
+
+  # Every year: collected is bounded between reporting and expected.
+  expect_true(all(cov$n_units_collected >= cov$n_units_reporting))
+  expect_true(all(cov$n_units_collected <= cov$n_units_expected))
+})
+
+test_that(".coverage_table() candidates a government collected-but-absent from the category result (uscogdata#36)", {
+  # Direct regression test for the mechanism itself: n_units_collected's
+  # candidate list must be the caller's full expected cohort (expected_ids),
+  # never derived from `result`/`rows`. A government with zero rows in the
+  # requested category across every requested year never appears in
+  # `result` at all, so deriving candidates from `result` would silently
+  # drop exactly the "collected but real zero" governments this counter
+  # exists to count -- collapsing it back to n_units_reporting.
+  con <- uscogdata:::.ensure_session()
+
+  # A real fixture govid, present in spending_long_harmonized for 2019 (in
+  # SOME category), but absent from this fake category-specific `result`.
+  govid <- "011021100004"
+  fake_result <- data.frame(canonical_govid = character(0), year = integer(0))
+
+  cov <- uscogdata:::.coverage_table(
+    fake_result, years = 2019L, n_expected = 1L,
+    con = con, long_view = "spending_long_harmonized",
+    expected_ids = govid
+  )
+  expect_equal(cov$n_units_collected, 1L)
+  expect_equal(cov$n_units_reporting, 0L)
+
+  # Without a connection, long_view, or expected_ids, the lookup is skipped
+  # rather than silently wrong.
+  no_con <- uscogdata:::.coverage_table(fake_result, years = 2019L, n_expected = 1L)
+  expect_true(is.na(no_con$n_units_collected))
+
+  no_ids <- uscogdata:::.coverage_table(
+    fake_result, years = 2019L, n_expected = 1L,
+    con = con, long_view = "spending_long_harmonized"
+  )
+  expect_true(is.na(no_ids$n_units_collected))
+})
+
+test_that("n_units_collected uses the resolved basis's long view, not a hardcoded harmonized one (uscogdata#36)", {
+  # spending_long_harmonized only exists when schema_version >= 5 (R/views.R
+  # gates the harmonization views on it); on an older corpus cog_spending()
+  # resolves basis = "raw" and queries spending_long instead. The coverage
+  # lookup must follow the SAME resolved basis, not a literal
+  # "spending_long_harmonized", or it hard-errors with a DuckDB catalog
+  # error on every schema_version < 5 corpus -- a vintage the package
+  # otherwise explicitly still supports (see test-manifest.R's dual-accept
+  # tests).
+  skip_if_no_corpus()
+  with_doctored_schema_version(4L, {
+    con <- cog_open()
+    ids <- DBI::dbGetQuery(con,
+      "SELECT DISTINCT canonical_govid FROM spending_long WHERE year = 2011 LIMIT 3"
+    )$canonical_govid
+    expect_gte(length(ids), 3L)
+
+    roll <- suppressMessages(cog_geographic_rollup(
+      list(city = ids), category = NULL, years = 2011L))
+    expect_equal(attr(roll, "provenance")$basis, "raw")
+    cov <- attr(roll, "provenance")$coverage
+    expect_false(is.na(cov$n_units_collected))
+    expect_equal(cov$n_units_collected, length(ids))
+
+    cmp <- suppressMessages(cog_peer_compare(
+      target_govid = ids[1], peers = ids[-1], category = NULL, years = 2011L))
+    expect_equal(attr(cmp, "provenance")$basis, "raw")
+    cov_peers <- attr(cmp, "provenance")$coverage
+    expect_false(is.na(cov_peers$n_units_collected))
+  })
+})

tests/testthat/test-duckdb-limits.R

@@ -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))
+})

tests/testthat/test-expenditure-concept.R

@@ -315,15 +315,17 @@ test_that("a mis-scoped cog_spending() call never attaches an M/L counterpart to
   # (SB194, cog_pipeline#64), so the recipe stopped being a candidate there.
   # FL state carries a real FY2011 B47 amount, so this exercises the guard
   # against a suggestion that genuinely fires.
+  #
+  # Issue #34: "IG Federal" maps to B-prefixed codes in summary_categories
+  # with category_type = 'revenue'. A spending verb (flow_prefixes E/F/G)
+  # now scopes its candidate query by category_type = 'expenditure', so it
+  # correctly finds NO candidates for this revenue-only category -- the
+  # suggestion machinery cannot fire, and no M/L counterpart is attached.
   r <- suppressMessages(
     cog_spending("120000226351", years = c(2005, 2011), category = "IG Federal")
   )
   sugg <- attr(r, "provenance")$suggestions
-  expect_gt(length(sugg), 0L)
-  ids <- vapply(sugg, function(s) s$recipe_id %||% "", character(1))
-  expect_true("ig_federal_b47_wide" %in% ids)
-  ig <- unlist(lapply(sugg, function(s) s$ig_recipe_id))
-  expect_length(ig, 0L)
+  expect_length(sugg, 0L)
 })
 
 test_that("C1: 'total' on a legacy aggregate-only family reports the IG-only figure honestly, not as Direct + IG", {

tests/testthat/test-explain.R

@@ -118,3 +118,17 @@ test_that("cog_explain prints denominator + popyear_range + counts", {
     expect_false(grepl("popyear range: 19-20", out, fixed = TRUE))
   })
 })
+
+test_that("cog_explain reports units collected alongside units reporting (uscogdata#36)", {
+  skip_if_no_corpus()
+  wi <- cog_gov_search(name = NULL, state = "WI", type = "city")
+  roll <- suppressMessages(cog_geographic_rollup(
+    govids = list(city = wi$canonical_govid), category = "Police",
+    years = 2012L))
+  out <- paste(c(
+    capture.output(cog_explain(roll)),
+    capture.output(cog_explain(roll), type = "message")
+  ), collapse = "\n")
+  expect_true(grepl("597 of 608 units collected", out, fixed = TRUE))
+  expect_true(grepl("485 reporting in this category", out, fixed = TRUE))
+})

tests/testthat/test-long-files.R

@@ -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))
+})

tests/testthat/test-recipes.R

@@ -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)
@@ -371,34 +371,45 @@ test_that("uscogdata#9: the revenue verb inherits the same trigger", {
   expect_null(sugg[[1]]$ig_recipe_id)
 })
 
-test_that("I1: cog_revenue never fabricates suppressed dollars for an expenditure-only recipe", {
+test_that("I1 + #34: cog_revenue never suggests expenditure-only recipes", {
   # uscogdata#9 review, finding I1: Corrections is an expenditure-only
-  # category (E04/E05). cog_revenue() naturally returns zero rows for it, so
-  # corrections_combined still fires as an empty_year suggestion (its own
-  # generic join finds real E04/E05 data for this government) -- but before
-  # the flow_prefixes fix, .suppressed_components() measured E04/E05 against
-  # cog_revenue()'s OWN view (which can never contain an E-coded row by
-  # construction) and reported the full $3,631,945,000 as "suppressed",
-  # when cog_spending() for the same gov/years/category actually returns
-  # $3,691,029,000 -- nothing was suppressed at all.
+  # category (E04/E05). Before the flow_prefixes fix (#9), .suppressed_components()
+  # measured E04/E05 against cog_revenue()'s OWN view and reported $3.6B as
+  # "suppressed" -- nothing was suppressed at all.
+  #
+  # Issue #34 builds on that: the candidate query now also filters by
+  # category_type ('revenue'), so expenditure-only recipes like corrections_combined
+  # (whose components E04/E05 are classified as 'expenditure' in summary_categories)
+  # are never even considered for a revenue verb. This is stronger than just
+  # suppressing the dollar claim -- it prevents the suggestion from firing at all.
   skip_if_no_corpus()
   r <- suppressMessages(
     cog_revenue("061037123085", years = 2019:2020, category = "Corrections"))
   sugg <- attr(r, "provenance")$suggestions
-  ids <- vapply(sugg, function(s) s$recipe_id, character(1))
-  expect_true("corrections_combined" %in% ids)
+  ids <- vapply(sugg, function(s) s$recipe_id %||% "", character(1))
 
-  hit <- sugg[[which(ids == "corrections_combined")]]
-  expect_equal(hit$suppressed_amount, 0)
-  expect_equal(hit$suppressed_years, integer(0))
-  expect_equal(hit$suppressed_codes, character(0))
+  # corrections_combined should NOT appear -- its components are expenditure-only.
+  expect_false("corrections_combined" %in% ids)
+})
 
-  # And cog_spending() for the identical gov/years/category is unaffected --
-  # it actually finds the E04/E05 dollars the buggy measurement claimed were
-  # excluded.
-  sp <- suppressMessages(
-    cog_spending("061037123085", years = 2019:2020, category = "Corrections"))
-  expect_equal(sum(sp$amt_nominal), 3691029000)
+test_that(".query_candidate_recipes() scopes candidates by category_type (#34)", {
+  # Direct assertion on the mechanism the two tests above exercise
+  # end-to-end: corrections_combined's own components (E04/E05) are
+  # category_type = 'expenditure' in summary_categories, so an
+  # expenditure-flavored flow_prefixes call must surface it and a
+  # revenue-flavored one must not. This queries only summary_categories/
+  # harmonization_recipes (no government data), so it runs against the
+  # bundled fixture with no skip_if_no_corpus() needed.
+  con <- uscogdata:::.ensure_session()
+
+  expenditure <- uscogdata:::.query_candidate_recipes(
+    con, category = "Corrections", flow_prefixes = c("E", "F", "G"))
+  expect_true("corrections_combined" %in% expenditure)
+
+  revenue <- uscogdata:::.query_candidate_recipes(
+    con, category = "Corrections",
+    flow_prefixes = c("T", "A", "U", "B", "C", "D"))
+  expect_false("corrections_combined" %in% revenue)
 })
 
 test_that("uscogdata#9: no partial-coverage fire in a modern year", {

tests/testthat/test-search-balances-pagination.R

@@ -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))))
+})