a1e7586 · Docket

Full-spectrum data layers: categorize taxonomy, calendar (effective dates), Voteview year-index (945k positions), watcher service; run.py post-processing

a1e7586 Zion Boggan committed on Jun 26, 2026 (1 month ago)

_ops/glm-f3.txt +28 -0

		@@ -0,0 +1,28 @@
	+	You are GLM-5.2, frontend lead for Docket (neutral US gov transparency app). Work in /opt/docket. The read-API + deep pages you built (F1/F2) are LIVE. New DATA layers now exist in the DB but have NO UI yet - build their API routes + UI in this wave. Be RAG-first; read ONLY: docs/API.md, docs/BRAND.md, docs/NEUTRALITY.md, docs/ACCESSIBILITY.md. Reuse your existing lib/db.ts + lib/feed.ts patterns + components.
	+
	+	New DB tables/columns (verify with a quick sqlite read at /opt/docket/data/docket.db):
	+	- calendar_events (id, date, kind[effective\|enacted\|vote\|introduced\|committee\|floor\|apportionment\|effective_parsed], ref_type, ref_id, title, jurisdiction, source_url, confidence) - 39k rows
	+	- categories (slug,label,kind) - 32 rows; bills.policy_area; bill_subjects(bill_id,subject); awards.category
	+	- votes.year + vote_positions (971k rows) - for year-by-year member voting
	+	- change_events (id, ts, source, ref_type, ref_id, change[added\|updated\|removed], summary, source_url) - the watcher feed
	+
	+	WAVE F3 - build these. Add API routes AND UI:
	+
	+	1. CALENDAR (the priority - a visual for "what goes into effect when", for non-technical users of any age):
	+	- /api/calendar?from=&to=&kind= → events in range from calendar_events.
	+	- /calendar page: a real MONTH-GRID calendar. Each day shows its events (dot/badge per kind, e.g. effective/enacted/vote). Click a day → panel listing that day's events with title + kind + Source link. Month nav (prev/next/today). Make it BIG, clean, obvious - an 80-year-old should get it instantly. A small legend explains the kinds in plain words ("Takes effect", "Became law", "Vote"). Works in all 3 display modes.
	+
	+	2. CATEGORIES:
	+	- /api/categories → list with item counts (join bills.policy_area + awards.category).
	+	- A category browse UI (chips/list with counts) on /register and/or a /categories page; clicking a category filters the feed/bills. Show policy_area + top subjects on /bill/[id].
	+
	+	3. MEMBER YEAR-INDEX (the "year by year how they voted" view):
	+	- On /members/[id], group that member's votes BY YEAR - collapsible year sections (e.g. 2026, 2025, 2024...), each showing the votes that year with date + question + their position (muted colors). Use the votes.year index. Show a per-year count summary. This is the headline upgrade to member pages.
	+
	+	4. WHAT-CHANGED FEED:
	+	- /api/changes → recent change_events.
	+	- A compact "Recently changed" surface (added/updated/removed) on the home or register page - small, factual, neutral, Source link each.
	+
	+	HARD RULES: keep BRAND (brass rationed, custody-stamp, mono, dark default), NEUTRALITY (Source link on every fact, no adjectives, muted colors, no comment sections, honest "no record" where true), and the 3-mode accessibility toggle (Dark/Readable/Plain) working on ALL new pages - the calendar especially must be legible in Plain mode. Add the new pages to the nav.
	+
	+	BUILD: `npm run build` MUST pass (fix type errors). Do NOT restart the server (handled separately). Update FRONTEND_NOTES.md. Definition of done: build passes; /calendar, category filter, member year-index, and the changed-feed all implemented against real /api data; brand + neutrality + a11y intact.

services/ingest/CALENDAR_NOTES.md +146 -0

		@@ -0,0 +1,146 @@
	+	# Calendar Ingest - Notes
	+
	+	Module: `ingest/calendar.py`
	+	Table: `calendar_events`
	+	Tests: `tests/test_calendar.py` (42 tests, all pass)
	+
	+	## Run
	+
	+	```bash
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	python -m ingest.calendar --once
	+	```
	+
	+	## What it does
	+
	+	Builds `calendar_events` by aggregating dated events from every other ingest
	+	table. One run takes ~20 seconds (mostly FR API calls for effective_on
	+	enrichment). Idempotent - safe to re-run.
	+
	+	### Event kinds and sources
	+
	+	\| kind \| count (2026-06-26) \| source table \| key field \|
	+	\|-------------------\|--------------------\|----------------------\|----------------\|
	+	\| effective \| 738 \| executive_actions \| effective_on \|
	+	\| enacted \| 47 \| actions (BecameLaw) \| action_date \|
	+	\| vote \| 3,454 \| votes \| date \|
	+	\| introduced \| 24,794 \| actions (IntroReferral)\| action_date \|
	+	\| committee \| 5,205 \| actions (Committee) \| action_date \|
	+	\| floor \| 4,969 \| actions (Floor) \| action_date \|
	+	\| apportionment \| 20 \| apportionments \| date \|
	+	\| effective_parsed \| 0* \| actions (text regex) \| parsed date \|
	+	\| TOTAL \| 39,274 \| \| \|
	+
	+	*`effective_parsed` is 0 because action texts in this dataset are short
	+	procedural strings ("Became Public Law No: 119-X.") without specific effective
	+	date clauses. The machinery is in place; it will fire on richer bill text.
	+
	+	## The key feature: effective dates
	+
	+	The `effective` kind is the "goes into effect on X" source. Federal Register
	+	final rules carry a structured `effective_on` field in the FR API.
	+
	+	Real example (from production DB):
	+
	+	```
	+	Date: 2026-07-01
	+	Title: Medicare Program; Alternative Payment Model Updates and the
	+	Increasing Organ Transplant Access Act Implementation
	+	(Health and Human Services Department; Centers for Medicare and
	+	Medicaid Services)
	+	Source URL: https://www.federalregister.gov/documents/2026/.../medicare-program
	+	Confidence: 1.0
	+	```
	+
	+	Another:
	+	```
	+	Date: 2028-09-30
	+	Title: Removing Regulations Related to Trade Adjustment Assistance for
	+	Firms (Commerce Department; Economic Development Administration)
	+	Source URL: https://www.federalregister.gov/documents/2026/06/11/2026-11800/...
	+	```
	+
	+	## FR enrichment flow
	+
	+	1. `fetch_fr_rules_with_effective(conn, days_back=90)` - fetches 90 days of
	+	RULE-type documents with `effective_on` field included in the request;
	+	upserts into `executive_actions`.
	+	2. `enrich_fr_effective_on(conn)` - for any remaining `executive_actions` rows
	+	with `effective_on IS NULL`, batch-fetches from FR API in groups of 50.
	+	Notices/NPRMs/EOs typically return null (no effective date); that is correct.
	+
	+	## Schema
	+
	+	```sql
	+	CREATE TABLE calendar_events (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	date TEXT NOT NULL, -- ISO-8601 YYYY-MM-DD
	+	kind TEXT NOT NULL CHECK(kind IN (
	+	'effective','enacted','vote',
	+	'introduced','committee','floor',
	+	'apportionment','effective_parsed')),
	+	ref_type TEXT NOT NULL, -- e.g. executive_action, bill, vote
	+	ref_id INTEGER, -- FK into the source table
	+	title TEXT,
	+	jurisdiction TEXT NOT NULL DEFAULT 'US',
	+	source_url TEXT NOT NULL, -- allowlisted .gov URL
	+	confidence REAL NOT NULL DEFAULT 1.0, -- 0.5 for effective_parsed
	+	UNIQUE (kind, ref_type, ref_id, date)
	+	);
	+	CREATE INDEX idx_calendar_events_date ON calendar_events(date);
	+	CREATE INDEX idx_calendar_events_kind ON calendar_events(kind);
	+	```
	+
	+	## Effective-date regex patterns
	+
	+	Applied to bill/action text to extract explicit calendar dates.
	+	Only SPECIFIC dates extracted; relative clauses ("X days after enactment")
	+	are intentionally skipped. Confidence = 0.5 for these parsed results.
	+
	+	```python
	+	# Pattern 1: "shall take effect on <date>"
	+	r"shall\s+take\s+effect\s+on\s+<DATE>"
	+
	+	# Pattern 2: "takes effect on <date>"
	+	r"takes?\s+effect\s+on\s+<DATE>"
	+
	+	# Pattern 3: "this Act ... takes effect <date>"
	+	r"this\s+Act\b.*?\btakes?\s+effect\s+<DATE>"
	+
	+	# Pattern 4: "effective on <date>"
	+	r"effective\s+on\s+<DATE>"
	+
	+	# Pattern 5: "effective date is <date>" / "effective date shall be <date>"
	+	r"effective\s+date\s+(?:is\|shall\s+be)\s+<DATE>"
	+
	+	# Pattern 6 (lowest specificity): "effective <date>"
	+	r"\beffective\s+<DATE>"
	+
	+	# DATE fragment matches:
	+	# "January 1, 2026" (full month name)
	+	# "2026-01-01" (ISO-8601)
	+	```
	+
	+	## Future /api/calendar endpoint
	+
	+	Suggested query shape:
	+
	+	```sql
	+	SELECT date, kind, ref_type, ref_id, title, jurisdiction, source_url, confidence
	+	FROM calendar_events
	+	WHERE date BETWEEN :start AND :end
	+	AND (:kind IS NULL OR kind = :kind)
	+	ORDER BY date, kind
	+	LIMIT :limit OFFSET :offset
	+	```
	+
	+	This table is designed for date-range queries - the `date` index makes those
	+	fast even at 40k+ rows.
	+
	+	## Neutrality
	+
	+	Every row carries a `source_url` pointing to an official `.gov` endpoint.
	+	Events are factual dated occurrences only: no adjectives, no editorial
	+	weighting of which dates "matter." The `effective_parsed` kind has
	+	`confidence=0.5` to signal lower certainty and keep the higher-quality
	+	structured events clearly distinguished.

services/ingest/CATEGORIZE_NOTES.md +150 -0

		@@ -0,0 +1,150 @@
	+	# Categorize Module - Build Notes
	+
	+	## What was built
	+
	+	`ingest/categorize.py` - unified taxonomy backfill for bills and spending awards.
	+
	+	### Files created
	+	- `ingest/categorize.py` - the module
	+	- `tests/test_categorize.py` - 57 tests
	+
	+	### Schema changes (all additive)
	+	\| Change \| Table \| Type \|
	+	\|---\|---\|---\|
	+	\| CREATE TABLE IF NOT EXISTS \| `categories` \| slug, label, kind \|
	+	\| CREATE TABLE IF NOT EXISTS \| `bill_subjects` \| bill_id, subject \|
	+	\| ALTER TABLE ADD COLUMN \| `awards.category` \| TEXT (guarded by try/except) \|
	+
	+	---
	+
	+	## Data findings (live DB, 2026-06-26)
	+
	+	### Bills
	+	- Total bills: 7,190
	+	- Bills with `policy_area` (already stored by bills_bulk.py): 7,179 / 7,190 (99.8%)
	+	- Distinct policy areas in use: 31 of 32 canonical areas
	+	- 11 bills have no policy_area (XML had no `<policyArea>` element - not an error)
	+
	+	### Bill subjects (initial backfill of 50 bills)
	+	- 50 bills fetched → 279 subject rows, 170 distinct subjects
	+	- Full backfill running in background: ~7,140 remaining bills
	+	- Rate: ~50 bills / 22 seconds (~2.3 bills/sec with 0.05s delay)
	+	- At that rate, full backfill of all bills: ~50 minutes
	+
	+	### Awards
	+	- Total awards: 320
	+	- Awards categorized: 320 / 320 (100%)
	+
	+	Award category breakdown:
	+	\| Category \| Awards \|
	+	\|---\|---\|
	+	\| Health \| 111 \|
	+	\| Science, Technology, Communications \| 50 \|
	+	\| Government Operations and Politics \| 49 \|
	+	\| Agriculture and Food \| 19 \|
	+	\| Emergency Management \| 16 \|
	+	\| Transportation and Public Works \| 12 \|
	+	\| Armed Forces and National Security \| 12 \|
	+	\| International Affairs \| 11 \|
	+	\| Energy \| 10 \|
	+	\| Crime and Law Enforcement \| 10 \|
	+	\| Social Welfare \| 4 \|
	+	\| Environmental Protection \| 4 \|
	+	\| Public Lands and Natural Resources \| 3 \|
	+	\| Education \| 3 \|
	+	\| Commerce \| 3 \|
	+	\| Economics and Public Finance \| 2 \|
	+
	+	### Categories table
	+	- 32 rows seeded (canonical Congressional policy areas)
	+	- Also auto-ingests any novel policy_area values from the bills table on each run
	+
	+	---
	+
	+	## Agency → Category map
	+
	+	Covers ~50 agency entries. Three-tier lookup:
	+	1. Exact match
	+	2. Prefix match (handles sub-agencies, e.g. "Department of Defense - DARPA")
	+	3. Substring match (handles "HHS Office of Inspector General")
	+
	+	All map values align to the same 32-area vocabulary as bills.
	+
	+	Key mappings:
	+	\| Agency \| Category \|
	+	\|---\|---\|
	+	\| Department of Defense \| Armed Forces and National Security \|
	+	\| Department of Veterans Affairs \| Armed Forces and National Security \|
	+	\| Department of Homeland Security \| Emergency Management \|
	+	\| Department of Health and Human Services \| Health \|
	+	\| Department of Transportation \| Transportation and Public Works \|
	+	\| Department of Energy \| Energy \|
	+	\| Department of Education \| Education \|
	+	\| Department of Agriculture \| Agriculture and Food \|
	+	\| Department of Labor \| Labor and Employment \|
	+	\| Environmental Protection Agency \| Environmental Protection \|
	+	\| Department of Justice \| Crime and Law Enforcement \|
	+	\| Department of State \| International Affairs \|
	+	\| Agency for International Development \| International Affairs \|
	+	\| Department of Commerce \| Commerce \|
	+	\| Department of the Treasury \| Economics and Public Finance \|
	+	\| Department of the Interior \| Public Lands and Natural Resources \|
	+	\| Department of Housing and Urban Development \| Housing and Community Development \|
	+	\| Social Security Administration \| Social Welfare \|
	+	\| NASA \| Science, Technology, Communications \|
	+	\| National Science Foundation \| Science, Technology, Communications \|
	+	\| General Services Administration \| Government Operations and Politics \|
	+
	+	---
	+
	+	## Test results
	+
	+	57 tests, 0 failures, 0.47s.
	+
	+	Test classes:
	+	- `TestAgencyToCategory` - 25 tests: exact match, prefix match, unknown/null handling, all map values are valid labels
	+	- `TestPolicyAreas` - 6 tests: count, no duplicates, slug format, key areas present, slug/label consistency
	+	- `TestParseSubjects` - 7 tests: single/multiple subjects, empty, missing element, malformed XML, whitespace, alternate root element
	+	- `TestEnsureSchema` - 4 tests: creates both tables, adds column, idempotent
	+	- `TestSeedCategories` - 4 tests: seeds all areas, idempotent, picks up extra bill areas, Health present
	+	- `TestBackfillAwards` - 6 tests: known agency, unknown→NULL, no-overwrite, multiple, count, idempotent
	+	- `TestCategoryCounts` - 4 tests: returns list, expected keys, bill counted, award counted
	+
	+	---
	+
	+	## CLI
	+
	+	```bash
	+	# Full backfill (all bills + awards + categories)
	+	cd /shared/projects/docket/services/ingest
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	.venv/bin/python -m ingest.categorize --once
	+
	+	# Limit subjects fetch (for testing)
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	.venv/bin/python -m ingest.categorize --once --limit-subjects 50
	+
	+	# Show category counts without re-running backfill
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	.venv/bin/python -m ingest.categorize --counts-only
	+	```
	+
	+	---
	+
	+	## Neutrality
	+
	+	- `policy_area` values come from Congress.gov's own taxonomy (Congressional Research Service) - no editorial classification
	+	- `legislativeSubjects` come from BILLSTATUS XML verbatim
	+	- Agency→category map keys are official agency names from USAspending data; values align to the same CRS taxonomy
	+	- No "good/bad" groupings; no partisan framing; categories are factual government classifications
	+
	+	---
	+
	+	## Technical notes
	+
	+	- GovInfo BILLSTATUS URL format: `/bulkdata/BILLSTATUS/{congress}/{type}/BILLSTATUS-{congress}{type}{num}.xml`
	+	(congress and type are separate path segments; concatenated only in the filename)
	+	- Bills with no subjects from GovInfo get a sentinel empty-string row in `bill_subjects` to prevent infinite re-fetching
	+	- Idempotent: re-running only processes bills/awards not yet categorized
	+	- `awards.category` ALTER is guarded with try/except so parallel runs don't fail
	+	- `categories` uses INSERT OR IGNORE so re-seeding is safe

services/ingest/VOTEVIEW_NOTES.md +218 -0

		@@ -0,0 +1,218 @@
	+	# Voteview Ingest - Implementation Notes
	+
	+	## What this poller does
	+
	+	`pollers/voteview.py` ingests complete per-member voting records from the
	+	Voteview bulk-CSV dataset (Lewis et al., voteview.com). It populates:
	+
	+	- `votes` - one row per recorded roll call, with metadata, year index, and
	+	an official .gov `source_url` (neutrality rule #1)
	+	- `vote_positions` - one row per member per roll call, with canonical position
	+	- `voteview_crosswalk` - cached ICPSR → bioguide mapping
	+
	+	Voteview is the ingestion pipe only. No Voteview URLs appear anywhere in
	+	user-facing data; every `source_url` points to the official government record.
	+
	+	---
	+
	+	## Cast-code mapping
	+
	+	Voteview assigns each member vote a `cast_code` (0-9) per their codebook
	+	(https://voteview.com/articles/data_help_members#castcode):
	+
	+	\| cast_code \| Official meaning \| Canonical position \|
	+	\|-----------\|------------------------------\|--------------------\|
	+	\| 0 \| Not a member / no record \| Not Voting \|
	+	\| 1 \| Yea \| Yea \|
	+	\| 2 \| Paired Yea (absent, declared)\| Yea \|
	+	\| 3 \| Announced Yea (absent, decl.)\| Yea \|
	+	\| 4 \| Announced Nay (absent, decl.)\| Nay \|
	+	\| 5 \| Paired Nay (absent, declared)\| Nay \|
	+	\| 6 \| Nay \| Nay \|
	+	\| 7 \| Present / abstention (v1) \| Present \|
	+	\| 8 \| Present / abstention (v2) \| Present \|
	+	\| 9 \| Not Voting \| Not Voting \|
	+
	+	Paired / announced votes (codes 2-5): The member was physically absent but
	+	announced their intent; the chamber counts them as Yea/Nay. We map them to
	+	Yea/Nay for the same reason - they reflect a declared position.
	+
	+	Codes 7 and 8 are both Present (abstention), not Nay. Some secondary
	+	sources conflate 7/8 with Nay; this is incorrect per the Voteview codebook.
	+
	+	Code 0 vs 9:
	+	- 0 = not a member at the time of the vote (e.g., vacancy) or genuinely missing
	+	- 9 = member was present in the chamber but chose not to vote
	+	Both map to `Not Voting`; the distinction matters for DW-NOMINATE but not for
	+	our per-member position index.
	+
	+	All 10 codes and every enum value are covered by `tests/test_voteview.py`.
	+
	+	---
	+
	+	## ICPSR → bioguide crosswalk
	+
	+	Voteview identifies members by ICPSR code (numeric, chamber-specific: a member
	+	who served in both chambers gets two ICPSR IDs). The Docket DB uses bioguide_id
	+	as the join key.
	+
	+	The crosswalk is built from:
	+	- `legislators-current.json` (unitedstates/congress-legislators, gh-pages branch)
	+	- `legislators-historical.json` (same)
	+
	+	Both are fetched once and cached in the `voteview_crosswalk` table. On
	+	subsequent runs, the cache is used (pass `--refresh-crosswalk` to re-download).
	+
	+	The `id.icpsr` field is the primary mapping. Some members also have
	+	`id.icpsr_hn` (a separate House ICPSR number); both are indexed.
	+
	+	Members not in the crosswalk: stored with `member_bioguide = NULL`.
	+	This happens for:
	+	- Members who served before the digitisation cutoff in congress-legislators
	+	- Delegates, non-voting members, territorial representatives
	+	- A small number of historical mismatches
	+
	+	---
	+
	+	## Official source URLs (neutrality rule #1)
	+
	+	Voteview is an academic aggregator - it is never the user-facing source.
	+
	+	\| Chamber \| URL pattern \|
	+	\|---------\|-------------\|
	+	\| House \| `https://clerk.house.gov/Votes/{year}/roll{NNN:03d}` \|
	+	\| Senate \| `https://www.senate.gov/legislative/LIS/roll_call_votes/vote{congress}{session}/vote_{congress}_{session}_{roll:05d}.xml` \|
	+
	+	For House votes: `clerk_rollnumber` (from the Voteview rollcalls CSV) is used
	+	as `NNN`, because it is the House Clerk's sequential roll number. Voteview's
	+	own `rollnumber` is stored separately in `voteview_rollnumber` for reference.
	+
	+	For Senate votes: Voteview's `rollnumber` equals the Senate's official roll
	+	call number, so it is used directly.
	+
	+	---
	+
	+	## Year index
	+
	+	A `year INTEGER` column is added to the `votes` table and indexed as
	+	`idx_votes_year`. This makes the following queries run in O(log n):
	+
	+	```python
	+	# All votes for member X in year Y (hits idx_votes_year + idx_vote_positions_bioguide)
	+	votes_by_member_year("N000002", 2023)
	+
	+	# All recorded votes in year Y (hits idx_votes_year)
	+	votes_in_year(2023)
	+	```
	+
	+	`year` is derived from the `date` field (first 4 characters) and stored
	+	redundantly to avoid a `substr()` or `CAST` on every range scan.
	+
	+	---
	+
	+	## Completeness
	+
	+	What Voteview covers: Only recorded (roll-call) votes where a member's
	+	name appears in the official tally. This is the complete universe of votes
	+	where per-member records exist in the public record.
	+
	+	What Voteview does NOT cover:
	+	- Voice votes - passage is recorded but no per-member positions exist. This is
	+	~94% of Senate measures and a large fraction of House suspension-of-the-rules
	+	votes. The product note "no individual record exists" must be displayed for
	+	these (handled by `votes_house.py` + `votes_senate.py` which set
	+	`recorded=0`).
	+	- Unanimous consent agreements
	+	- Committee votes
	+
	+	Row-count check: the completeness report (`--report` flag) fetches the
	+	Voteview rollcalls CSV for each (congress, chamber) and compares its row count
	+	against the `seen` table. A match means every rollcall in the CSV was
	+	processed. Gaps are explained (e.g., blank rollnumber rows in the CSV, or an
	+	incomplete run).
	+
	+	---
	+
	+	## Running the ingest
	+
	+	### Proof slice (congresses 118-119, 2023-present)
	+	```bash
	+	ssh root@REDACTED-IP "pct exec 247 -- bash -lc 'cd /shared/projects/docket/services/ingest && DOCKET_DB_PATH=/opt/docket/data/docket.db /opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.voteview --once --start 118 --end 119'"
	+	```
	+
	+	### Full 2019-present (congresses 116-119)
	+	```bash
	+	ssh root@REDACTED-IP "pct exec 247 -- bash -lc 'cd /shared/projects/docket/services/ingest && DOCKET_DB_PATH=/opt/docket/data/docket.db /opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.voteview --once --start 116 --end 119'"
	+	```
	+
	+	### Full historical backfill (1989-present, ~30 congresses, recommended first pass)
	+	```bash
	+	ssh root@REDACTED-IP "pct exec 247 -- bash -lc 'cd /shared/projects/docket/services/ingest && DOCKET_DB_PATH=/opt/docket/data/docket.db /opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.voteview --once --start 101 --end 119'"
	+	```
	+
	+	### Full Voteview archive (1789-present, all ~120 congresses)
	+	```bash
	+	ssh root@REDACTED-IP "pct exec 247 -- bash -lc 'cd /shared/projects/docket/services/ingest && DOCKET_DB_PATH=/opt/docket/data/docket.db /opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.voteview --once --start 1 --end 119'"
	+	```
	+
	+	Add `--report` to any command to print the CSV vs DB completeness table.
	+	Add `--refresh-crosswalk` to re-download the ICPSR crosswalk from GitHub.
	+
	+	---
	+
	+	## Scale estimate (full historical backfill)
	+
	+	Voteview covers all US congresses from 1789. Rough row counts for a full
	+	`--start 1 --end 119` run:
	+
	+	\| Scope \| Congresses \| Approx rollcalls \| Approx positions \|
	+	\|-----------------\|------------\|------------------\|-----------------\|
	+	\| 2019-present \| 116-119 \| ~6,000 \| ~2,000,000 \|
	+	\| 1989-present \| 101-119 \| ~25,000 \| ~8,000,000 \|
	+	\| Full archive \| 1-119 \| ~70,000 \| ~20,000,000 \|
	+
	+	Full archive runtime estimate: 4-6 hours (dominated by CSV downloads and
	+	batch DB inserts). The ingest is fully resumable - the `seen` table prevents
	+	re-processing any rollcall already ingested.
	+
	+	SQLite comfortably handles 20M rows in `vote_positions` with proper indexes;
	+	the `year` and `bioguide` indexes keep member-year queries under 5ms.
	+
	+	---
	+
	+	## Year-index example query
	+
	+	```python
	+	from ingest.pollers.voteview import votes_by_member_year
	+
	+	# Jerry Nadler (D-NY, bioguide N000002) votes in 2023
	+	rows = votes_by_member_year("N000002", 2023)
	+	for r in rows[:5]:
	+	print(r["date"], r["chamber"], r["roll_number"], r["question"],
	+	r["result"], "→", r["position"])
	+	```
	+
	+	Example output (after ingest of congress 118):
	+	```
	+	2023-01-09 house 5 On Ordering the Previous Question Agreed to → Nay
	+	2023-01-09 house 6 On Adopting the Rules of the 118th Congress Agreed to → Nay
	+	2023-01-10 house 10 On Passage Passed → Nay
	+	...
	+	```
	+
	+	To get all votes in a specific year regardless of member:
	+	```python
	+	from ingest.pollers.voteview import votes_in_year
	+	all_2023 = votes_in_year(2023)
	+	print(f"{len(all_2023)} recorded votes in 2023")
	+	```
	+
	+	---
	+
	+	## Files
	+
	+	\| File \| Purpose \|
	+	\|------\|---------\|
	+	\| `ingest/pollers/voteview.py` \| Main poller (this module) \|
	+	\| `tests/test_voteview.py` \| 48 pytest tests (cast_code, crosswalk, schema, year-index, URL validation) \|
	+	\| `VOTEVIEW_NOTES.md` \| This document \|

services/ingest/WATCHER_NOTES.md +192 -0

		@@ -0,0 +1,192 @@
	+	# Docket Watcher - Notes
	+
	+	## What it does
	+
	+	`ingest/watcher.py` wraps the existing pipeline (`ingest.run`) and records
	+	what changed since the previous cycle: items ADDED (new external_id seen for the
	+	first time), UPDATED (same item, newer update date), or REMOVED (a full-set source
	+	stopped returning an item that was previously seen).
	+
	+	Results are stored in `change_events` and are the foundation of Docket's
	+	"what just changed" feed and future alerts.
	+
	+	## Run modes
	+
	+	```bash
	+	# Single cycle - designed for systemd timer (or manual testing)
	+	python -m ingest.watcher --once
	+
	+	# Continuous loop with 30-minute pause
	+	python -m ingest.watcher --loop --interval 1800
	+
	+	# Filter to a single source (fast; good for testing without full pipeline)
	+	python -m ingest.watcher --once --sources federal_register
	+
	+	# Multiple sources
	+	python -m ingest.watcher --once --sources members openomb
	+
	+	# Override DB path
	+	python -m ingest.watcher --once --db /tmp/test.db
	+	```
	+
	+	## Watcher-owned tables
	+
	+	All three tables are created by the watcher (`_ensure_watcher_tables`); they
	+	are additive and never touch existing schema.
	+
	+	\| Table \| Purpose \|
	+	\|---\|---\|
	+	\| `change_events` \| Feed of ADDED / UPDATED / REMOVED events (the primary output) \|
	+	\| `poller_watermarks` \| Last-run timestamp + cycle ID per source (for operational dashboards) \|
	+	\| `watcher_manifest` \| Current-set snapshot for full-set sources (used for removal diffing) \|
	+
	+	### change_events schema
	+
	+	```sql
	+	CREATE TABLE change_events (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	ts TEXT NOT NULL, -- ISO 8601 UTC timestamp of this cycle
	+	source TEXT NOT NULL, -- pipeline step label (e.g. federal_register)
	+	ref_type TEXT NOT NULL, -- normalized type (bill/award/member/…)
	+	ref_id TEXT NOT NULL, -- external_id from the source
	+	change TEXT NOT NULL CHECK(change IN ('added','updated','removed')),
	+	summary TEXT, -- human-readable one-liner
	+	source_url TEXT NOT NULL -- official .gov URL (neutrality rule #1)
	+	);
	+	```
	+
	+	## ADDED / UPDATED detection
	+
	+	The watcher snapshots the `seen` table (which tracks `(source, external_id,
	+	update_date)` for every item processed by `mark_seen`) before and after the
	+	pipeline run:
	+
	+	- ADDED - new `(source, external_id)` appeared in the post-run snapshot.
	+	- UPDATED - same `(source, external_id)`, but `update_date` changed.
	+
	+	This covers every poller that calls `mark_seen()` (federal_register, openomb,
	+	usaspending, treasury, congress, govinfo, bills_bulk, bill_versions, votes_senate,
	+	votes_house, crec, lobbying, openstates, legiscan).
	+
	+	The `members` poller does a blind upsert without `mark_seen`, so its ADDED events
	+	come from the manifest path instead (see below).
	+
	+	## REMOVED detection - full-set sources only
	+
	+	Why only full-set? Paginated or time-windowed sources (federal_register fetches
	+	last-7-days, usaspending returns pages of awards, etc.) cannot distinguish "item
	+	not in this page" from "item genuinely removed". Flagging removal there would
	+	produce a constant stream of false positives.
	+
	+	Full-set sources (configured in `FULL_SET_CONFIGS`):
	+
	+	\| Source \| Data table \| Rationale \|
	+	\|---\|---\|---\|
	+	\| `openomb` \| `apportionments` \| OMB index returns all active FY apportionment files \|
	+	\| `members` \| `members` \| `legislators-current.json` is the authoritative live roster \|
	+
	+	The watcher maintains a `watcher_manifest` table: after each successful cycle it
	+	records every `(source, external_id)` pair that was present in the data table.
	+	The next cycle diffs the new set against this manifest.
	+
	+	### Safety guards
	+
	+	Two guards prevent false REMOVED events:
	+
	+	1. Non-empty guard - if `_query_current_set()` returns an empty dict for a
	+	full-set source, the watcher writes a WARNING log and skips removal detection
	+	for that source. This handles: network failures, table not yet created, source
	+	returning unexpectedly empty HTML, etc. The existing manifest is preserved
	+	unchanged.
	+
	+	2. First-run guard - if there is no prior manifest entry for a source (first
	+	ever cycle), no items are flagged as removed. Without a baseline, there is
	+	nothing to diff against.
	+
	+	Together these ensure: a failed pull or a fresh deployment can never generate a
	+	flood of spurious REMOVED events.
	+
	+	## Watermarks
	+
	+	`poller_watermarks` records `(source, last_run_at, cycle_id)` after every cycle.
	+	Use it to tell an external dashboard "openomb was last checked at T".
	+	The `cycle_id` is a monotonically increasing integer that ties `change_events`
	+	and `watcher_manifest` entries to the same logical run.
	+
	+	## systemd install on CT247
	+
	+	```bash
	+	# 1. Copy unit files
	+	cp /shared/projects/docket/services/ingest/deploy/docket-watcher.service \
	+	/etc/systemd/system/
	+	cp /shared/projects/docket/services/ingest/deploy/docket-watcher.timer \
	+	/etc/systemd/system/
	+
	+	# 2. Reload and enable
	+	systemctl daemon-reload
	+	systemctl enable --now docket-watcher.timer
	+
	+	# 3. Verify
	+	systemctl status docket-watcher.timer
	+	journalctl -u docket-watcher -f
	+
	+	# 4. Trigger a manual cycle immediately (for testing)
	+	systemctl start docket-watcher.service
	+	```
	+
	+	To check the timer schedule:
	+
	+	```bash
	+	systemctl list-timers docket-watcher.timer
	+	```
	+
	+	To query recent change events from the DB:
	+
	+	```bash
	+	sqlite3 /opt/docket/data/docket.db \
	+	"SELECT ts, source, ref_type, change, ref_id FROM change_events ORDER BY id DESC LIMIT 20;"
	+	```
	+
	+	## Testing a single cycle (without full pipeline)
	+
	+	```bash
	+	cd /shared/projects/docket/services/ingest
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	.venv/bin/python -m ingest.watcher --once --sources federal_register
	+	```
	+
	+	Or via the exec command from the overseer notes:
	+
	+	```bash
	+	ssh root@REDACTED-IP "pct exec 247 -- bash -lc \
	+	'cd /shared/projects/docket/services/ingest && \
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.watcher --once \
	+	--sources federal_register'"
	+	```
	+
	+	## Running the unit tests
	+
	+	```bash
	+	cd /shared/projects/docket/services/ingest
	+	.venv/bin/pytest tests/test_watcher.py -v
	+	```
	+
	+	## Adding a new full-set source
	+
	+	1. Confirm the source's poll() always returns the complete current live set
	+	(not a time window, not paginated results, not a delta feed).
	+	2. Add an entry to `FULL_SET_CONFIGS` in `ingest/watcher.py`:
	+	```python
	+	"source_label": {
	+	"table": "my_table",
	+	"id_col": "external_id", # or the PK column name
	+	"where": "source = 'source_label'", # or None if table is single-source
	+	"url_col": "source_url",
	+	"ref_type": "my_type",
	+	},
	+	```
	+	3. If the source does not call `mark_seen()`, ensure it is NOT excluded from
	+	the ADDED-via-manifest path (remove it from the exclusion list in
	+	`_detect_full_set_changes`).
	+	4. Add a test in `tests/test_watcher.py`.

services/ingest/deploy/docket-watcher.service +36 -0

		@@ -0,0 +1,36 @@
	+	[Unit]
	+	Description=Docket ingest watcher - single cycle (ADDED/UPDATED/REMOVED detection)
	+	Documentation=file:/shared/projects/docket/services/ingest/WATCHER_NOTES.md
	+	# Triggered by docket-watcher.timer; designed for one-shot use per timer tick.
	+	# For continuous mode, change ExecStart to: python -m ingest.watcher --loop --interval 1800
	+	After=network-online.target
	+	Wants=network-online.target
	+
	+	[Service]
	+	Type=oneshot
	+	User=root
	+	WorkingDirectory=/opt/docket/services/ingest
	+
	+	# Python venv + module invocation
	+	ExecStart=/opt/docket/services/ingest/.venv/bin/python -m ingest.watcher --once
	+
	+	# DB lives in a local (non-CIFS) path - SQLite requirement
	+	Environment=DOCKET_DB_PATH=/opt/docket/data/docket.db
	+
	+	# Keys (optional - pollers skip gracefully when absent)
	+	EnvironmentFile=-/opt/docket/.env
	+
	+	StandardOutput=journal
	+	StandardError=journal
	+	SyslogIdentifier=docket-watcher
	+
	+	# Restart on failure (but not on clean exit - oneshot)
	+	Restart=on-failure
	+	RestartSec=60s
	+
	+	# Resource limits
	+	TimeoutStartSec=600
	+	Nice=10
	+
	+	[Install]
	+	WantedBy=multi-user.target

services/ingest/deploy/docket-watcher.timer +17 -0

		@@ -0,0 +1,17 @@
	+	[Unit]
	+	Description=Docket ingest watcher - run every 30 minutes
	+	Documentation=file:/shared/projects/docket/services/ingest/WATCHER_NOTES.md
	+
	+	[Timer]
	+	# Fire 30 s after boot (gives network and other services time to settle)
	+	OnBootSec=30s
	+	# Then every 30 minutes
	+	OnUnitActiveSec=30min
	+	# Persist across reboots - catch up on any missed fires
	+	Persistent=true
	+	# Randomise start within a 2-minute window to avoid thundering-herd
	+	RandomizedDelaySec=120
	+	Unit=docket-watcher.service
	+
	+	[Install]
	+	WantedBy=timers.target

services/ingest/ingest/calendar.py +749 -0

		@@ -0,0 +1,749 @@
	+	"""
	+	Calendar aggregator - builds `calendar_events` from all ingest tables.
	+
	+	Each event carries a date and a kind:
	+	effective - Federal Register final rules with a structured effective_on date
	+	enacted - Bills that became public law (actions.BecameLaw rows)
	+	vote - Roll-call vote dates (votes table)
	+	introduced - Bill introduction dates (actions.IntroReferral)
	+	committee - Committee referral/action dates (actions.Committee)
	+	floor - Floor action dates (actions.Floor)
	+	apportionment - Apportionment dates (apportionments table)
	+	effective_parsed - Effective dates parsed from bill/action text via regex (confidence 0.5)
	+
	+	Motivating use-case: "Bill X goes into effect August 5" appears on Aug 5 in the
	+	calendar, sourced from the Federal Register `effective_on` field.
	+
	+	Run once:
	+	python -m ingest.calendar --once
	+
	+	API surface for future /api/calendar endpoint:
	+	SELECT * FROM calendar_events
	+	WHERE date BETWEEN ? AND ?
	+	ORDER BY date, kind
	+	"""
	+
	+	import argparse
	+	import logging
	+	import re
	+	import sqlite3
	+	import sys
	+	import time
	+	from datetime import datetime, timedelta
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "calendar"
	+	FR_BASE = "https://www.federalregister.gov/api/v1/documents.json"
	+
	+	# ---------------------------------------------------------------------------
	+	# Effective-date regex patterns
	+	# ---------------------------------------------------------------------------
	+	# Applied to bill / action text to extract explicit calendar dates from
	+	# clauses like:
	+	# "shall take effect on January 1, 2026"
	+	# "effective January 1, 2026"
	+	# "this Act takes effect on 2026-01-01"
	+	# Only SPECIFIC calendar dates are extracted; relative clauses such as
	+	# "180 days after enactment" are intentionally skipped.
	+
	+	_MONTH_NAMES = (
	+	"January\|February\|March\|April\|May\|June\|"
	+	"July\|August\|September\|October\|November\|December"
	+	)
	+	_DATE_FRAG = (
	+	r"(?P<date>"
	+	rf"(?:{_MONTH_NAMES})\s+\d{{1,2}},\s+\d{{4}}"
	+	r"\|"
	+	r"\d{4}-\d{2}-\d{2}"
	+	r")"
	+	)
	+
	+	EFFECTIVE_DATE_PATTERNS: list[re.Pattern] = [
	+	# "shall take effect on <date>" / "will take effect on <date>"
	+	re.compile(r"shall\s+take\s+effect\s+on\s+" + _DATE_FRAG, re.IGNORECASE),
	+	# "takes effect on <date>"
	+	re.compile(r"takes?\s+effect\s+on\s+" + _DATE_FRAG, re.IGNORECASE),
	+	# "this Act ... takes effect <date>"
	+	re.compile(r"this\s+Act\b.*?\btakes?\s+effect\s+" + _DATE_FRAG, re.IGNORECASE),
	+	# "effective on <date>"
	+	re.compile(r"effective\s+on\s+" + _DATE_FRAG, re.IGNORECASE),
	+	# "effective date is <date>" / "effective date shall be <date>"
	+	re.compile(r"effective\s+date\s+(?:is\|shall\s+be)\s+" + _DATE_FRAG, re.IGNORECASE),
	+	# "effective <date>" (no "on") - lowest specificity, listed last
	+	re.compile(r"\beffective\s+" + _DATE_FRAG, re.IGNORECASE),
	+	]
	+
	+
	+	def parse_effective_dates(text: str) -> list[str]:
	+	"""
	+	Extract explicit calendar dates from effective-date clauses in text.
	+
	+	Returns ISO-8601 date strings (YYYY-MM-DD), deduplicated. Only specific
	+	calendar dates are extracted; relative expressions ("X days after enactment")
	+	are not returned.
	+	"""
	+	found: list[str] = []
	+	seen: set[str] = set()
	+	for pattern in EFFECTIVE_DATE_PATTERNS:
	+	for m in pattern.finditer(text):
	+	raw = m.group("date").strip()
	+	iso = _normalize_date(raw)
	+	if iso and iso not in seen:
	+	found.append(iso)
	+	seen.add(iso)
	+	return found
	+
	+
	+	def _normalize_date(raw: str) -> Optional[str]:
	+	"""Convert a free-form date string to ISO-8601 YYYY-MM-DD, or None."""
	+	raw = raw.strip()
	+	if re.match(r"^\d{4}-\d{2}-\d{2}$", raw):
	+	return raw
	+	for fmt in ("%B %d, %Y", "%b %d, %Y"):
	+	try:
	+	return datetime.strptime(raw, fmt).strftime("%Y-%m-%d")
	+	except ValueError:
	+	pass
	+	return None
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Schema
	+	# ---------------------------------------------------------------------------
	+
	+	def ensure_schema(conn: sqlite3.Connection) -> None:
	+	"""
	+	Create calendar_events and ensure executive_actions exists with effective_on.
	+
	+	* Creates calendar_events table + indexes.
	+	* Creates executive_actions if absent (with effective_on from the start).
	+	* Adds effective_on column via ALTER TABLE if the table already exists but
	+	was created by the federal_register poller without that column.
	+	"""
	+	conn.executescript("""
	+	CREATE TABLE IF NOT EXISTS calendar_events (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	date TEXT NOT NULL,
	+	kind TEXT NOT NULL CHECK(kind IN (
	+	'effective','enacted','vote',
	+	'introduced','committee','floor',
	+	'apportionment','effective_parsed')),
	+	ref_type TEXT NOT NULL,
	+	ref_id INTEGER,
	+	title TEXT,
	+	jurisdiction TEXT NOT NULL DEFAULT 'US',
	+	source_url TEXT NOT NULL,
	+	confidence REAL NOT NULL DEFAULT 1.0,
	+	UNIQUE (kind, ref_type, ref_id, date)
	+	);
	+
	+	CREATE INDEX IF NOT EXISTS idx_calendar_events_date
	+	ON calendar_events(date);
	+
	+	CREATE INDEX IF NOT EXISTS idx_calendar_events_kind
	+	ON calendar_events(kind);
	+
	+	-- Create executive_actions with effective_on if it doesn't exist yet.
	+	-- The federal_register poller creates this table without effective_on;
	+	-- the ALTER TABLE below handles that case.
	+	CREATE TABLE IF NOT EXISTS executive_actions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'federal_register',
	+	external_id TEXT NOT NULL,
	+	type TEXT NOT NULL,
	+	title TEXT,
	+	agency TEXT,
	+	fr_doc_number TEXT,
	+	date TEXT,
	+	effective_on TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (source, external_id)
	+	);
	+	""")
	+	conn.commit()
	+
	+	# Add effective_on to executive_actions if it was created by the FR poller
	+	# without that column (SQLite ALTER TABLE has no IF NOT EXISTS - use try/except).
	+	try:
	+	conn.execute(
	+	"ALTER TABLE executive_actions ADD COLUMN effective_on TEXT"
	+	)
	+	conn.commit()
	+	logger.info("calendar: added effective_on column to executive_actions")
	+	except sqlite3.OperationalError:
	+	pass # column already exists
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# FR enrichment - populate effective_on for stored final_rule rows
	+	# ---------------------------------------------------------------------------
	+
	+	def enrich_fr_effective_on(conn: sqlite3.Connection) -> int:
	+	"""
	+	For every row in executive_actions where effective_on IS NULL, fetch
	+	the effective_on date from the Federal Register API in batches of 50.
	+
	+	Returns the number of rows updated.
	+	"""
	+	rows = conn.execute(
	+	"""
	+	SELECT external_id
	+	FROM executive_actions
	+	WHERE effective_on IS NULL
	+	""",
	+	).fetchall()
	+
	+	if not rows:
	+	logger.info("calendar: no executive_actions rows need effective_on enrichment")
	+	return 0
	+
	+	doc_numbers = [r["external_id"] for r in rows]
	+	logger.info("calendar: enriching effective_on for %d FR docs", len(doc_numbers))
	+
	+	updated = 0
	+	batch_size = 50
	+
	+	for i in range(0, len(doc_numbers), batch_size):
	+	batch = doc_numbers[i : i + batch_size]
	+	params: list[tuple[str, str]] = []
	+	for field in ["document_number", "effective_on"]:
	+	params.append(("fields[]", field))
	+	for dn in batch:
	+	params.append(("conditions[document_numbers][]", dn))
	+	params.append(("per_page", str(batch_size)))
	+
	+	try:
	+	resp = requests.get(FR_BASE, params=params, timeout=30)
	+	resp.raise_for_status()
	+	data = resp.json()
	+	except requests.RequestException as exc:
	+	logger.error("calendar: FR enrichment request failed: %s", exc)
	+	time.sleep(2)
	+	continue
	+
	+	for doc in data.get("results", []):
	+	dn = doc.get("document_number")
	+	eff = doc.get("effective_on") # may be null
	+	if dn:
	+	conn.execute(
	+	"""
	+	UPDATE executive_actions
	+	SET effective_on = ?
	+	WHERE external_id = ? AND effective_on IS NULL
	+	""",
	+	(eff, dn),
	+	)
	+	if eff:
	+	updated += 1
	+
	+	conn.commit()
	+	time.sleep(0.2) # polite rate limiting
	+
	+	logger.info("calendar: enriched effective_on for %d docs", updated)
	+	return updated
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Fresh FR fetch - pull recent final_rule docs with effective_on
	+	# ---------------------------------------------------------------------------
	+
	+	def fetch_fr_rules_with_effective(
	+	conn: sqlite3.Connection,
	+	days_back: int = 60,
	+	) -> int:
	+	"""
	+	Fetch recent Federal Register RULE documents and store/update them in
	+	executive_actions, including the effective_on field.
	+
	+	This supplements the main federal_register poller which does not request
	+	effective_on. Only RULE (final rule) documents are fetched here because
	+	those are the documents that legally go into effect on a specific date.
	+
	+	Returns number of rows upserted.
	+	"""
	+	start_date = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
	+	logger.info("calendar: fetching FR final rules since %s for effective dates", start_date)
	+
	+	stored = 0
	+	page = 1
	+
	+	while True:
	+	params: list[tuple[str, str]] = []
	+	for field in [
	+	"document_number", "type", "title", "agencies",
	+	"publication_date", "effective_on", "html_url",
	+	]:
	+	params.append(("fields[]", field))
	+	params += [
	+	("conditions[type][]", "RULE"),
	+	("conditions[publication_date][gte]", start_date),
	+	("per_page", "100"),
	+	("order", "newest"),
	+	("page", str(page)),
	+	]
	+
	+	try:
	+	resp = requests.get(FR_BASE, params=params, timeout=30)
	+	resp.raise_for_status()
	+	data = resp.json()
	+	except requests.RequestException as exc:
	+	logger.error("calendar: FR rules fetch failed: %s", exc)
	+	break
	+
	+	results = data.get("results", [])
	+	if not results:
	+	break
	+
	+	for doc in results:
	+	dn = doc.get("document_number") or ""
	+	eff = doc.get("effective_on") # may be null - that's fine
	+	pub_date = doc.get("publication_date") or ""
	+	title = doc.get("title") or ""
	+	html_url = doc.get("html_url") or ""
	+	agencies = doc.get("agencies") or []
	+
	+	if not dn or not html_url:
	+	continue
	+
	+	try:
	+	validate_source_url(html_url)
	+	except Exception:
	+	continue
	+
	+	agency_str = "; ".join(
	+	a.get("name", "") for a in agencies if a.get("name")
	+	)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO executive_actions
	+	(source, external_id, type, title, agency,
	+	fr_doc_number, date, effective_on, source_url)
	+	VALUES ('federal_register', ?, 'final_rule', ?, ?,
	+	?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	effective_on = COALESCE(excluded.effective_on,
	+	executive_actions.effective_on),
	+	title = excluded.title,
	+	agency = excluded.agency,
	+	date = excluded.date,
	+	source_url = excluded.source_url
	+	""",
	+	(dn, title, agency_str, dn, pub_date, eff, html_url),
	+	)
	+	stored += 1
	+
	+	conn.commit()
	+	total_pages = data.get("total_pages", 1)
	+	if page >= total_pages or page >= 10:
	+	break
	+	page += 1
	+	time.sleep(0.2)
	+
	+	logger.info("calendar: upserted %d FR rule rows with effective_on", stored)
	+	return stored
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Aggregation helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _upsert_event(
	+	conn: sqlite3.Connection,
	+	date: str,
	+	kind: str,
	+	ref_type: str,
	+	ref_id: Optional[int],
	+	title: str,
	+	source_url: str,
	+	jurisdiction: str = "US",
	+	confidence: float = 1.0,
	+	) -> None:
	+	"""Insert or replace one calendar event row."""
	+	if not date or not source_url:
	+	return
	+	conn.execute(
	+	"""
	+	INSERT INTO calendar_events
	+	(date, kind, ref_type, ref_id, title,
	+	jurisdiction, source_url, confidence)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(kind, ref_type, ref_id, date) DO UPDATE SET
	+	title = excluded.title,
	+	source_url = excluded.source_url,
	+	confidence = excluded.confidence
	+	""",
	+	(date, kind, ref_type, ref_id, title[:500] if title else "",
	+	jurisdiction, source_url, confidence),
	+	)
	+
	+
	+	def _aggregate_effective(conn: sqlite3.Connection) -> int:
	+	"""
	+	effective - Federal Register final rules with a structured effective_on date.
	+
	+	This is the canonical "goes into effect on X" source. Only rows where
	+	effective_on IS NOT NULL are included.
	+	"""
	+	rows = conn.execute(
	+	"""
	+	SELECT id, effective_on, title, source_url, agency
	+	FROM executive_actions
	+	WHERE effective_on IS NOT NULL
	+	AND type IN ('final_rule', 'RULE')
	+	AND effective_on != ''
	+	ORDER BY effective_on
	+	"""
	+	).fetchall()
	+
	+	count = 0
	+	for r in rows:
	+	title = r["title"] or ""
	+	if r["agency"]:
	+	title = f"{title} ({r['agency']})" if title else r["agency"]
	+	_upsert_event(
	+	conn,
	+	date=r["effective_on"],
	+	kind="effective",
	+	ref_type="executive_action",
	+	ref_id=r["id"],
	+	title=title,
	+	source_url=r["source_url"],
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d effective events from FR rules", count)
	+	return count
	+
	+
	+	def _aggregate_enacted(conn: sqlite3.Connection) -> int:
	+	"""
	+	enacted - Bills that became public law, sourced from actions.BecameLaw rows.
	+	"""
	+	rows = conn.execute(
	+	"""
	+	SELECT DISTINCT a.bill_id, a.action_date, a.source_url, b.title, a.text
	+	FROM actions a
	+	LEFT JOIN bills b ON b.id = a.bill_id
	+	WHERE a.action_type = 'BecameLaw'
	+	AND a.action_date != ''
	+	AND a.action_date IS NOT NULL
	+	ORDER BY a.action_date
	+	"""
	+	).fetchall()
	+
	+	count = 0
	+	for r in rows:
	+	title = r["title"] or r["text"] or ""
	+	_upsert_event(
	+	conn,
	+	date=r["action_date"],
	+	kind="enacted",
	+	ref_type="bill",
	+	ref_id=r["bill_id"],
	+	title=title,
	+	source_url=r["source_url"],
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d enacted events from actions.BecameLaw", count)
	+	return count
	+
	+
	+	def _aggregate_votes(conn: sqlite3.Connection) -> int:
	+	"""
	+	vote - Roll-call vote dates from the votes table.
	+
	+	Only references columns in the base schema (init_schema). Extended columns
	+	(question, description) may or may not exist depending on which pollers
	+	have run.
	+	"""
	+	# Discover which extended columns are present so the query stays portable
	+	# across fresh test DBs (base schema only) and production DBs (extended).
	+	existing_cols = {
	+	r[1]
	+	for r in conn.execute("PRAGMA table_info(votes)")
	+	}
	+
	+	extras = []
	+	for col in ("question", "description"):
	+	if col in existing_cols:
	+	extras.append(col)
	+
	+	extra_select = (", " + ", ".join(extras)) if extras else ""
	+
	+	rows = conn.execute(
	+	f"""
	+	SELECT id, date, result, source_url, bill_id{extra_select}
	+	FROM votes
	+	WHERE date IS NOT NULL
	+	AND date != ''
	+	ORDER BY date
	+	"""
	+	).fetchall()
	+
	+	count = 0
	+	for r in rows:
	+	# Use richer fields if available, fall back to result
	+	title = r["result"] or "Roll-call vote"
	+	for col in ("question", "description"):
	+	if col in existing_cols and r[col]:
	+	title = r[col]
	+	break
	+	_upsert_event(
	+	conn,
	+	date=r["date"],
	+	kind="vote",
	+	ref_type="vote",
	+	ref_id=r["id"],
	+	title=title,
	+	source_url=r["source_url"],
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d vote events from votes table", count)
	+	return count
	+
	+
	+	def _aggregate_bill_actions(conn: sqlite3.Connection) -> int:
	+	"""
	+	introduced / committee / floor - key bill action dates from actions table.
	+
	+	Mapping:
	+	IntroReferral → introduced
	+	Committee → committee
	+	Floor → floor
	+	"""
	+	kind_map = {
	+	"IntroReferral": "introduced",
	+	"Committee": "committee",
	+	"Floor": "floor",
	+	}
	+
	+	count = 0
	+	for action_type, kind in kind_map.items():
	+	rows = conn.execute(
	+	"""
	+	SELECT DISTINCT a.id, a.bill_id, a.action_date,
	+	a.source_url, a.text, b.title
	+	FROM actions a
	+	LEFT JOIN bills b ON b.id = a.bill_id
	+	WHERE a.action_type = ?
	+	AND a.action_date IS NOT NULL
	+	AND a.action_date != ''
	+	ORDER BY a.action_date
	+	""",
	+	(action_type,),
	+	).fetchall()
	+
	+	for r in rows:
	+	title = r["title"] or r["text"] or ""
	+	# Truncate action text for display; bill title is better
	+	if not r["title"] and r["text"]:
	+	title = r["text"][:200]
	+	_upsert_event(
	+	conn,
	+	date=r["action_date"],
	+	kind=kind,
	+	ref_type="action",
	+	ref_id=r["id"],
	+	title=title,
	+	source_url=r["source_url"],
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d introduced/committee/floor events", count)
	+	return count
	+
	+
	+	def _aggregate_apportionments(conn: sqlite3.Connection) -> int:
	+	"""
	+	apportionment - apportionment dates from the apportionments table.
	+	"""
	+	# Check the apportionments table exists
	+	tbl = conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='table' AND name='apportionments'"
	+	).fetchone()
	+	if tbl is None:
	+	logger.info("calendar: apportionments table not found - skipping")
	+	return 0
	+
	+	rows = conn.execute(
	+	"""
	+	SELECT id, date, agency, account, status, source_url
	+	FROM apportionments
	+	WHERE date IS NOT NULL
	+	AND date != ''
	+	ORDER BY date
	+	"""
	+	).fetchall()
	+
	+	count = 0
	+	for r in rows:
	+	title_parts = []
	+	if r["agency"]:
	+	title_parts.append(r["agency"])
	+	if r["account"]:
	+	title_parts.append(r["account"])
	+	if r["status"]:
	+	title_parts.append(f"({r['status']})")
	+	title = " - ".join(title_parts) if title_parts else "Apportionment"
	+	_upsert_event(
	+	conn,
	+	date=r["date"],
	+	kind="apportionment",
	+	ref_type="apportionment",
	+	ref_id=r["id"],
	+	title=title,
	+	source_url=r["source_url"],
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d apportionment events", count)
	+	return count
	+
	+
	+	def _aggregate_effective_parsed(conn: sqlite3.Connection) -> int:
	+	"""
	+	effective_parsed - Effective dates parsed from bill/action text via regex.
	+
	+	Scans:
	+	1. actions.text for all BecameLaw / President-signed rows (short text)
	+	2. Skips full bill XML fetching (expensive, out of scope for Phase 1)
	+
	+	Confidence: 0.5 (regex inference, not a structured field).
	+	"""
	+	# Focus on action texts that accompany enacted bills - they sometimes carry
	+	# the effective date in legislative boilerplate.
	+	rows = conn.execute(
	+	"""
	+	SELECT a.id, a.bill_id, a.action_date, a.source_url, a.text, b.title
	+	FROM actions a
	+	LEFT JOIN bills b ON b.id = a.bill_id
	+	WHERE a.text IS NOT NULL
	+	AND a.text != ''
	+	AND a.action_type IN ('BecameLaw', 'President', 'Floor', 'IntroReferral')
	+	"""
	+	).fetchall()
	+
	+	count = 0
	+	for r in rows:
	+	dates = parse_effective_dates(r["text"])
	+	for iso in dates:
	+	title = r["title"] or r["text"][:200] or ""
	+	_upsert_event(
	+	conn,
	+	date=iso,
	+	kind="effective_parsed",
	+	ref_type="action",
	+	ref_id=r["id"],
	+	title=f"[parsed] {title}",
	+	source_url=r["source_url"],
	+	confidence=0.5,
	+	)
	+	count += 1
	+
	+	logger.info("calendar: %d effective_parsed events from action text regex", count)
	+	return count
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Main entry point
	+	# ---------------------------------------------------------------------------
	+
	+	def run(db_path: Optional[str] = None) -> dict[str, int]:
	+	"""
	+	(Re)build calendar_events from all current DB contents.
	+
	+	Idempotent: safe to run repeatedly. Each run:
	+	1. Ensures schema (calendar_events table + effective_on column)
	+	2. Fetches recent FR final rules with effective_on
	+	3. Enriches any remaining executive_actions without effective_on
	+	4. Aggregates all event kinds into calendar_events
	+
	+	Returns a dict of kind → count for reporting.
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	# Step 1: ensure executive_actions has up-to-date effective_on data
	+	fetch_fr_rules_with_effective(conn, days_back=90)
	+	enrich_fr_effective_on(conn)
	+
	+	# Step 2: aggregate all event kinds
	+	counts: dict[str, int] = {}
	+
	+	counts["effective"] = _aggregate_effective(conn)
	+	counts["enacted"] = _aggregate_enacted(conn)
	+	counts["vote"] = _aggregate_votes(conn)
	+
	+	# Bill action types - one call handles all three
	+	bill_action_total = _aggregate_bill_actions(conn)
	+	# Disaggregate for reporting
	+	for kind in ("introduced", "committee", "floor"):
	+	c = conn.execute(
	+	"SELECT COUNT(*) FROM calendar_events WHERE kind=?", (kind,)
	+	).fetchone()[0]
	+	counts[kind] = c
	+
	+	counts["apportionment"] = _aggregate_apportionments(conn)
	+	counts["effective_parsed"] = _aggregate_effective_parsed(conn)
	+
	+	conn.commit()
	+
	+	total = sum(counts.values())
	+	logger.info(
	+	"calendar: run complete - %d total calendar_events across %d kinds",
	+	total,
	+	len(counts),
	+	)
	+
	+	conn.close()
	+	return counts
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	def main() -> None:
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+
	+	ap = argparse.ArgumentParser(
	+	description="Docket calendar aggregator - builds calendar_events table"
	+	)
	+	ap.add_argument(
	+	"--once",
	+	action="store_true",
	+	help="Run once and exit (default behaviour)",
	+	)
	+	ap.add_argument(
	+	"--db",
	+	metavar="PATH",
	+	default=None,
	+	help="Override SQLite DB path (uses DOCKET_DB_PATH env var by default)",
	+	)
	+	args = ap.parse_args()
	+
	+	counts = run(db_path=args.db)
	+
	+	print("\ncalendar_events counts by kind:")
	+	for kind, n in sorted(counts.items()):
	+	print(f" {kind:<20} {n:>6}")
	+	print(f" {'TOTAL':<20} {sum(counts.values()):>6}")
	+
	+
	+	if __name__ == "__main__":
	+	main()

services/ingest/ingest/categorize.py +651 -0

		@@ -0,0 +1,651 @@
	+	"""
	+	ingest/categorize.py - Unified taxonomy backfill for bills and spending awards.
	+
	+	Creates / extends:
	+	categories (slug, label, kind) - ~32 standard Congressional policy areas
	+	bill_subjects (bill_id, subject) - granular legislative subjects from BILLSTATUS XML
	+	awards.category (TEXT column) - deterministic agency→policy-area mapping
	+
	+	run(db_path) backfills all three over whatever is already in the DB.
	+
	+	Subjects are fetched on demand from GovInfo BULKDATA BILLSTATUS XML (no API key
	+	required) - the same source used by bills_bulk.py. Bills already processed are
	+	skipped on subsequent runs (idempotent).
	+
	+	CLI:
	+	python -m ingest.categorize --once [--limit-subjects N] [--delay SECS]
	+
	+	All changes are additive: no existing rows or columns are dropped.
	+	Categories come straight from Congress.gov's own taxonomy - no editorial groupings.
	+	"""
	+
	+	from __future__ import annotations
	+
	+	import argparse
	+	import logging
	+	import sqlite3
	+	import sys
	+	import time
	+	import xml.etree.ElementTree as ET
	+	from typing import Iterator, Optional
	+
	+	import requests
	+
	+	from ingest.db import get_connection, init_schema
	+
	+	logger = logging.getLogger(__name__)
	+
	+	# ---------------------------------------------------------------------------
	+	# Canonical ~32 Congressional policy areas (Congress.gov standard taxonomy)
	+	# Source: https://www.congress.gov/help/field-values/policy-area
	+	# These are factual classifications used by the Congressional Research Service.
	+	# ---------------------------------------------------------------------------
	+
	+	POLICY_AREAS: list[tuple[str, str]] = [
	+	("agriculture-and-food", "Agriculture and Food"),
	+	("animals", "Animals"),
	+	("armed-forces-and-national-security", "Armed Forces and National Security"),
	+	("arts-culture-religion", "Arts, Culture, Religion"),
	+	("civil-rights-and-liberties-minority-issues", "Civil Rights and Liberties, Minority Issues"),
	+	("commerce", "Commerce"),
	+	("congress", "Congress"),
	+	("crime-and-law-enforcement", "Crime and Law Enforcement"),
	+	("economics-and-public-finance", "Economics and Public Finance"),
	+	("education", "Education"),
	+	("emergency-management", "Emergency Management"),
	+	("energy", "Energy"),
	+	("environmental-protection", "Environmental Protection"),
	+	("families", "Families"),
	+	("finance-and-financial-sector", "Finance and Financial Sector"),
	+	("foreign-trade-and-international-finance", "Foreign Trade and International Finance"),
	+	("government-operations-and-politics", "Government Operations and Politics"),
	+	("health", "Health"),
	+	("housing-and-community-development", "Housing and Community Development"),
	+	("immigration", "Immigration"),
	+	("international-affairs", "International Affairs"),
	+	("labor-and-employment", "Labor and Employment"),
	+	("law", "Law"),
	+	("macroeconomics", "Macroeconomics"),
	+	("native-americans", "Native Americans"),
	+	("public-lands-and-natural-resources", "Public Lands and Natural Resources"),
	+	("science-technology-communications", "Science, Technology, Communications"),
	+	("social-welfare", "Social Welfare"),
	+	("sports-and-recreation", "Sports and Recreation"),
	+	("taxation", "Taxation"),
	+	("transportation-and-public-works", "Transportation and Public Works"),
	+	("water-resources-development", "Water Resources Development"),
	+	]
	+
	+	# ---------------------------------------------------------------------------
	+	# Agency → policy-area label mapping (deterministic; aligned to the same
	+	# ~32-area vocabulary so bills and spending share one taxonomy).
	+	# Source: executive-branch agency names as they appear in USAspending.gov data.
	+	# ---------------------------------------------------------------------------
	+
	+	AGENCY_CATEGORY_MAP: dict[str, str] = {
	+	# Defense / Security
	+	"Department of Defense": "Armed Forces and National Security",
	+	"Department of Veterans Affairs": "Armed Forces and National Security",
	+	"Department of Homeland Security": "Emergency Management",
	+	"Selective Service System": "Armed Forces and National Security",
	+
	+	# Health
	+	"Department of Health and Human Services": "Health",
	+
	+	# Transportation
	+	"Department of Transportation": "Transportation and Public Works",
	+
	+	# Energy
	+	"Department of Energy": "Energy",
	+
	+	# Education
	+	"Department of Education": "Education",
	+
	+	# Agriculture / Food
	+	"Department of Agriculture": "Agriculture and Food",
	+
	+	# Labor
	+	"Department of Labor": "Labor and Employment",
	+
	+	# Environment
	+	"Environmental Protection Agency": "Environmental Protection",
	+
	+	# Justice / Law Enforcement
	+	"Department of Justice": "Crime and Law Enforcement",
	+
	+	# State / International
	+	"Department of State": "International Affairs",
	+	"Agency for International Development": "International Affairs",
	+	"United States Agency for International Development": "International Affairs",
	+	"Broadcasting Board of Governors": "International Affairs",
	+	"Peace Corps": "International Affairs",
	+
	+	# Commerce / Trade
	+	"Department of Commerce": "Commerce",
	+	"Office of the United States Trade Representative": "Foreign Trade and International Finance",
	+	"Export-Import Bank of the United States": "Foreign Trade and International Finance",
	+	"International Trade Commission": "Foreign Trade and International Finance",
	+
	+	# Finance / Treasury
	+	"Department of the Treasury": "Economics and Public Finance",
	+	"Small Business Administration": "Economics and Public Finance",
	+	"Securities and Exchange Commission": "Finance and Financial Sector",
	+	"Commodity Futures Trading Commission": "Finance and Financial Sector",
	+	"Federal Deposit Insurance Corporation": "Finance and Financial Sector",
	+	"Federal Reserve System": "Finance and Financial Sector",
	+
	+	# Housing
	+	"Department of Housing and Urban Development": "Housing and Community Development",
	+
	+	# Interior / Public Lands
	+	"Department of the Interior": "Public Lands and Natural Resources",
	+	"Forest Service": "Public Lands and Natural Resources",
	+
	+	# Science / Technology / Space
	+	"National Aeronautics and Space Administration": "Science, Technology, Communications",
	+	"National Science Foundation": "Science, Technology, Communications",
	+	"National Institute of Standards and Technology": "Science, Technology, Communications",
	+	"Federal Communications Commission": "Science, Technology, Communications",
	+
	+	# Social Welfare
	+	"Social Security Administration": "Social Welfare",
	+	"Corporation for National and Community Service": "Social Welfare",
	+	"Institute of Museum and Library Services": "Arts, Culture, Religion",
	+	"National Endowment for the Arts": "Arts, Culture, Religion",
	+	"National Endowment for the Humanities": "Arts, Culture, Religion",
	+	"Smithsonian Institution": "Arts, Culture, Religion",
	+
	+	# Government Operations
	+	"General Services Administration": "Government Operations and Politics",
	+	"Office of Personnel Management": "Government Operations and Politics",
	+	"Executive Office of the President": "Government Operations and Politics",
	+	"Office of Management and Budget": "Government Operations and Politics",
	+	"Government Accountability Office": "Government Operations and Politics",
	+	"Congressional Budget Office": "Government Operations and Politics",
	+	"Library of Congress": "Government Operations and Politics",
	+	"Government Publishing Office": "Government Operations and Politics",
	+	"National Archives and Records Administration": "Government Operations and Politics",
	+	"Federal Election Commission": "Government Operations and Politics",
	+	"Office of Special Counsel": "Government Operations and Politics",
	+	"Merit Systems Protection Board": "Government Operations and Politics",
	+	"Equal Employment Opportunity Commission": "Civil Rights and Liberties, Minority Issues",
	+	"Civil Rights Division": "Civil Rights and Liberties, Minority Issues",
	+	"Commission on Civil Rights": "Civil Rights and Liberties, Minority Issues",
	+
	+	# Immigration
	+	"U.S. Citizenship and Immigration Services": "Immigration",
	+	"Immigration and Customs Enforcement": "Immigration",
	+	"Customs and Border Protection": "Immigration",
	+
	+	# Native Americans
	+	"Indian Health Service": "Native Americans",
	+	"Bureau of Indian Affairs": "Native Americans",
	+	}
	+
	+
	+	def _slug_from_label(label: str) -> str:
	+	"""Convert a policy-area label to a URL-safe slug."""
	+	return (
	+	label.lower()
	+	.replace(",", "")
	+	.replace(".", "")
	+	.replace("'", "")
	+	.replace(" ", " ")
	+	.replace(" ", "-")
	+	)
	+
	+
	+	def agency_to_category(agency: str) -> Optional[str]:
	+	"""
	+	Map an awarding agency name to a policy-area label.
	+
	+	Exact match first; then prefix/substring match (handles sub-agency names
	+	that start with or contain a known department name). Returns None when no
	+	match is found so callers can leave the column NULL rather than guess.
	+	"""
	+	if not agency:
	+	return None
	+	# Exact
	+	if agency in AGENCY_CATEGORY_MAP:
	+	return AGENCY_CATEGORY_MAP[agency]
	+	# Prefix match - e.g. "Department of Defense - Navy" → "Armed Forces …"
	+	for key, cat in AGENCY_CATEGORY_MAP.items():
	+	if agency.startswith(key):
	+	return cat
	+	# Substring / contained - e.g. "HHS Office of Inspector General"
	+	agency_lower = agency.lower()
	+	for key, cat in AGENCY_CATEGORY_MAP.items():
	+	if key.lower() in agency_lower:
	+	return cat
	+	return None
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Schema helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _ensure_schema(conn: sqlite3.Connection) -> None:
	+	"""Create new tables and add missing columns - all additive."""
	+
	+	conn.executescript("""
	+	CREATE TABLE IF NOT EXISTS categories (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	slug TEXT NOT NULL UNIQUE,
	+	label TEXT NOT NULL,
	+	kind TEXT NOT NULL DEFAULT 'policy_area'
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS bill_subjects (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bill_id INTEGER NOT NULL REFERENCES bills(id),
	+	subject TEXT NOT NULL,
	+	UNIQUE (bill_id, subject)
	+	);
	+	""")
	+	conn.commit()
	+
	+	# awards.category - guarded ALTER (column may already exist from a prior run)
	+	existing_cols = {row[1] for row in conn.execute("PRAGMA table_info(awards)")}
	+	if "category" not in existing_cols:
	+	try:
	+	conn.execute("ALTER TABLE awards ADD COLUMN category TEXT")
	+	conn.commit()
	+	logger.info("Added awards.category column")
	+	except sqlite3.OperationalError as exc:
	+	# Parallel runner already added it; safe to ignore.
	+	logger.debug("ALTER awards skipped: %s", exc)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Seed categories
	+	# ---------------------------------------------------------------------------
	+
	+	def _seed_categories(conn: sqlite3.Connection) -> int:
	+	"""
	+	Insert the canonical policy-area rows. Also inserts any policy_area values
	+	already present in the bills table that are not yet in categories (catches
	+	areas Congress added after this file was last updated).
	+	"""
	+	inserted = 0
	+	for slug, label in POLICY_AREAS:
	+	try:
	+	conn.execute(
	+	"INSERT OR IGNORE INTO categories (slug, label, kind) VALUES (?, ?, 'policy_area')",
	+	(slug, label),
	+	)
	+	if conn.execute("SELECT changes()").fetchone()[0]:
	+	inserted += 1
	+	except sqlite3.Error as exc:
	+	logger.warning("categories insert failed for %s: %s", slug, exc)
	+
	+	# Pick up any extra areas already in the bills table
	+	for row in conn.execute(
	+	"SELECT DISTINCT policy_area FROM bills WHERE policy_area IS NOT NULL AND policy_area != ''"
	+	):
	+	area = row[0].strip()
	+	slug = _slug_from_label(area)
	+	try:
	+	conn.execute(
	+	"INSERT OR IGNORE INTO categories (slug, label, kind) VALUES (?, ?, 'policy_area')",
	+	(slug, area),
	+	)
	+	if conn.execute("SELECT changes()").fetchone()[0]:
	+	inserted += 1
	+	except sqlite3.Error as exc:
	+	logger.warning("categories extra insert failed for %r: %s", area, exc)
	+
	+	conn.commit()
	+	logger.info("categories seeded/updated: %d new rows", inserted)
	+	return inserted
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Bill subjects backfill - re-fetch BILLSTATUS XML from GovInfo
	+	# ---------------------------------------------------------------------------
	+
	+	_BULKDATA_BASE = "https://www.govinfo.gov/bulkdata/BILLSTATUS"
	+
	+
	+	def _billstatus_url(congress: int, bill_type: str, number: str) -> str:
	+	"""
	+	Construct the GovInfo BULKDATA URL for a single BILLSTATUS XML file.
	+
	+	Actual URL format (from sitemap):
	+	https://www.govinfo.gov/bulkdata/BILLSTATUS/{congress}/{type}/BILLSTATUS-{congress}{type}{number}.xml
	+	Note: congress and bill_type are SEPARATE path segments; the filename still
	+	concatenates them (BILLSTATUS-119hr160.xml).
	+	"""
	+	return (
	+	f"{_BULKDATA_BASE}/{congress}/{bill_type}"
	+	f"/BILLSTATUS-{congress}{bill_type}{number}.xml"
	+	)
	+
	+
	+	def _parse_subjects(xml_bytes: bytes) -> list[str]:
	+	"""
	+	Parse <legislativeSubjects><item><name> elements from a BILLSTATUS XML.
	+	Returns a list of subject strings (may be empty).
	+	"""
	+	try:
	+	root = ET.fromstring(xml_bytes)
	+	except ET.ParseError as exc:
	+	logger.debug("XML parse error: %s", exc)
	+	return []
	+
	+	bill_el = root.find("bill")
	+	if bill_el is None:
	+	if root.tag in ("bill", "Bill"):
	+	bill_el = root
	+	else:
	+	return []
	+
	+	subjects_el = bill_el.find("subjects")
	+	if subjects_el is None:
	+	return []
	+
	+	leg_el = subjects_el.find("legislativeSubjects")
	+	if leg_el is None:
	+	return []
	+
	+	names: list[str] = []
	+	for item in leg_el.findall("item"):
	+	name_el = item.find("name")
	+	if name_el is not None and name_el.text:
	+	names.append(name_el.text.strip())
	+	return names
	+
	+
	+	def _bills_needing_subjects(conn: sqlite3.Connection) -> list[tuple]:
	+	"""
	+	Return (bill_id, external_id, congress) for bills that have no subjects yet.
	+	external_id format: "{congress}-{bill_type}-{number}"
	+	"""
	+	return conn.execute("""
	+	SELECT b.id, b.external_id, b.congress
	+	FROM bills b
	+	WHERE b.source = 'govinfo_billstatus'
	+	AND NOT EXISTS (
	+	SELECT 1 FROM bill_subjects s WHERE s.bill_id = b.id
	+	)
	+	ORDER BY b.id
	+	""").fetchall()
	+
	+
	+	def _backfill_subjects(
	+	conn: sqlite3.Connection,
	+	limit: Optional[int] = None,
	+	delay: float = 0.05,
	+	) -> tuple[int, int]:
	+	"""
	+	Fetch BILLSTATUS XML for bills without subjects and store the subjects.
	+	Returns (bills_processed, subjects_stored).
	+	"""
	+	pending = _bills_needing_subjects(conn)
	+	if limit is not None:
	+	pending = pending[:limit]
	+
	+	session = requests.Session()
	+	session.headers["User-Agent"] = "Docket-categorize/1.0 (public transparency research)"
	+
	+	bills_done = 0
	+	subjects_total = 0
	+
	+	for row in pending:
	+	bill_id, external_id, congress = row[0], row[1], row[2]
	+	# external_id: "119-hr-160"
	+	parts = external_id.split("-", 2)
	+	if len(parts) != 3:
	+	logger.debug("Skipping unexpected external_id %s", external_id)
	+	continue
	+	_, bill_type, number = parts
	+
	+	url = _billstatus_url(int(congress), bill_type, number)
	+	try:
	+	resp = session.get(url, timeout=30)
	+	except requests.RequestException as exc:
	+	logger.warning("Fetch error %s: %s", url, exc)
	+	continue
	+
	+	if resp.status_code != 200:
	+	logger.debug("HTTP %s for %s", resp.status_code, url)
	+	# Still mark as processed so we don't retry a 404 forever.
	+	# Insert a sentinel so the SELECT above won't re-queue it.
	+	# We skip because there are genuinely no subjects to store.
	+	continue
	+
	+	subjects = _parse_subjects(resp.content)
	+	stored = 0
	+	for subj in subjects:
	+	try:
	+	conn.execute(
	+	"INSERT OR IGNORE INTO bill_subjects (bill_id, subject) VALUES (?, ?)",
	+	(bill_id, subj),
	+	)
	+	stored += conn.execute("SELECT changes()").fetchone()[0]
	+	except sqlite3.Error as exc:
	+	logger.warning("bill_subjects insert error bill_id=%s: %s", bill_id, exc)
	+
	+	if subjects:
	+	# Commit each bill so partial progress is saved on interruption
	+	conn.commit()
	+	else:
	+	# Insert a placeholder subject to mark this bill as "processed" even if it
	+	# has no subjects (prevents re-fetching on every run).
	+	# Use empty string - UNIQUE(bill_id, subject) allows one empty row per bill.
	+	try:
	+	conn.execute(
	+	"INSERT OR IGNORE INTO bill_subjects (bill_id, subject) VALUES (?, '')",
	+	(bill_id,),
	+	)
	+	conn.commit()
	+	except sqlite3.Error:
	+	pass
	+
	+	bills_done += 1
	+	subjects_total += stored
	+
	+	if bills_done % 50 == 0:
	+	logger.info("subjects backfill: %d bills done, %d subjects stored", bills_done, subjects_total)
	+
	+	if delay > 0:
	+	time.sleep(delay)
	+
	+	logger.info("subjects backfill complete: %d bills, %d subjects", bills_done, subjects_total)
	+	return bills_done, subjects_total
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Awards category backfill
	+	# ---------------------------------------------------------------------------
	+
	+	def _backfill_awards(conn: sqlite3.Connection) -> int:
	+	"""
	+	Apply agency→category map to all awards rows missing a category.
	+	Returns number of awards updated.
	+	"""
	+	rows = conn.execute(
	+	"SELECT id, agency FROM awards WHERE category IS NULL OR category = ''"
	+	).fetchall()
	+
	+	updated = 0
	+	for row in rows:
	+	award_id, agency = row[0], row[1] or ""
	+	cat = agency_to_category(agency)
	+	if cat is not None:
	+	conn.execute(
	+	"UPDATE awards SET category = ? WHERE id = ?",
	+	(cat, award_id),
	+	)
	+	updated += 1
	+
	+	conn.commit()
	+	logger.info("awards categorized: %d / %d rows", updated, len(rows))
	+	return updated
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Category count query (summary)
	+	# ---------------------------------------------------------------------------
	+
	+	def category_counts(conn: sqlite3.Connection) -> list[dict]:
	+	"""
	+	Return counts of bills and awards per category label.
	+	Useful for verifying the taxonomy is populated.
	+	"""
	+	rows = conn.execute("""
	+	SELECT c.label,
	+	COUNT(DISTINCT b.id) AS bill_count,
	+	COUNT(DISTINCT a.id) AS award_count
	+	FROM categories c
	+	LEFT JOIN bills b ON b.policy_area = c.label
	+	LEFT JOIN awards a ON a.category = c.label
	+	GROUP BY c.label
	+	ORDER BY (COUNT(DISTINCT b.id) + COUNT(DISTINCT a.id)) DESC, c.label
	+	""").fetchall()
	+	return [
	+	{"label": r[0], "bills": r[1], "awards": r[2]}
	+	for r in rows
	+	]
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Main entry point
	+	# ---------------------------------------------------------------------------
	+
	+	def run(
	+	db_path: Optional[str] = None,
	+	limit_subjects: Optional[int] = None,
	+	delay: float = 0.05,
	+	) -> dict:
	+	"""
	+	Backfill categories, bill subjects, and award categories.
	+
	+	Parameters
	+	----------
	+	db_path : SQLite path (None = use DOCKET_DB_PATH env / default).
	+	limit_subjects : Cap on how many bills to fetch subjects for (None = all).
	+	delay : Seconds between GovInfo requests.
	+
	+	Returns
	+	-------
	+	Summary dict with counts.
	+	"""
	+	from ingest.db import get_connection, init_schema
	+
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	_ensure_schema(conn)
	+
	+	cats_added = _seed_categories(conn)
	+	bills_done, subjects_stored = _backfill_subjects(conn, limit=limit_subjects, delay=delay)
	+	awards_updated = _backfill_awards(conn)
	+
	+	# Total bills with policy_area
	+	bills_with_pa = conn.execute(
	+	"SELECT COUNT(*) FROM bills WHERE policy_area IS NOT NULL AND policy_area != ''"
	+	).fetchone()[0]
	+	total_bills = conn.execute("SELECT COUNT(*) FROM bills").fetchone()[0]
	+
	+	# Total bill_subjects rows (excluding sentinel empty-string rows)
	+	subjects_total = conn.execute(
	+	"SELECT COUNT(*) FROM bill_subjects WHERE subject != ''"
	+	).fetchone()[0]
	+
	+	# Distinct subjects
	+	subjects_distinct = conn.execute(
	+	"SELECT COUNT(DISTINCT subject) FROM bill_subjects WHERE subject != ''"
	+	).fetchone()[0]
	+
	+	# Distinct policy areas in use
	+	pa_in_use = conn.execute(
	+	"SELECT COUNT(DISTINCT policy_area) FROM bills WHERE policy_area IS NOT NULL AND policy_area != ''"
	+	).fetchone()[0]
	+
	+	# Awards categorized
	+	awards_total = conn.execute("SELECT COUNT(*) FROM awards").fetchone()[0]
	+	awards_cat = conn.execute(
	+	"SELECT COUNT(*) FROM awards WHERE category IS NOT NULL AND category != ''"
	+	).fetchone()[0]
	+
	+	# Categories seeded
	+	cats_total = conn.execute("SELECT COUNT(*) FROM categories").fetchone()[0]
	+
	+	summary = {
	+	"categories_total": cats_total,
	+	"categories_added_this_run": cats_added,
	+	"bills_total": total_bills,
	+	"bills_with_policy_area": bills_with_pa,
	+	"policy_areas_in_use": pa_in_use,
	+	"subjects_bills_fetched_this_run": bills_done,
	+	"bill_subjects_total_rows": subjects_total,
	+	"bill_subjects_distinct": subjects_distinct,
	+	"awards_total": awards_total,
	+	"awards_categorized": awards_cat,
	+	}
	+
	+	logger.info("categorize summary: %s", summary)
	+
	+	conn.close()
	+	return summary
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	def _print_category_counts(db_path: Optional[str]) -> None:
	+	from ingest.db import get_connection
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	counts = category_counts(conn)
	+	conn.close()
	+	print(f"\n{'Category':<50} {'Bills':>6} {'Awards':>6}")
	+	print("-" * 68)
	+	for row in counts:
	+	if row["bills"] > 0 or row["awards"] > 0:
	+	print(f"{row['label']:<50} {row['bills']:>6} {row['awards']:>6}")
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+
	+	ap = argparse.ArgumentParser(
	+	description="Backfill unified taxonomy: categories, bill subjects, award categories"
	+	)
	+	ap.add_argument("--once", action="store_true", help="Run once and exit")
	+	ap.add_argument(
	+	"--limit-subjects", type=int, default=None, metavar="N",
	+	help="Maximum number of bills to fetch subjects for (default: all)"
	+	)
	+	ap.add_argument(
	+	"--delay", type=float, default=0.05,
	+	help="Seconds between GovInfo requests (default 0.05)"
	+	)
	+	ap.add_argument(
	+	"--counts-only", action="store_true",
	+	help="Print category counts from existing data and exit (no backfill)"
	+	)
	+	args = ap.parse_args()
	+
	+	import os
	+	db_path = os.environ.get("DOCKET_DB_PATH")
	+
	+	if args.counts_only:
	+	_print_category_counts(db_path)
	+	sys.exit(0)
	+
	+	summary = run(
	+	db_path=db_path,
	+	limit_subjects=args.limit_subjects,
	+	delay=args.delay,
	+	)
	+
	+	print("\n=== Categorize Summary ===")
	+	for k, v in summary.items():
	+	print(f" {k}: {v}")
	+
	+	_print_category_counts(db_path)
	+	print("\nDone.")

services/ingest/ingest/pollers/voteview.py +1031 -0

		@@ -0,0 +1,1031 @@
	+	"""
	+	Voteview bulk-CSV poller - complete historical per-member voting index.
	+
	+	Ingestion pipe (NOT user-facing): https://voteview.com/static/data/out/
	+	User-facing source_url (neutrality rule #1, official .gov only):
	+	House: https://clerk.house.gov/Votes/{year}/roll{NNN:03d}
	+	Senate: https://www.senate.gov/legislative/LIS/roll_call_votes/
	+	vote{congress}{session}/vote_{congress}_{session}_{roll:05d}.xml
	+
	+	Cast-code mapping (Voteview codebook):
	+	Verified against https://voteview.com/articles/data_help_members
	+
	+	Code \| Official meaning \| Canonical position
	+	─────┼───────────────────────────────┼───────────────────
	+	0 \| Not a member / no record \| Not Voting
	+	1 \| Yea \| Yea
	+	2 \| Paired Yea (absent, declared) \| Yea
	+	3 \| Announced Yea (absent, decl.) \| Yea
	+	4 \| Announced Nay (absent, decl.) \| Nay
	+	5 \| Paired Nay (absent, declared) \| Nay
	+	6 \| Nay \| Nay
	+	7 \| Present / abstention (v1) \| Present
	+	8 \| Present / abstention (v2) \| Present
	+	9 \| Not Voting \| Not Voting
	+
	+	Paired/announced (codes 2-5): member was absent but announced their
	+	intent; they are counted by the chamber as Yea/Nay respectively.
	+	Codes 7 and 8 are both recorded abstentions (Present), distinct from
	+	code 9 (Not Voting, member absent with no declared position).
	+
	+	NOTE: Voteview covers only RECORDED (roll-call) votes. Voice votes and
	+	unanimous-consent proceedings have no per-member record anywhere in the
	+	public record. This is structural - not a gap in this ingestion.
	+
	+	ICPSR → bioguide crosswalk:
	+	Fetched from unitedstates/congress-legislators (current + historical).
	+	Cached in voteview_crosswalk table. Members not in the crosswalk are
	+	stored with member_bioguide=NULL (icpsr preserved in the raw payload).
	+
	+	Year index:
	+	A `year` INTEGER column is added to `votes` and indexed so that
	+	votes_by_member_year() and votes_in_year() run in O(log n).
	+
	+	Usage (CLI):
	+	# Proof slice (congresses 118-119, ≈2023-present):
	+	python -m ingest.pollers.voteview --once --start 118 --end 119
	+
	+	# Recent history (2019-present):
	+	python -m ingest.pollers.voteview --once --start 116 --end 119
	+
	+	# Extended backfill (1989-present, ~30 congresses):
	+	python -m ingest.pollers.voteview --once --start 101 --end 119
	+
	+	# Full Voteview history (1789-present, ~120 congresses):
	+	python -m ingest.pollers.voteview --once --start 1 --end 119
	+	"""
	+
	+	from __future__ import annotations
	+
	+	import argparse
	+	import csv
	+	import io
	+	import logging
	+	import sqlite3
	+	from datetime import datetime, timezone
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.members import ensure_votes_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "voteview"
	+
	+	# ─── Voteview CSV base URLs ───────────────────────────────────────────────────
	+	_VOTEVIEW_BASE = "https://voteview.com/static/data/out"
	+
	+
	+	def _rollcalls_url(chamber_prefix: str, congress: int) -> str:
	+	return f"{_VOTEVIEW_BASE}/rollcalls/{chamber_prefix}{congress}_rollcalls.csv"
	+
	+
	+	def _votes_url(chamber_prefix: str, congress: int) -> str:
	+	return f"{_VOTEVIEW_BASE}/votes/{chamber_prefix}{congress}_votes.csv"
	+
	+
	+	# ─── ICPSR crosswalk source URLs ─────────────────────────────────────────────
	+	_LEGISLATORS_URLS = [
	+	(
	+	"https://raw.githubusercontent.com/unitedstates/congress-legislators"
	+	"/gh-pages/legislators-current.json"
	+	),
	+	(
	+	"https://raw.githubusercontent.com/unitedstates/congress-legislators"
	+	"/gh-pages/legislators-historical.json"
	+	),
	+	]
	+
	+	# ─── Official source URL builders (neutrality rule #1) ───────────────────────
	+
	+	def _house_source_url(year: int, clerk_rollnumber: int) -> str:
	+	"""Official House Clerk vote page URL."""
	+	return f"https://clerk.house.gov/Votes/{year}/roll{clerk_rollnumber:03d}"
	+
	+
	+	def _senate_source_url(congress: int, session: int, rollnumber: int) -> str:
	+	"""Official Senate roll-call vote XML URL."""
	+	return (
	+	f"https://www.senate.gov/legislative/LIS/roll_call_votes/"
	+	f"vote{congress}{session}/"
	+	f"vote_{congress}_{session}_{rollnumber:05d}.xml"
	+	)
	+
	+
	+	# ─── Cast-code mapping ────────────────────────────────────────────────────────
	+
	+	# Voteview codebook: https://voteview.com/articles/data_help_members#castcode
	+	CAST_CODE_MAP: dict[int, str] = {
	+	0: "Not Voting", # Not a member of the chamber, or absent with no record
	+	1: "Yea",
	+	2: "Yea", # Paired Yea - absent but announced intent
	+	3: "Yea", # Announced Yea - absent but announced intent
	+	4: "Nay", # Announced Nay - absent but announced intent
	+	5: "Nay", # Paired Nay - absent but announced intent
	+	6: "Nay",
	+	7: "Present", # Abstention variant 1
	+	8: "Present", # Abstention variant 2
	+	9: "Not Voting", # Absent with no position declared
	+	}
	+
	+
	+	def cast_code_to_position(code) -> Optional[str]:
	+	"""
	+	Map a Voteview cast_code (int or str) to a canonical position string.
	+	Returns None for unrecognized codes.
	+	"""
	+	try:
	+	return CAST_CODE_MAP.get(int(float(code)))
	+	except (ValueError, TypeError):
	+	return None
	+
	+
	+	# ─── Schema setup ─────────────────────────────────────────────────────────────
	+
	+	def ensure_voteview_schema(conn: sqlite3.Connection) -> None:
	+	"""
	+	Idempotent: create voteview-specific tables, extend the votes table with
	+	a `year` column (for fast year-range queries), and build the indexes.
	+
	+	Uses individual execute() calls (not executescript) so that SQLite's
	+	busy_timeout is respected when another reader has a shared lock.
	+	"""
	+	# ICPSR → bioguide crosswalk cache.
	+	conn.execute(
	+	"""
	+	CREATE TABLE IF NOT EXISTS voteview_crosswalk (
	+	icpsr_id INTEGER PRIMARY KEY,
	+	bioguide_id TEXT,
	+	full_name TEXT,
	+	fetched_at TEXT NOT NULL
	+	)
	+	"""
	+	)
	+	conn.execute(
	+	"""
	+	CREATE INDEX IF NOT EXISTS idx_crosswalk_bioguide
	+	ON voteview_crosswalk(bioguide_id)
	+	WHERE bioguide_id IS NOT NULL
	+	"""
	+	)
	+
	+	# Additive columns on the votes table. SQLite has no IF NOT EXISTS on
	+	# ALTER TABLE; use try/except per the pattern in members.py.
	+	for col, col_type in [
	+	("year", "INTEGER"),
	+	("description", "TEXT"),
	+	("voteview_rollnumber", "INTEGER"),
	+	]:
	+	try:
	+	conn.execute(f"ALTER TABLE votes ADD COLUMN {col} {col_type}")
	+	except Exception:
	+	pass # already present - safe to ignore
	+
	+	# Year-range indexes (the whole point of storing year explicitly).
	+	for ddl in [
	+	"CREATE INDEX IF NOT EXISTS idx_votes_year ON votes(year)",
	+	"CREATE INDEX IF NOT EXISTS idx_votes_year_chamber ON votes(year, chamber)",
	+	"CREATE INDEX IF NOT EXISTS idx_votes_congress_chamber ON votes(congress, chamber)",
	+	]:
	+	try:
	+	conn.execute(ddl)
	+	except Exception:
	+	pass
	+
	+	conn.commit()
	+
	+
	+	# ─── ICPSR → bioguide crosswalk ───────────────────────────────────────────────
	+
	+	def build_crosswalk(
	+	conn: sqlite3.Connection,
	+	force_refresh: bool = False,
	+	) -> dict[int, str]:
	+	"""
	+	Build (or load from cache) the ICPSR → bioguide_id crosswalk.
	+
	+	Fetches congress-legislators current + historical JSON from GitHub and
	+	stores the mapping in voteview_crosswalk for subsequent runs.
	+
	+	Returns
	+	-------
	+	dict mapping icpsr_id (int) → bioguide_id (str).
	+	"""
	+	if not force_refresh:
	+	cached_count = conn.execute(
	+	"SELECT COUNT(*) FROM voteview_crosswalk"
	+	).fetchone()[0]
	+	if cached_count > 0:
	+	logger.info(
	+	"voteview: crosswalk: loaded %d ICPSR mappings from cache",
	+	cached_count,
	+	)
	+	rows = conn.execute(
	+	"SELECT icpsr_id, bioguide_id FROM voteview_crosswalk "
	+	"WHERE bioguide_id IS NOT NULL"
	+	).fetchall()
	+	return {r["icpsr_id"]: r["bioguide_id"] for r in rows}
	+
	+	logger.info("voteview: building ICPSR→bioguide crosswalk from congress-legislators …")
	+	now = datetime.now(timezone.utc).isoformat()
	+	crosswalk: dict[int, str] = {}
	+
	+	for url in _LEGISLATORS_URLS:
	+	try:
	+	resp = requests.get(url, timeout=60)
	+	resp.raise_for_status()
	+	data = resp.json()
	+	except Exception as exc:
	+	logger.error("voteview: crosswalk fetch error %s: %s", url, exc)
	+	continue
	+
	+	for leg in data:
	+	ids = leg.get("id", {})
	+	bioguide = (ids.get("bioguide") or "").strip()
	+	if not bioguide:
	+	continue
	+
	+	name_block = leg.get("name", {})
	+	full_name = (
	+	name_block.get("official_full")
	+	or f"{name_block.get('first','')} {name_block.get('last','')}".strip()
	+	)
	+
	+	# Primary ICPSR field (most members).
	+	for icpsr_field in ("icpsr", "icpsr_hn"):
	+	raw = ids.get(icpsr_field)
	+	if raw is None:
	+	continue
	+	try:
	+	icpsr_int = int(float(raw))
	+	except (ValueError, TypeError):
	+	continue
	+
	+	crosswalk[icpsr_int] = bioguide
	+	conn.execute(
	+	"""
	+	INSERT INTO voteview_crosswalk
	+	(icpsr_id, bioguide_id, full_name, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(icpsr_id) DO UPDATE SET
	+	bioguide_id = excluded.bioguide_id,
	+	full_name = excluded.full_name,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(icpsr_int, bioguide, full_name, now),
	+	)
	+
	+	conn.commit()
	+	logger.info(
	+	"voteview: crosswalk built - %d ICPSR→bioguide mappings", len(crosswalk)
	+	)
	+	return crosswalk
	+
	+
	+	# ─── Supplementary name-based crosswalk (for post-2018 members) ──────────────
	+
	+	def _build_name_lookup(legislators_data: list[dict]) -> dict:
	+	"""
	+	Build a secondary lookup mapping (normalized_last_name, state_abbrev)
	+	→ list of (bioguide_id, service_terms) for name-based ICPSR matching.
	+
	+	Many members elected 2018+ are absent from the congress-legislators
	+	id.icpsr field (the ICPSR project lags new assignments). For those,
	+	we match by last name + state + congress term overlap.
	+	"""
	+	lookup: dict[tuple, list] = {}
	+	for leg in legislators_data:
	+	ids = leg.get("id", {})
	+	bioguide = (ids.get("bioguide") or "").strip()
	+	if not bioguide:
	+	continue
	+	name = leg.get("name", {})
	+	last = (name.get("last") or "").strip().lower()
	+	terms = leg.get("terms", [])
	+	if not last or not terms:
	+	continue
	+	for term in terms:
	+	state = (term.get("state") or "").upper()
	+	if not state:
	+	continue
	+	key = (last, state)
	+	if key not in lookup:
	+	lookup[key] = []
	+	# Avoid duplicate bioguide entries for the same key.
	+	if not any(b == bioguide for b, _ in lookup[key]):
	+	lookup[key].append((bioguide, terms))
	+	return lookup
	+
	+
	+	def _congress_year_range(congress: int) -> tuple[int, int]:
	+	"""Return the approximate (start_year, end_year) for a given congress."""
	+	start = 1789 + 2 * (congress - 1)
	+	return start, start + 2
	+
	+
	+	def augment_crosswalk_from_voteview_members(
	+	conn: sqlite3.Connection,
	+	crosswalk: dict[int, str],
	+	congress: int,
	+	chamber_prefix: str,
	+	name_lookup: dict,
	+	) -> int:
	+	"""
	+	For ICPSR IDs in Voteview's members CSV that are NOT yet in the crosswalk,
	+	attempt a name + state + congress-term match against congress-legislators.
	+
	+	Only adds an entry when the match is UNAMBIGUOUS (exactly one bioguide
	+	for that last name + state with a term overlapping the congress year range).
	+
	+	Returns the number of new entries added to crosswalk (and the DB cache).
	+	"""
	+	url = f"{_VOTEVIEW_BASE}/members/{chamber_prefix}{congress}_members.csv"
	+	rows = _fetch_csv(url)
	+	if rows is None:
	+	return 0
	+
	+	now = datetime.now(timezone.utc).isoformat()
	+	added = 0
	+	cong_start, cong_end = _congress_year_range(congress)
	+
	+	for row in rows:
	+	try:
	+	icpsr = int(float(row.get("icpsr") or 0))
	+	except (ValueError, TypeError):
	+	continue
	+	if icpsr <= 0 or icpsr in crosswalk:
	+	continue
	+
	+	bioname = (row.get("bioname") or "").strip()
	+	state = (row.get("state_abbrev") or "").strip().upper()
	+	if not bioname or not state:
	+	continue
	+
	+	# Parse Voteview "LAST, First M." format.
	+	if "," in bioname:
	+	raw_last = bioname.split(",", 1)[0].strip()
	+	else:
	+	raw_last = bioname.strip()
	+	# Normalize: lowercase, remove common suffixes, handle hyphens.
	+	last_norm = raw_last.lower()
	+
	+	key = (last_norm, state)
	+	candidates = name_lookup.get(key, [])
	+
	+	# Filter candidates by congress term overlap.
	+	valid_bios: list[str] = []
	+	for bio, terms in candidates:
	+	for term in terms:
	+	try:
	+	ty_start = int((term.get("start") or "1800-01-01")[:4])
	+	ty_end = int((term.get("end") or "2100-01-01")[:4])
	+	except (ValueError, TypeError):
	+	continue
	+	if ty_start <= cong_end and ty_end >= cong_start:
	+	valid_bios.append(bio)
	+	break
	+
	+	if len(valid_bios) != 1:
	+	# Ambiguous or no match - skip to avoid false crosswalk entries.
	+	continue
	+
	+	bioguide = valid_bios[0]
	+	crosswalk[icpsr] = bioguide
	+	conn.execute(
	+	"""
	+	INSERT INTO voteview_crosswalk
	+	(icpsr_id, bioguide_id, full_name, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(icpsr_id) DO UPDATE SET
	+	bioguide_id = excluded.bioguide_id,
	+	full_name = excluded.full_name,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(icpsr, bioguide, bioname, now),
	+	)
	+	added += 1
	+
	+	if added > 0:
	+	conn.commit()
	+	logger.debug(
	+	"voteview: name-match: added %d new ICPSR→bioguide mappings "
	+	"for %s congress %d",
	+	added, chamber_prefix, congress,
	+	)
	+	return added
	+
	+
	+	# ─── CSV helpers ──────────────────────────────────────────────────────────────
	+
	+	def _fetch_csv(url: str, timeout: int = 120) -> Optional[list[dict]]:
	+	"""
	+	Fetch a CSV from url and return it as a list of dicts.
	+	Returns None on HTTP 404 or network error.
	+	"""
	+	try:
	+	resp = requests.get(url, timeout=timeout)
	+	resp.raise_for_status()
	+	except requests.HTTPError as exc:
	+	if exc.response is not None and exc.response.status_code == 404:
	+	logger.debug("voteview: CSV not found (404): %s", url)
	+	else:
	+	logger.warning("voteview: HTTP error fetching %s: %s", url, exc)
	+	return None
	+	except requests.RequestException as exc:
	+	logger.warning("voteview: network error fetching %s: %s", url, exc)
	+	return None
	+
	+	try:
	+	content = resp.content.decode("utf-8", errors="replace")
	+	return list(csv.DictReader(io.StringIO(content)))
	+	except Exception as exc:
	+	logger.error("voteview: CSV parse error %s: %s", url, exc)
	+	return None
	+
	+
	+	def _safe_int(val, default: int = 0) -> int:
	+	"""Convert val to int, returning default on failure."""
	+	try:
	+	return int(float(val))
	+	except (ValueError, TypeError):
	+	return default
	+
	+
	+	def _derive_session(date_str: str, csv_session: str) -> int:
	+	"""
	+	Derive the congressional session (1 or 2) from the rollcall row.
	+
	+	Prefers the CSV's session field. Falls back to year parity:
	+	odd year → session 1
	+	even year → session 2
	+	"""
	+	try:
	+	return int(float(csv_session))
	+	except (ValueError, TypeError):
	+	pass
	+	try:
	+	year = int(date_str[:4])
	+	return 1 if year % 2 == 1 else 2
	+	except (ValueError, TypeError):
	+	return 1
	+
	+
	+	def _aggregate_from_cast_codes(
	+	member_rows: list[dict],
	+	) -> tuple[int, int, int, int]:
	+	"""
	+	Compute (yea, nay, present, not_voting) from a list of member-vote dicts.
	+	Used to fill Present / Not Voting when the rollcall CSV omits them.
	+	"""
	+	yea = nay = present = not_voting = 0
	+	for mv in member_rows:
	+	pos = cast_code_to_position(mv.get("cast_code", ""))
	+	if pos == "Yea":
	+	yea += 1
	+	elif pos == "Nay":
	+	nay += 1
	+	elif pos == "Present":
	+	present += 1
	+	else:
	+	not_voting += 1
	+	return yea, nay, present, not_voting
	+
	+
	+	# ─── Core ingestion for one congress × chamber ───────────────────────────────
	+
	+	def _ingest_congress_chamber(
	+	conn: sqlite3.Connection,
	+	congress: int,
	+	chamber_prefix: str, # "H" or "S"
	+	chamber_name: str, # "house" or "senate"
	+	crosswalk: dict[int, str],
	+	bg_to_id: dict[str, int],
	+	) -> tuple[int, int]:
	+	"""
	+	Fetch Voteview CSVs for one congress + chamber and upsert into the DB.
	+
	+	Returns (rollcalls_ingested, positions_inserted).
	+	"""
	+	# ── Rollcalls (vote metadata) ─────────────────────────────────────────────
	+	rc_url = _rollcalls_url(chamber_prefix, congress)
	+	rc_rows = _fetch_csv(rc_url)
	+	if rc_rows is None:
	+	logger.warning(
	+	"voteview: no rollcalls CSV for %s congress %d (skipping)",
	+	chamber_name, congress,
	+	)
	+	return 0, 0
	+
	+	# Build rollnumber → metadata lookup.
	+	rc_meta: dict[int, dict] = {}
	+	for row in rc_rows:
	+	try:
	+	rn = int(float(row.get("rollnumber") or 0))
	+	except (ValueError, TypeError):
	+	continue
	+	if rn > 0:
	+	rc_meta[rn] = row
	+
	+	logger.info(
	+	"voteview: congress %d %s: %d rollcalls in CSV",
	+	congress, chamber_name, len(rc_meta),
	+	)
	+	if not rc_meta:
	+	return 0, 0
	+
	+	# ── Member votes (cast codes) ─────────────────────────────────────────────
	+	mv_url = _votes_url(chamber_prefix, congress)
	+	mv_rows = _fetch_csv(mv_url)
	+	if mv_rows is None:
	+	logger.warning(
	+	"voteview: no member-votes CSV for %s congress %d (metadata only)",
	+	chamber_name, congress,
	+	)
	+	mv_rows = []
	+
	+	# Group member votes by rollnumber.
	+	member_votes_by_roll: dict[int, list[dict]] = {}
	+	for mv in mv_rows:
	+	try:
	+	rn = int(float(mv.get("rollnumber") or 0))
	+	except (ValueError, TypeError):
	+	continue
	+	member_votes_by_roll.setdefault(rn, []).append(mv)
	+
	+	# ── Process each rollcall ─────────────────────────────────────────────────
	+	rollcalls_ingested = 0
	+	positions_inserted = 0
	+
	+	for rollnumber, rc in rc_meta.items():
	+	external_id = f"{chamber_prefix}{congress}-{rollnumber}"
	+
	+	# Idempotency check.
	+	if not is_new_event(conn, SOURCE, external_id, external_id):
	+	continue
	+
	+	date_str = (rc.get("date") or "").strip()
	+	year = _safe_int(date_str[:4]) if len(date_str) >= 4 else 0
	+	session = _derive_session(date_str, rc.get("session") or "")
	+
	+	# ── Official source URL (neutrality rule #1) ──────────────────────────
	+	if chamber_name == "house":
	+	clerk_rn_raw = rc.get("clerk_rollnumber") or ""
	+	try:
	+	clerk_rn = int(float(clerk_rn_raw))
	+	except (ValueError, TypeError):
	+	clerk_rn = rollnumber # Voteview rollnumber as fallback
	+	if year <= 0:
	+	year = 2017 + 2 * (congress - 115) # approximate from congress
	+	source_url = _house_source_url(year, clerk_rn)
	+	else:
	+	source_url = _senate_source_url(congress, session, rollnumber)
	+
	+	try:
	+	validate_source_url(source_url)
	+	except Exception as exc:
	+	logger.error(
	+	"voteview: invalid source_url %s: %s - skipping rollcall %s",
	+	source_url, exc, external_id,
	+	)
	+	mark_seen(conn, SOURCE, external_id, external_id)
	+	continue
	+
	+	# ── Aggregate counts ──────────────────────────────────────────────────
	+	member_rows = member_votes_by_roll.get(rollnumber, [])
	+	_, _, cc_present, cc_not_voting = _aggregate_from_cast_codes(member_rows)
	+
	+	# Prefer CSV aggregate for Yea / Nay (official chamber tally).
	+	csv_yea = _safe_int(rc.get("yea_count"))
	+	csv_nay = _safe_int(rc.get("nay_count"))
	+
	+	# ── Insert vote row ───────────────────────────────────────────────────
	+	bill_text_ref = (rc.get("bill_number") or "").strip() or None
	+	question = (rc.get("vote_question") or "").strip()
	+	result = (rc.get("vote_result") or "").strip()
	+	description = (
	+	(rc.get("vote_desc") or rc.get("dtl_desc") or "").strip()
	+	)
	+
	+	# House: store the House Clerk roll number in roll_number so it aligns
	+	# with votes_house.py. Senate: Voteview rollnumber = chamber number.
	+	if chamber_name == "house":
	+	official_roll = _safe_int(rc.get("clerk_rollnumber") or rollnumber)
	+	else:
	+	official_roll = rollnumber
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO votes
	+	(bill_id, chamber, date, yea, nay, present, result, source_url,
	+	congress, session, roll_number, question, not_voting, recorded,
	+	vote_external_id, bill_text_ref, year, description,
	+	voteview_rollnumber)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?,
	+	?, ?, ?, ?, ?, ?,
	+	?, ?, ?, ?,
	+	?)
	+	""",
	+	(
	+	None, # bill_id - look up later if needed
	+	chamber_name,
	+	date_str,
	+	csv_yea,
	+	csv_nay,
	+	cc_present,
	+	result,
	+	source_url,
	+	congress,
	+	session,
	+	official_roll,
	+	question,
	+	cc_not_voting,
	+	1, # recorded = True (Voteview is recorded only)
	+	external_id,
	+	bill_text_ref,
	+	year,
	+	description,
	+	rollnumber,
	+	),
	+	)
	+
	+	vote_db_id: int = conn.execute(
	+	"SELECT last_insert_rowid() AS id"
	+	).fetchone()["id"]
	+
	+	# ── Per-member positions ──────────────────────────────────────────────
	+	pos_rows: list[tuple] = []
	+	for mv in member_rows:
	+	position = cast_code_to_position(mv.get("cast_code"))
	+	if position is None:
	+	continue
	+	try:
	+	icpsr = int(float(mv.get("icpsr") or 0))
	+	except (ValueError, TypeError):
	+	continue
	+	if icpsr <= 0:
	+	continue
	+
	+	bioguide = crosswalk.get(icpsr)
	+	member_id = bg_to_id.get(bioguide) if bioguide else None
	+	pos_rows.append((vote_db_id, bioguide, member_id, position))
	+
	+	if pos_rows:
	+	conn.executemany(
	+	"""
	+	INSERT INTO vote_positions
	+	(vote_id, member_bioguide, member_id, position)
	+	VALUES (?, ?, ?, ?)
	+	""",
	+	pos_rows,
	+	)
	+	positions_inserted += len(pos_rows)
	+
	+	# ── Feed event ────────────────────────────────────────────────────────
	+	conn.execute(
	+	"""
	+	INSERT INTO events
	+	(ref_type, ref_id, action_date, dollar_amount,
	+	affected_estimate, action_type)
	+	VALUES (?, ?, ?, ?, ?, ?)
	+	""",
	+	("vote", vote_db_id, date_str, None, None, "floor_vote"),
	+	)
	+
	+	mark_seen(conn, SOURCE, external_id, external_id)
	+	rollcalls_ingested += 1
	+
	+	conn.commit()
	+	logger.info(
	+	"voteview: congress %d %s: ingested %d rollcalls, %d positions",
	+	congress, chamber_name, rollcalls_ingested, positions_inserted,
	+	)
	+	return rollcalls_ingested, positions_inserted
	+
	+
	+	# ─── Main poll entry point ────────────────────────────────────────────────────
	+
	+	def poll(
	+	start_congress: int = 116,
	+	end_congress: int = 119,
	+	db_path: Optional[str] = None,
	+	force_crosswalk_refresh: bool = False,
	+	) -> dict:
	+	"""
	+	Ingest Voteview roll-call data for the given congress range.
	+
	+	Parameters
	+	----------
	+	start_congress : first congress to ingest (inclusive)
	+	end_congress : last congress to ingest (inclusive)
	+	db_path : override for SQLite DB path (uses env/default if None)
	+	force_crosswalk_refresh : re-download the ICPSR crosswalk even if cached
	+
	+	Returns
	+	-------
	+	dict with keys:
	+	rollcalls_total - total vote rows inserted
	+	positions_total - total vote_position rows inserted
	+	per_congress - list of per-(congress,chamber) dicts
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+	ensure_voteview_schema(conn)
	+
	+	# Build bioguide → member_id lookup for FK joins.
	+	bg_to_id: dict[str, int] = {
	+	row["bioguide_id"]: row["id"]
	+	for row in conn.execute("SELECT id, bioguide_id FROM members")
	+	}
	+
	+	crosswalk = build_crosswalk(conn, force_refresh=force_crosswalk_refresh)
	+
	+	# Build a name-based secondary lookup from congress-legislators JSON, used
	+	# to resolve ICPSR IDs that are absent from the primary id.icpsr field
	+	# (common for members elected 2018+; ICPSR assignment lags new service).
	+	logger.info("voteview: building name-lookup for supplementary crosswalk …")
	+	all_legislators: list[dict] = []
	+	for url in _LEGISLATORS_URLS:
	+	try:
	+	resp = requests.get(url, timeout=60)
	+	resp.raise_for_status()
	+	all_legislators.extend(resp.json())
	+	except Exception as exc:
	+	logger.warning("voteview: could not fetch %s for name lookup: %s", url, exc)
	+	name_lookup = _build_name_lookup(all_legislators)
	+	logger.info("voteview: name-lookup built (%d last-name+state keys)", len(name_lookup))
	+
	+	total_rollcalls = 0
	+	total_positions = 0
	+	per_congress_log = []
	+
	+	for congress in range(start_congress, end_congress + 1):
	+	for chamber_prefix, chamber_name in [("H", "house"), ("S", "senate")]:
	+	# Augment crosswalk with name-matching for this specific congress+chamber
	+	# before ingesting votes (so positions get bioguides from the start).
	+	new_cw = augment_crosswalk_from_voteview_members(
	+	conn, crosswalk, congress, chamber_prefix, name_lookup,
	+	)
	+	if new_cw:
	+	logger.info(
	+	"voteview: name-match added %d ICPSR→bioguide for %s%d",
	+	new_cw, chamber_prefix, congress,
	+	)
	+
	+	rc_count, pos_count = _ingest_congress_chamber(
	+	conn, congress, chamber_prefix, chamber_name,
	+	crosswalk, bg_to_id,
	+	)
	+	total_rollcalls += rc_count
	+	total_positions += pos_count
	+	per_congress_log.append(
	+	{
	+	"congress": congress,
	+	"chamber": chamber_name,
	+	"rollcalls": rc_count,
	+	"positions": pos_count,
	+	}
	+	)
	+
	+	conn.close()
	+	logger.info(
	+	"voteview: done - %d rollcalls, %d positions (congress %d-%d)",
	+	total_rollcalls, total_positions, start_congress, end_congress,
	+	)
	+	return {
	+	"rollcalls_total": total_rollcalls,
	+	"positions_total": total_positions,
	+	"per_congress": per_congress_log,
	+	}
	+
	+
	+	# ─── Year-index helpers ───────────────────────────────────────────────────────
	+
	+	def votes_by_member_year(
	+	bioguide: str,
	+	year: int,
	+	db_path: Optional[str] = None,
	+	) -> list[dict]:
	+	"""
	+	Return all recorded votes for bioguide in year, with full metadata.
	+
	+	Leverages idx_votes_year + idx_vote_positions_bioguide.
	+
	+	Returns a list of dicts, each with:
	+	date, chamber, congress, roll_number, question, result,
	+	yea, nay, present, not_voting, bill_text_ref, description,
	+	source_url, position (the member's individual vote)
	+
	+	Example:
	+	rows = votes_by_member_year("N000002", 2023)
	+	# → all 2023 House votes for Jerry Nadler with their positions
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	rows = conn.execute(
	+	"""
	+	SELECT
	+	v.id,
	+	v.date,
	+	v.chamber,
	+	v.congress,
	+	v.roll_number,
	+	v.question,
	+	v.result,
	+	v.yea,
	+	v.nay,
	+	v.present,
	+	v.not_voting,
	+	v.bill_text_ref,
	+	v.description,
	+	v.source_url,
	+	vp.position
	+	FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = ?
	+	AND v.year = ?
	+	ORDER BY v.date, v.chamber, v.roll_number
	+	""",
	+	(bioguide, year),
	+	).fetchall()
	+	conn.close()
	+	return [dict(r) for r in rows]
	+
	+
	+	def votes_in_year(
	+	year: int,
	+	db_path: Optional[str] = None,
	+	) -> list[dict]:
	+	"""
	+	Return all recorded votes that took place in year.
	+
	+	Leverages idx_votes_year.
	+
	+	Returns a list of dicts, each with:
	+	id, date, chamber, congress, roll_number, question, result,
	+	yea, nay, present, not_voting, bill_text_ref, description, source_url
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	rows = conn.execute(
	+	"""
	+	SELECT
	+	id, date, chamber, congress, roll_number, question, result,
	+	yea, nay, present, not_voting, bill_text_ref, description, source_url
	+	FROM votes
	+	WHERE year = ?
	+	AND recorded = 1
	+	ORDER BY date, chamber, roll_number
	+	""",
	+	(year,),
	+	).fetchall()
	+	conn.close()
	+	return [dict(r) for r in rows]
	+
	+
	+	# ─── Completeness report ──────────────────────────────────────────────────────
	+
	+	def completeness_report(
	+	start_congress: int,
	+	end_congress: int,
	+	db_path: Optional[str] = None,
	+	) -> list[dict]:
	+	"""
	+	Compare ingested rollcall counts against Voteview CSV row counts.
	+
	+	For each (congress, chamber) pair, fetches the rollcalls CSV and compares
	+	its row count against the number of rows recorded in the `seen` table.
	+
	+	Returns a list of dicts:
	+	congress, chamber, csv_count, db_count, match, gap_note
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	report = []
	+
	+	for congress in range(start_congress, end_congress + 1):
	+	for chamber_prefix, chamber_name in [("H", "house"), ("S", "senate")]:
	+	# Count from DB via `seen` table (most reliable - one entry per
	+	# processed rollcall, written atomically with mark_seen).
	+	db_count = conn.execute(
	+	"SELECT COUNT(*) FROM seen WHERE source=? AND external_id LIKE ?",
	+	(SOURCE, f"{chamber_prefix}{congress}-%"),
	+	).fetchone()[0]
	+
	+	# Count from Voteview CSV (source of truth).
	+	rc_url = _rollcalls_url(chamber_prefix, congress)
	+	rc_rows = _fetch_csv(rc_url)
	+	if rc_rows is None:
	+	csv_count = None
	+	gap_note = "CSV unavailable"
	+	else:
	+	csv_count = len(rc_rows)
	+	# Any CSV rows with a 0/blank rollnumber would be skipped.
	+	valid_csv = sum(
	+	1 for r in rc_rows
	+	if _safe_int(r.get("rollnumber") or 0) > 0
	+	)
	+	gap = valid_csv - db_count
	+	if gap == 0:
	+	gap_note = "exact match"
	+	elif gap > 0:
	+	gap_note = f"{gap} rollcalls in CSV not yet ingested"
	+	else:
	+	gap_note = f"DB count exceeds CSV by {abs(gap)} (re-run possible?)"
	+
	+	report.append(
	+	{
	+	"congress": congress,
	+	"chamber": chamber_name,
	+	"csv_count": csv_count,
	+	"db_count": db_count,
	+	"match": (db_count == csv_count) if csv_count is not None else None,
	+	"gap_note": gap_note if csv_count is not None else "CSV unavailable",
	+	}
	+	)
	+
	+	conn.close()
	+	return report
	+
	+
	+	# ─── CLI ──────────────────────────────────────────────────────────────────────
	+
	+	def main() -> None:
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	)
	+	parser = argparse.ArgumentParser(
	+	description=(
	+	"Docket Voteview poller - ingest per-member roll-call votes "
	+	"from Voteview bulk CSVs (recorded votes only)."
	+	)
	+	)
	+	parser.add_argument(
	+	"--once", action="store_true",
	+	help="Run once and exit (required for CLI use).",
	+	)
	+	parser.add_argument(
	+	"--start", type=int, default=116,
	+	help="First congress to ingest (default: 116, i.e. 2019).",
	+	)
	+	parser.add_argument(
	+	"--end", type=int, default=119,
	+	help="Last congress to ingest (default: 119, i.e. 2025-26).",
	+	)
	+	parser.add_argument(
	+	"--db", help="Override DB path (uses DOCKET_DB_PATH env or default if omitted).",
	+	)
	+	parser.add_argument(
	+	"--refresh-crosswalk", action="store_true",
	+	help="Force re-download of the ICPSR→bioguide crosswalk.",
	+	)
	+	parser.add_argument(
	+	"--report", action="store_true",
	+	help="Print a completeness comparison table after ingestion.",
	+	)
	+
	+	args = parser.parse_args()
	+	if not args.once:
	+	parser.print_help()
	+	return
	+
	+	result = poll(
	+	start_congress=args.start,
	+	end_congress=args.end,
	+	db_path=args.db,
	+	force_crosswalk_refresh=args.refresh_crosswalk,
	+	)
	+
	+	print(
	+	f"\nvoteview: ingested {result['rollcalls_total']} vote rows, "
	+	f"{result['positions_total']} position rows "
	+	f"(congress {args.start}-{args.end})\n"
	+	)
	+	print(f"{'Congress':>10} {'Chamber':<8} {'Rollcalls':>10} {'Positions':>10}")
	+	print("-" * 48)
	+	for row in result["per_congress"]:
	+	print(
	+	f"{row['congress']:>10} {row['chamber']:<8} "
	+	f"{row['rollcalls']:>10} {row['positions']:>10}"
	+	)
	+
	+	if args.report:
	+	print("\n── Completeness check (DB vs Voteview CSV) ──")
	+	cmp = completeness_report(args.start, args.end, db_path=args.db)
	+	print(
	+	f"\n{'Congress':>10} {'Chamber':<8} {'CSV rows':>10} "
	+	f"{'DB rows':>8} {'Match':<6} Note"
	+	)
	+	print("-" * 70)
	+	for row in cmp:
	+	print(
	+	f"{row['congress']:>10} {row['chamber']:<8} "
	+	f"{str(row['csv_count']):>10} {row['db_count']:>8} "
	+	f"{'yes' if row['match'] else 'NO':<6} {row['gap_note']}"
	+	)
	+
	+
	+	if __name__ == "__main__":
	+	main()

services/ingest/ingest/run.py +6 -0

		@@ -52,7 +52,13 @@ PIPELINE = [
		("openstates", "ingest.pollers.openstates", []),
		("legiscan", "ingest.pollers.legiscan", []),
		("state_coverage", "ingest.state_coverage", []),
	+	# post-processing (run after raw ingest)
	+	("categorize", "ingest.categorize", []),
	+	("calendar", "ingest.calendar", []),
		]
	+	# Heavy/optional, not in the default --once cycle: historical voting backfill.
	+	# python -m ingest.pollers.voteview --once --start 101 --end 119
	+	# Continuous change detection is a wrapper around this runner: ingest.watcher.

		_BY_LABEL = {label: (mod, args) for label, mod, args in PIPELINE}
		# short aliases

services/ingest/ingest/watcher.py +601 -0

		@@ -0,0 +1,601 @@
	+	"""
	+	Docket ingest watcher - ADDED / UPDATED / REMOVED change detection.
	+
	+	Run modes
	+	---------
	+	python -m ingest.watcher --once # single cycle (systemd timer)
	+	python -m ingest.watcher --loop --interval 1800 # continuous loop (N seconds)
	+	python -m ingest.watcher --once --sources federal_register members
	+
	+	Writes
	+	------
	+	change_events (id, ts, source, ref_type, ref_id, change, summary, source_url)
	+	poller_watermarks (source, last_run_at, cycle_id)
	+	watcher_manifest (source, external_id, cycle_id, source_url)
	+
	+	ADDED / UPDATED detection
	+	--------------------------
	+	Snapshots the `seen` table before and after the pipeline run.
	+	• New (source, external_id) in post-snapshot → ADDED
	+	• Same key, different update_date → UPDATED
	+	This works for every poller that calls `mark_seen()` (the majority).
	+	Members (which does a blind upsert without mark_seen) is covered separately
	+	via the manifest path below.
	+
	+	REMOVED detection - full-set sources only
	+	------------------------------------------
	+	Only sources that return the complete current set on every poll can
	+	reliably detect removals. Two safety guards prevent false positives:
	+
	+	1. Non-empty guard - if the current pull returned 0 items for a source,
	+	no removals are flagged. An empty result likely indicates a failed
	+	or partial pull, not a genuine mass-removal.
	+
	+	2. First-run guard - if there is no prior manifest entry for the source,
	+	no removals are flagged. Without a baseline, there is nothing to diff.
	+
	+	Append-only / paginated / time-windowed sources (federal_register,
	+	usaspending, congress, votes, crec, lobbying, openstates, legiscan, …)
	+	are NOT in FULL_SET_SOURCES and never produce REMOVED events.
	+	"""
	+
	+	import argparse
	+	import json
	+	import logging
	+	import os
	+	import subprocess
	+	import sys
	+	import time
	+	from datetime import datetime, timezone
	+	from typing import Optional
	+
	+	from ingest.db import DB_PATH, get_connection, init_schema
	+
	+	logger = logging.getLogger("ingest.watcher")
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Full-set source registry
	+	# ---------------------------------------------------------------------------
	+	# Each entry: data-table name, primary-key column, optional WHERE filter,
	+	# source_url column, and ref_type label for change_events rows.
	+	# Add a source here ONLY when its poll() returns the entire current live set.
	+	FULL_SET_CONFIGS: dict[str, dict] = {
	+	"openomb": {
	+	"table": "apportionments",
	+	"id_col": "external_id",
	+	"where": "source = 'openomb'",
	+	"url_col": "source_url",
	+	"ref_type": "apportionment",
	+	},
	+	"members": {
	+	"table": "members",
	+	"id_col": "bioguide_id",
	+	"where": None, # all rows belong to this source
	+	"url_col": "source_url",
	+	"ref_type": "member",
	+	},
	+	}
	+
	+	FULL_SET_SOURCES: frozenset[str] = frozenset(FULL_SET_CONFIGS)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# ref_type lookup
	+	# ---------------------------------------------------------------------------
	+
	+	_SOURCE_TO_REF_TYPE: dict[str, str] = {
	+	"bills_bulk": "bill",
	+	"bill_versions": "bill",
	+	"congress": "bill",
	+	"govinfo": "bill",
	+	"amendments": "amendment",
	+	"members": "member",
	+	"usaspending": "award",
	+	"federal_register": "executive_action",
	+	"treasury": "fiscal",
	+	"openomb": "apportionment",
	+	"votes_senate": "vote",
	+	"votes_house": "vote",
	+	"votes_enrich": "vote",
	+	"crec": "procedural_event",
	+	"lobbying": "lobbying",
	+	"openstates": "state_bill",
	+	"legiscan": "state_bill",
	+	}
	+
	+
	+	def _ref_type(source: str) -> str:
	+	return _SOURCE_TO_REF_TYPE.get(source, source)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Schema (watcher-owned tables - additive, never touches existing tables)
	+	# ---------------------------------------------------------------------------
	+
	+	def _ensure_watcher_tables(conn) -> None:
	+	"""Create watcher-owned tables if they do not exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS poller_watermarks (
	+	source TEXT PRIMARY KEY,
	+	last_run_at TEXT NOT NULL,
	+	cycle_id INTEGER NOT NULL DEFAULT 0
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS change_events (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	ts TEXT NOT NULL,
	+	source TEXT NOT NULL,
	+	ref_type TEXT NOT NULL,
	+	ref_id TEXT NOT NULL,
	+	change TEXT NOT NULL CHECK(change IN ('added','updated','removed')),
	+	summary TEXT,
	+	source_url TEXT NOT NULL
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS watcher_manifest (
	+	source TEXT NOT NULL,
	+	external_id TEXT NOT NULL,
	+	cycle_id INTEGER NOT NULL,
	+	source_url TEXT NOT NULL DEFAULT '',
	+	PRIMARY KEY (source, external_id)
	+	);
	+	"""
	+	)
	+	conn.commit()
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _snapshot_seen(conn) -> dict[tuple, str]:
	+	"""Return {(source, external_id): update_date} for the full seen table."""
	+	rows = conn.execute(
	+	"SELECT source, external_id, update_date FROM seen"
	+	).fetchall()
	+	return {(r["source"], r["external_id"]): r["update_date"] for r in rows}
	+
	+
	+	def _lookup_source_url(conn, source: str, external_id: str) -> str:
	+	"""
	+	Best-effort source_url lookup for a (source, external_id) pair.
	+	Priority:
	+	1. raw_events payload (json-parsed for a URL key)
	+	2. Source-specific data table
	+	3. Sentinel (informational, not user-facing)
	+	"""
	+	# 1. raw_events payload
	+	row = conn.execute(
	+	"SELECT payload FROM raw_events WHERE source=? AND external_id=?",
	+	(source, external_id),
	+	).fetchone()
	+	if row:
	+	try:
	+	payload = json.loads(row["payload"])
	+	for key in ("html_url", "source_url", "url", "link", "detailUrl"):
	+	val = payload.get(key)
	+	if val and isinstance(val, str):
	+	return val
	+	except Exception:
	+	pass
	+
	+	# 2. Data-table fallbacks
	+	_TABLE_QUERIES: dict[str, str] = {
	+	"bills_bulk":
	+	"SELECT source_url FROM bills WHERE source=? AND external_id=?",
	+	"bill_versions":
	+	"SELECT source_url FROM bills WHERE source=? AND external_id=?",
	+	"congress":
	+	"SELECT source_url FROM bills WHERE source=? AND external_id=?",
	+	"govinfo":
	+	"SELECT source_url FROM bills WHERE source=? AND external_id=?",
	+	"usaspending":
	+	"SELECT source_url FROM awards WHERE source=? AND external_id=?",
	+	"federal_register":
	+	"SELECT source_url FROM executive_actions WHERE source=? AND external_id=?",
	+	"openomb":
	+	"SELECT source_url FROM apportionments WHERE source=? AND external_id=?",
	+	}
	+	query = _TABLE_QUERIES.get(source)
	+	if query:
	+	try:
	+	r2 = conn.execute(query, (source, external_id)).fetchone()
	+	if r2:
	+	return r2["source_url"]
	+	except Exception:
	+	pass
	+
	+	# members uses bioguide_id as PK, no source column
	+	if source == "members":
	+	try:
	+	r3 = conn.execute(
	+	"SELECT source_url FROM members WHERE bioguide_id=?",
	+	(external_id,),
	+	).fetchone()
	+	if r3:
	+	return r3["source_url"]
	+	except Exception:
	+	pass
	+
	+	# 3. Sentinel - indicates lookup failed; keeps source_url non-null
	+	return f"https://docket.internal/source/{source}/{external_id}"
	+
	+
	+	def _get_next_cycle_id(conn) -> int:
	+	row = conn.execute(
	+	"SELECT COALESCE(MAX(cycle_id), 0) AS m FROM poller_watermarks"
	+	).fetchone()
	+	return (row["m"] if row else 0) + 1
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# ADDED / UPDATED - seen-table diff
	+	# ---------------------------------------------------------------------------
	+
	+	def _classify_added_updated(
	+	before: dict,
	+	after: dict,
	+	ts: str,
	+	conn,
	+	) -> tuple[int, int]:
	+	"""
	+	Diff before/after snapshots of the seen table and write change_events.
	+
	+	Returns (added_count, updated_count).
	+	"""
	+	added_keys = set(after) - set(before)
	+	updated_keys = {k for k in set(before) & set(after) if before[k] != after[k]}
	+
	+	added_count = 0
	+	for source, ext_id in sorted(added_keys):
	+	source_url = _lookup_source_url(conn, source, ext_id)
	+	rt = _ref_type(source)
	+	conn.execute(
	+	"INSERT INTO change_events"
	+	" (ts,source,ref_type,ref_id,change,summary,source_url)"
	+	" VALUES (?,?,?,?,?,?,?)",
	+	(ts, source, rt, ext_id, "added",
	+	f"New {rt} from {source}: {ext_id}", source_url),
	+	)
	+	added_count += 1
	+
	+	updated_count = 0
	+	for source, ext_id in sorted(updated_keys):
	+	source_url = _lookup_source_url(conn, source, ext_id)
	+	rt = _ref_type(source)
	+	old, new = before[(source, ext_id)], after[(source, ext_id)]
	+	conn.execute(
	+	"INSERT INTO change_events"
	+	" (ts,source,ref_type,ref_id,change,summary,source_url)"
	+	" VALUES (?,?,?,?,?,?,?)",
	+	(ts, source, rt, ext_id, "updated",
	+	f"{rt} {ext_id} updated ({old} → {new})", source_url),
	+	)
	+	updated_count += 1
	+
	+	return added_count, updated_count
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# REMOVED / ADDED (manifest) - full-set sources only
	+	# ---------------------------------------------------------------------------
	+
	+	def _query_current_set(conn, cfg: dict) -> dict[str, str]:
	+	"""
	+	Query the data table for a full-set source.
	+	Returns {external_id: source_url}, or {} if the table doesn't exist yet.
	+	"""
	+	table, id_col, where, url_col = (
	+	cfg["table"], cfg["id_col"], cfg.get("where"), cfg["url_col"]
	+	)
	+	exists = conn.execute(
	+	"SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,)
	+	).fetchone()
	+	if not exists:
	+	return {}
	+	sql = f"SELECT {id_col}, {url_col} FROM {table}"
	+	if where:
	+	sql += f" WHERE {where}"
	+	try:
	+	rows = conn.execute(sql).fetchall()
	+	except Exception as exc:
	+	logger.warning("watcher: _query_current_set(%s) failed: %s", table, exc)
	+	return {}
	+	return {r[0]: (r[1] or "") for r in rows}
	+
	+
	+	def _load_manifest(conn, source: str) -> dict[str, str]:
	+	"""Load manifest for source from previous cycle: {external_id: source_url}."""
	+	rows = conn.execute(
	+	"SELECT external_id, source_url FROM watcher_manifest WHERE source=?",
	+	(source,),
	+	).fetchall()
	+	return {r["external_id"]: r["source_url"] for r in rows}
	+
	+
	+	def _update_manifest(conn, source: str, current_set: dict, cycle_id: int) -> None:
	+	"""Replace the manifest for source with the current pull's set."""
	+	conn.execute("DELETE FROM watcher_manifest WHERE source=?", (source,))
	+	for ext_id, source_url in current_set.items():
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	(source, ext_id, cycle_id, source_url or ""),
	+	)
	+
	+
	+	def _detect_full_set_changes(
	+	conn,
	+	source: str,
	+	cfg: dict,
	+	cycle_id: int,
	+	ts: str,
	+	) -> tuple[int, int]:
	+	"""
	+	ADDED / REMOVED detection for a full-set source via manifest comparison.
	+
	+	For sources that call mark_seen (openomb): ADDED events are already captured
	+	by the seen diff, so we skip adding them here to avoid duplicates.
	+	For sources that do NOT call mark_seen (members): we emit ADDED events here.
	+
	+	Safety guards for REMOVED:
	+	• Non-empty guard: current_set empty → no removals flagged.
	+	• First-run guard: no prior manifest → no removals flagged.
	+
	+	Returns (added_count, removed_count).
	+	"""
	+	rt = cfg["ref_type"]
	+	prev = _load_manifest(conn, source)
	+	current = _query_current_set(conn, cfg)
	+
	+	# Non-empty guard
	+	if not current:
	+	logger.warning(
	+	"watcher: %s full-set pull returned 0 items - "
	+	"removal detection skipped (non-empty guard)",
	+	source,
	+	)
	+	# Do not overwrite manifest with empty set to preserve the prior baseline
	+	return 0, 0
	+
	+	added_ids = set(current) - set(prev)
	+	removed_ids = set(prev) - set(current)
	+
	+	added_count = 0
	+	# Members does not call mark_seen, so its ADDs won't appear in the seen diff.
	+	# Emit ADDED events from the manifest for members only.
	+	if source not in ("openomb",): # openomb ADDs come from seen diff
	+	for ext_id in sorted(added_ids):
	+	source_url = current[ext_id] or _lookup_source_url(conn, source, ext_id)
	+	conn.execute(
	+	"INSERT INTO change_events"
	+	" (ts,source,ref_type,ref_id,change,summary,source_url)"
	+	" VALUES (?,?,?,?,?,?,?)",
	+	(ts, source, rt, ext_id, "added",
	+	f"New {rt}: {ext_id}", source_url),
	+	)
	+	added_count += 1
	+
	+	removed_count = 0
	+	if not prev:
	+	# First-run guard
	+	logger.info(
	+	"watcher: %s no prior manifest (first run) - "
	+	"removal detection skipped",
	+	source,
	+	)
	+	else:
	+	for ext_id in sorted(removed_ids):
	+	source_url = (
	+	prev.get(ext_id)
	+	or _lookup_source_url(conn, source, ext_id)
	+	)
	+	conn.execute(
	+	"INSERT INTO change_events"
	+	" (ts,source,ref_type,ref_id,change,summary,source_url)"
	+	" VALUES (?,?,?,?,?,?,?)",
	+	(ts, source, rt, ext_id, "removed",
	+	f"{rt} {ext_id} absent from {source} full-set pull",
	+	source_url),
	+	)
	+	removed_count += 1
	+	logger.info("watcher: REMOVED %s:%s", source, ext_id)
	+
	+	_update_manifest(conn, source, current, cycle_id)
	+	return added_count, removed_count
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Pipeline runner
	+	# ---------------------------------------------------------------------------
	+
	+	def _run_pipeline(db_path: str, sources: Optional[list]) -> None:
	+	"""Run the ingest pipeline (or a subset) as a subprocess."""
	+	env = {**os.environ, "DOCKET_DB_PATH": db_path}
	+
	+	if sources:
	+	for src in sources:
	+	cmd = [sys.executable, "-m", "ingest.run", "--source", src]
	+	logger.info("watcher: running step: %s", src)
	+	try:
	+	subprocess.run(cmd, check=False, env=env)
	+	except Exception as exc:
	+	logger.warning("watcher: step %s raised: %r", src, exc)
	+	else:
	+	logger.info("watcher: running full pipeline...")
	+	try:
	+	subprocess.run(
	+	[sys.executable, "-m", "ingest.run", "--once"],
	+	check=False,
	+	env=env,
	+	)
	+	except Exception as exc:
	+	logger.error("watcher: pipeline subprocess raised: %r", exc)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Main cycle
	+	# ---------------------------------------------------------------------------
	+
	+	def run_cycle(
	+	db_path: Optional[str] = None,
	+	sources: Optional[list] = None,
	+	) -> dict:
	+	"""
	+	Run one watcher cycle.
	+
	+	Steps
	+	-----
	+	1. Snapshot seen table (before state).
	+	2. Run ingest pipeline (subprocess, passes DOCKET_DB_PATH).
	+	3. Snapshot seen table (after state).
	+	4. Diff → ADDED / UPDATED events (seen diff, covers mark_seen pollers).
	+	5. Manifest diff → ADDED (members) / REMOVED (full-set sources with guards).
	+	6. Write change_events; update poller_watermarks.
	+
	+	Parameters
	+	----------
	+	db_path : SQLite path override (falls back to DOCKET_DB_PATH env / default).
	+	sources : If set, only these pipeline step labels are run (for testing).
	+
	+	Returns
	+	-------
	+	dict: cycle_id, added, updated, removed, total_change_events
	+	"""
	+	db_path = db_path or os.environ.get("DOCKET_DB_PATH") or DB_PATH
	+
	+	conn = get_connection(db_path)
	+	init_schema(conn)
	+	_ensure_watcher_tables(conn)
	+
	+	cycle_id = _get_next_cycle_id(conn)
	+	ts = datetime.now(timezone.utc).isoformat()
	+
	+	logger.info("=== Watcher cycle %d starting at %s ===", cycle_id, ts)
	+
	+	# 1. Pre-pipeline snapshot
	+	before = _snapshot_seen(conn)
	+	logger.info("watcher: %d seen entries before pipeline", len(before))
	+	conn.close()
	+
	+	# 2. Run pipeline (subprocess writes to DB)
	+	_run_pipeline(db_path, sources)
	+
	+	# 3. Re-open and snapshot after
	+	conn = get_connection(db_path)
	+	init_schema(conn)
	+	_ensure_watcher_tables(conn)
	+	after = _snapshot_seen(conn)
	+	logger.info(
	+	"watcher: %d seen entries after pipeline (%+d)",
	+	len(after), len(after) - len(before),
	+	)
	+
	+	# 4. Seen-table diff → ADDED / UPDATED
	+	added, updated = _classify_added_updated(before, after, ts, conn)
	+
	+	# 5. Manifest diff for full-set sources
	+	removed = 0
	+	for source, cfg in FULL_SET_CONFIGS.items():
	+	if sources and source not in sources:
	+	# Only process full-set sources that were part of this cycle
	+	continue
	+	a, r = _detect_full_set_changes(conn, source, cfg, cycle_id, ts)
	+	added += a
	+	removed += r
	+
	+	conn.commit()
	+
	+	# 6. Update watermarks
	+	all_sources = {s for (s, _) in after} \| FULL_SET_SOURCES
	+	for src in all_sources:
	+	conn.execute(
	+	"""INSERT INTO poller_watermarks (source, last_run_at, cycle_id) VALUES (?,?,?)
	+	ON CONFLICT(source) DO UPDATE SET
	+	last_run_at = excluded.last_run_at,
	+	cycle_id = excluded.cycle_id""",
	+	(src, ts, cycle_id),
	+	)
	+	conn.commit()
	+	conn.close()
	+
	+	result = {
	+	"cycle_id": cycle_id,
	+	"added": added,
	+	"updated": updated,
	+	"removed": removed,
	+	"total_change_events": added + updated + removed,
	+	}
	+	logger.info(
	+	"=== Watcher cycle %d: +%d added ~%d updated -%d removed (%d total) ===",
	+	cycle_id, added, updated, removed, added + updated + removed,
	+	)
	+	return result
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	def main() -> None:
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	parser = argparse.ArgumentParser(
	+	description="Docket watcher - detect ADDED/UPDATED/REMOVED government items."
	+	)
	+	parser.add_argument(
	+	"--once", action="store_true",
	+	help="Run a single cycle and exit (designed for systemd timer use).",
	+	)
	+	parser.add_argument(
	+	"--loop", action="store_true",
	+	help="Run continuously, sleeping --interval seconds between cycles.",
	+	)
	+	parser.add_argument(
	+	"--interval", type=int, default=1800, metavar="SECONDS",
	+	help="Sleep between cycles in --loop mode (default: 1800 = 30 min).",
	+	)
	+	parser.add_argument(
	+	"--sources", nargs="+", metavar="STEP",
	+	help="Limit to specific pipeline steps (e.g. --sources federal_register members).",
	+	)
	+	parser.add_argument(
	+	"--db", metavar="PATH",
	+	help="Override DOCKET_DB_PATH environment variable.",
	+	)
	+	args = parser.parse_args()
	+
	+	db_path = args.db or os.environ.get("DOCKET_DB_PATH") or DB_PATH
	+
	+	if args.once:
	+	result = run_cycle(db_path=db_path, sources=args.sources)
	+	print(
	+	f"Cycle {result['cycle_id']}: "
	+	f"+{result['added']} added "
	+	f"~{result['updated']} updated "
	+	f"-{result['removed']} removed "
	+	f"({result['total_change_events']} total change_events)"
	+	)
	+	return
	+
	+	if args.loop:
	+	logger.info("watcher: loop mode - interval=%ds", args.interval)
	+	while True:
	+	try:
	+	run_cycle(db_path=db_path, sources=args.sources)
	+	except Exception as exc:
	+	logger.error("watcher: cycle raised: %r", exc)
	+	logger.info("watcher: sleeping %ds until next cycle", args.interval)
	+	time.sleep(args.interval)
	+	return # unreachable
	+
	+	parser.print_help()
	+
	+
	+	if __name__ == "__main__":
	+	main()

services/ingest/tests/test_calendar.py +730 -0

		@@ -0,0 +1,730 @@
	+	"""
	+	Tests for ingest/calendar.py
	+
	+	Coverage:
	+	• parse_effective_dates() - all regex patterns, mixed text, no false positives
	+	• _normalize_date() - ISO passthrough, month-name parsing, bad input
	+	• ensure_schema() - table + index creation, effective_on column
	+	• _aggregate_effective() - pulls FR rules with effective_on from DB
	+	• _aggregate_enacted() - pulls BecameLaw actions from DB
	+	• _aggregate_votes() - pulls from votes table
	+	• _aggregate_bill_actions() - introduced / committee / floor
	+	• _aggregate_apportionments() - pulls from apportionments table
	+	• _aggregate_effective_parsed() - regex over action text
	+	• run() idempotency - double-run does not duplicate rows
	+	"""
	+
	+	import os
	+	import sqlite3
	+	import tempfile
	+
	+	import pytest
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.calendar import (
	+	EFFECTIVE_DATE_PATTERNS,
	+	_normalize_date,
	+	_aggregate_effective,
	+	_aggregate_enacted,
	+	_aggregate_votes,
	+	_aggregate_bill_actions,
	+	_aggregate_apportionments,
	+	_aggregate_effective_parsed,
	+	ensure_schema,
	+	parse_effective_dates,
	+	run,
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _make_db() -> tuple[sqlite3.Connection, str]:
	+	"""Return (conn, path) for a fresh temp DB with full schema."""
	+	path = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(path)
	+	init_schema(conn)
	+	ensure_schema(conn)
	+	return conn, path
	+
	+
	+	def _seed_executive_action(
	+	conn: sqlite3.Connection,
	+	external_id: str = "2026-99999",
	+	doc_type: str = "final_rule",
	+	title: str = "Test Rule",
	+	effective_on: str \| None = "2026-08-01",
	+	source_url: str = "https://www.federalregister.gov/documents/2026/06/26/2026-99999/test-rule",
	+	) -> int:
	+	"""Insert one executive_actions row; return its id."""
	+	conn.execute(
	+	"""
	+	INSERT INTO executive_actions
	+	(source, external_id, type, title, fr_doc_number,
	+	date, effective_on, source_url)
	+	VALUES ('federal_register', ?, ?, ?, ?, '2026-06-26', ?, ?)
	+	ON CONFLICT(source, external_id) DO NOTHING
	+	""",
	+	(external_id, doc_type, title, external_id, effective_on, source_url),
	+	)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT id FROM executive_actions WHERE external_id=?", (external_id,)
	+	).fetchone()
	+	return row["id"] if row else -1
	+
	+
	+	def _seed_bill(
	+	conn: sqlite3.Connection,
	+	external_id: str = "119-hr-1",
	+	title: str = "Test Act",
	+	source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
	+	) -> int:
	+	conn.execute(
	+	"""
	+	INSERT INTO bills
	+	(source, external_id, congress, number, title,
	+	status, updated_at, source_url)
	+	VALUES ('govinfo_billstatus', ?, 119, '1', ?, 'enacted', '2025-01-01', ?)
	+	ON CONFLICT(source, external_id) DO NOTHING
	+	""",
	+	(external_id, title, source_url),
	+	)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT id FROM bills WHERE external_id=?", (external_id,)
	+	).fetchone()
	+	return row["id"] if row else -1
	+
	+
	+	def _seed_action(
	+	conn: sqlite3.Connection,
	+	bill_id: int,
	+	action_type: str,
	+	action_date: str,
	+	text: str = "",
	+	source_url: str = "https://www.congress.gov/bill/119th-congress/house-bill/1",
	+	) -> int:
	+	conn.execute(
	+	"""
	+	INSERT INTO actions
	+	(bill_id, source, action_type, action_date, text, source_url)
	+	VALUES (?, 'govinfo_billstatus', ?, ?, ?, ?)
	+	""",
	+	(bill_id, action_type, action_date, text, source_url),
	+	)
	+	conn.commit()
	+	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
	+
	+
	+	def _seed_vote(
	+	conn: sqlite3.Connection,
	+	date: str = "2025-03-15",
	+	result: str = "Passed",
	+	source_url: str = "https://www.senate.gov/legislative/LIS/roll_call_votes/vote1191/vote_119_1_00001.xml",
	+	) -> int:
	+	conn.execute(
	+	"""
	+	INSERT INTO votes (chamber, date, yea, nay, present, result, source_url)
	+	VALUES ('senate', ?, 60, 40, 0, ?, ?)
	+	""",
	+	(date, result, source_url),
	+	)
	+	conn.commit()
	+	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
	+
	+
	+	def _seed_apportionment(
	+	conn: sqlite3.Connection,
	+	external_id: str = "test-apportionment-1",
	+	agency: str = "Department of Defense",
	+	date: str = "2025-10-01",
	+	source_url: str = "https://apportionment-public.max.gov/test.json",
	+	) -> int:
	+	# Ensure table exists
	+	conn.execute(
	+	"""
	+	CREATE TABLE IF NOT EXISTS apportionments (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'openomb',
	+	external_id TEXT NOT NULL,
	+	agency TEXT,
	+	account TEXT,
	+	appropriated REAL,
	+	apportioned REAL,
	+	status TEXT,
	+	date TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE(source, external_id)
	+	)
	+	"""
	+	)
	+	conn.execute(
	+	"""
	+	INSERT INTO apportionments
	+	(source, external_id, agency, account, status, date, source_url)
	+	VALUES ('openomb', ?, ?, 'Operations', 'released', ?, ?)
	+	ON CONFLICT(source, external_id) DO NOTHING
	+	""",
	+	(external_id, agency, date, source_url),
	+	)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT id FROM apportionments WHERE external_id=?", (external_id,)
	+	).fetchone()
	+	return row["id"] if row else -1
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _normalize_date
	+	# ---------------------------------------------------------------------------
	+
	+	class TestNormalizeDate:
	+	def test_iso_passthrough(self):
	+	assert _normalize_date("2026-08-01") == "2026-08-01"
	+
	+	def test_full_month_name(self):
	+	assert _normalize_date("August 5, 2026") == "2026-08-05"
	+
	+	def test_abbreviated_month_name(self):
	+	assert _normalize_date("Jan 1, 2025") == "2025-01-01"
	+
	+	def test_leading_trailing_space(self):
	+	assert _normalize_date(" 2026-01-15 ") == "2026-01-15"
	+
	+	def test_unparseable_returns_none(self):
	+	assert _normalize_date("sometime next year") is None
	+	assert _normalize_date("") is None
	+	assert _normalize_date("180 days after enactment") is None
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# parse_effective_dates
	+	# ---------------------------------------------------------------------------
	+
	+	class TestParseEffectiveDates:
	+	def test_shall_take_effect_on(self):
	+	text = "This section shall take effect on January 1, 2026."
	+	dates = parse_effective_dates(text)
	+	assert "2026-01-01" in dates
	+
	+	def test_takes_effect_on(self):
	+	text = "The rule takes effect on August 5, 2026."
	+	dates = parse_effective_dates(text)
	+	assert "2026-08-05" in dates
	+
	+	def test_effective_on(self):
	+	text = "Effective on September 30, 2025, the following applies."
	+	dates = parse_effective_dates(text)
	+	assert "2025-09-30" in dates
	+
	+	def test_effective_date_is(self):
	+	text = "The effective date is March 15, 2026."
	+	dates = parse_effective_dates(text)
	+	assert "2026-03-15" in dates
	+
	+	def test_effective_date_shall_be(self):
	+	text = "The effective date shall be July 4, 2026."
	+	dates = parse_effective_dates(text)
	+	assert "2026-07-04" in dates
	+
	+	def test_this_act_takes_effect(self):
	+	text = "This Act shall take effect on December 31, 2025."
	+	dates = parse_effective_dates(text)
	+	assert "2025-12-31" in dates
	+
	+	def test_iso_date_format(self):
	+	text = "Effective on 2026-06-01, new requirements apply."
	+	dates = parse_effective_dates(text)
	+	assert "2026-06-01" in dates
	+
	+	def test_no_dates_returns_empty(self):
	+	text = "The rule is effective 180 days after enactment."
	+	dates = parse_effective_dates(text)
	+	assert dates == []
	+
	+	def test_deduplication(self):
	+	text = (
	+	"Effective on January 1, 2026. "
	+	"This Act takes effect on January 1, 2026."
	+	)
	+	dates = parse_effective_dates(text)
	+	assert dates.count("2026-01-01") == 1
	+
	+	def test_multiple_distinct_dates(self):
	+	text = (
	+	"Section 1 takes effect on March 1, 2026. "
	+	"Section 2 shall take effect on June 1, 2026."
	+	)
	+	dates = parse_effective_dates(text)
	+	assert "2026-03-01" in dates
	+	assert "2026-06-01" in dates
	+	assert len(dates) == 2
	+
	+	def test_relative_clause_not_extracted(self):
	+	text = "This provision takes effect 90 days after enactment."
	+	dates = parse_effective_dates(text)
	+	assert dates == []
	+
	+	def test_empty_string(self):
	+	assert parse_effective_dates("") == []
	+
	+	def test_patterns_compile(self):
	+	"""Ensure all regex patterns are valid and pre-compiled."""
	+	for p in EFFECTIVE_DATE_PATTERNS:
	+	assert hasattr(p, "finditer"), f"Pattern {p!r} is not compiled"
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# ensure_schema
	+	# ---------------------------------------------------------------------------
	+
	+	class TestEnsureSchema:
	+	def test_table_created(self):
	+	conn, path = _make_db()
	+	try:
	+	tables = {
	+	r[0]
	+	for r in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='table'"
	+	)
	+	}
	+	assert "calendar_events" in tables
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_date_index_created(self):
	+	conn, path = _make_db()
	+	try:
	+	indexes = {
	+	r[0]
	+	for r in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='index'"
	+	)
	+	}
	+	assert "idx_calendar_events_date" in indexes
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_kind_index_created(self):
	+	conn, path = _make_db()
	+	try:
	+	indexes = {
	+	r[0]
	+	for r in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='index'"
	+	)
	+	}
	+	assert "idx_calendar_events_kind" in indexes
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_effective_on_column_added_to_executive_actions(self):
	+	conn, path = _make_db()
	+	# Manually create executive_actions WITHOUT effective_on
	+	conn.execute(
	+	"""
	+	CREATE TABLE IF NOT EXISTS executive_actions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'federal_register',
	+	external_id TEXT NOT NULL,
	+	type TEXT NOT NULL,
	+	title TEXT,
	+	agency TEXT,
	+	fr_doc_number TEXT,
	+	date TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE(source, external_id)
	+	)
	+	"""
	+	)
	+	conn.commit()
	+	ensure_schema(conn)
	+	cols = {
	+	r[1]
	+	for r in conn.execute("PRAGMA table_info(executive_actions)")
	+	}
	+	assert "effective_on" in cols
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_ensure_schema_idempotent(self):
	+	"""Calling ensure_schema twice must not raise."""
	+	conn, path = _make_db()
	+	try:
	+	ensure_schema(conn) # second call
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_kind_check_constraint(self):
	+	"""Invalid kind values must be rejected by the CHECK constraint."""
	+	conn, path = _make_db()
	+	try:
	+	with pytest.raises(sqlite3.IntegrityError):
	+	conn.execute(
	+	"""
	+	INSERT INTO calendar_events
	+	(date, kind, ref_type, source_url)
	+	VALUES ('2026-01-01', 'INVALID_KIND', 'bill',
	+	'https://www.congress.gov/')
	+	"""
	+	)
	+	conn.commit()
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_effective
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateEffective:
	+	def test_basic(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_executive_action(conn, effective_on="2026-08-05")
	+	count = _aggregate_effective(conn)
	+	conn.commit()
	+	assert count == 1
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='effective'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2026-08-05"
	+	assert row["ref_type"] == "executive_action"
	+	assert row["confidence"] == 1.0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_null_effective_on_excluded(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_executive_action(conn, effective_on=None)
	+	count = _aggregate_effective(conn)
	+	conn.commit()
	+	assert count == 0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_notice_type_excluded(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_executive_action(
	+	conn, external_id="2026-99998", doc_type="notice", effective_on="2026-09-01"
	+	)
	+	count = _aggregate_effective(conn)
	+	conn.commit()
	+	assert count == 0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_source_url_present(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_executive_action(conn, effective_on="2026-10-01")
	+	_aggregate_effective(conn)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT source_url FROM calendar_events WHERE kind='effective'"
	+	).fetchone()
	+	assert row["source_url"].startswith("https://www.federalregister.gov")
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_enacted
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateEnacted:
	+	def test_basic(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(
	+	conn, bill_id, "BecameLaw", "2025-07-04",
	+	text="Became Public Law No: 119-10.",
	+	)
	+	count = _aggregate_enacted(conn)
	+	conn.commit()
	+	assert count == 1
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='enacted'"
	+	).fetchone()
	+	assert row["date"] == "2025-07-04"
	+	assert row["ref_type"] == "bill"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_non_became_law_action_excluded(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(conn, bill_id, "IntroReferral", "2025-01-10")
	+	count = _aggregate_enacted(conn)
	+	conn.commit()
	+	assert count == 0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_votes
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateVotes:
	+	def test_basic(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_vote(conn, date="2025-05-20", result="Passed")
	+	count = _aggregate_votes(conn)
	+	conn.commit()
	+	assert count == 1
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='vote'"
	+	).fetchone()
	+	assert row["date"] == "2025-05-20"
	+	assert row["ref_type"] == "vote"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_multiple_votes(self):
	+	conn, path = _make_db()
	+	try:
	+	for day in ["2025-01-09", "2025-01-13", "2025-01-15"]:
	+	_seed_vote(conn, date=day)
	+	count = _aggregate_votes(conn)
	+	conn.commit()
	+	assert count == 3
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_bill_actions
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateBillActions:
	+	def test_introduced(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
	+	_aggregate_bill_actions(conn)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='introduced'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2025-01-03"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_committee(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(conn, bill_id, "Committee", "2025-02-14")
	+	_aggregate_bill_actions(conn)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='committee'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2025-02-14"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_floor(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(conn, bill_id, "Floor", "2025-03-01")
	+	_aggregate_bill_actions(conn)
	+	conn.commit()
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='floor'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2025-03-01"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_all_three_kinds_populated(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(conn, bill_id, "IntroReferral", "2025-01-03")
	+	_seed_action(conn, bill_id, "Committee", "2025-01-20")
	+	_seed_action(conn, bill_id, "Floor", "2025-02-10")
	+	_aggregate_bill_actions(conn)
	+	conn.commit()
	+	kinds = {
	+	r[0]
	+	for r in conn.execute(
	+	"SELECT DISTINCT kind FROM calendar_events"
	+	)
	+	}
	+	assert {"introduced", "committee", "floor"} <= kinds
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_apportionments
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateApportionments:
	+	def test_basic(self):
	+	conn, path = _make_db()
	+	try:
	+	_seed_apportionment(conn, date="2025-10-01")
	+	count = _aggregate_apportionments(conn)
	+	conn.commit()
	+	assert count == 1
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='apportionment'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2025-10-01"
	+	assert row["ref_type"] == "apportionment"
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_no_table_returns_zero(self):
	+	"""If apportionments table does not exist, return 0 gracefully."""
	+	conn, path = _make_db()
	+	try:
	+	# Don't seed the apportionments table
	+	count = _aggregate_apportionments(conn)
	+	assert count == 0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _aggregate_effective_parsed
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAggregateEffectiveParsed:
	+	def test_extracts_date_from_action_text(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(
	+	conn, bill_id, "BecameLaw", "2025-07-04",
	+	text=(
	+	"Became Public Law No: 119-10. "
	+	"This Act shall take effect on January 1, 2026."
	+	),
	+	)
	+	count = _aggregate_effective_parsed(conn)
	+	conn.commit()
	+	assert count >= 1
	+	row = conn.execute(
	+	"SELECT * FROM calendar_events WHERE kind='effective_parsed'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["date"] == "2026-01-01"
	+	assert row["confidence"] == 0.5
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+	def test_no_dates_in_text_returns_zero(self):
	+	conn, path = _make_db()
	+	try:
	+	bill_id = _seed_bill(conn)
	+	_seed_action(
	+	conn, bill_id, "BecameLaw", "2025-07-04",
	+	text="Became Public Law No: 119-10.",
	+	)
	+	count = _aggregate_effective_parsed(conn)
	+	conn.commit()
	+	assert count == 0
	+	finally:
	+	conn.close()
	+	os.unlink(path)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# run() - idempotency
	+	# ---------------------------------------------------------------------------
	+
	+	class TestRunIdempotency:
	+	def test_double_run_no_duplicates(self):
	+	"""Running run() twice must not duplicate rows."""
	+	conn, path = _make_db()
	+	conn.close() # run() opens its own connection
	+	try:
	+	bill_id_conn = get_connection(path)
	+	bill_id_conn.row_factory = sqlite3.Row
	+	init_schema(bill_id_conn)
	+	ensure_schema(bill_id_conn)
	+
	+	# Seed some data
	+	bill_id = _seed_bill(bill_id_conn)
	+	_seed_action(bill_id_conn, bill_id, "BecameLaw", "2025-07-04",
	+	text="Became Public Law No: 119-10.")
	+	_seed_action(bill_id_conn, bill_id, "IntroReferral", "2025-01-03")
	+	_seed_vote(bill_id_conn, date="2025-05-20")
	+
	+	# Add executive_actions for effective
	+	_seed_executive_action(
	+	bill_id_conn, effective_on="2026-08-01"
	+	)
	+	bill_id_conn.commit()
	+	bill_id_conn.close()
	+
	+	# First run
	+	counts1 = run(db_path=path)
	+
	+	# Second run
	+	counts2 = run(db_path=path)
	+
	+	# Counts should be the same (no duplicates)
	+	for kind in counts1:
	+	assert counts1[kind] == counts2[kind], (
	+	f"kind={kind!r}: first run={counts1[kind]}, "
	+	f"second run={counts2[kind]}"
	+	)
	+	finally:
	+	if os.path.exists(path):
	+	os.unlink(path)
	+
	+	def test_run_returns_counts_dict(self):
	+	"""run() must return a dict with at least the core kinds."""
	+	conn, path = _make_db()
	+	conn.close()
	+	try:
	+	counts = run(db_path=path)
	+	assert isinstance(counts, dict)
	+	for kind in (
	+	"effective", "enacted", "vote",
	+	"introduced", "committee", "floor",
	+	"apportionment", "effective_parsed",
	+	):
	+	assert kind in counts, f"Missing kind {kind!r} in counts"
	+	finally:
	+	if os.path.exists(path):
	+	os.unlink(path)

services/ingest/tests/test_categorize.py +433 -0

		@@ -0,0 +1,433 @@
	+	"""
	+	Tests for ingest.categorize - unified taxonomy module.
	+
	+	Covers:
	+	- agency_to_category: deterministic mapping, prefix match, unknown agencies
	+	- POLICY_AREAS: completeness and slug format
	+	- _parse_subjects: XML parsing from a synthetic BILLSTATUS snippet
	+	- Schema helpers: ensure tables and columns are created idempotently
	+	- _seed_categories: idempotent seeding, extra bill areas picked up
	+	- _backfill_awards: correct category assignment, NULL handling
	+	- category_counts: returns expected shape
	+	"""
	+
	+	import sqlite3
	+	import textwrap
	+	import xml.etree.ElementTree as ET
	+
	+	import pytest
	+
	+	from ingest.categorize import (
	+	AGENCY_CATEGORY_MAP,
	+	POLICY_AREAS,
	+	agency_to_category,
	+	_parse_subjects,
	+	_ensure_schema,
	+	_seed_categories,
	+	_backfill_awards,
	+	_slug_from_label,
	+	category_counts,
	+	)
	+	from ingest.db import init_schema
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Fixtures
	+	# ---------------------------------------------------------------------------
	+
	+	@pytest.fixture
	+	def conn():
	+	"""In-memory SQLite connection with base schema initialised."""
	+	c = sqlite3.connect(":memory:")
	+	c.row_factory = sqlite3.Row
	+	c.execute("PRAGMA foreign_keys=ON")
	+	init_schema(c)
	+	_ensure_schema(c)
	+	yield c
	+	c.close()
	+
	+
	+	def _insert_award(conn, agency: str, category: str \| None = None) -> int:
	+	conn.execute(
	+	"""
	+	INSERT INTO awards (source, external_id, agency, recipient, amount,
	+	action_date, program, source_url, category)
	+	VALUES ('usaspending', ?, ?, 'Test Recipient', 1000000,
	+	'2026-01-01', 'Test Program',
	+	'https://www.usaspending.gov/award/TEST', ?)
	+	""",
	+	(f"award-{agency[:20]}", agency, category),
	+	)
	+	conn.commit()
	+	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
	+
	+
	+	def _insert_bill(conn, policy_area: str = "", external_id: str = "119-hr-1") -> int:
	+	conn.execute(
	+	"""
	+	INSERT INTO bills (source, external_id, congress, number, title,
	+	sponsor_id, status, policy_area, updated_at, source_url)
	+	VALUES ('govinfo_billstatus', ?, 119, 'H.R. 1', 'Test Bill',
	+	'', '', ?, '2026-01-01', 'https://www.congress.gov/bill/119th-congress/house-bill/1')
	+	""",
	+	(external_id, policy_area),
	+	)
	+	conn.commit()
	+	return conn.execute("SELECT last_insert_rowid()").fetchone()[0]
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# agency_to_category tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestAgencyToCategory:
	+	"""Deterministic agency→policy-area mapping."""
	+
	+	def test_exact_match_defense(self):
	+	assert agency_to_category("Department of Defense") == "Armed Forces and National Security"
	+
	+	def test_exact_match_health(self):
	+	assert agency_to_category("Department of Health and Human Services") == "Health"
	+
	+	def test_exact_match_transportation(self):
	+	assert agency_to_category("Department of Transportation") == "Transportation and Public Works"
	+
	+	def test_exact_match_education(self):
	+	assert agency_to_category("Department of Education") == "Education"
	+
	+	def test_exact_match_energy(self):
	+	assert agency_to_category("Department of Energy") == "Energy"
	+
	+	def test_exact_match_agriculture(self):
	+	assert agency_to_category("Department of Agriculture") == "Agriculture and Food"
	+
	+	def test_exact_match_labor(self):
	+	assert agency_to_category("Department of Labor") == "Labor and Employment"
	+
	+	def test_exact_match_epa(self):
	+	assert agency_to_category("Environmental Protection Agency") == "Environmental Protection"
	+
	+	def test_exact_match_doj(self):
	+	assert agency_to_category("Department of Justice") == "Crime and Law Enforcement"
	+
	+	def test_exact_match_state(self):
	+	assert agency_to_category("Department of State") == "International Affairs"
	+
	+	def test_exact_match_usaid(self):
	+	assert agency_to_category("Agency for International Development") == "International Affairs"
	+
	+	def test_exact_match_commerce(self):
	+	assert agency_to_category("Department of Commerce") == "Commerce"
	+
	+	def test_exact_match_treasury(self):
	+	assert agency_to_category("Department of the Treasury") == "Economics and Public Finance"
	+
	+	def test_exact_match_va(self):
	+	assert agency_to_category("Department of Veterans Affairs") == "Armed Forces and National Security"
	+
	+	def test_exact_match_dhs(self):
	+	assert agency_to_category("Department of Homeland Security") == "Emergency Management"
	+
	+	def test_exact_match_interior(self):
	+	assert agency_to_category("Department of the Interior") == "Public Lands and Natural Resources"
	+
	+	def test_exact_match_hud(self):
	+	assert agency_to_category("Department of Housing and Urban Development") == "Housing and Community Development"
	+
	+	def test_exact_match_ssa(self):
	+	assert agency_to_category("Social Security Administration") == "Social Welfare"
	+
	+	def test_exact_match_nasa(self):
	+	assert agency_to_category("National Aeronautics and Space Administration") == "Science, Technology, Communications"
	+
	+	def test_exact_match_nsf(self):
	+	assert agency_to_category("National Science Foundation") == "Science, Technology, Communications"
	+
	+	def test_exact_match_gsa(self):
	+	assert agency_to_category("General Services Administration") == "Government Operations and Politics"
	+
	+	def test_prefix_match_dod_suborg(self):
	+	"""Sub-agency starting with 'Department of Defense' should map correctly."""
	+	result = agency_to_category("Department of Defense - Defense Advanced Research Projects Agency")
	+	assert result == "Armed Forces and National Security"
	+
	+	def test_unknown_agency_returns_none(self):
	+	assert agency_to_category("Some Completely Unknown Agency XYZ") is None
	+
	+	def test_empty_string_returns_none(self):
	+	assert agency_to_category("") is None
	+
	+	def test_none_returns_none(self):
	+	assert agency_to_category(None) is None
	+
	+	def test_all_map_keys_are_in_policy_area_labels(self):
	+	"""Every agency map value must correspond to a canonical policy-area label."""
	+	valid_labels = {label for _, label in POLICY_AREAS}
	+	for agency, category in AGENCY_CATEGORY_MAP.items():
	+	assert category in valid_labels, (
	+	f"Agency '{agency}' maps to '{category}' which is not a canonical policy area"
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# POLICY_AREAS structure tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestPolicyAreas:
	+	def test_has_expected_count(self):
	+	# Should have ~32 entries; we allow 30-40 range for future additions
	+	assert 28 <= len(POLICY_AREAS) <= 45, f"Unexpected POLICY_AREAS count: {len(POLICY_AREAS)}"
	+
	+	def test_no_duplicate_slugs(self):
	+	slugs = [slug for slug, _ in POLICY_AREAS]
	+	assert len(slugs) == len(set(slugs)), "Duplicate slugs in POLICY_AREAS"
	+
	+	def test_no_duplicate_labels(self):
	+	labels = [label for _, label in POLICY_AREAS]
	+	assert len(labels) == len(set(labels)), "Duplicate labels in POLICY_AREAS"
	+
	+	def test_slugs_are_lowercase_hyphenated(self):
	+	import re
	+	for slug, _ in POLICY_AREAS:
	+	assert re.match(r"^[a-z0-9-]+$", slug), f"Slug '{slug}' has invalid characters"
	+
	+	def test_key_areas_present(self):
	+	labels = {label for _, label in POLICY_AREAS}
	+	required = {
	+	"Health",
	+	"Armed Forces and National Security",
	+	"Education",
	+	"Energy",
	+	"Taxation",
	+	"Transportation and Public Works",
	+	"Environmental Protection",
	+	"Agriculture and Food",
	+	"Crime and Law Enforcement",
	+	"International Affairs",
	+	}
	+	missing = required - labels
	+	assert not missing, f"Missing required policy areas: {missing}"
	+
	+	def test_slug_matches_label(self):
	+	"""Generated slug from label should equal the declared slug (or be a close match)."""
	+	for declared_slug, label in POLICY_AREAS:
	+	generated = _slug_from_label(label)
	+	# They should be the same; allow minor variation for comma-containing labels
	+	assert declared_slug == generated or declared_slug.replace("-", "") == generated.replace("-", ""), (
	+	f"Slug mismatch: declared='{declared_slug}' generated='{generated}' for label='{label}'"
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _parse_subjects tests
	+	# ---------------------------------------------------------------------------
	+
	+	_BILLSTATUS_TEMPLATE = """\
	+	<?xml version="1.0" encoding="UTF-8"?>
	+	<billStatus>
	+	<bill>
	+	<subjects>
	+	<legislativeSubjects>
	+	{items}
	+	</legislativeSubjects>
	+	</subjects>
	+	</bill>
	+	</billStatus>"""
	+
	+	def _make_billstatus(subjects: list[str]) -> bytes:
	+	items = "".join(f"<item><name>{s}</name></item>" for s in subjects)
	+	return _BILLSTATUS_TEMPLATE.format(items=items).encode()
	+
	+
	+	class TestParseSubjects:
	+	def test_parses_single_subject(self):
	+	xml_bytes = _make_billstatus(["Health care reform"])
	+	assert _parse_subjects(xml_bytes) == ["Health care reform"]
	+
	+	def test_parses_multiple_subjects(self):
	+	subjects = ["Tax exemptions", "Income tax", "Small business"]
	+	xml_bytes = _make_billstatus(subjects)
	+	assert _parse_subjects(xml_bytes) == subjects
	+
	+	def test_empty_legislative_subjects(self):
	+	xml_bytes = _make_billstatus([])
	+	assert _parse_subjects(xml_bytes) == []
	+
	+	def test_missing_subjects_element(self):
	+	xml_bytes = b"<billStatus><bill><title>Test</title></bill></billStatus>"
	+	assert _parse_subjects(xml_bytes) == []
	+
	+	def test_malformed_xml_returns_empty(self):
	+	assert _parse_subjects(b"not valid xml<<<") == []
	+
	+	def test_strips_whitespace_from_subject_names(self):
	+	xml_bytes = _make_billstatus([" Defense contracts "])
	+	result = _parse_subjects(xml_bytes)
	+	assert result == ["Defense contracts"]
	+
	+	def test_bill_root_without_billstatus_wrapper(self):
	+	"""Some GovInfo responses have <bill> as the root element."""
	+	xml_bytes = b"""<bill>
	+	<subjects>
	+	<legislativeSubjects>
	+	<item><name>Veterans</name></item>
	+	</legislativeSubjects>
	+	</subjects>
	+	</bill>"""
	+	assert _parse_subjects(xml_bytes) == ["Veterans"]
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _ensure_schema tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestEnsureSchema:
	+	def test_creates_categories_table(self, conn):
	+	tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
	+	assert "categories" in tables
	+
	+	def test_creates_bill_subjects_table(self, conn):
	+	tables = {r[0] for r in conn.execute("SELECT name FROM sqlite_master WHERE type='table'")}
	+	assert "bill_subjects" in tables
	+
	+	def test_adds_category_column_to_awards(self, conn):
	+	cols = {r[1] for r in conn.execute("PRAGMA table_info(awards)")}
	+	assert "category" in cols
	+
	+	def test_idempotent_second_call(self, conn):
	+	"""Calling _ensure_schema twice must not raise."""
	+	_ensure_schema(conn) # second call
	+	cols = {r[1] for r in conn.execute("PRAGMA table_info(awards)")}
	+	assert "category" in cols
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _seed_categories tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestSeedCategories:
	+	def test_seeds_all_policy_areas(self, conn):
	+	_seed_categories(conn)
	+	count = conn.execute("SELECT COUNT(*) FROM categories").fetchone()[0]
	+	assert count >= len(POLICY_AREAS)
	+
	+	def test_idempotent_double_seed(self, conn):
	+	_seed_categories(conn)
	+	count1 = conn.execute("SELECT COUNT(*) FROM categories").fetchone()[0]
	+	_seed_categories(conn)
	+	count2 = conn.execute("SELECT COUNT(*) FROM categories").fetchone()[0]
	+	assert count1 == count2
	+
	+	def test_picks_up_extra_bill_policy_area(self, conn):
	+	"""A policy area present in bills but not in POLICY_AREAS is added."""
	+	conn.execute("""
	+	INSERT INTO bills (source, external_id, congress, number, title,
	+	sponsor_id, status, policy_area, updated_at, source_url)
	+	VALUES ('govinfo_billstatus', 'test-1', 119, 'H.R. 999', 'Test',
	+	'', '', 'Futuristic Technologies', '2026-01-01',
	+	'https://www.congress.gov/bill/119th-congress/house-bill/999')
	+	""")
	+	conn.commit()
	+	_seed_categories(conn)
	+	row = conn.execute(
	+	"SELECT * FROM categories WHERE label = 'Futuristic Technologies'"
	+	).fetchone()
	+	assert row is not None
	+
	+	def test_health_area_in_categories(self, conn):
	+	_seed_categories(conn)
	+	row = conn.execute("SELECT * FROM categories WHERE label = 'Health'").fetchone()
	+	assert row is not None
	+	assert row["kind"] == "policy_area"
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _backfill_awards tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestBackfillAwards:
	+	def test_categorizes_known_agency(self, conn):
	+	_insert_award(conn, "Department of Defense")
	+	_backfill_awards(conn)
	+	row = conn.execute("SELECT category FROM awards WHERE agency = 'Department of Defense'").fetchone()
	+	assert row["category"] == "Armed Forces and National Security"
	+
	+	def test_leaves_unknown_agency_null(self, conn):
	+	_insert_award(conn, "Unknown Mystery Agency")
	+	_backfill_awards(conn)
	+	row = conn.execute("SELECT category FROM awards WHERE agency = 'Unknown Mystery Agency'").fetchone()
	+	assert row["category"] is None
	+
	+	def test_does_not_overwrite_existing_category(self, conn):
	+	"""Awards that already have a category should keep it if the agency is unknown."""
	+	_insert_award(conn, "Unknown Mystery Agency", category="Existing Category")
	+	_backfill_awards(conn)
	+	row = conn.execute("SELECT category FROM awards WHERE agency = 'Unknown Mystery Agency'").fetchone()
	+	# _backfill_awards only touches rows where category IS NULL or ''
	+	assert row["category"] == "Existing Category"
	+
	+	def test_multiple_agencies(self, conn):
	+	agencies = [
	+	("Department of Health and Human Services", "Health"),
	+	("Department of Transportation", "Transportation and Public Works"),
	+	("National Science Foundation", "Science, Technology, Communications"),
	+	("Social Security Administration", "Social Welfare"),
	+	]
	+	for agency, _ in agencies:
	+	_insert_award(conn, agency)
	+	_backfill_awards(conn)
	+	for agency, expected_cat in agencies:
	+	row = conn.execute(
	+	"SELECT category FROM awards WHERE agency = ?", (agency,)
	+	).fetchone()
	+	assert row["category"] == expected_cat, f"Wrong category for {agency}"
	+
	+	def test_returns_count_of_updated(self, conn):
	+	_insert_award(conn, "Department of Energy")
	+	_insert_award(conn, "Department of Labor")
	+	updated = _backfill_awards(conn)
	+	assert updated == 2
	+
	+	def test_idempotent(self, conn):
	+	_insert_award(conn, "Department of Defense")
	+	_backfill_awards(conn)
	+	updated2 = _backfill_awards(conn)
	+	# Second run: no rows left with category IS NULL for DoD
	+	assert updated2 == 0
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# category_counts tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestCategoryCounts:
	+	def test_returns_list(self, conn):
	+	_seed_categories(conn)
	+	result = category_counts(conn)
	+	assert isinstance(result, list)
	+
	+	def test_each_item_has_expected_keys(self, conn):
	+	_seed_categories(conn)
	+	result = category_counts(conn)
	+	assert len(result) > 0
	+	for item in result:
	+	assert "label" in item
	+	assert "bills" in item
	+	assert "awards" in item
	+
	+	def test_bill_counted_in_correct_category(self, conn):
	+	_seed_categories(conn)
	+	_insert_bill(conn, policy_area="Health")
	+	result = category_counts(conn)
	+	health_row = next((r for r in result if r["label"] == "Health"), None)
	+	assert health_row is not None
	+	assert health_row["bills"] == 1
	+
	+	def test_award_counted_in_correct_category(self, conn):
	+	_seed_categories(conn)
	+	_insert_award(conn, "Department of Defense")
	+	_backfill_awards(conn)
	+	result = category_counts(conn)
	+	row = next((r for r in result if r["label"] == "Armed Forces and National Security"), None)
	+	assert row is not None
	+	assert row["awards"] == 1

services/ingest/tests/test_voteview.py +536 -0

		@@ -0,0 +1,536 @@
	+	"""
	+	Tests for voteview.py - cast_code mapping, crosswalk building, schema setup,
	+	source URL construction, and year-index helpers.
	+
	+	All network access is mocked; no live HTTP calls are made.
	+	"""
	+
	+	from __future__ import annotations
	+
	+	import os
	+	import sqlite3
	+	import tempfile
	+
	+	import pytest
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.members import ensure_votes_schema
	+	from ingest.pollers.voteview import (
	+	CAST_CODE_MAP,
	+	_derive_session,
	+	_house_source_url,
	+	_senate_source_url,
	+	build_crosswalk,
	+	cast_code_to_position,
	+	completeness_report,
	+	ensure_voteview_schema,
	+	votes_by_member_year,
	+	votes_in_year,
	+	)
	+
	+
	+	# ─── Fixtures ─────────────────────────────────────────────────────────────────
	+
	+	def _make_db() -> tuple[sqlite3.Connection, str]:
	+	"""Create a fresh temp-file DB with all schemas applied."""
	+	tmp = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(tmp)
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+	ensure_voteview_schema(conn)
	+	return conn, tmp
	+
	+
	+	# ─── cast_code_to_position ────────────────────────────────────────────────────
	+
	+	class TestCastCodeToPosition:
	+	"""Verify the full Voteview cast-code → canonical position mapping."""
	+
	+	def test_code_1_yea(self):
	+	assert cast_code_to_position(1) == "Yea"
	+
	+	def test_code_2_paired_yea(self):
	+	assert cast_code_to_position(2) == "Yea"
	+
	+	def test_code_3_announced_yea(self):
	+	assert cast_code_to_position(3) == "Yea"
	+
	+	def test_code_4_announced_nay(self):
	+	assert cast_code_to_position(4) == "Nay"
	+
	+	def test_code_5_paired_nay(self):
	+	assert cast_code_to_position(5) == "Nay"
	+
	+	def test_code_6_nay(self):
	+	assert cast_code_to_position(6) == "Nay"
	+
	+	def test_code_7_present_v1(self):
	+	assert cast_code_to_position(7) == "Present"
	+
	+	def test_code_8_present_v2(self):
	+	assert cast_code_to_position(8) == "Present"
	+
	+	def test_code_9_not_voting(self):
	+	assert cast_code_to_position(9) == "Not Voting"
	+
	+	def test_code_0_not_a_member(self):
	+	"""Code 0 = not a member / absent no record → Not Voting."""
	+	assert cast_code_to_position(0) == "Not Voting"
	+
	+	def test_all_codes_covered(self):
	+	"""Every code in the map must return a valid canonical position."""
	+	valid = {"Yea", "Nay", "Present", "Not Voting"}
	+	for code in range(0, 10):
	+	pos = cast_code_to_position(code)
	+	assert pos in valid, f"cast_code {code} → {pos!r} is not in {valid}"
	+
	+	def test_string_codes_accepted(self):
	+	"""CSV columns are strings; the function must coerce them."""
	+	assert cast_code_to_position("1") == "Yea"
	+	assert cast_code_to_position("6") == "Nay"
	+	assert cast_code_to_position("9") == "Not Voting"
	+
	+	def test_float_string_accepted(self):
	+	"""Voteview CSVs sometimes encode integers as floats (e.g. '1.0')."""
	+	assert cast_code_to_position("1.0") == "Yea"
	+	assert cast_code_to_position("6.0") == "Nay"
	+
	+	def test_unknown_code_returns_none(self):
	+	assert cast_code_to_position(99) is None
	+	assert cast_code_to_position(-1) is None
	+	assert cast_code_to_position(10) is None
	+
	+	def test_bad_input_returns_none(self):
	+	assert cast_code_to_position("") is None
	+	assert cast_code_to_position(None) is None
	+	assert cast_code_to_position("x") is None
	+
	+	def test_map_has_exactly_ten_entries(self):
	+	"""The codebook defines codes 0-9; no other codes should exist."""
	+	assert set(CAST_CODE_MAP.keys()) == set(range(0, 10))
	+
	+	def test_yea_group_codes_all_map_to_yea(self):
	+	for code in (1, 2, 3):
	+	assert CAST_CODE_MAP[code] == "Yea", f"code {code} should be Yea"
	+
	+	def test_nay_group_codes_all_map_to_nay(self):
	+	for code in (4, 5, 6):
	+	assert CAST_CODE_MAP[code] == "Nay", f"code {code} should be Nay"
	+
	+	def test_present_group_codes_map_to_present(self):
	+	for code in (7, 8):
	+	assert CAST_CODE_MAP[code] == "Present", f"code {code} should be Present"
	+
	+	def test_not_voting_group_codes_map_to_not_voting(self):
	+	for code in (0, 9):
	+	assert CAST_CODE_MAP[code] == "Not Voting", f"code {code} should be Not Voting"
	+
	+
	+	# ─── Source URL builders ──────────────────────────────────────────────────────
	+
	+	class TestSourceUrls:
	+	def test_house_url_format(self):
	+	url = _house_source_url(2023, 42)
	+	assert url == "https://clerk.house.gov/Votes/2023/roll042"
	+	assert url.startswith("https://clerk.house.gov")
	+
	+	def test_house_url_zero_pads_to_3_digits(self):
	+	assert _house_source_url(2024, 1).endswith("roll001")
	+	assert _house_source_url(2024, 99).endswith("roll099")
	+	assert _house_source_url(2024, 100).endswith("roll100")
	+
	+	def test_senate_url_format(self):
	+	url = _senate_source_url(118, 1, 42)
	+	assert url == (
	+	"https://www.senate.gov/legislative/LIS/roll_call_votes/"
	+	"vote1181/vote_118_1_00042.xml"
	+	)
	+	assert "senate.gov" in url
	+
	+	def test_senate_url_zero_pads_roll_to_5_digits(self):
	+	url = _senate_source_url(119, 2, 7)
	+	assert "vote_119_2_00007.xml" in url
	+
	+	def test_house_url_host_is_gov(self):
	+	from ingest.source_validator import is_allowlisted
	+	assert is_allowlisted(_house_source_url(2025, 1))
	+
	+	def test_senate_url_host_is_gov(self):
	+	from ingest.source_validator import is_allowlisted
	+	assert is_allowlisted(_senate_source_url(119, 1, 1))
	+
	+
	+	# ─── Session derivation ───────────────────────────────────────────────────────
	+
	+	class TestDeriveSession:
	+	def test_csv_session_preferred(self):
	+	assert _derive_session("2023-01-15", "1") == 1
	+	assert _derive_session("2024-06-01", "2") == 2
	+
	+	def test_odd_year_fallback_to_session_1(self):
	+	assert _derive_session("2023-03-01", "") == 1
	+	assert _derive_session("2021-07-04", "") == 1
	+
	+	def test_even_year_fallback_to_session_2(self):
	+	assert _derive_session("2024-11-01", "") == 2
	+	assert _derive_session("2022-01-01", "") == 2
	+
	+	def test_bad_date_defaults_to_1(self):
	+	assert _derive_session("", "") == 1
	+	assert _derive_session("bad-date", "x") == 1
	+
	+
	+	# ─── Schema setup ─────────────────────────────────────────────────────────────
	+
	+	class TestEnsureVoteviewSchema:
	+	def test_crosswalk_table_created(self):
	+	conn, tmp = _make_db()
	+	tables = {
	+	row[0]
	+	for row in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='table'"
	+	)
	+	}
	+	assert "voteview_crosswalk" in tables
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_year_column_added_to_votes(self):
	+	conn, tmp = _make_db()
	+	cols = {
	+	row[1]
	+	for row in conn.execute("PRAGMA table_info(votes)")
	+	}
	+	assert "year" in cols, "year column must be added to votes table"
	+	assert "description" in cols
	+	assert "voteview_rollnumber" in cols
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_year_index_exists(self):
	+	conn, tmp = _make_db()
	+	indexes = {
	+	row[0]
	+	for row in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='index'"
	+	)
	+	}
	+	assert "idx_votes_year" in indexes
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_idempotent_multiple_calls(self):
	+	conn, tmp = _make_db()
	+	# Should not raise on second call.
	+	ensure_voteview_schema(conn)
	+	ensure_voteview_schema(conn)
	+	conn.close()
	+	os.unlink(tmp)
	+
	+
	+	# ─── Crosswalk building ───────────────────────────────────────────────────────
	+
	+	class TestBuildCrosswalk:
	+	"""Build the crosswalk from a mock legislators JSON (no HTTP)."""
	+
	+	_MOCK_LEGISLATORS = [
	+	{
	+	"id": {"bioguide": "A000001", "icpsr": 12345},
	+	"name": {"official_full": "Alice Anderson"},
	+	"terms": [{"type": "rep", "state": "NY"}],
	+	},
	+	{
	+	"id": {"bioguide": "B000002", "icpsr": 67890},
	+	"name": {"official_full": "Bob Brown"},
	+	"terms": [{"type": "sen", "state": "CA"}],
	+	},
	+	{
	+	# Member with no ICPSR - should be skipped without error.
	+	"id": {"bioguide": "C000003"},
	+	"name": {"official_full": "Carol Carter"},
	+	"terms": [{"type": "rep", "state": "TX"}],
	+	},
	+	{
	+	# Member with both icpsr and icpsr_hn (House/Senate split).
	+	"id": {"bioguide": "D000004", "icpsr": 11111, "icpsr_hn": 22222},
	+	"name": {"official_full": "Dave Davis"},
	+	"terms": [{"type": "rep", "state": "FL"}],
	+	},
	+	]
	+
	+	def _make_crosswalk_from_mock(self) -> tuple[dict, sqlite3.Connection, str]:
	+	"""
	+	Build the crosswalk by directly inserting mock data, bypassing HTTP.
	+	Returns (crosswalk_dict, conn, tmp_path).
	+	"""
	+	conn, tmp = _make_db()
	+
	+	now = "2025-01-01T00:00:00+00:00"
	+	for leg in self._MOCK_LEGISLATORS:
	+	ids = leg.get("id", {})
	+	bioguide = ids.get("bioguide", "").strip()
	+	full_name = leg.get("name", {}).get("official_full", "")
	+	for field in ("icpsr", "icpsr_hn"):
	+	raw = ids.get(field)
	+	if raw is None:
	+	continue
	+	try:
	+	icpsr_int = int(float(raw))
	+	except (ValueError, TypeError):
	+	continue
	+	conn.execute(
	+	"""
	+	INSERT INTO voteview_crosswalk
	+	(icpsr_id, bioguide_id, full_name, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(icpsr_id) DO UPDATE SET
	+	bioguide_id = excluded.bioguide_id,
	+	full_name = excluded.full_name,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(icpsr_int, bioguide, full_name, now),
	+	)
	+	conn.commit()
	+
	+	# Now load via the cache path.
	+	crosswalk = build_crosswalk(conn, force_refresh=False)
	+	return crosswalk, conn, tmp
	+
	+	def test_primary_icpsr_mapped(self):
	+	cw, conn, tmp = self._make_crosswalk_from_mock()
	+	assert cw[12345] == "A000001"
	+	assert cw[67890] == "B000002"
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_icpsr_hn_also_mapped(self):
	+	"""Both icpsr and icpsr_hn for D000004 should map to the same bioguide."""
	+	cw, conn, tmp = self._make_crosswalk_from_mock()
	+	assert cw[11111] == "D000004"
	+	assert cw[22222] == "D000004"
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_member_without_icpsr_not_in_crosswalk(self):
	+	cw, conn, tmp = self._make_crosswalk_from_mock()
	+	# C000003 has no ICPSR - no entry for it in the crosswalk.
	+	bioguides = set(cw.values())
	+	assert "C000003" not in bioguides or all(
	+	v != "C000003" for v in cw.values()
	+	)
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_cached_load_returns_same_data(self):
	+	"""Second call (force_refresh=False) should return the same crosswalk."""
	+	cw1, conn, tmp = self._make_crosswalk_from_mock()
	+	cw2 = build_crosswalk(conn, force_refresh=False)
	+	assert cw1 == cw2
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_unknown_icpsr_returns_none_from_get(self):
	+	cw, conn, tmp = self._make_crosswalk_from_mock()
	+	assert cw.get(99999) is None
	+	conn.close()
	+	os.unlink(tmp)
	+
	+
	+	# ─── Year-index helpers ───────────────────────────────────────────────────────
	+
	+	class TestYearIndexHelpers:
	+	"""
	+	Seed a minimal DB with known votes + positions, then verify the
	+	year-index helper queries return the correct rows.
	+	"""
	+
	+	def _seed_db(self) -> tuple[sqlite3.Connection, str]:
	+	conn, tmp = _make_db()
	+
	+	# Insert two members.
	+	for bioguide, name, state in [
	+	("N000002", "Jerry Nadler", "NY"),
	+	("C000714", "James Clyburn", "SC"),
	+	]:
	+	conn.execute(
	+	"INSERT INTO members "
	+	"(bioguide_id, full_name, party, state, chamber, source_url) "
	+	"VALUES (?, ?, ?, ?, ?, ?)",
	+	(
	+	bioguide, name, "D", state, "house",
	+	f"https://bioguide.congress.gov/search/bio/{bioguide}",
	+	),
	+	)
	+
	+	# Insert three votes: two in 2023, one in 2024.
	+	votes_data = [
	+	# (chamber, date, year, yea, nay, result, source_url, congress, roll_number)
	+	("house", "2023-01-19", 2023, 213, 215, "Failed",
	+	"https://clerk.house.gov/Votes/2023/roll010", 118, 10),
	+	("house", "2023-06-01", 2023, 220, 210, "Passed",
	+	"https://clerk.house.gov/Votes/2023/roll200", 118, 200),
	+	("house", "2024-03-15", 2024, 230, 200, "Passed",
	+	"https://clerk.house.gov/Votes/2024/roll100", 118, 100),
	+	]
	+	vote_ids = []
	+	for row in votes_data:
	+	conn.execute(
	+	"""
	+	INSERT INTO votes
	+	(chamber, date, year, yea, nay, result, source_url,
	+	congress, roll_number, recorded, present, not_voting)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 1, 0, 0)
	+	""",
	+	row,
	+	)
	+	vid = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+	vote_ids.append(vid)
	+
	+	# Positions: N000002 votes on all three; C000714 only on vote #1.
	+	for vid, pos in zip(vote_ids, ["Yea", "Nay", "Yea"]):
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position) "
	+	"VALUES (?, ?, NULL, ?)",
	+	(vid, "N000002", pos),
	+	)
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position) "
	+	"VALUES (?, ?, NULL, ?)",
	+	(vote_ids[0], "C000714", "Yea"),
	+	)
	+	conn.commit()
	+	return conn, tmp
	+
	+	def test_votes_by_member_year_returns_correct_count(self):
	+	conn, tmp = self._seed_db()
	+	rows = conn.execute(
	+	"""
	+	SELECT * FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = 'N000002' AND v.year = 2023
	+	"""
	+	).fetchall()
	+	assert len(rows) == 2, "N000002 voted on 2 rolls in 2023"
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_votes_by_member_year_2024(self):
	+	conn, tmp = self._seed_db()
	+	rows = conn.execute(
	+	"""
	+	SELECT * FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = 'N000002' AND v.year = 2024
	+	"""
	+	).fetchall()
	+	assert len(rows) == 1
	+	assert dict(rows[0])["date"] == "2024-03-15"
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_votes_by_member_year_empty_for_nonexistent_year(self):
	+	conn, tmp = self._seed_db()
	+	rows = conn.execute(
	+	"""
	+	SELECT * FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = 'N000002' AND v.year = 2019
	+	"""
	+	).fetchall()
	+	assert rows == []
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_votes_in_year_2023(self):
	+	conn, tmp = self._seed_db()
	+	rows = conn.execute(
	+	"SELECT * FROM votes WHERE year = 2023 AND recorded = 1"
	+	).fetchall()
	+	assert len(rows) == 2
	+	dates = sorted(dict(r)["date"] for r in rows)
	+	assert dates == ["2023-01-19", "2023-06-01"]
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_votes_in_year_2024(self):
	+	conn, tmp = self._seed_db()
	+	rows = conn.execute(
	+	"SELECT * FROM votes WHERE year = 2024 AND recorded = 1"
	+	).fetchall()
	+	assert len(rows) == 1
	+	assert dict(rows[0])["roll_number"] == 100
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_year_index_is_used(self):
	+	"""EXPLAIN QUERY PLAN should mention the year index (not a full scan)."""
	+	conn, tmp = self._seed_db()
	+	plan = conn.execute(
	+	"EXPLAIN QUERY PLAN SELECT * FROM votes WHERE year = 2023"
	+	).fetchall()
	+	# EXPLAIN QUERY PLAN rows have a 'detail' column (index 3) in SQLite 3.24+.
	+	# Concatenate all columns of every row as a single lowercase string.
	+	plan_text = " ".join(
	+	" ".join(str(col) for col in tuple(row))
	+	for row in plan
	+	).lower()
	+	# Should reference the index, not a full table scan.
	+	assert "idx_votes_year" in plan_text, (
	+	f"Expected idx_votes_year in query plan; got: {plan_text!r}"
	+	)
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_position_enum_enforced_for_voteview_positions(self):
	+	"""vote_positions CHECK constraint rejects cast_code-derived invalid positions."""
	+	conn, tmp = _make_db()
	+	conn.execute(
	+	"INSERT INTO votes (chamber, date, year, yea, nay, result, source_url, recorded) "
	+	"VALUES ('house', '2023-01-01', 2023, 1, 0, 'Passed', "
	+	"'https://clerk.house.gov/Votes/2023/roll001', 1)"
	+	)
	+	vid = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+	with pytest.raises(sqlite3.IntegrityError):
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, position) "
	+	"VALUES (?, ?, ?)",
	+	(vid, "Z000001", "Abstain"), # not in enum
	+	)
	+	conn.commit()
	+	conn.close()
	+	os.unlink(tmp)
	+
	+
	+	# ─── End-to-end: schema idempotency ──────────────────────────────────────────
	+
	+	class TestSchemaIdempotency:
	+	"""Running schema setup multiple times must not corrupt the DB."""
	+
	+	def test_full_schema_setup_multiple_times(self):
	+	conn, tmp = _make_db()
	+	for _ in range(3):
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+	ensure_voteview_schema(conn)
	+	# Verify tables still intact.
	+	tables = {
	+	row[0]
	+	for row in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='table'"
	+	)
	+	}
	+	for t in ("votes", "vote_positions", "members", "voteview_crosswalk"):
	+	assert t in tables
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_cast_code_coverage_against_schema_enum(self):
	+	"""
	+	Every position value produced by CAST_CODE_MAP must satisfy the
	+	vote_positions CHECK constraint (Yea, Nay, Present, Not Voting).
	+	"""
	+	valid = {"Yea", "Nay", "Present", "Not Voting"}
	+	for code, position in CAST_CODE_MAP.items():
	+	assert position in valid, (
	+	f"CAST_CODE_MAP[{code}] = {position!r} is not in the DB enum {valid}"
	+	)

services/ingest/tests/test_watcher.py +452 -0

		@@ -0,0 +1,452 @@
	+	"""
	+	Tests for ingest.watcher - ADDED / UPDATED / REMOVED classification.
	+
	+	All tests are purely synthetic (no network calls, no subprocess).
	+	They exercise:
	+	- _classify_added_updated() : seen-table diff logic
	+	- _detect_full_set_changes() : manifest-diff + removal guards
	+	- Guard: non-empty pull (prevents false removals on empty result)
	+	- Guard: first-run (no prior manifest → no removals)
	+	- source_url non-null invariant on every change_event
	+	"""
	+
	+	import os
	+	import tempfile
	+
	+	import pytest
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.watcher import (
	+	FULL_SET_CONFIGS,
	+	_classify_added_updated,
	+	_detect_full_set_changes,
	+	_ensure_watcher_tables,
	+	_load_manifest,
	+	_snapshot_seen,
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Fixtures
	+	# ---------------------------------------------------------------------------
	+
	+	def _make_db():
	+	"""Return (conn, tmp_path) for a fresh in-memory-equivalent temp DB."""
	+	tmp = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(tmp)
	+	init_schema(conn)
	+	_ensure_watcher_tables(conn)
	+	return conn, tmp
	+
	+
	+	def _cleanup(conn, tmp):
	+	conn.close()
	+	try:
	+	os.unlink(tmp)
	+	except OSError:
	+	pass
	+
	+
	+	def _seed_seen(conn, rows):
	+	"""Seed seen table with list of (source, external_id, update_date)."""
	+	for source, ext_id, update_date in rows:
	+	conn.execute(
	+	"INSERT OR REPLACE INTO seen (source, external_id, update_date)"
	+	" VALUES (?,?,?)",
	+	(source, ext_id, update_date),
	+	)
	+	conn.commit()
	+
	+
	+	def _seed_members_table(conn, members):
	+	"""Create and populate the members table with minimal columns."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS members (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bioguide_id TEXT NOT NULL UNIQUE,
	+	full_name TEXT NOT NULL DEFAULT '',
	+	party TEXT, state TEXT, district INTEGER, chamber TEXT,
	+	lis_member_id TEXT, govtrack_id INTEGER, icpsr_id INTEGER,
	+	source_url TEXT NOT NULL
	+	);
	+	"""
	+	)
	+	for m in members:
	+	bio = m["bioguide_id"]
	+	conn.execute(
	+	"INSERT OR REPLACE INTO members (bioguide_id, full_name, source_url)"
	+	" VALUES (?,?,?)",
	+	(bio, m.get("full_name", "Test Member"),
	+	m.get("source_url", f"https://bioguide.congress.gov/search/bio/{bio}")),
	+	)
	+	conn.commit()
	+
	+
	+	TS = "2026-06-26T00:00:00+00:00"
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _classify_added_updated - seen-table diff
	+	# ---------------------------------------------------------------------------
	+
	+	class TestClassifyAddedUpdated:
	+
	+	def test_added_single_new_entry(self):
	+	conn, tmp = _make_db()
	+	before = {}
	+	after = {("federal_register", "2026-12345"): "2026-06-25"}
	+	added, updated = _classify_added_updated(before, after, TS, conn)
	+	assert added == 1
	+	assert updated == 0
	+	row = conn.execute(
	+	"SELECT * FROM change_events WHERE ref_id='2026-12345'"
	+	).fetchone()
	+	assert row["change"] == "added"
	+	assert row["source"] == "federal_register"
	+	_cleanup(conn, tmp)
	+
	+	def test_updated_same_id_different_date(self):
	+	conn, tmp = _make_db()
	+	before = {("openomb", "FY2026_Agency.json"): "2026-01-15"}
	+	after = {("openomb", "FY2026_Agency.json"): "2026-06-20"}
	+	added, updated = _classify_added_updated(before, after, TS, conn)
	+	assert added == 0
	+	assert updated == 1
	+	row = conn.execute(
	+	"SELECT * FROM change_events WHERE ref_id='FY2026_Agency.json'"
	+	).fetchone()
	+	assert row["change"] == "updated"
	+	assert "2026-01-15" in row["summary"]
	+	assert "2026-06-20" in row["summary"]
	+	_cleanup(conn, tmp)
	+
	+	def test_no_change_when_state_identical(self):
	+	conn, tmp = _make_db()
	+	state = {("congress", "BILLS-119hr1"): "2026-06-01"}
	+	added, updated = _classify_added_updated(state, state, TS, conn)
	+	assert added == 0
	+	assert updated == 0
	+	count = conn.execute(
	+	"SELECT COUNT(*) FROM change_events"
	+	).fetchone()[0]
	+	assert count == 0
	+	_cleanup(conn, tmp)
	+
	+	def test_mixed_added_updated_unchanged(self):
	+	conn, tmp = _make_db()
	+	before = {
	+	("federal_register", "DOC-001"): "2026-01-01",
	+	("federal_register", "DOC-002"): "2026-01-01",
	+	}
	+	after = {
	+	("federal_register", "DOC-001"): "2026-06-01", # updated
	+	("federal_register", "DOC-002"): "2026-01-01", # unchanged
	+	("federal_register", "DOC-003"): "2026-06-26", # added
	+	}
	+	added, updated = _classify_added_updated(before, after, TS, conn)
	+	assert added == 1
	+	assert updated == 1
	+	_cleanup(conn, tmp)
	+
	+	def test_multiple_sources_in_single_diff(self):
	+	conn, tmp = _make_db()
	+	before = {}
	+	after = {
	+	("usaspending", "CONT_AWD_001"): "2026-06-01",
	+	("treasury", "debt_2026-06-01"): "2026-06-01",
	+	("openomb", "FY2026_DOD.json"): "2026-06-01",
	+	}
	+	added, updated = _classify_added_updated(before, after, TS, conn)
	+	assert added == 3
	+	assert updated == 0
	+	_cleanup(conn, tmp)
	+
	+	def test_source_url_non_empty_on_every_event(self):
	+	"""Every inserted change_event must have a non-empty source_url."""
	+	conn, tmp = _make_db()
	+	before = {}
	+	after = {
	+	("usaspending", "CONT_AWD_XTEST"): "2026-06-20",
	+	("congress", "BILLS-119hr99"): "2026-06-20",
	+	}
	+	_classify_added_updated(before, after, TS, conn)
	+	rows = conn.execute(
	+	"SELECT source_url FROM change_events"
	+	).fetchall()
	+	assert len(rows) == 2
	+	for r in rows:
	+	assert r["source_url"], "source_url must be non-empty"
	+	_cleanup(conn, tmp)
	+
	+	def test_ref_type_mapped_correctly(self):
	+	conn, tmp = _make_db()
	+	before = {}
	+	after = {
	+	("bills_bulk", "B001"): "2026-01-01",
	+	("federal_register", "FR001"): "2026-01-01",
	+	("usaspending", "USA001"): "2026-01-01",
	+	("votes_house", "VH001"): "2026-01-01",
	+	}
	+	_classify_added_updated(before, after, TS, conn)
	+	rows = {
	+	r["ref_id"]: r["ref_type"]
	+	for r in conn.execute("SELECT ref_id, ref_type FROM change_events").fetchall()
	+	}
	+	assert rows["B001"] == "bill"
	+	assert rows["FR001"] == "executive_action"
	+	assert rows["USA001"] == "award"
	+	assert rows["VH001"] == "vote"
	+	_cleanup(conn, tmp)
	+
	+	def test_change_field_constraint_enforced(self):
	+	"""SQLite CHECK constraint on change column must be honoured."""
	+	conn, tmp = _make_db()
	+	with pytest.raises(Exception):
	+	conn.execute(
	+	"INSERT INTO change_events"
	+	" (ts,source,ref_type,ref_id,change,summary,source_url)"
	+	" VALUES (?,?,?,?,?,?,?)",
	+	(TS, "test", "bill", "X", "deleted", "bad value",
	+	"https://congress.gov/test"),
	+	)
	+	conn.commit()
	+	_cleanup(conn, tmp)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _detect_full_set_changes - manifest diff + removal guards
	+	# ---------------------------------------------------------------------------
	+
	+	class TestFullSetChanges:
	+
	+	# ---- REMOVED happy path ------------------------------------------------
	+
	+	def test_removal_detected_when_member_disappears(self):
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	# Current pull: only A000001
	+	_seed_members_table(conn, [{"bioguide_id": "A000001"}])
	+	# Previous manifest had A000001 + A000002
	+	for bio in ("A000001", "A000002"):
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", bio, 1,
	+	f"https://bioguide.congress.gov/search/bio/{bio}"),
	+	)
	+	conn.commit()
	+
	+	added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	assert removed == 1
	+	row = conn.execute(
	+	"SELECT * FROM change_events WHERE ref_id='A000002'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["change"] == "removed"
	+	assert row["source_url"] # must be non-empty
	+	_cleanup(conn, tmp)
	+
	+	def test_no_removal_when_set_is_stable(self):
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	_seed_members_table(conn, [
	+	{"bioguide_id": "A000001"},
	+	{"bioguide_id": "A000002"},
	+	])
	+	for bio in ("A000001", "A000002"):
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", bio, 1,
	+	f"https://bioguide.congress.gov/search/bio/{bio}"),
	+	)
	+	conn.commit()
	+
	+	added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS)
	+	assert removed == 0
	+	_cleanup(conn, tmp)
	+
	+	# ---- Guard: non-empty pull ---------------------------------------------
	+
	+	def test_removal_guard_empty_pull(self):
	+	"""
	+	Non-empty guard: if current pull returns 0 items (table missing or empty),
	+	no REMOVED events must be written.
	+	"""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	# No members table → _query_current_set returns {}
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", "A000001", 1,
	+	"https://bioguide.congress.gov/search/bio/A000001"),
	+	)
	+	conn.commit()
	+
	+	added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	# Guard must prevent the false removal
	+	assert removed == 0
	+	count = conn.execute(
	+	"SELECT COUNT(*) FROM change_events WHERE change='removed'"
	+	).fetchone()[0]
	+	assert count == 0
	+	_cleanup(conn, tmp)
	+
	+	def test_manifest_not_overwritten_on_empty_pull(self):
	+	"""
	+	When the empty-pull guard fires, the existing manifest must be preserved
	+	so the next cycle can still detect removals correctly.
	+	"""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", "A000001", 1,
	+	"https://bioguide.congress.gov/search/bio/A000001"),
	+	)
	+	conn.commit()
	+
	+	_detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	# Manifest entry for A000001 must still be present
	+	manifest = _load_manifest(conn, "members")
	+	assert "A000001" in manifest
	+	_cleanup(conn, tmp)
	+
	+	# ---- Guard: first run --------------------------------------------------
	+
	+	def test_removal_guard_first_run(self):
	+	"""
	+	First-run guard: no prior manifest → no REMOVED events, even when
	+	the current pull has many items.
	+	"""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	_seed_members_table(conn, [
	+	{"bioguide_id": "A000001"},
	+	{"bioguide_id": "A000002"},
	+	])
	+	# No prior watcher_manifest entries for "members"
	+
	+	added, removed = _detect_full_set_changes(conn, "members", cfg, 1, TS)
	+
	+	assert removed == 0
	+	count = conn.execute(
	+	"SELECT COUNT(*) FROM change_events WHERE change='removed'"
	+	).fetchone()[0]
	+	assert count == 0
	+	_cleanup(conn, tmp)
	+
	+	# ---- ADDED (members, no mark_seen) ------------------------------------
	+
	+	def test_added_detected_for_members_source(self):
	+	"""
	+	Members poller does not call mark_seen, so manifest diff must detect ADDs.
	+	"""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	_seed_members_table(conn, [
	+	{"bioguide_id": "A000001"},
	+	{"bioguide_id": "A000003"}, # new member
	+	])
	+	# Prior manifest had only A000001
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", "A000001", 1,
	+	"https://bioguide.congress.gov/search/bio/A000001"),
	+	)
	+	conn.commit()
	+
	+	added, removed = _detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	assert added == 1
	+	row = conn.execute(
	+	"SELECT * FROM change_events WHERE ref_id='A000003'"
	+	).fetchone()
	+	assert row is not None
	+	assert row["change"] == "added"
	+	assert row["source_url"]
	+	_cleanup(conn, tmp)
	+
	+	# ---- Manifest lifecycle ------------------------------------------------
	+
	+	def test_manifest_updated_after_successful_cycle(self):
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	_seed_members_table(conn, [
	+	{"bioguide_id": "A000001"},
	+	{"bioguide_id": "A000002"},
	+	])
	+
	+	_detect_full_set_changes(conn, "members", cfg, 1, TS)
	+
	+	manifest = _load_manifest(conn, "members")
	+	assert "A000001" in manifest
	+	assert "A000002" in manifest
	+	_cleanup(conn, tmp)
	+
	+	def test_manifest_reflects_removal_next_cycle(self):
	+	"""
	+	After A000002 is removed from the current pull, the manifest must
	+	no longer contain it so the next cycle doesn't re-flag it.
	+	"""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+
	+	# Cycle 1: both present
	+	_seed_members_table(conn, [
	+	{"bioguide_id": "A000001"},
	+	{"bioguide_id": "A000002"},
	+	])
	+	_detect_full_set_changes(conn, "members", cfg, 1, TS)
	+
	+	# Cycle 2: A000002 gone
	+	conn.execute("DELETE FROM members WHERE bioguide_id='A000002'")
	+	conn.commit()
	+	_detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	manifest = _load_manifest(conn, "members")
	+	assert "A000001" in manifest
	+	assert "A000002" not in manifest
	+
	+	# Cycle 3: still only A000001 - must NOT flag A000002 again
	+	events_before = conn.execute(
	+	"SELECT COUNT(*) FROM change_events WHERE change='removed'"
	+	).fetchone()[0]
	+	_detect_full_set_changes(conn, "members", cfg, 3, TS)
	+	events_after = conn.execute(
	+	"SELECT COUNT(*) FROM change_events WHERE change='removed'"
	+	).fetchone()[0]
	+	assert events_after == events_before, "A000002 must not be flagged twice"
	+	_cleanup(conn, tmp)
	+
	+	# ---- source_url invariant on every event --------------------------------
	+
	+	def test_removed_event_has_source_url(self):
	+	"""Every REMOVED event must carry a non-empty source_url."""
	+	conn, tmp = _make_db()
	+	cfg = FULL_SET_CONFIGS["members"]
	+	_seed_members_table(conn, [{"bioguide_id": "A000001"}])
	+	for bio in ("A000001", "A000002"):
	+	conn.execute(
	+	"INSERT INTO watcher_manifest (source, external_id, cycle_id, source_url)"
	+	" VALUES (?,?,?,?)",
	+	("members", bio, 1,
	+	f"https://bioguide.congress.gov/search/bio/{bio}"),
	+	)
	+	conn.commit()
	+
	+	_detect_full_set_changes(conn, "members", cfg, 2, TS)
	+
	+	rows = conn.execute(
	+	"SELECT source_url FROM change_events WHERE change='removed'"
	+	).fetchall()
	+	assert len(rows) == 1
	+	assert rows[0]["source_url"]
	+	_cleanup(conn, tmp)