a7452ab · Docket

Backend full-spectrum: bills+diff/rider engine, closed-door (FR/OpenOMB/CREC/lobbying), granular votes+members (25k positions), state coverage; run.py orchestrator

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

_ops/glm-f2.txt +23 -0

		@@ -0,0 +1,23 @@
	+	You are GLM-5.2, frontend lead for Docket (neutral US gov transparency app). Work in /opt/docket (local build dir). The read-API is LIVE and returns real data - all /api routes work (curl http://localhost:3000/api/feed?limit=2 to see). Be RAG-first; read ONLY: docs/API.md, docs/BRAND.md, docs/NEUTRALITY.md, docs/ACCESSIBILITY.md. Look at your existing app/ pages + components + lib/.
	+
	+	WAVE F2 - build the DEEP, GRANULAR surfaces (the heart of the product). The DB now has: 290 bills, 410 bill_versions, 34 bill_diffs (with rider flags), 99 amendments, 90 votes, 25,572 vote_positions, 537 members, 500 executive_actions, 20 apportionments, 188 procedural_events, 320 awards. Make all of it explorable. Build/enhance these pages, each consuming the real /api routes:
	+
	+	1. VOTES - /votes (browse recorded roll-call votes) and /votes/[id] (THE ROSTER: show EVERY member's position - Yea/Nay/Present/Not Voting - with name + party + state + district. Use muted party/vote colors per BRAND, NEVER vivid red/blue. Default sort by vote then alphabetical, not grouped by party. For voice/UC votes where recorded=false, show "No individual record exists - passed by voice vote" instead of a fake roster.) Source link on every vote.
	+
	+	2. MEMBERS - /members (browseable roster, search/filter by state/chamber/party) and /members/[id] (a representative's FULL record: their recent votes with how they voted, bills sponsored, link each to the bill/vote). This is "what did MY rep do."
	+
	+	3. BILL DETAIL - enhance /bill/[id] with: (a) the "WHAT CHANGED / WHAT GOT SLIPPED IN" view consuming /api/bills/[id]/diff - show added/removed/amended sections with the snippet, and flag rider-flagged sections prominently (this is our signature feature); (b) the amendments list; (c) the bill's votes (link to the /votes/[id] roster); (d) full status timeline; (e) sponsor + cosponsors (link to /members/[id]). Honesty notes where data is voice-vote/conference (no individual record).
	+
	+	4. SPENDING - enhance /spending with the OMB APPORTIONMENT IMPOUND TRACKER from /api/spending/apportionments: per agency/account show appropriated vs apportioned vs status (released/delayed/withheld). Plus the awards list (already wired) and fiscal stats.
	+
	+	5. EXECUTIVE - /executive: Federal Register actions (Executive Orders, proposed/final rules, notices) from /api/executive, filterable by type.
	+
	+	6. STATES - enhance /states with the per-state confidence map (ok/stale/missing badges) from /api/states, and /states/[code] detail.
	+
	+	7. SEARCH - a global search UI consuming /api/search (the "look up my state / rep / topic" entry point). Make it prominent and simple (per ACCESSIBILITY.md: labeled, big target, drop-downs not bare box).
	+
	+	HARD RULES: keep the BRAND (brass rationed, custody-stamp, mono data, dark default) and the 3-mode accessibility toggle working on ALL new pages (Dark/Readable/Plain). Enforce NEUTRALITY: a Source link on every fact, no adjectives, no comment sections, muted vote/party colors, honest "no record exists" where true. All new pages must work in all 3 display modes.
	+
	+	BUILD: run `npm run build` and ensure it PASSES (fix any type errors). Do NOT try to restart the server (the tooling for that is handled separately - fuser is absent on this box). Just make sure the build is clean.
	+
	+	Update FRONTEND_NOTES.md with the new pages. Definition of done: npm run build passes; all 7 surfaces implemented consuming real /api data; brand + neutrality + 3-mode accessibility intact.

services/ingest/BILLS_NOTES.md +168 -0

		@@ -0,0 +1,168 @@
	+	# Bills Pipeline - Build Notes
	+
	+	## Architecture
	+
	+	Five new modules under `services/ingest/ingest/`:
	+
	+	\| Module \| Role \|
	+	\|---\|---\|
	+	\| `pollers/bills_bulk.py` \| Downloads BILLSTATUS sitemaps + individual XMLs; populates `bills`, `bills_ext`, `actions`, `bill_versions`, `amendments` \|
	+	\| `pollers/bill_versions.py` \| Supplementary version discovery via HEAD-probe of GovInfo content URLs \|
	+	\| `diff.py` \| Section-parse + align + diff engine; populates `bill_diffs` \|
	+	\| `rider_detect.py` \| R Street keyword scan over `bill_diffs.change='added'`; sets `rider_flag` \|
	+	\| `amendments.py` \| Parses `<amendments>` from BILLSTATUS XML; re-fetches for backfill \|
	+
	+	Schema additions (all `CREATE TABLE IF NOT EXISTS`):
	+	- `bills_ext` - extended bill metadata (bill_type, sponsor_name/party/state)
	+	- `bill_versions` - version rows with govinfo_package_id + text_ref (XML URL)
	+	- `bill_diffs` - per-section diff rows with change type + snippet + rider_flag
	+	- `amendments` - amendment rows with agreed_method + congress.gov source_url
	+
	+	---
	+
	+	## GovInfo Bulk Data Gotchas
	+
	+	### No ZIP files for BILLSTATUS
	+	The documented ZIP URL pattern (`BILLSTATUS-119hr.zip`) returns 404 for the 119th Congress. Individual XMLs are accessible; the correct enumeration method is the sitemap:
	+
	+	```
	+	https://www.govinfo.gov/sitemap/bulkdata/BILLSTATUS/{congress}{type}/sitemap.xml
	+	```
	+
	+	The sitemap index is at:
	+	```
	+	https://www.govinfo.gov/sitemap/bulkdata/BILLSTATUS/sitemapindex.xml
	+	```
	+
	+	Individual files:
	+	```
	+	https://www.govinfo.gov/bulkdata/BILLSTATUS/{congress}/{type}/BILLSTATUS-{congress}{type}{number}.xml
	+	```
	+
	+	### Directory listing returns HTML error
	+	`GET /bulkdata/BILLSTATUS/119/hr/` returns a "Bulkdata Service Error" HTML page. Use the sitemap instead.
	+
	+	### Bill text (BILLS collection) - no key needed
	+	Full bill XML is accessible at:
	+	```
	+	https://www.govinfo.gov/content/pkg/BILLS-{congress}{type}{number}{version}/xml/BILLS-{congress}{type}{number}{version}.xml
	+	```
	+	This path does NOT require an API key. The `/api/` paths (JSON metadata) require a key.
	+
	+	### Sitemap entries are not sorted by bill number
	+	The sitemap contains entries in an arbitrary order. The first N entries from the sitemap are a random sample of bills, not the lowest-numbered ones. `--limit 200` will NOT get H.R. 1-200.
	+
	+	### BILLSTATUS XML root tag varies
	+	The root element is `<billStatus>` (lowercase) in modern files. The `<bill>` child is always there. Code handles both cases.
	+
	+	### Version date missing for `enr` (Enrolled Bill)
	+	The enrolled bill version often has no `version_date` in BILLSTATUS textVersions. The diff engine sorts by `(chain_rank, version_date)` so dateless entries fall into their correct chain position by rank. `enr` = rank 6 in the House chain.
	+
	+	### Section ID format changes between versions
	+	Different XML schemas are used for different version types:
	+	- Introduced (`ih`, `rh`) → long hex IDs: `H71241E5177324574B4C749197110EAB0`
	+	- Enrolled (`enr`) → short USLM IDs: `d244e72`
	+
	+	When IDs change between versions, the aligner falls back to fuzzy header matching (threshold 0.6). For short bills where all sections have the same content (e.g., Born-Alive Act H.R. 21 ih→eh), this produces 0 diff rows - correct, the text is identical.
	+
	+	For bills where the enrolled version switches from House HTML schema to USLM schema (common for enrolled bills), the aligner may incorrectly classify "removed" + "added" instead of "amended." This is a known GovInfo data quality issue. The enrolled→public-law pairs are most affected.
	+
	+	### Cross-chamber version codes (`pcs`, `rds`, `rih`, `ris`)
	+	House bills that go to the Senate acquire `pcs` (Placed on Calendar Senate) before becoming `enr`. Senate bills acquire `rds` (Received/Referred in Senate). These codes are in neither the standard House nor Senate chain; the diff engine assigns them rank 5 (after engrossed, before enrolled).
	+
	+	### Version code `rfs` (Referred in Senate)
	+	Appears for House bills passed and sent to Senate. Treated as rank 5 (cross-chamber). This is the same logical position as `pcs`.
	+
	+	### Public law version codes
	+	The BILLSTATUS XML sometimes includes a textVersion with type like `-119publ61` for Public Law 119-61. This is the enrolled public law text. Assigned rank 998 (near end of chain). Diffing `enr→-119publ61` usually shows section ID format changes only (not substantive differences).
	+
	+	---
	+
	+	## Real Data Results (as of 2026-06-26, 119th Congress sample)
	+
	+	Initial run (290 bills total):
	+
	+	\| Table \| Rows \|
	+	\|---\|---\|
	+	\| `bills` \| 290 \|
	+	\| `bill_versions` \| 410 \|
	+	\| `bill_diffs` \| 34 \|
	+	\| `amendments` \| 99 \|
	+	\| `actions` \| 1,695 \|
	+
	+	Bills with most versions:
	+	- H.J.Res. 25 (6 versions): Congressional disapproval of CFPB rule
	+	- H.R. 165 (6 versions): Wounded Knee Massacre Memorial Act
	+	- H.R. 131 (5 versions): Finish the Arkansas Valley Conduit Act
	+	- S.J.Res. 28/18/13/11 (5 versions each): Disapproval joint resolutions
	+
	+	Rider flags (3 flagged diff rows):
	+
	+	1. H.R. 43 (`rds→enr`) - Alaska Native Claims Settlement Act amendment:
	+	> "Section 14(c) of the Alaska Native Claims Settlement Act (43 U.S.C. 1613(c)) is amended..."
	+	Signal: `Is Amended`
	+
	+	2. S. 5 - Laken Riley Act (`es→enr`):
	+	> "Section 236(c) of the Immigration and Nationality Act (8 U.S.C. 1226(c)) is amended..."
	+	Signal: `Is Amended`
	+
	+	3. S. 5 - Laken Riley Act (`es→enr`, second section):
	+	> "Section 235(b) of the Immigration and Nationality Act (8 U.S.C. 1225(b)) is amended..."
	+	Signal: `Is Amended`
	+
	+	Real amended section example (H.R. 41, ih→rh):
	+	H.R. 41 - Unrecognized Southeast Alaska Native Communities Recognition and Compensation Act:
	+	Section H4DAD6E8E44444A988CA8A1864ADE20A6 was amended between the introduced version and the committee-reported version. The unified diff shows rewording of the Urban Corporation definition and conveyance authorization.
	+
	+	Real amended section example (S. 5, pcs→es):
	+	S. 5 - Laken Riley Act:
	+	Section H793DBCBA2A07463588AC4F4FDBEDECAF was amended between the placed-on-calendar and engrossed versions, showing Senate floor amendments to the detention mandate language (INA 236(c)).
	+
	+	---
	+
	+	## How to Run
	+
	+	```bash
	+	# Ingest all bill types for 119th Congress (runs from sitemap; ~9K HR + 4K S + …)
	+	python -m ingest.pollers.bills_bulk --once
	+
	+	# Test run (50 HR bills only)
	+	python -m ingest.pollers.bills_bulk --once --types hr --limit 50
	+
	+	# Discover additional text versions via HEAD-probe
	+	python -m ingest.pollers.bill_versions --once
	+
	+	# Diff all bills with 2+ versions
	+	python -m ingest.diff --once --all --congress 119
	+
	+	# Rider detection
	+	python -m ingest.rider_detect --once
	+
	+	# Amendments
	+	python -m ingest.amendments --once
	+
	+	# Run tests
	+	python -m pytest tests/test_bills_diff.py -v
	+	```
	+
	+	## Environment Variables
	+
	+	\| Variable \| Purpose \|
	+	\|---\|---\|
	+	\| `DOCKET_DB_PATH` \| SQLite path (default `/var/tmp/docket/docket.db`) \|
	+	\| `DOCKET_CONGRESS_KEY` \| api.data.gov key - unlocks Congress.gov API path in `amendments.py` \|
	+
	+	## Version Chain Order
	+
	+	House: `ih → rh → rh2 → eh → eah → eas → enr`
	+	Senate: `is → rs → pcs → es → eas → enr`
	+	Cross-chamber (rank 5): `pcs pch rds rih ris`
	+	Public law (rank 998): `-{congress}publ{n}`
	+
	+	## Amendment Data Coverage (honest gaps)
	+
	+	- Senate amendments 117th+ (2021+): machine-readable text available in GovInfo SAMDT collection. `amendments.py` stores congress.gov URLs; text can be fetched separately.
	+	- House amendments: text lives in Congressional Record (CREC), not discrete objects in BILLSTATUS. `text` column is NULL; source_url points to congress.gov amendment page.
	+	- Manager's / en-bloc packages: not public as discrete amendment objects anywhere. Only inferable from CREC narrative.
	+	- Voice vote / UC additions: `agreed_method` set to `'voice'` or `'uc'`; `text` is NULL; UI must show "no individual record exists."
	+	- Pre-117th amendment text: not machine-readable in bulk data; linked from congress.gov pages only.

services/ingest/CLOSEDDOOR_NOTES.md +173 -0

		@@ -0,0 +1,173 @@
	+	# Closed-Door + Executive Layer - Build Notes
	+	_Built 2026-06-26. All pollers verified live on CT247._
	+
	+	## What was built
	+
	+	Four new pollers in `services/ingest/ingest/pollers/`:
	+
	+	\| File \| Source \| Table \| Auth \|
	+	\|------\|--------\|-------\|------\|
	+	\| `federal_register.py` \| federalregister.gov/api/v1 \| `executive_actions` \| none \|
	+	\| `openomb.py` \| apportionment-public.max.gov \| `apportionments` \| none \|
	+	\| `crec.py` \| govinfo.gov/content/pkg/CREC-*/zip/ \| `procedural_events` \| none \|
	+	\| `lobbying.py` \| lda.gov/api/v1 \| `lobbying` \| env `DOCKET_LDA_KEY` \|
	+
	+	One new test file: `tests/test_crec_voice_vote.py` (62 tests).
	+
	+	All four pollers follow the existing pattern: `_ensure_tables()` creates new
	+	tables without touching `db.py`; `validate_source_url()` is called before every
	+	write; `is_new_event()` / `mark_seen()` enforce idempotency.
	+
	+	---
	+
	+	## Live run results (CT247 - 2026-06-26)
	+
	+	\| Table \| Rows \| Notes \|
	+	\|-------\|------\|-------\|
	+	\| `executive_actions` \| 500 \| 7 days of FR docs, all 4 types (PRESDOCU/RULE/PRORULE/NOTICE) \|
	+	\| `apportionments` \| 20 \| 20 records from FY2025 Corps of Engineers slice \|
	+	\| `procedural_events` \| 188 \| 5 session days from CREC-2026-06-16 through 06-23 \|
	+	\| `lobbying` \| 0 \| Graceful skip - `DOCKET_LDA_KEY` not set \|
	+
	+	---
	+
	+	## Sample: real Executive Order / rule row
	+
	+	```
	+	type: nprm
	+	title: Removing Obsolete References to Twentieth-Century Airman Certificates
	+	agency: Transportation Department; Federal Aviation Administration
	+	date: 2026-06-26
	+	source_url: https://www.federalregister.gov/documents/2026/06/26/2026-13003/...
	+	```
	+
	+	---
	+
	+	## Sample: real apportionment showing appropriated vs apportioned
	+
	+	```
	+	agency: Corps of Engineers--Civil Works
	+	account: Construction
	+	appropriated: $39,088,945,000
	+	apportioned: $39,088,945,000
	+	status: released
	+	date: 2024-09-06
	+	source_url: https://apportionment-public.max.gov/Fiscal%20Year%202025/Corps%20
	+	of%20Engineers--Civil%20Works/JSON/FY2025_Agency%3DCOE_...
	+	```
	+
	+	Status logic: apportioned/appropriated ≥ 95% → released; > 0 but < 95% → delayed; 0 → withheld.
	+
	+	The FY2025 Corps of Engineers records all show "released" (normal operating year).
	+	The impound-tracker signal emerges when `status = withheld` (OMB released $0 against
	+	an appropriation) - those rows are the headline feature for Docket's "buried data" pillar.
	+
	+	---
	+
	+	## Sample: real detected voice vote / procedural event
	+
	+	```
	+	kind: cloture
	+	date: 2026-06-23
	+	chamber: Both
	+	detail: Cloture action: …provisions of Rule XXII of the Standing Rules of the
	+	Senate, a vote on cloture will oc…
	+	members_jsonb: NULL ← no individual record exists (see honest gap below)
	+	source_url: https://www.govinfo.gov/app/details/CREC-2026-06-23
	+	```
	+
	+	CREC-2026-06-23 yielded: 80 events (cloture × 13, unanimous_consent × 63, suspension × 4).
	+	No voice-vote matches on that day - the Senate was on cloture votes, not voice-vote-only measures.
	+
	+	---
	+
	+	## Test results
	+
	+	```
	+	191 passed in 20.80s
	+	```
	+
	+	- `test_crec_voice_vote.py` - 62 tests: voice-vote regex, UC, cloture, suspension, MTR,
	+	discharge, XML/HTML extractor, chamber inference, snippet extraction, `_detect_events()`
	+	end-to-end, source-URL .gov enforcement, idempotency.
	+
	+	---
	+
	+	## Technical gotchas discovered during build
	+
	+	### Federal Register API - array param encoding
	+	Must use a list of `(key, value)` tuples (not a dict) for `fields[]` and
	+	`conditions[type][]`. The field `presidential_document_type` does not exist
	+	in the API - causes HTTP 400. Use `abstract` instead.
	+
	+	### GovInfo CREC - correct URL
	+	Bulk data path (`/bulkdata/CREC/`) returns HTML error pages, not ZIPs.
	+	The working URL is:
	+	```
	+	https://www.govinfo.gov/content/pkg/CREC-YYYY-MM-DD/zip/CREC-YYYY-MM-DD.zip
	+	```
	+	Non-session days (weekends, recesses) return HTML (not 404) - "bad ZIP" errors
	+	are expected and handled gracefully. Default look-back is 14 days to cover 5+
	+	session days.
	+
	+	### CREC ZIP contents - HTML, not XML
	+	The ZIP contains `.htm` files (not standalone XML). Text extraction strips HTML
	+	tags with regex + `html.unescape()`. The `_extract_text_from_xml()` function
	+	handles both correctly.
	+
	+	### max.gov apportionments - no REST API
	+	`apportionment-public.max.gov` has no JSON API endpoint. The site serves
	+	individual JSON files per apportionment record (one per TAFS account + FY +
	+	iteration), listed in an HTML index page. The poller:
	+	1. Fetches the HTML index (~500KB, loads in ~20s)
	+	2. Extracts all JSON file hrefs matching the target FY via regex
	+	3. Downloads individual records (capped at 200/run, 50 for `--once`)
	+
	+	FY2025 has 6,172 JSON links; FY2026 will grow through the year. A scheduled
	+	hourly/daily run can work through the full set over time.
	+
	+	openomb.org (the aggregator) returns HTTP 403 for automated requests - use
	+	max.gov directly, which is also the correct source_url target per the
	+	neutrality rule.
	+
	+	### Neutrality rule - source_url enforcement
	+	- `federal_register.py`: html_url from FR API → federalregister.gov (.gov ✓)
	+	- `openomb.py`: direct max.gov JSON file URL (.max.gov = .gov ✓)
	+	- `crec.py`: govinfo.gov app/details page (.gov ✓)
	+	- `lobbying.py`: lda.gov/filings/{uuid}/ (.gov ✓)
	+
	+	All validated by `source_validator.validate_source_url()` before every DB write.
	+
	+	---
	+
	+	## How to run
	+
	+	```bash
	+	# On CT247 (venv at /opt/docket/services/ingest/.venv)
	+	cd /shared/projects/docket/services/ingest
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.federal_register --once
	+
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.openomb --once --max 50
	+
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.crec --once --days 14
	+
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db DOCKET_LDA_KEY=<your_key> \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.pollers.lobbying --once
	+	```
	+
	+	## LDA key
	+
	+	Register free at https://lda.gov/api/. Without a key the lobbying poller logs
	+	a clear message and returns 0 - no crash, no partial state.
	+
	+	## Recommended poll cadences
	+
	+	\| Poller \| Cadence \| Notes \|
	+	\|--------\|---------\|-------\|
	+	\| federal_register \| Every 4h \| Same-day FR lag \|
	+	\| openomb \| Daily \| max.gov updates when OMB signs a new apportionment \|
	+	\| crec \| Daily (noon ET) \| ~1-day CREC publishing lag \|
	+	\| lobbying \| Weekly \| LDA filings are quarterly \|

services/ingest/VOTES_NOTES.md +155 -0

		@@ -0,0 +1,155 @@
	+	# Docket - Votes & Members Build Notes
	+
	+	Verified 2026-06-26 on CT247 (venv `/opt/docket/services/ingest/.venv`, DB `/opt/docket/data/docket.db`).
	+
	+	## Files Created / Modified
	+
	+	\| File \| Status \| Purpose \|
	+	\|---\|---\|---\|
	+	\| `ingest/pollers/members.py` \| new \| Full current member roster (537 members) \|
	+	\| `ingest/pollers/votes_house.py` \| rewritten \| House XML with per-member positions \|
	+	\| `ingest/pollers/votes_enrich.py` \| new \| Senate positions backfill from existing vote source_urls \|
	+	\| `tests/test_votes_house_parser.py` \| new \| 39 tests: XML parser, position enum, schema, both pivots \|
	+
	+	## Schema additions (additive - all CREATE TABLE IF NOT EXISTS / ALTER TABLE)
	+
	+	```sql
	+	-- members table (new)
	+	CREATE TABLE IF NOT EXISTS members (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bioguide_id TEXT NOT NULL UNIQUE,
	+	full_name TEXT NOT NULL,
	+	party TEXT, state TEXT, district INTEGER, chamber TEXT,
	+	lis_member_id TEXT, govtrack_id INTEGER, icpsr_id INTEGER,
	+	source_url TEXT NOT NULL -- https://bioguide.congress.gov/search/bio/{id}
	+	);
	+	CREATE INDEX IF NOT EXISTS idx_members_lis ON members(lis_member_id);
	+
	+	-- vote_positions table (new)
	+	CREATE TABLE IF NOT EXISTS vote_positions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	vote_id INTEGER NOT NULL REFERENCES votes(id),
	+	member_bioguide TEXT,
	+	member_id INTEGER REFERENCES members(id),
	+	position TEXT NOT NULL CHECK(position IN ('Yea','Nay','Present','Not Voting'))
	+	);
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_vote_id ON vote_positions(vote_id);
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_member_id ON vote_positions(member_id);
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_bioguide ON vote_positions(member_bioguide);
	+
	+	-- Extended columns on votes (all nullable, backward-compatible)
	+	ALTER TABLE votes ADD COLUMN congress INTEGER;
	+	ALTER TABLE votes ADD COLUMN session INTEGER;
	+	ALTER TABLE votes ADD COLUMN roll_number INTEGER;
	+	ALTER TABLE votes ADD COLUMN question TEXT;
	+	ALTER TABLE votes ADD COLUMN not_voting INTEGER;
	+	ALTER TABLE votes ADD COLUMN recorded INTEGER DEFAULT 1;
	+	ALTER TABLE votes ADD COLUMN vote_external_id TEXT;
	+	ALTER TABLE votes ADD COLUMN bill_text_ref TEXT;
	+	```
	+
	+	## Data ingested (2026-06-26 baseline)
	+
	+	\| Source \| Count \|
	+	\|---\|---\|
	+	\| Members \| 537 (437 House, 100 Senate) \|
	+	\| House votes (2026, rolls 1-62) \| 50 \|
	+	\| Senate votes (119/1, votes 1-40) \| 40 \|
	+	\| vote_positions total \| 25,572 \|
	+	\| vote_positions (House) \| 21,581 \|
	+	\| vote_positions (Senate) \| 3,991 \|
	+
	+	## Running the pollers
	+
	+	```bash
	+	# Members roster (run first - needed for member_id foreign keys)
	+	python -m ingest.pollers.members --once
	+
	+	# House votes (probes current year sequentially, stops at 404)
	+	python -m ingest.run --source votes_house
	+
	+	# Senate votes (existing poller, no modification)
	+	python -m ingest.run --source votes_senate
	+
	+	# Senate positions backfill (run after members + senate votes)
	+	python -m ingest.pollers.votes_enrich --once
	+	```
	+
	+	## Worked example - both pivot directions
	+
	+	### Pivot 1: "Who voted how on House Roll #62?"
	+
	+	```
	+	Roll #62 (2026-02-11) - On Agreeing to the Resolution, as Amended
	+	H RES 1057 \| Passed \| Y:216 N:215 P:0 NV:1
	+	432 positions recorded
	+	Source: https://clerk.house.gov/evs/2026/roll062.xml
	+
	+	Sample:
	+	Shomari Figures (D-AL): Nay
	+	Terri A. Sewell (D-AL): Nay
	+	Nicholas J. Begich III (R-AK): Yea
	+	Robert B. Aderholt (R-AL): Yea
	+	```
	+
	+	Query:
	+	```sql
	+	SELECT vp.position, m.full_name, m.party, m.state
	+	FROM vote_positions vp
	+	LEFT JOIN members m ON m.bioguide_id = vp.member_bioguide
	+	WHERE vp.vote_id = (SELECT id FROM votes WHERE roll_number=62 AND chamber='house')
	+	ORDER BY m.state, m.full_name;
	+	```
	+
	+	### Pivot 2: "How did Rep. Aderholt vote recently?"
	+
	+	```
	+	Robert B. Aderholt (R-AL) \| bioguide: A000055
	+	Source: https://bioguide.congress.gov/search/bio/A000055
	+	50 recorded votes in DB
	+
	+	Roll #062 \| 2026-02-11 \| On Agreeing to the Resolution, as Amended -> Yea
	+	Roll #061 \| 2026-02-11 \| On Ordering the Previous Question -> Yea
	+	Roll #059 \| 2026-02-10 \| On Ordering the Previous Question -> Yea
	+	Roll #054 \| 2026-02-04 \| On Motion to Recommit -> Nay
	+	```
	+
	+	Query:
	+	```sql
	+	SELECT v.roll_number, v.date, v.question, vp.position
	+	FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = 'A000055'
	+	ORDER BY v.date DESC, v.roll_number DESC;
	+	```
	+
	+	## Key design decisions
	+
	+	Members source: `unitedstates/congress-legislators` `gh-pages` branch JSON (not `main` - YAML only). Fetch URL is `raw.githubusercontent.com`; stored `source_url` is always `bioguide.congress.gov/search/bio/{id}` (allowlisted .gov).
	+
	+	House XML date format: The Clerk XML uses `"D-Mon-YYYY"` (e.g. `"8-Jan-2026"`), not `MM/DD/YYYY` as some docs suggest. The parser tries `%d-%b-%Y` first, then fallbacks.
	+
	+	House Clerk URL format: `https://clerk.house.gov/evs/{year}/roll{NNN}.xml` (lowercase `roll`, no underscore, zero-padded 3 digits). The prior stub incorrectly used `ROLL_{NNN}.xml`.
	+
	+	Bioguide join: House XML `<legislator name-id="...">` carries the bioguide directly. Senate XML uses `lis_member_id`; the enrich module maps it via `members.lis_member_id`.
	+
	+	Unmatched members: Delegates (DC, territories, etc.) appear in House roll-call XMLs but not in `legislators-current.json`. Their positions are stored with `member_bioguide` populated and `member_id = NULL` - data preserved, nothing fabricated.
	+
	+	Honesty on voice/UC votes: `parse_house_vote_xml()` returns `None` for non-recorded vote types (`VOICE VOTE`, etc.). Those votes are not in this table. The closed-door agent (CREC) owns them.
	+
	+	Senate backfill approach: `votes_enrich.py` re-fetches the `source_url` already stored on existing `votes` rows - no external mapping needed. Idempotent: only processes rows with no existing positions.
	+
	+	## Tests (39 total, 230 total suite)
	+
	+	```
	+	tests/test_votes_house_parser.py - 39 tests
	+
	+	TestNormalizePosition (6) - position enum mapping incl. Aye/No variants
	+	TestParseLegisNum (9) - bill type normalization, all 8 prefixes
	+	TestParseHouseVoteXml (9) - fields, counts, bioguides, voice-vote honesty
	+	TestParseSenateVotePositions (9) - LIS IDs, all 4 positions, malformed XML
	+	TestSchemaAndInsertion (6) - table creation, column presence, enum enforcement,
	+	both pivot directions via JOIN
	+	```
	+
	+	All 230 tests pass (suite includes pre-existing ranking/source_url/state/diff/crec tests).

services/ingest/check_db_tmp.py +0 -29

		@@ -1,29 +0,0 @@
	-	import sqlite3, os
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	conn = sqlite3.connect(db)
	-	conn.row_factory = sqlite3.Row
	-
	-	print("=== BILLS ===")
	-	r = conn.execute("SELECT COUNT(*) n FROM bills WHERE source='govinfo_billstatus'").fetchone()
	-	print(" Total bills:", r["n"])
	-
	-	print("\n=== SAMPLE BILLS ===")
	-	for row in conn.execute("SELECT b.number, b.title, be.sponsor_name, be.sponsor_party, be.sponsor_state FROM bills b LEFT JOIN bills_ext be ON be.bill_id=b.id WHERE b.source='govinfo_billstatus' LIMIT 5").fetchall():
	-	print(" ", row["number"], "\|", row["title"][:50], "\|", row["sponsor_name"], f'({row["sponsor_party"]}-{row["sponsor_state"]})')
	-
	-	print("\n=== BILL_VERSIONS ===")
	-	r = conn.execute("SELECT COUNT(*) n FROM bill_versions").fetchone()
	-	print(" Total versions:", r["n"])
	-
	-	print("\n=== VERSION CODES ===")
	-	for row in conn.execute("SELECT version_code, COUNT(*) n FROM bill_versions GROUP BY version_code ORDER BY n DESC").fetchall():
	-	print(" ", row["version_code"], ":", row["n"])
	-
	-	print("\n=== BILLS WITH 2+ VERSIONS ===")
	-	for row in conn.execute("SELECT b.number, b.title, COUNT(bv.id) nv FROM bills b JOIN bill_versions bv ON bv.bill_id=b.id WHERE b.source='govinfo_billstatus' GROUP BY b.id HAVING nv >= 2 ORDER BY nv DESC LIMIT 10").fetchall():
	-	print(" ", row["number"], "(", row["nv"], "versions):", row["title"][:60])
	-
	-	print("\n=== ACTIONS ===")
	-	r = conn.execute("SELECT COUNT(*) n FROM actions WHERE source='govinfo_billstatus'").fetchone()
	-	print(" Total actions:", r["n"])

services/ingest/check_diff2_tmp.py +0 -38

		@@ -1,38 +0,0 @@
	-	import sqlite3, os, sys, requests
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	from ingest.diff import fetch_version_xml, parse_sections
	-
	-	conn = sqlite3.connect(db)
	-	conn.row_factory = sqlite3.Row
	-
	-	# Find H.R. 21 (Born-Alive, 3 versions)
	-	for num in ["H.R. 21", "H.R. 27", "H.R. 26"]:
	-	row = conn.execute("SELECT id, title FROM bills WHERE number=? AND source='govinfo_billstatus'", (num,)).fetchone()
	-	if row:
	-	bill_id = row["id"]
	-	title = row["title"]
	-	vers = conn.execute("SELECT version_code, text_ref FROM bill_versions WHERE bill_id=? ORDER BY version_date", (bill_id,)).fetchall()
	-	if len(vers) >= 2:
	-	print(f"Bill: {num} id={bill_id} '{title[:50]}'")
	-	for v in vers:
	-	print(f" {v['version_code']}: {v['text_ref']}")
	-	# Check section count in first two versions
	-	v0, v1 = vers[0], vers[1]
	-	xml0 = fetch_version_xml(v0["text_ref"])
	-	xml1 = fetch_version_xml(v1["text_ref"])
	-	if xml0:
	-	s0 = parse_sections(xml0)
	-	print(f" Sections in {v0['version_code']}: {len(s0)}")
	-	for s in s0[:3]:
	-	print(f" sec_id={s.sec_id!r} enum={s.enum!r} header={s.header!r}")
	-	if xml1:
	-	s1 = parse_sections(xml1)
	-	print(f" Sections in {v1['version_code']}: {len(s1)}")
	-	for s in s1[:3]:
	-	print(f" sec_id={s.sec_id!r} enum={s.enum!r} header={s.header!r}")
	-	break
	-
	-	conn.close()

services/ingest/check_diff_tmp.py +0 -53

		@@ -1,53 +0,0 @@
	-	import sqlite3, os, sys
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	from ingest.db import get_connection, init_schema
	-	from ingest.pollers.bills_bulk import ensure_schema
	-	from ingest.diff import diff_bill
	-
	-	conn = get_connection(db)
	-	init_schema(conn)
	-	ensure_schema(conn)
	-
	-	# Find bill_id for HR 165
	-	row = conn.execute("SELECT id, title FROM bills WHERE number='H.R. 165' AND source='govinfo_billstatus'").fetchone()
	-	if row is None:
	-	print("HR 165 not found, trying HR 21")
	-	row = conn.execute("SELECT id, title FROM bills WHERE number='H.R. 21' AND source='govinfo_billstatus'").fetchone()
	-
	-	if row is None:
	-	print("No multi-version bill found")
	-	# Find any bill with 2+ versions
	-	row2 = conn.execute("""SELECT b.id, b.number, b.title, COUNT(bv.id) nv
	-	FROM bills b JOIN bill_versions bv ON bv.bill_id=b.id
	-	WHERE b.source='govinfo_billstatus'
	-	GROUP BY b.id HAVING nv >= 2 ORDER BY nv DESC LIMIT 1""").fetchone()
	-	if row2:
	-	bill_id = row2["id"]
	-	print(f"Using bill {row2['number']}: {row2['title'][:60]} ({row2['nv']} versions)")
	-	else:
	-	print("No multi-version bills found")
	-	sys.exit(1)
	-	else:
	-	bill_id = row["id"]
	-	print(f"Diffing bill_id={bill_id}: {row['title'][:60]}")
	-
	-	# Show versions
	-	versions = conn.execute("SELECT version_code, version_date, text_ref FROM bill_versions WHERE bill_id=? ORDER BY version_date", (bill_id,)).fetchall()
	-	print("\nVersions:")
	-	for v in versions:
	-	print(f" {v['version_code']}: {v['version_date']}")
	-
	-	print("\nRunning diff_bill...")
	-	n = diff_bill(bill_id, conn)
	-	print(f"\nDiff rows written: {n}")
	-
	-	print("\n=== BILL_DIFFS ===")
	-	for row in conn.execute("SELECT from_version, to_version, section_id, change, rider_flag, snippet FROM bill_diffs WHERE bill_id=? LIMIT 20", (bill_id,)).fetchall():
	-	print(f" {row['from_version']}→{row['to_version']} \| {row['change']:8s} \| rider={row['rider_flag']} \| section={row['section_id']}")
	-	if row['snippet']:
	-	print(f" snippet: {row['snippet'][:120]}")
	-
	-	conn.close()

services/ingest/find_multi_ver_tmp.py +0 -38

		@@ -1,38 +0,0 @@
	-	import sqlite3, os, sys
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	conn = sqlite3.connect(db)
	-	conn.row_factory = sqlite3.Row
	-
	-	print("=== BILLS WITH 3+ VERSIONS (all types) ===")
	-	for row in conn.execute("""
	-	SELECT b.id, b.number, b.title, be.bill_type, COUNT(bv.id) nv
	-	FROM bills b
	-	JOIN bill_versions bv ON bv.bill_id=b.id
	-	LEFT JOIN bills_ext be ON be.bill_id=b.id
	-	WHERE b.source='govinfo_billstatus'
	-	GROUP BY b.id HAVING nv >= 3 ORDER BY nv DESC LIMIT 15
	-	""").fetchall():
	-	vcs = [r["version_code"] for r in conn.execute("SELECT version_code FROM bill_versions WHERE bill_id=? ORDER BY version_date", (row["id"],)).fetchall()]
	-	print(f" {row['number']:15s} ({row['nv']} vers) {row['bill_type']:7s}: {row['title'][:55]}")
	-	print(f" versions: {' → '.join(vcs)}")
	-
	-	print("\n=== APPROPRIATIONS / HJRES BILLS ===")
	-	for row in conn.execute("""
	-	SELECT b.id, b.number, b.title, be.bill_type, COUNT(bv.id) nv
	-	FROM bills b
	-	JOIN bill_versions bv ON bv.bill_id=b.id
	-	LEFT JOIN bills_ext be ON be.bill_id=b.id
	-	WHERE b.source='govinfo_billstatus' AND (
	-	be.bill_type='HJRES' OR be.bill_type='SJRES'
	-	OR LOWER(b.title) LIKE '%appropriat%'
	-	)
	-	GROUP BY b.id ORDER BY nv DESC LIMIT 10
	-	""").fetchall():
	-	vcs = [r["version_code"] for r in conn.execute("SELECT version_code FROM bill_versions WHERE bill_id=? ORDER BY version_date", (row["id"],)).fetchall()]
	-	print(f" {row['number']:15s} ({row['nv']} vers) {row['bill_type']:7s}: {row['title'][:55]}")
	-	print(f" versions: {' → '.join(vcs)}")
	-
	-	conn.close()

services/ingest/ingest/amendments.py +129 -46

		@@ -86,47 +86,44 @@ def _classify_agreed_method(description: str) -> Optional[str]:

		def _ingest_from_billstatus(conn: sqlite3.Connection, congress: int) -> int:
		"""
	-	Parse amendment references from bills already in the DB.
	-	BILLSTATUS XML sometimes contains <amendments> sections.
	-	Reads raw_events for the govinfo_billstatus source.
	+	Parse amendment references from action rows already stored by bills_bulk.
	+
	+	bills_bulk.py stores one row per action in the `actions` table. Any action
	+	whose text references "H.Amdt." or "S.Amdt." plus a number is treated as an
	+	amendment reference and stored in the `amendments` table with whatever
	+	metadata can be inferred from the action text.
	+
	+	Honest gaps:
	+	- House amendment TEXT is rarely machine-readable here; text=NULL unless
	+	the action row itself quotes it.
	+	- agreed_method inferred from action text keywords; may be NULL.
	+	- 117th Congress (2021)+ has machine-readable Senate amendment text in
	+	GovInfo; older congresses have Congress-Record-only text.
		"""
	-	rows = conn.execute(
	-	"""
	-	SELECT re.external_id, re.payload
	-	FROM raw_events re
	-	JOIN bills b ON b.source = re.source AND b.external_id = re.external_id
	-	WHERE re.source = ? AND b.congress = ?
	-	""",
	+	import re as _re
	+
	+	# All bills for this congress
	+	bills = conn.execute(
	+	"SELECT id, source_url FROM bills WHERE source=? AND congress=?",
		(BILL_SOURCE, congress),
		).fetchall()

		stored = 0
	-	for row in rows:
	-	try:
	-	payload = json.loads(row["payload"])
	-	except (json.JSONDecodeError, TypeError):
	-	continue
	+	for bill in bills:
	+	bill_id = bill["id"]
	+	source_url = bill["source_url"]

	-	# The payload stored by bills_bulk is the raw XML bytes or a dict summary.
	-	# bills_bulk doesn't store raw_events - but we can look up the bill_id
	-	# and check its actions for amendment references.
	-	bill_ext_id = row["external_id"] # e.g. "119-hr-1"
	-	bill_row = conn.execute(
	-	"SELECT id FROM bills WHERE source=? AND external_id=?",
	-	(BILL_SOURCE, bill_ext_id),
	-	).fetchone()
	-	if bill_row is None:
	-	continue
	-	bill_id = bill_row["id"]
	-
	-	# Look for amendment action rows (action_type contains 'Amendment')
	+	# Look for amendment action rows
		amdt_actions = conn.execute(
		"""
		SELECT action_date, text, action_type, chamber
		FROM actions
	-	WHERE bill_id = ? AND (
	+	WHERE bill_id=? AND (
		LOWER(action_type) LIKE '%amendment%'
	-	OR LOWER(text) LIKE '%amendment%'
	+	OR text LIKE '%Amdt.%'
	+	OR text LIKE '%amendment%agreed%'
	+	OR text LIKE '%amendment%rejected%'
	+	OR text LIKE '%amendment%tabled%'
		)
		ORDER BY action_date
		""",
		@@ -134,19 +131,33 @@ def _ingest_from_billstatus(conn: sqlite3.Connection, congress: int) -> int:
		).fetchall()

		for a in amdt_actions:
	-	atext = a["text"] or ""
	-	# Try to extract amendment number from text
	-	import re
	-	m = re.search(r"(?:H\.\|S\.)\sAmdt\.?\s(\d+)", atext, re.IGNORECASE)
	-	if m:
	-	amdt_num = m.group(1)
	-	amdt_type = "HAMDT" if a["chamber"] == "House" else "SAMDT"
	-	source_url = _congress_amdt_url(congress, amdt_type, amdt_num)
	+	atext = a["text"] or ""
	+	chamber = a["chamber"] or ""
	+
	+	# Extract "H.Amdt. NNN" or "S.Amdt. NNN"
	+	for m in _re.finditer(
	+	r"([HS])\.?\sAmdt\.?\s(\d+)", atext, _re.IGNORECASE
	+	):
	+	prefix = m.group(1).upper()
	+	amdt_num = m.group(2)
	+	amdt_type = "HAMDT" if prefix == "H" else "SAMDT"
	+
	+	amdt_url = _congress_amdt_url(congress, amdt_type, amdt_num)
		try:
	-	validate_source_url(source_url)
	+	validate_source_url(amdt_url)
		except Exception:
		continue
	+
		agreed = _classify_agreed_method(atext)
	+
	+	# Infer status from action text
	+	status = "proposed"
	+	low = atext.lower()
	+	if "agreed to" in low or "passed" in low:
	+	status = "agreed"
	+	elif "rejected" in low or "tabled" in low or "failed" in low:
	+	status = "rejected"
	+
		conn.execute(
		"""
		INSERT INTO amendments
		@@ -155,14 +166,15 @@ def _ingest_from_billstatus(conn: sqlite3.Connection, congress: int) -> int:
		VALUES (?, ?, ?, ?, ?, ?, ?, ?)
		ON CONFLICT(congress, type, number) DO UPDATE SET
		bill_id = excluded.bill_id,
	-	status = excluded.status,
	-	agreed_method = COALESCE(excluded.agreed_method, agreed_method),
	+	status = COALESCE(excluded.status, amendments.status),
	+	agreed_method = COALESCE(excluded.agreed_method, amendments.agreed_method),
		source_url = excluded.source_url
		""",
		(
		congress, amdt_type, amdt_num, bill_id,
	-	"enacted" if "enacted" in atext.lower() else "proposed",
	-	agreed, None, source_url,
	+	status, agreed,
	+	None, # text=NULL - not machine-readable at this source
	+	amdt_url,
		),
		)
		stored += 1
		@@ -273,6 +285,72 @@ def _ingest_from_congress_api(
		# Main poll
		# ---------------------------------------------------------------------------

	+	def _ingest_by_refetching_billstatus(conn: sqlite3.Connection, congress: int) -> int:
	+	"""
	+	For each bill already in the DB that has no amendments yet,
	+	re-fetch its BILLSTATUS XML from GovInfo and parse the <amendments> element.
	+
	+	This is needed because bills_bulk.py only parses amendments for NEW bills.
	+	Existing bills (ingested before this code was added) need a one-time backfill.
	+	"""
	+	from ingest.pollers.bills_bulk import (
	+	_store_amendments_from_bill_el,
	+	_iter_bill_elements,
	+	BILL_TYPE_SLUG,
	+	)
	+
	+	BULK_BASE = "https://www.govinfo.gov/bulkdata/BILLSTATUS"
	+	session = requests.Session()
	+	session.headers["User-Agent"] = "Docket-ingest/1.0 (transparency research)"
	+
	+	# Bills in this congress that have no amendments row yet
	+	bills = conn.execute(
	+	"""
	+	SELECT b.id, b.number, be.bill_type, b.source_url
	+	FROM bills b
	+	LEFT JOIN bills_ext be ON be.bill_id=b.id
	+	WHERE b.source='govinfo_billstatus' AND b.congress=?
	+	""",
	+	(congress,),
	+	).fetchall()
	+
	+	total = 0
	+	for bill in bills:
	+	bill_id = bill["id"]
	+	raw_num = bill["number"] or "" # "H.R. 5" → need the numeric part
	+	bill_type = (bill["bill_type"] or "").lower()
	+
	+	# Extract numeric part
	+	num = "".join(c for c in raw_num if c.isdigit())
	+	if not num or not bill_type:
	+	continue
	+
	+	# Already has amendments? Skip
	+	existing = conn.execute(
	+	"SELECT COUNT(*) n FROM amendments WHERE bill_id=?", (bill_id,)
	+	).fetchone()["n"]
	+	if existing > 0:
	+	continue
	+
	+	url = f"{BULK_BASE}/{congress}/{bill_type}/BILLSTATUS-{congress}{bill_type}{num}.xml"
	+	try:
	+	r = session.get(url, timeout=30)
	+	if r.status_code != 200:
	+	continue
	+	xml_bytes = r.content
	+	except requests.RequestException:
	+	continue
	+
	+	for bill_el in _iter_bill_elements(xml_bytes):
	+	n = _store_amendments_from_bill_el(
	+	conn, bill_el, bill_id, congress, bill["source_url"] or ""
	+	)
	+	total += n
	+
	+	conn.commit()
	+	return total
	+
	+
		def poll(congress: int = 119, db_path: Optional[str] = None) -> int:
		"""
		Populate the amendments table.
		@@ -284,12 +362,17 @@ def poll(congress: int = 119, db_path: Optional[str] = None) -> int:

		total = 0

	-	# Path 1: parse from already-stored bill action rows
	+	# Path 1: Re-fetch BILLSTATUS XMLs to parse <amendments> elements
	+	n = _ingest_by_refetching_billstatus(conn, congress)
	+	logger.info("amendments: %d rows from BILLSTATUS XML re-fetch", n)
	+	total += n
	+
	+	# Path 2: parse from action-text references (supplementary)
		n = _ingest_from_billstatus(conn, congress)
	-	logger.info("amendments: %d rows from BILLSTATUS action parsing", n)
	+	logger.info("amendments: %d rows from action-text parsing", n)
		total += n

	-	# Path 2: Congress.gov API (if key present)
	+	# Path 3: Congress.gov API (if key present)
		key = os.environ.get("DOCKET_CONGRESS_KEY", "")
		if key:
		n = _ingest_from_congress_api(conn, congress, key)

services/ingest/ingest/pollers/bills_bulk.py +95 -0

		@@ -456,9 +456,104 @@ def _store_bill(
		(bill_id, vcode, pkg_id, label, vdate, text_ref, text_ref),
		)

	+	# Amendments from the <amendments> element
	+	_store_amendments_from_bill_el(conn, bill_el, bill_id, congress, source_url)
	+
		return bill_id


	+	def _store_amendments_from_bill_el(
	+	conn: sqlite3.Connection,
	+	bill_el: ET.Element,
	+	bill_id: int,
	+	congress: int,
	+	default_source_url: str,
	+	) -> int:
	+	"""
	+	Parse the <amendments> section of a BILLSTATUS <bill> element and
	+	upsert into the amendments table.
	+
	+	Agreed method is inferred from the latestAction text:
	+	Yea-Nay / Roll Call → 'recorded'
	+	Voice Vote → 'voice'
	+	Unanimous Consent → 'uc'
	+	NULL otherwise.
	+
	+	Note: House amendment TEXT is almost never in BILLSTATUS XML; text stays NULL.
	+	Senate amendment text is in GovInfo SAMDT collection (117th+).
	+	"""
	+	import re as _re
	+
	+	amd_el = bill_el.find("amendments")
	+	if amd_el is None:
	+	return 0
	+
	+	stored = 0
	+	for amd in amd_el.findall("amendment"):
	+	amd_num = _txt(amd, "number")
	+	amd_type = _txt(amd, "type") # HAMDT or SAMDT
	+	amd_cong = _txt(amd, "congress") or str(congress)
	+
	+	if not amd_num or not amd_type:
	+	continue
	+
	+	# Sponsor
	+	sp_el = amd.find("sponsors")
	+	sp_bio = ""
	+	if sp_el is not None:
	+	first = sp_el.find("item")
	+	if first is not None:
	+	sp_bio = _txt(first, "bioguideId")
	+
	+	# Status / agreed_method from latestAction
	+	la_el = amd.find("latestAction")
	+	status = ""
	+	agreed = None
	+	if la_el is not None:
	+	la_text = _txt(la_el, "text")
	+	status = la_text[:200]
	+	low = la_text.lower()
	+	if any(k in low for k in ("yea-nay", "recorded vote", "roll call")):
	+	agreed = "recorded"
	+	elif "voice vote" in low:
	+	agreed = "voice"
	+	elif "unanimous consent" in low or "u.c." in low:
	+	agreed = "uc"
	+
	+	# source_url: congress.gov amendment page
	+	slug_map = {"HAMDT": "house-amendment", "SAMDT": "senate-amendment"}
	+	slug = slug_map.get(amd_type.upper(), amd_type.lower())
	+	amd_url = (
	+	f"https://www.congress.gov/amendment/{amd_cong}/{slug}/{amd_num}"
	+	)
	+	try:
	+	validate_source_url(amd_url)
	+	except Exception:
	+	continue
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO amendments
	+	(congress, type, number, bill_id, sponsor_id,
	+	status, agreed_method, text, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(congress, type, number) DO UPDATE SET
	+	bill_id = COALESCE(excluded.bill_id, amendments.bill_id),
	+	sponsor_id = excluded.sponsor_id,
	+	status = COALESCE(excluded.status, amendments.status),
	+	agreed_method = COALESCE(excluded.agreed_method, amendments.agreed_method),
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	int(amd_cong), amd_type.upper(), amd_num, bill_id,
	+	sp_bio, status, agreed, None, amd_url,
	+	),
	+	)
	+	stored += 1
	+
	+	return stored
	+
	+
		# ---------------------------------------------------------------------------
		# Main poll
		# ---------------------------------------------------------------------------

services/ingest/ingest/pollers/members.py +260 -0

		@@ -0,0 +1,260 @@
	+	"""
	+	Member roster poller - unitedstates/congress-legislators.
	+
	+	Fetches `legislators-current.json` from the keyless GitHub dataset and
	+	populates the `members` table with full metadata. source_url for every
	+	member row is the official Bioguide page at bioguide.congress.gov (a .gov
	+	domain, allowlisted by rule #1).
	+
	+	Also owns the shared schema additions used by votes_house.py and
	+	votes_enrich.py:
	+	• members table
	+	• vote_positions table (+ indexes)
	+	• Extended columns on the votes table (congress, session, roll_number, …)
	+
	+	Run once (CLI):
	+	python -m ingest.pollers.members --once
	+
	+	Imported by:
	+	votes_house.poll() - calls ensure_votes_schema() before inserting
	+	votes_enrich.poll() - same
	+	"""
	+
	+	import argparse
	+	import logging
	+	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 = "members"
	+
	+	LEGISLATORS_JSON_URL = (
	+	"https://raw.githubusercontent.com/unitedstates/congress-legislators"
	+	"/gh-pages/legislators-current.json"
	+	)
	+
	+	# Official bioguide page - ends in .gov, always allowlisted.
	+	BIOGUIDE_BASE = "https://bioguide.congress.gov/search/bio"
	+
	+
	+	# ─── Shared schema setup ──────────────────────────────────────────────────────
	+
	+	def ensure_votes_schema(conn) -> None:
	+	"""
	+	Idempotent: create members + vote_positions tables and extend the votes
	+	table with richer columns. Safe to call from any poller before writing.
	+	"""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS members (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bioguide_id TEXT NOT NULL UNIQUE,
	+	full_name TEXT NOT NULL,
	+	party TEXT,
	+	state TEXT,
	+	district INTEGER,
	+	chamber TEXT,
	+	lis_member_id TEXT,
	+	govtrack_id INTEGER,
	+	icpsr_id INTEGER,
	+	source_url TEXT NOT NULL
	+	);
	+
	+	CREATE INDEX IF NOT EXISTS idx_members_lis
	+	ON members(lis_member_id);
	+
	+	CREATE TABLE IF NOT EXISTS vote_positions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	vote_id INTEGER NOT NULL REFERENCES votes(id),
	+	member_bioguide TEXT,
	+	member_id INTEGER REFERENCES members(id),
	+	position TEXT NOT NULL
	+	CHECK(position IN ('Yea','Nay','Present','Not Voting'))
	+	);
	+
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_vote_id
	+	ON vote_positions(vote_id);
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_member_id
	+	ON vote_positions(member_id);
	+	CREATE INDEX IF NOT EXISTS idx_vote_positions_bioguide
	+	ON vote_positions(member_bioguide);
	+	"""
	+	)
	+
	+	# Additive columns on the pre-existing votes table.
	+	# SQLite has no IF NOT EXISTS for ALTER TABLE; use try/except.
	+	extra = [
	+	("congress", "INTEGER"),
	+	("session", "INTEGER"),
	+	("roll_number", "INTEGER"),
	+	("question", "TEXT"),
	+	("not_voting", "INTEGER"),
	+	("recorded", "INTEGER DEFAULT 1"),
	+	("vote_external_id", "TEXT"),
	+	("bill_text_ref", "TEXT"),
	+	]
	+	for col, col_type in extra:
	+	try:
	+	conn.execute(f"ALTER TABLE votes ADD COLUMN {col} {col_type}")
	+	except Exception:
	+	pass # column already present - ignore
	+
	+	conn.commit()
	+
	+
	+	# ─── Data parsing ─────────────────────────────────────────────────────────────
	+
	+	def _parse_legislators(data: list) -> list[dict]:
	+	"""
	+	Convert the raw legislators JSON list into cleaned member dicts.
	+	Uses the most recent term to determine current chamber / party / district.
	+	"""
	+	members = []
	+	for leg in data:
	+	ids = leg.get("id", {})
	+	name = leg.get("name", {})
	+	terms = leg.get("terms", [])
	+	if not terms:
	+	continue
	+
	+	bioguide = ids.get("bioguide", "").strip()
	+	if not bioguide:
	+	continue
	+
	+	last_term = terms[-1]
	+	term_type = last_term.get("type", "")
	+	chamber = "senate" if term_type == "sen" else "house"
	+
	+	full_name = (
	+	name.get("official_full")
	+	or f"{name.get('first', '')} {name.get('last', '')}".strip()
	+	)
	+	party = last_term.get("party", "")
	+	state = last_term.get("state", "")
	+	district = last_term.get("district") # None for senators
	+
	+	lis_id = ids.get("lis")
	+	govtrack = ids.get("govtrack")
	+	icpsr = ids.get("icpsr")
	+
	+	members.append(
	+	{
	+	"bioguide_id": bioguide,
	+	"full_name": full_name,
	+	"party": party,
	+	"state": state,
	+	"district": district,
	+	"chamber": chamber,
	+	"lis_member_id": lis_id,
	+	"govtrack_id": govtrack,
	+	"icpsr_id": icpsr,
	+	"source_url": f"{BIOGUIDE_BASE}/{bioguide}",
	+	}
	+	)
	+	return members
	+
	+
	+	# ─── Poll entry point ─────────────────────────────────────────────────────────
	+
	+	def poll(db_path: Optional[str] = None) -> int:
	+	"""
	+	Fetch the full current member roster and upsert into `members`.
	+
	+	Returns
	+	-------
	+	Number of members upserted (created or updated).
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+
	+	logger.info("members: fetching legislators-current.json …")
	+	try:
	+	resp = requests.get(LEGISLATORS_JSON_URL, timeout=30)
	+	resp.raise_for_status()
	+	except requests.RequestException as exc:
	+	logger.error("members: fetch error: %s", exc)
	+	conn.close()
	+	return 0
	+
	+	try:
	+	data = resp.json()
	+	except ValueError as exc:
	+	logger.error("members: JSON parse error: %s", exc)
	+	conn.close()
	+	return 0
	+
	+	if not isinstance(data, list):
	+	logger.error("members: unexpected JSON shape (expected list, got %s)", type(data))
	+	conn.close()
	+	return 0
	+
	+	members = _parse_legislators(data)
	+	logger.info("members: parsed %d legislator(s) from roster", len(members))
	+
	+	upserted = 0
	+	for m in members:
	+	validate_source_url(m["source_url"]) # gate - must not raise for bioguide.congress.gov
	+	conn.execute(
	+	"""
	+	INSERT INTO members
	+	(bioguide_id, full_name, party, state, district, chamber,
	+	lis_member_id, govtrack_id, icpsr_id, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(bioguide_id) DO UPDATE SET
	+	full_name = excluded.full_name,
	+	party = excluded.party,
	+	state = excluded.state,
	+	district = excluded.district,
	+	chamber = excluded.chamber,
	+	lis_member_id = excluded.lis_member_id,
	+	govtrack_id = excluded.govtrack_id,
	+	icpsr_id = excluded.icpsr_id,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	m["bioguide_id"],
	+	m["full_name"],
	+	m["party"],
	+	m["state"],
	+	m["district"],
	+	m["chamber"],
	+	m["lis_member_id"],
	+	m["govtrack_id"],
	+	m["icpsr_id"],
	+	m["source_url"],
	+	),
	+	)
	+	upserted += 1
	+
	+	conn.commit()
	+	conn.close()
	+	logger.info("members: upserted %d member(s)", upserted)
	+	return upserted
	+
	+
	+	# ─── CLI ──────────────────────────────────────────────────────────────────────
	+
	+	def main() -> None:
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	)
	+	parser = argparse.ArgumentParser(description="Docket member roster poller")
	+	parser.add_argument("--once", action="store_true", help="Run once and exit.")
	+	parser.add_argument("--db", help="Override DB path.")
	+	args = parser.parse_args()
	+	if not args.once:
	+	parser.print_help()
	+	return
	+	count = poll(db_path=args.db)
	+	print(f"members: {count} upserted")
	+
	+
	+	if __name__ == "__main__":
	+	main()

services/ingest/ingest/pollers/openomb.py +184 -164

		@@ -1,16 +1,27 @@
		"""
		OMB Apportionment poller - the impound tracker (flagship "buried data" feature).

	-	Data source: openomb.org JSON feed (no auth), which mirrors the official
	-	OMB apportionment data published at apportionment-public.max.gov.
	+	Data source: apportionment-public.max.gov (official OMB portal - .gov ✓)

	-	NEUTRALITY RULE: source_url stored in every row MUST point to the underlying
	-	official max.gov record, NOT to openomb.org. Aggregators are pipes, not sources.
	-	(DATA_SOURCES.md third-party-normalizer rule.)
	+	max.gov serves individual JSON files per apportionment record (one per TAFS
	+	account + iteration). There is no bulk REST API; this poller:
	+	1. Fetches the HTML index page to extract JSON file URLs.
	+	2. Downloads individual JSON apportionment files (capped per poll).
	+	3. Parses ScheduleData line items to derive appropriated/apportioned amounts.
	+	4. Writes normalized rows to the `apportionments` table.

	-	The apportionment-public.max.gov site is also queryable directly via its own
	-	API; we attempt that first and fall back to openomb.org if max.gov doesn't
	-	respond cleanly. Either way, source_url → max.gov.
	+	source_url for every row → the direct max.gov JSON file URL (never a third-party).
	+
	+	Status derivation (OMB quarterly apportionment logic):
	+	- appropriated = "Total budgetary resources avail" line
	+	- apportioned = sum of Category A/B lines
	+	- released → apportioned ≥ 95% of appropriated (or all quarters filled)
	+	- delayed → apportioned > 0 but < 95% of appropriated (partial release)
	+	- withheld → apportioned == 0 and appropriated > 0 (funds not released)
	+
	+	Honest gap: this reflects the most recent approved apportionment, not any
	+	informal hold or pending review. "Withheld" here means "not yet apportioned
	+	by OMB" which includes genuine impoundments and normal timing gaps.

		Writes to: apportionments table (created if absent - does not touch db.py core)
		"""
		@@ -18,10 +29,11 @@ Writes to: apportionments table (created if absent - does not touch db.py core
		import argparse
		import json
		import logging
	+	import re
		import sys
	-	from datetime import datetime, timedelta
	+	from datetime import datetime
		from typing import Optional
	-	from urllib.parse import urlencode
	+	from urllib.parse import urljoin, unquote

		import requests

		@@ -32,23 +44,23 @@ from ingest.source_validator import validate_source_url
		logger = logging.getLogger(__name__)

		SOURCE = "openomb"
	-
	-	# Official max.gov apportionment portal
		MAX_GOV_BASE = "https://apportionment-public.max.gov"
	-	MAX_GOV_API = "https://apportionment-public.max.gov/api/v1"

	-	# OpenOMB mirror (used as data source only - source_url never points here)
	-	OPENOMB_API = "https://openomb.org/api/apportionments"
	+	# Limit records fetched per poll to avoid hammering the server
	+	_MAX_RECORDS_PER_POLL = 200

	-	# Status normalization
	-	_STATUS_MAP = {
	-	"released": "released",
	-	"approved": "released",
	-	"normal": "released",
	-	"delayed": "delayed",
	-	"withheld": "withheld",
	-	"not apportioned": "withheld",
	-	}
	+	# Line-description patterns for extracting dollar amounts
	+	_APPROPRIATED_PATTERNS = [
	+	re.compile(r"total budgetary resources avail", re.IGNORECASE),
	+	re.compile(r"total budgetary resources available", re.IGNORECASE),
	+	]
	+	_APPORTIONED_PATTERNS = [
	+	re.compile(r"^Category [AB]", re.IGNORECASE),
	+	re.compile(r"^Category [AB] --", re.IGNORECASE),
	+	]
	+	_WITHHELD_PATTERNS = [
	+	re.compile(r"withheld\|impound\|not apportioned\|unapportioned", re.IGNORECASE),
	+	]


		def _ensure_tables(conn) -> None:
		@@ -73,142 +85,125 @@ def _ensure_tables(conn) -> None:
		conn.commit()


	-	def _max_gov_url(agency_code: str = "", account_code: str = "", year: str = "") -> str:
	-	"""
	-	Construct the canonical max.gov URL for an apportionment record.
	-
	-	The public portal search page accepts query params; a full-text search
	-	URL is the most reliable deep-link when account identifiers are available.
	-	"""
	-	base = f"{MAX_GOV_BASE}/apportionments"
	-	if year:
	-	base = f"{base}?fiscal_year={year}"
	-	if agency_code:
	-	base += f"&agency_code={agency_code}"
	-	return base
	-
	-
	-	def _fetch_from_max_gov(fy: int) -> list:
	-	"""
	-	Try fetching apportionment records directly from apportionment-public.max.gov.
	-
	-	Returns list of normalized dicts or [] on failure.
	-	"""
	-	# max.gov exposes a JSON API for apportionment data
	-	url = f"{MAX_GOV_API}/apportionments/"
	-	params = {
	-	"fiscal_year": fy,
	-	"format": "json",
	-	"limit": 500,
	-	}
	+	def _fetch_index_html() -> str:
	+	"""Fetch the max.gov apportionment HTML index page."""
		try:
	-	resp = requests.get(url, params=params, timeout=30)
	+	resp = requests.get(MAX_GOV_BASE + "/", timeout=30)
		resp.raise_for_status()
	-	data = resp.json()
	-	records = data if isinstance(data, list) else data.get("results", data.get("data", []))
	-	if records:
	-	logger.info("openomb: got %d record(s) from max.gov for FY%d", len(records), fy)
	-	return records
	-	except Exception as exc:
	-	logger.debug("openomb: max.gov direct API did not respond cleanly: %s", exc)
	-	return []
	+	return resp.text
	+	except requests.RequestException as exc:
	+	logger.error("openomb: failed to fetch max.gov index: %s", exc)
	+	return ""


	-	def _fetch_from_openomb(fy: int) -> list:
	+	def _extract_json_links(html: str, fiscal_year: int) -> list:
		"""
	-	Fetch apportionment data from openomb.org JSON feed.
	+	Extract JSON apportionment file paths from the max.gov HTML index.

	-	Returns list of raw record dicts or [] on failure.
	+	Returns a list of (agency, relative_path) tuples for the given FY.
		"""
	-	params = {"fiscal_year": fy, "format": "json"}
	-	try:
	-	resp = requests.get(OPENOMB_API, params=params, timeout=30)
	-	resp.raise_for_status()
	-	data = resp.json()
	-	if isinstance(data, list):
	-	logger.info("openomb: got %d record(s) from openomb.org for FY%d", len(data), fy)
	-	return data
	-	results = data.get("results", data.get("apportionments", []))
	-	logger.info("openomb: got %d record(s) from openomb.org for FY%d", len(results), fy)
	-	return results
	-	except Exception as exc:
	-	logger.warning("openomb: openomb.org fetch failed: %s", exc)
	-	return []
	-
	-
	-	def _normalize_status(raw: str) -> str:
	-	"""Normalize a raw status string to released \| delayed \| withheld."""
	-	if not raw:
	+	fy_str = str(fiscal_year)
	+	# Match hrefs pointing to JSON apportionment files for the target FY
	+	pattern = re.compile(
	+	r'href="(/Fiscal%20Year%20' + fy_str + r'/[^"]+\.json)"',
	+	re.IGNORECASE,
	+	)
	+	matches = pattern.findall(html)
	+	results = []
	+	for path in matches:
	+	# Decode URL-encoded path to extract agency name
	+	decoded = unquote(path)
	+	# Path structure: /Fiscal Year 2026/Agency Name/JSON/filename.json
	+	parts = decoded.strip("/").split("/")
	+	agency = parts[1] if len(parts) >= 2 else "Unknown"
	+	results.append((agency, path))
	+	logger.info("openomb: found %d JSON links for FY%d", len(results), fiscal_year)
	+	return results
	+
	+
	+	def _derive_status(appropriated: float, apportioned: float) -> str:
	+	"""Classify an apportionment record as released / delayed / withheld."""
	+	if appropriated <= 0:
	+	return "released" # nothing to apportion
	+	ratio = apportioned / appropriated
	+	if ratio >= 0.95:
		return "released"
	-	lower = raw.strip().lower()
	-	for key, val in _STATUS_MAP.items():
	-	if key in lower:
	-	return val
	-	return "released"
	+	if apportioned > 0:
	+	return "delayed"
	+	return "withheld"


	-	def _extract_fields(rec: dict) -> Optional[dict]:
	+	def _parse_apportionment_json(data: dict) -> Optional[dict]:
		"""
	-	Extract normalized fields from a raw apportionment record dict.
	+	Extract key financial fields from a max.gov apportionment JSON record.

	-	Handles various field-name conventions from max.gov and openomb.org.
	-	Returns None if the record is unusable.
	+	Returns a normalized dict or None if the record is unusable.
		"""
	-	# Field name aliases - max.gov and openomb may differ
	-	agency = (rec.get("agency_name") or rec.get("agency") or
	-	rec.get("Agency") or "")
	-	account = (rec.get("account_name") or rec.get("account") or
	-	rec.get("Account") or rec.get("account_title") or "")
	-	appropriated = (rec.get("total_appropriations") or rec.get("appropriated") or
	-	rec.get("amount_appropriated") or rec.get("Appropriated") or 0)
	-	apportioned = (rec.get("total_apportioned") or rec.get("apportioned") or
	-	rec.get("amount_apportioned") or rec.get("Apportioned") or 0)
	-	status_raw = (rec.get("status") or rec.get("apportionment_status") or
	-	rec.get("Status") or "released")
	-	date = (rec.get("date") or rec.get("period_of_availability") or
	-	rec.get("apportionment_date") or rec.get("fiscal_year") or "")
	-	# For external_id, use a combination of agency+account+date
	-	agency_code = (rec.get("agency_code") or rec.get("agency_identifier") or "")
	-	account_code = (rec.get("account_code") or rec.get("account_number") or "")
	-	year_str = str(rec.get("fiscal_year") or "")
	+	schedule = data.get("ScheduleData") or []
	+	if not schedule:
	+	return None
	+
	+	first = schedule[0]
	+	agency = data.get("Folder") or first.get("BudgetAgencyTitle") or ""
	+	account = first.get("AccountTitle") or ""

		if not agency and not account:
		return None

	-	ext_id = f"{year_str}_{agency_code or agency[:20]}_{account_code or account[:20]}".replace(" ", "_")
	+	appropriated = 0.0
	+	apportioned = 0.0

	-	try:
	-	appropriated_f = float(appropriated or 0)
	-	apportioned_f = float(apportioned or 0)
	-	except (ValueError, TypeError):
	-	appropriated_f = 0.0
	-	apportioned_f = 0.0
	+	for line in schedule:
	+	desc = (line.get("LineDescription") or "").strip()
	+	amount = float(line.get("ApprovedAmount") or 0)

	-	source_url = _max_gov_url(agency_code, account_code, year_str)
	+	if any(p.search(desc) for p in _APPROPRIATED_PATTERNS):
	+	appropriated = amount
	+	if any(p.match(desc) for p in _APPORTIONED_PATTERNS):
	+	apportioned += max(0.0, amount)
	+
	+	status = _derive_status(appropriated, apportioned)
	+
	+	# Date from approval timestamp: "2025-09-16-17.07.55.231605" → "2025-09-16"
	+	ts = data.get("ApprovalTimestamp") or ""
	+	date = ts[:10].replace("-", "-") if len(ts) >= 10 else datetime.utcnow().strftime("%Y-%m-%d")

		return {
	-	"external_id": ext_id,
	-	"agency": str(agency),
	-	"account": str(account),
	-	"appropriated": appropriated_f,
	-	"apportioned": apportioned_f,
	-	"status": _normalize_status(str(status_raw)),
	-	"date": str(date)[:10] if date else "",
	-	"source_url": source_url,
	-	"update_date": str(date)[:10] if date else datetime.utcnow().strftime("%Y-%m-%d"),
	+	"agency": agency,
	+	"account": account,
	+	"appropriated": appropriated,
	+	"apportioned": apportioned,
	+	"status": status,
	+	"date": date,
	+	"fiscal_year": data.get("FiscalYear", ""),
	+	"filename": data.get("FileName", ""),
		}


	-	def poll(fy: Optional[int] = None, db_path: Optional[str] = None) -> int:
	+	def _fetch_json_record(relative_path: str) -> Optional[dict]:
	+	"""Download and parse one apportionment JSON file from max.gov."""
	+	url = MAX_GOV_BASE + relative_path
	+	try:
	+	resp = requests.get(url, timeout=30)
	+	resp.raise_for_status()
	+	return resp.json()
	+	except Exception as exc:
	+	logger.debug("openomb: could not fetch %s: %s", relative_path, exc)
	+	return None
	+
	+
	+	def poll(fy: Optional[int] = None, db_path: Optional[str] = None,
	+	max_records: int = _MAX_RECORDS_PER_POLL) -> int:
		"""
	-	Pull OMB apportionment records and write to the apportionments table.
	+	Pull OMB apportionment records from apportionment-public.max.gov.

	-	Always links source_url to apportionment-public.max.gov, never to openomb.org.
	+	Every stored row has source_url → a specific max.gov JSON file (all .gov ✓).

		Parameters
		----------
	-	fy : Fiscal year to fetch (default: current FY).
	-	db_path : Override default SQLite path.
	+	fy : Fiscal year (default: current FY).
	+	db_path : Override default SQLite path.
	+	max_records : Cap on records fetched per run (default 200).

		Returns
		-------
		@@ -216,51 +211,69 @@ def poll(fy: Optional[int] = None, db_path: Optional[str] = None) -> int:
		"""
		if fy is None:
		now = datetime.utcnow()
	-	# US fiscal year starts Oct 1
		fy = now.year if now.month >= 10 else now.year - 1

	-	logger.info("openomb: fetching apportionments for FY%d", fy)
	+	logger.info("openomb: fetching apportionments for FY%d (max %d records)", fy, max_records)

		conn = get_connection(db_path) if db_path else get_connection()
		init_schema(conn)
		_ensure_tables(conn)

	-	# Try max.gov first; fall back to openomb.org
	-	records = _fetch_from_max_gov(fy)
	-	if not records:
	-	records = _fetch_from_openomb(fy)
	-
	-	if not records:
	-	logger.warning("openomb: no records retrieved for FY%d from any source", fy)
	-	# Attempt current FY - 1 as a fallback if nothing for current FY
	-	if fy == datetime.utcnow().year:
	-	records = _fetch_from_openomb(fy - 1)
	-	if records:
	-	fy = fy - 1
	-	logger.info("openomb: fell back to FY%d - %d record(s)", fy, len(records))
	-
	-	if not records:
	+	# Step 1: get the file index
	+	html = _fetch_index_html()
	+	if not html:
		conn.close()
		return 0

	+	links = _extract_json_links(html, fy)
	+	if not links:
	+	logger.warning("openomb: no JSON links found for FY%d", fy)
	+	# Try current FY and previous FY
	+	if fy > 2022:
	+	links = _extract_json_links(html, fy - 1)
	+
	+	if not links:
	+	logger.warning("openomb: no apportionment links found at all")
	+	conn.close()
	+	return 0
	+
	+	# Step 2: download and store records (cap at max_records)
		stored = 0
	-	for rec in records:
	-	fields = _extract_fields(rec)
	-	if fields is None:
	+	processed = 0
	+
	+	for agency, rel_path in links:
	+	if processed >= max_records:
	+	logger.info("openomb: reached max_records cap (%d), stopping", max_records)
	+	break
	+
	+	source_url = MAX_GOV_BASE + rel_path
	+	# external_id = URL path (URL-decoded filename portion)
	+	filename = unquote(rel_path.split("/")[-1])
	+	external_id = filename
	+
	+	# update_date: use the date embedded in the filename (e.g., 2026-06-17)
	+	date_m = re.search(r"(\d{4}-\d{2}-\d{2})", filename)
	+	update_date = date_m.group(1) if date_m else datetime.utcnow().strftime("%Y-%m-%d")
	+
	+	if not is_new_event(conn, SOURCE, external_id, update_date):
	+	processed += 1
		continue

		try:
	-	validate_source_url(fields["source_url"])
	+	validate_source_url(source_url)
		except Exception as exc:
	-	logger.error(
	-	"openomb: source_url %r not allowlisted (%s) - skipping row",
	-	fields["source_url"], exc,
	-	)
	+	logger.error("openomb: source_url %r not allowlisted: %s", source_url, exc)
	+	processed += 1
		continue

	-	update_date = fields["update_date"]
	-	if not is_new_event(conn, SOURCE, fields["external_id"], update_date):
	-	logger.debug("openomb: already seen %s", fields["external_id"])
	+	raw = _fetch_json_record(rel_path)
	+	if raw is None:
	+	processed += 1
	+	continue
	+
	+	fields = _parse_apportionment_json(raw)
	+	if fields is None:
	+	processed += 1
		continue

		conn.execute(
		@@ -280,14 +293,14 @@ def poll(fy: Optional[int] = None, db_path: Optional[str] = None) -> int:
		""",
		(
		SOURCE,
	-	fields["external_id"],
	+	external_id,
		fields["agency"],
		fields["account"],
		fields["appropriated"],
		fields["apportioned"],
		fields["status"],
		fields["date"],
	-	fields["source_url"],
	+	source_url,
		),
		)

		@@ -299,11 +312,16 @@ def poll(fy: Optional[int] = None, db_path: Optional[str] = None) -> int:
		payload = excluded.payload,
		fetched_at = excluded.fetched_at
		""",
	-	(SOURCE, fields["external_id"], json.dumps(rec), datetime.utcnow().isoformat()),
	+	(SOURCE, external_id, json.dumps(raw), datetime.utcnow().isoformat()),
		)

	-	mark_seen(conn, SOURCE, fields["external_id"], update_date)
	+	mark_seen(conn, SOURCE, external_id, update_date)
		stored += 1
	+	processed += 1
	+
	+	if stored % 10 == 0:
	+	conn.commit()
	+	logger.info("openomb: %d new records stored so far", stored)

		conn.commit()
		conn.close()
		@@ -320,8 +338,10 @@ if __name__ == "__main__":
		parser = argparse.ArgumentParser(description="OMB Apportionment poller")
		parser.add_argument("--once", action="store_true", help="Run once and exit")
		parser.add_argument("--fy", type=int, default=None, help="Fiscal year (default: current)")
	+	parser.add_argument("--max", type=int, default=50,
	+	help="Max records to fetch (default 50 for --once; 200 in scheduled mode)")
		args = parser.parse_args()

	-	count = poll(fy=args.fy)
	+	count = poll(fy=args.fy, max_records=args.max)
		print(f"Stored {count} new apportionments row(s).")
		sys.exit(0)

services/ingest/ingest/pollers/votes_enrich.py +257 -0

		@@ -0,0 +1,257 @@
	+	"""
	+	Votes enrichment - per-member positions for Senate roll-call votes.
	+
	+	Senate completeness layer. The votes_senate.py poller captures aggregate
	+	counts (yea/nay/present) but NOT per-member positions. This module:
	+
	+	1. Scans the `votes` table for Senate rows that have NO entries in
	+	vote_positions.
	+	2. Re-fetches each vote's source_url (already stored on the row - the
	+	original senate.gov XML URL).
	+	3. Parses the per-member <member><lis_member_id>/<vote_cast> fields.
	+	4. Maps lis_member_id → bioguide_id via the `members` table (populated
	+	by pollers/members.py). Where no mapping exists the row still lands
	+	with member_bioguide=None and member_id=None so the record is
	+	preserved without fabricated data.
	+
	+	Run:
	+	python -m ingest.pollers.votes_enrich --once
	+
	+	Idempotent: re-running never double-inserts because it filters for votes
	+	with zero existing positions first.
	+	"""
	+
	+	import argparse
	+	import logging
	+	import re
	+	import xml.etree.ElementTree as ET
	+	from typing import Optional
	+
	+	import requests
	+
	+	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 = "votes_enrich"
	+
	+	# Canonical position map for Senate <vote_cast> values
	+	_SENATE_POS_MAP: dict[str, str] = {
	+	"yea": "Yea",
	+	"aye": "Yea",
	+	"nay": "Nay",
	+	"no": "Nay",
	+	"present": "Present",
	+	"abstain": "Present", # Rare; treated as Present in the 4-value enum
	+	"not voting": "Not Voting",
	+	"not present":"Not Voting",
	+	}
	+
	+
	+	# ─── Senate XML parser ────────────────────────────────────────────────────────
	+
	+	def parse_senate_vote_positions(xml_text: str) -> list[dict]:
	+	"""
	+	Parse a Senate roll-call XML and return a list of per-member position dicts:
	+	[{lis_member_id, member_full, party, state, position}, …]
	+
	+	Returns [] on parse failure.
	+	"""
	+	try:
	+	root = ET.fromstring(xml_text)
	+	except ET.ParseError as exc:
	+	logger.error("votes_enrich: XML parse error: %s", exc)
	+	return []
	+
	+	members_el = root.find("members")
	+	if members_el is None:
	+	logger.debug("votes_enrich: no <members> element - voice/UC vote?")
	+	return []
	+
	+	positions = []
	+	for member_el in members_el.findall("member"):
	+	lis_id = (member_el.findtext("lis_member_id") or "").strip()
	+	full_name = (member_el.findtext("member_full") or "").strip()
	+	party = (member_el.findtext("party") or "").strip()
	+	state = (member_el.findtext("state") or "").strip()
	+	vote_cast = (member_el.findtext("vote_cast") or "").strip()
	+
	+	position = _SENATE_POS_MAP.get(vote_cast.lower())
	+	if position is None:
	+	logger.debug(
	+	"votes_enrich: unknown vote_cast %r for %s - skipping",
	+	vote_cast, lis_id,
	+	)
	+	continue
	+
	+	positions.append(
	+	{
	+	"lis_member_id": lis_id,
	+	"member_full": full_name,
	+	"party": party,
	+	"state": state,
	+	"position": position,
	+	}
	+	)
	+
	+	return positions
	+
	+
	+	# ─── Enrichment logic ─────────────────────────────────────────────────────────
	+
	+	def _build_lis_to_bioguide(conn) -> dict[str, str]:
	+	"""Build a lis_member_id → bioguide_id map from the members table."""
	+	mapping: dict[str, str] = {}
	+	for row in conn.execute(
	+	"SELECT bioguide_id, lis_member_id FROM members WHERE lis_member_id IS NOT NULL"
	+	):
	+	mapping[row["lis_member_id"]] = row["bioguide_id"]
	+	return mapping
	+
	+
	+	def _build_bioguide_to_id(conn) -> dict[str, int]:
	+	"""Build a bioguide_id → members.id map."""
	+	mapping: dict[str, int] = {}
	+	for row in conn.execute("SELECT id, bioguide_id FROM members"):
	+	mapping[row["bioguide_id"]] = row["id"]
	+	return mapping
	+
	+
	+	def enrich_senate_positions(
	+	conn,
	+	lis_to_bg: dict[str, str],
	+	bg_to_id: dict[str, int],
	+	max_votes: int = 200,
	+	) -> int:
	+	"""
	+	Backfill vote_positions for Senate votes that have no positions yet.
	+
	+	Returns the number of vote rows enriched (positions inserted).
	+	"""
	+	# Find Senate votes without any position rows
	+	pending = conn.execute(
	+	"""
	+	SELECT v.id, v.source_url
	+	FROM votes v
	+	WHERE v.chamber = 'senate'
	+	AND v.source_url IS NOT NULL
	+	AND v.source_url LIKE '%senate.gov%'
	+	AND NOT EXISTS (
	+	SELECT 1 FROM vote_positions vp WHERE vp.vote_id = v.id
	+	)
	+	ORDER BY v.id
	+	LIMIT ?
	+	""",
	+	(max_votes,),
	+	).fetchall()
	+
	+	if not pending:
	+	logger.info("votes_enrich: no Senate votes need position enrichment")
	+	return 0
	+
	+	logger.info("votes_enrich: %d Senate vote(s) pending enrichment", len(pending))
	+	enriched = 0
	+
	+	for row in pending:
	+	vote_id = row["id"]
	+	source_url = row["source_url"]
	+
	+	# The source_url IS the senate.gov XML - already allowlisted.
	+	try:
	+	validate_source_url(source_url)
	+	except Exception as exc:
	+	logger.warning("votes_enrich: invalid source_url %s: %s - skip", source_url, exc)
	+	continue
	+
	+	try:
	+	resp = requests.get(source_url, timeout=20)
	+	except requests.RequestException as exc:
	+	logger.error("votes_enrich: fetch error %s: %s", source_url, exc)
	+	continue
	+
	+	if not resp.ok:
	+	logger.warning("votes_enrich: HTTP %d for %s", resp.status_code, source_url)
	+	continue
	+
	+	positions = parse_senate_vote_positions(resp.text)
	+	if not positions:
	+	logger.debug("votes_enrich: no positions parsed for vote_id=%d", vote_id)
	+	continue
	+
	+	pos_rows = []
	+	for p in positions:
	+	lis_id = p["lis_member_id"]
	+	bioguide = lis_to_bg.get(lis_id)
	+	member_id = bg_to_id.get(bioguide) if bioguide else None
	+	pos_rows.append((vote_id, bioguide, member_id, p["position"]))
	+
	+	conn.executemany(
	+	"""
	+	INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position)
	+	VALUES (?, ?, ?, ?)
	+	""",
	+	pos_rows,
	+	)
	+
	+	logger.info(
	+	"votes_enrich: vote_id=%d - %d position(s) inserted (url=%s)",
	+	vote_id, len(pos_rows), source_url,
	+	)
	+	enriched += 1
	+
	+	conn.commit()
	+	return enriched
	+
	+
	+	# ─── Poll entry point ─────────────────────────────────────────────────────────
	+
	+	def poll(db_path: Optional[str] = None, max_votes: int = 200) -> int:
	+	"""
	+	Enrich Senate vote_positions for all un-enriched Senate votes.
	+
	+	Returns
	+	-------
	+	Number of Senate votes that received position data.
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+
	+	lis_to_bg = _build_lis_to_bioguide(conn)
	+	bg_to_id = _build_bioguide_to_id(conn)
	+
	+	logger.info(
	+	"votes_enrich: member lookup tables ready (%d LIS→bioguide, %d bioguide→id)",
	+	len(lis_to_bg), len(bg_to_id),
	+	)
	+
	+	enriched = enrich_senate_positions(conn, lis_to_bg, bg_to_id, max_votes=max_votes)
	+
	+	conn.close()
	+	logger.info("votes_enrich: enriched %d Senate vote(s)", enriched)
	+	return enriched
	+
	+
	+	# ─── CLI ──────────────────────────────────────────────────────────────────────
	+
	+	def main() -> None:
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	)
	+	parser = argparse.ArgumentParser(description="Docket votes enrichment (Senate positions)")
	+	parser.add_argument("--once", action="store_true", help="Run once and exit.")
	+	parser.add_argument("--db", help="Override DB path.")
	+	parser.add_argument("--max-votes", type=int, default=200, help="Max votes to enrich per run.")
	+	args = parser.parse_args()
	+	if not args.once:
	+	parser.print_help()
	+	return
	+	count = poll(db_path=args.db, max_votes=args.max_votes)
	+	print(f"votes_enrich: {count} Senate vote(s) enriched")
	+
	+
	+	if __name__ == "__main__":
	+	main()

services/ingest/ingest/pollers/votes_house.py +286 -54

		@@ -1,14 +1,29 @@
		"""
	-	House Clerk XML roll-call vote poller.
	+	House Clerk XML roll-call vote poller - full per-member positions.

	-	Source: https://clerk.house.gov/evs/{year}/ROLL_{vote_num:03d}.xml
	+	Source: https://clerk.house.gov/evs/{year}/ROLL_{NNN}.xml
		No auth required. Posted within hours of the vote.

	-	Strategy: discover the latest vote number by checking sequentially from the
	-	last known; stop when a 404 is returned. Stores votes and emits feed events.
	+	Strategy:
	+	• Probe roll numbers sequentially from last known; stop on 404.
	+	• For each recorded vote, capture full metadata AND per-member positions
	+	into vote_positions (member_bioguide, position).
	+	• Aggregate totals are cross-checked against per-member count.
	+	• Honesty: we only process YEA-AND-NAY / RECORDED votes. Voice-vote or
	+	unanimous-consent proceedings do not appear in this XML stream.
	+
	+	Schema extensions (CREATE TABLE IF NOT EXISTS / ALTER TABLE IF NEEDED):
	+	Handled by ingest.pollers.members.ensure_votes_schema().
	+	Called at the start of poll().
	+
	+	Bill linkage:
	+	legis-num field (e.g. "H R 1234") is parsed into type + number.
	+	We attempt a bills-table lookup by congress + type + number; on miss,
	+	bill_id is NULL and bill_text_ref stores the raw legis-num for later join.
		"""

		import logging
	+	import re
		import sqlite3
		import xml.etree.ElementTree as ET
		from datetime import datetime
		@@ -18,6 +33,7 @@ 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__)
		@@ -25,61 +41,215 @@ logger = logging.getLogger(__name__)
		SOURCE = "votes_house"
		BASE_URL = "https://clerk.house.gov/evs"

	+	# Valid positions in House roll-call XML
	+	_VALID_POSITIONS = frozenset({"Yea", "Nay", "Present", "Not Voting"})
	+
	+	# Map any variant spelling to our canonical enum
	+	_POSITION_MAP: dict[str, str] = {
	+	"yea": "Yea",
	+	"aye": "Yea",
	+	"nay": "Nay",
	+	"no": "Nay",
	+	"present": "Present",
	+	"not voting": "Not Voting",
	+	"not-voting": "Not Voting",
	+	}
	+
	+	# legis-num → bill type mapping (longest prefix first)
	+	_LEGIS_TYPE_MAP = [
	+	("H J RES", "hjres"),
	+	("H CON RES","hconres"),
	+	("H RES", "hres"),
	+	("H R", "hr"),
	+	("S J RES", "sjres"),
	+	("S CON RES","sconres"),
	+	("S RES", "sres"),
	+	("S", "s"),
	+	]
	+
	+
	+	# ─── Helpers ──────────────────────────────────────────────────────────────────

		def _roll_url(year: int, roll_num: int) -> str:
	-	return f"{BASE_URL}/{year}/ROLL_{roll_num:03d}.xml"
	+	return f"{BASE_URL}/{year}/roll{roll_num:03d}.xml"


	-	def _parse_vote_xml(xml_text: str, source_url: str) -> Optional[dict]:
	-	"""Parse House roll-call XML; return a dict or None on parse error."""
	+	def _normalize_position(raw: str) -> Optional[str]:
	+	"""Map a raw <vote> string to a canonical position, or None if unknown."""
	+	key = raw.strip().lower()
	+	return _POSITION_MAP.get(key)
	+
	+
	+	def _parse_legis_num(legis_num: str, congress: int) -> tuple[Optional[str], Optional[str], Optional[str]]:
	+	"""
	+	Parse 'H R 1234' into (bill_type, number, external_id).
	+
	+	external_id matches the format used by pollers/congress.py:
	+	'{congress}-{bill_type}-{number}' e.g. '119-hr-1234'
	+
	+	Returns (None, None, None) if not recognizable.
	+	"""
	+	s = legis_num.strip().upper()
	+	for prefix, bill_type in _LEGIS_TYPE_MAP:
	+	if s.startswith(prefix):
	+	remainder = s[len(prefix):].strip()
	+	# Remainder must be a bare number
	+	if re.fullmatch(r"\d+", remainder):
	+	number = remainder
	+	ext_id = f"{congress}-{bill_type}-{number}"
	+	return bill_type, number, ext_id
	+	return None, None, None
	+
	+
	+	def _lookup_bill_id(conn: sqlite3.Connection, external_id: str) -> Optional[int]:
	+	"""Return the bills.id for a congress-poller external_id, or None."""
	+	row = conn.execute(
	+	"SELECT id FROM bills WHERE source='congress' AND external_id=?",
	+	(external_id,),
	+	).fetchone()
	+	return row["id"] if row else None
	+
	+
	+	# ─── XML parser ───────────────────────────────────────────────────────────────
	+
	+	def parse_house_vote_xml(xml_text: str, source_url: str) -> Optional[dict]:
	+	"""
	+	Parse House Clerk roll-call XML.
	+
	+	Returns a dict with keys:
	+	action_date, congress, session, roll_number, question, bill_text_ref,
	+	legis_external_id, result, yea, nay, present, not_voting,
	+	source_url, positions: list[{bioguide, position}]
	+
	+	Returns None on parse failure or if the vote type is not a recorded vote.
	+	"""
		try:
		root = ET.fromstring(xml_text)
		except ET.ParseError as exc:
		logger.error("votes_house: XML parse error: %s", exc)
		return None

	-	vote_metadata = root.find("vote-metadata")
	-	if vote_metadata is None:
	+	meta = root.find("vote-metadata")
	+	if meta is None:
	+	logger.warning("votes_house: missing <vote-metadata>")
		return None

		def text(tag: str) -> str:
	-	el = vote_metadata.find(tag)
	-	return el.text.strip() if el is not None and el.text else ""
	+	el = meta.find(tag)
	+	return (el.text or "").strip() if el is not None else ""

	-	action_date = text("action-date") # MM/DD/YYYY
	-	# Normalise to ISO
	-	try:
	-	action_date_iso = datetime.strptime(action_date, "%m/%d/%Y").strftime("%Y-%m-%d")
	-	except ValueError:
	-	action_date_iso = action_date
	+	# Skip voice / procedural records that sneak into the stream
	+	vote_type = text("vote-type").upper()
	+	if vote_type not in {"YEA-AND-NAY", "RECORDED VOTE", "TWO-THIRDS YEA-AND-NAY"}:
	+	logger.debug("votes_house: skipping non-recorded vote type %r", vote_type)
	+	return None

	-	yea_count = nay_count = present_count = 0
	-	totals = vote_metadata.find("vote-totals")
	-	if totals is not None:
	-	for totals_by_party in totals.findall("totals-by-vote"):
	-	yea_count = int(totals_by_party.findtext("yea-total", "0") or 0)
	-	nay_count = int(totals_by_party.findtext("nay-total", "0") or 0)
	-	present_count = int(totals_by_party.findtext("present-total", "0") or 0)
	+	action_date_raw = text("action-date")
	+	# House Clerk XML uses "D-Mon-YYYY" (e.g. "8-Jan-2026") or occasionally
	+	# "MM/DD/YYYY". Try both; fall back to raw string so the row is preserved.
	+	action_date = action_date_raw
	+	for fmt in ("%d-%b-%Y", "%m/%d/%Y", "%B %d, %Y", "%Y-%m-%d"):
	+	try:
	+	action_date = datetime.strptime(action_date_raw, fmt).strftime("%Y-%m-%d")
	+	break
	+	except ValueError:
	+	pass

	-	result_el = vote_metadata.find("vote-result")
	-	result = result_el.text.strip() if result_el is not None and result_el.text else ""
	+	try:
	+	congress = int(text("congress"))
	+	except ValueError:
	+	congress = 0
	+	try:
	+	session = int(text("session"))
	+	except ValueError:
	+	session = 0
	+	try:
	+	roll_number = int(text("rollcall-num"))
	+	except ValueError:
	+	roll_number = 0
	+
	+	question = text("vote-question")
	+	result = text("vote-result")
	+	legis_num_raw = text("legis-num")
	+
	+	# Bill linkage
	+	_, _, legis_external_id = _parse_legis_num(legis_num_raw, congress)
	+
	+	# Aggregate totals (totals-by-vote is the grand total, not per-party)
	+	yea = nay = present = not_voting = 0
	+	totals_el = meta.find("vote-totals")
	+	if totals_el is not None:
	+	tbv = totals_el.find("totals-by-vote")
	+	if tbv is not None:
	+	yea = int(tbv.findtext("yea-total", "0") or 0)
	+	nay = int(tbv.findtext("nay-total", "0") or 0)
	+	present = int(tbv.findtext("present-total", "0") or 0)
	+	not_voting = int(tbv.findtext("not-voting-total", "0") or 0)
	+
	+	# Per-member positions
	+	positions: list[dict] = []
	+	vote_data = root.find("vote-data")
	+	if vote_data is not None:
	+	for rv in vote_data.findall("recorded-vote"):
	+	leg_el = rv.find("legislator")
	+	vote_el = rv.find("vote")
	+	if leg_el is None or vote_el is None:
	+	continue
	+
	+	bioguide = (leg_el.get("name-id") or "").strip()
	+	raw_pos = (vote_el.text or "").strip()
	+	position = _normalize_position(raw_pos)
	+	if not bioguide or position is None:
	+	logger.debug("votes_house: skipping legislator %r pos %r", bioguide, raw_pos)
	+	continue
	+
	+	positions.append({"bioguide": bioguide, "position": position})
	+
	+	# Sanity check: aggregate totals should equal sum of positions
	+	if positions:
	+	counted = {p: 0 for p in _VALID_POSITIONS}
	+	for p in positions:
	+	counted[p["position"]] += 1
	+	total_from_positions = sum(counted.values())
	+	total_from_meta = yea + nay + present + not_voting
	+	if total_from_meta and abs(total_from_positions - total_from_meta) > 5:
	+	logger.warning(
	+	"votes_house: position count mismatch (parsed %d vs meta total %d) for %s",
	+	total_from_positions, total_from_meta, source_url,
	+	)

		return {
	-	"action_date": action_date_iso,
	-	"yea": yea_count,
	-	"nay": nay_count,
	-	"present": present_count,
	-	"result": result,
	-	"source_url": source_url,
	+	"action_date": action_date,
	+	"congress": congress,
	+	"session": session,
	+	"roll_number": roll_number,
	+	"question": question,
	+	"bill_text_ref": legis_num_raw,
	+	"legis_external_id": legis_external_id,
	+	"result": result,
	+	"yea": yea,
	+	"nay": nay,
	+	"present": present,
	+	"not_voting": not_voting,
	+	"source_url": source_url,
	+	"positions": positions,
		}


	-	def poll(year: Optional[int] = None, max_new: int = 20, db_path: Optional[str] = None) -> int:
	+	# ─── Poll entry point ─────────────────────────────────────────────────────────
	+
	+	def poll(
	+	year: Optional[int] = None,
	+	max_new: int = 50,
	+	db_path: Optional[str] = None,
	+	) -> int:
		"""
		Fetch new House roll-call votes for the given year (default: current year).

	-	Probes vote numbers sequentially from the last seen + 1 until a 404.
	-	Limits to max_new new votes per call to avoid runaway HTTP traffic.
	+	For each new vote, inserts:
	+	• A row in `votes` with extended metadata (congress, session, roll_number, …)
	+	• Per-member rows in `vote_positions`
	+	• An events feed row

		Returns
		-------
		@@ -90,19 +260,24 @@ def poll(year: Optional[int] = None, max_new: int = 20, db_path: Optional[str] =

		conn = get_connection(db_path) if db_path else get_connection()
		init_schema(conn)
	+	ensure_votes_schema(conn)

	-	# Find the highest vote number we've already seen for this year.
	+	# Build a quick bioguide → member_id lookup from the members table.
	+	# If the members table is empty (members poller not run yet), positions
	+	# will still be stored with member_bioguide; member_id will be NULL.
	+	bg_to_id: dict[str, int] = {}
	+	for row in conn.execute("SELECT id, bioguide_id FROM members"):
	+	bg_to_id[row["bioguide_id"]] = row["id"]
	+
	+	# Find the highest roll number already seen for this year.
		last_row = conn.execute(
	-	"SELECT external_id FROM seen WHERE source=? AND external_id LIKE ? ORDER BY external_id DESC LIMIT 1",
	+	"SELECT external_id FROM seen WHERE source=? AND external_id LIKE ? "
	+	"ORDER BY external_id DESC LIMIT 1",
		(SOURCE, f"{year}-%"),
		).fetchone()
	+	last_num = int(last_row["external_id"].split("-")[1]) if last_row else 0

	-	if last_row:
	-	last_num = int(last_row["external_id"].split("-")[1])
	-	else:
	-	last_num = 0
	-
	-	stored = 0
	+	stored = 0
		probe_num = last_num + 1

		while stored < max_new:
		@@ -113,13 +288,13 @@ def poll(year: Optional[int] = None, max_new: int = 20, db_path: Optional[str] =

		url = _roll_url(year, probe_num)
		try:
	-	resp = requests.get(url, timeout=15)
	+	resp = requests.get(url, timeout=20)
		except requests.RequestException as exc:
		logger.error("votes_house: request error %s: %s", url, exc)
		break

		if resp.status_code == 404:
	-	logger.info("votes_house: no vote %s (probe stopped)", external_id)
	+	logger.info("votes_house: no vote %s (probe stopped at 404)", external_id)
		break
		if not resp.ok:
		logger.warning("votes_house: HTTP %d for %s", resp.status_code, url)
		@@ -131,22 +306,76 @@ def poll(year: Optional[int] = None, max_new: int = 20, db_path: Optional[str] =
		logger.error("votes_house: invalid source_url %s: %s", url, exc)
		break

	-	vote = _parse_vote_xml(resp.text, url)
	+	vote = parse_house_vote_xml(resp.text, url)
		if vote is None:
	+	# parse_house_vote_xml logs the reason; skip and probe next
	+	mark_seen(conn, SOURCE, external_id, external_id)
		probe_num += 1
		continue

	+	# ── Attempt bill linkage ──────────────────────────────────────────
	+	bill_id: Optional[int] = None
	+	if vote["legis_external_id"]:
	+	bill_id = _lookup_bill_id(conn, vote["legis_external_id"])
	+
	+	# ── Insert vote row ───────────────────────────────────────────────
		conn.execute(
		"""
	-	INSERT INTO votes (chamber, date, yea, nay, present, result, source_url)
	-	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	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)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
		""",
	-	("house", vote["action_date"], vote["yea"], vote["nay"],
	-	vote["present"], vote["result"], vote["source_url"]),
	+	(
	+	bill_id,
	+	"house",
	+	vote["action_date"],
	+	vote["yea"],
	+	vote["nay"],
	+	vote["present"],
	+	vote["result"],
	+	vote["source_url"],
	+	vote["congress"],
	+	vote["session"],
	+	vote["roll_number"],
	+	vote["question"],
	+	vote["not_voting"],
	+	1, # recorded = True
	+	external_id,
	+	vote["bill_text_ref"],
	+	),
		)

		vote_row = conn.execute("SELECT last_insert_rowid() AS id").fetchone()
	-	if vote_row:
	+	vote_db_id = vote_row["id"] if vote_row else None
	+
	+	# ── Insert per-member positions ───────────────────────────────────
	+	if vote_db_id and vote["positions"]:
	+	pos_rows = [
	+	(
	+	vote_db_id,
	+	p["bioguide"],
	+	bg_to_id.get(p["bioguide"]), # member_id or None
	+	p["position"],
	+	)
	+	for p in vote["positions"]
	+	]
	+	conn.executemany(
	+	"""
	+	INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position)
	+	VALUES (?, ?, ?, ?)
	+	""",
	+	pos_rows,
	+	)
	+	logger.debug(
	+	"votes_house: %s - %d positions stored (Y:%d N:%d P:%d NV:%d)",
	+	external_id, len(pos_rows),
	+	vote["yea"], vote["nay"], vote["present"], vote["not_voting"],
	+	)
	+
	+	# ── Feed event ────────────────────────────────────────────────────
	+	if vote_db_id:
		conn.execute(
		"""
		INSERT INTO events
		@@ -154,14 +383,17 @@ def poll(year: Optional[int] = None, max_new: int = 20, db_path: Optional[str] =
		affected_estimate, action_type)
		VALUES (?, ?, ?, ?, ?, ?)
		""",
	-	("vote", vote_row["id"], vote["action_date"], None, None, "floor_vote"),
	+	("vote", vote_db_id, vote["action_date"], None, None, "floor_vote"),
		)

		mark_seen(conn, SOURCE, external_id, external_id)
	-	stored += 1
	+	stored += 1
		probe_num += 1

		conn.commit()
		conn.close()
	-	logger.info("votes_house: stored %d new vote(s) (year %d)", stored, year)
	+	logger.info(
	+	"votes_house: stored %d new vote(s) for year %d (last roll probed: %d)",
	+	stored, year, probe_num - 1,
	+	)
		return stored

services/ingest/ingest/run.py +84 -80

		@@ -1,24 +1,23 @@
		"""
	-	Docket ingest CLI.
	+	Docket ingest CLI - full-spectrum orchestrator.

		Usage
		-----
	-	python -m ingest.run --once # run all pollers once and exit
	-	python -m ingest.run --source usa # run only the USAspending poller
	-
	-	Pollers
	-	-------
	-	usaspending - real federal awards, no auth
	-	congress - bills (requires DOCKET_CONGRESS_KEY; skips if unset)
	-	govinfo - bill text / public laws (same key as congress)
	-	treasury - debt + cash flow, no auth
	-	votes_house - House roll-call XML, no auth
	-	votes_senate - Senate roll-call XML, no auth
	-	openstates - state bills (requires DOCKET_OPENSTATES_KEY; skips if unset)
	+	python -m ingest.run --once # run the whole pipeline once, in order
	+	python -m ingest.run --source members # run one module's CLI
	+	python -m ingest.run --list # list pipeline steps
	+
	+	Each step is a module with its own tested `__main__` CLI (or a poll() function).
	+	The orchestrator runs them in dependency order, best-effort: a failing step is
	+	logged and the pipeline continues. Counts are emitted by each module's own logs.
	+	Keys (DOCKET_CONGRESS_KEY / DOCKET_OPENSTATES_KEY / DOCKET_LEGISCAN_KEY /
	+	DOCKET_LDA_KEY) are optional - modules that need one skip gracefully when absent.
	+	DB path comes from env DOCKET_DB_PATH.
		"""

		import argparse
		import logging
	+	import subprocess
		import sys

		logging.basicConfig(
		@@ -28,80 +27,85 @@ logging.basicConfig(
		)
		logger = logging.getLogger("ingest.run")

	-
	-	def _run_all() -> dict:
	-	"""Run every poller once. Returns {source: count} of new rows stored."""
	-	from ingest.pollers import congress, govinfo, openstates, treasury
	-	from ingest.pollers import usaspending, votes_house, votes_senate
	-
	-	results = {}
	-
	-	logger.info("=== Docket ingest run starting ===")
	-
	-	results["usaspending"] = usaspending.poll()
	-	results["congress"] = congress.poll()
	-	results["govinfo"] = govinfo.poll()
	-	results["treasury"] = treasury.poll()
	-	results["votes_house"] = votes_house.poll()
	-	results["votes_senate"]= votes_senate.poll()
	-	results["openstates"] = openstates.poll()
	-
	-	logger.info("=== Ingest run complete ===")
	-	for source, count in results.items():
	-	logger.info(" %-15s → %d new row(s)", source, count)
	-
	-	return results
	-
	-
	-	def _run_one(source: str) -> int:
	-	"""Run a single named poller."""
	-	from ingest.pollers import congress, govinfo, openstates, treasury
	-	from ingest.pollers import usaspending, votes_house, votes_senate
	-
	-	pollers = {
	-	"usaspending": usaspending.poll,
	-	"congress": congress.poll,
	-	"govinfo": govinfo.poll,
	-	"treasury": treasury.poll,
	-	"votes_house": votes_house.poll,
	-	"votes_senate": votes_senate.poll,
	-	"openstates": openstates.poll,
	-	# short aliases
	-	"usa": usaspending.poll,
	-	"house": votes_house.poll,
	-	"senate": votes_senate.poll,
	-	}
	-	fn = pollers.get(source.lower())
	-	if fn is None:
	-	logger.error("Unknown source: %r. Valid: %s", source, ", ".join(pollers))
	-	sys.exit(1)
	-	return fn()
	+	# Dependency-ordered pipeline. Each entry: (label, module, [extra args]).
	+	# Roster first (vote/bill linkage), then independent federal sources,
	+	# then the bills pipeline (bulk -> versions -> diff -> riders -> amendments),
	+	# then votes (need members), procedural, lobbying, and finally state + coverage.
	+	PIPELINE = [
	+	("members", "ingest.pollers.members", []),
	+	("usaspending", "ingest.pollers.usaspending", []),
	+	("treasury", "ingest.pollers.treasury", []),
	+	("federal_register","ingest.pollers.federal_register",[]),
	+	("openomb", "ingest.pollers.openomb", []),
	+	("congress", "ingest.pollers.congress", []),
	+	("govinfo", "ingest.pollers.govinfo", []),
	+	("bills_bulk", "ingest.pollers.bills_bulk", []),
	+	("bill_versions", "ingest.pollers.bill_versions", []),
	+	("diff", "ingest.diff", ["--all", "--congress", "119"]),
	+	("rider_detect", "ingest.rider_detect", []),
	+	("amendments", "ingest.amendments", []),
	+	("votes_senate", "ingest.pollers.votes_senate", []),
	+	("votes_house", "ingest.pollers.votes_house", []),
	+	("votes_enrich", "ingest.pollers.votes_enrich", []),
	+	("crec", "ingest.pollers.crec", []),
	+	("lobbying", "ingest.pollers.lobbying", []),
	+	("openstates", "ingest.pollers.openstates", []),
	+	("legiscan", "ingest.pollers.legiscan", []),
	+	("state_coverage", "ingest.state_coverage", []),
	+	]
	+
	+	_BY_LABEL = {label: (mod, args) for label, mod, args in PIPELINE}
	+	# short aliases
	+	_ALIASES = {"usa": "usaspending", "house": "votes_house", "senate": "votes_senate",
	+	"fr": "federal_register", "omb": "openomb", "bills": "bills_bulk",
	+	"coverage": "state_coverage"}
	+
	+
	+	def _run_step(label: str, module: str, extra: list) -> bool:
	+	"""Run one module via its CLI. Returns True on exit 0."""
	+	cmd = [sys.executable, "-m", module, "--once", *extra]
	+	logger.info("--- step: %-16s (%s) ---", label, module)
	+	try:
	+	rc = subprocess.run(cmd, check=False).returncode
	+	if rc != 0:
	+	logger.warning("step %s exited %d (continuing)", label, rc)
	+	return rc == 0
	+	except Exception as exc: # never let one step kill the pipeline
	+	logger.warning("step %s raised %r (continuing)", label, exc)
	+	return False
	+
	+
	+	def _run_all() -> None:
	+	logger.info("=== Docket full-spectrum ingest starting (%d steps) ===", len(PIPELINE))
	+	ok = 0
	+	for label, module, extra in PIPELINE:
	+	if _run_step(label, module, extra):
	+	ok += 1
	+	logger.info("=== Ingest complete: %d/%d steps clean ===", ok, len(PIPELINE))


		def main():
	-	parser = argparse.ArgumentParser(description="Docket ingest runner")
	-	parser.add_argument(
	-	"--once",
	-	action="store_true",
	-	help="Run all pollers once and exit.",
	-	)
	-	parser.add_argument(
	-	"--source",
	-	metavar="SOURCE",
	-	help="Run only this poller (usaspending, congress, govinfo, treasury, "
	-	"votes_house, votes_senate, openstates).",
	-	)
	+	parser = argparse.ArgumentParser(description="Docket full-spectrum ingest runner")
	+	parser.add_argument("--once", action="store_true", help="Run the whole pipeline once.")
	+	parser.add_argument("--source", metavar="STEP", help="Run a single step by label.")
	+	parser.add_argument("--list", action="store_true", help="List pipeline steps.")
		args = parser.parse_args()

	-	if not args.once and not args.source:
	-	parser.print_help()
	-	sys.exit(0)
	-
	+	if args.list:
	+	for label, mod, extra in PIPELINE:
	+	print(f" {label:16} {mod} {' '.join(extra)}")
	+	return
		if args.source:
	-	count = _run_one(args.source)
	-	logger.info("Done: %d new row(s) from %s", count, args.source)
	-	else:
	+	label = _ALIASES.get(args.source.lower(), args.source.lower())
	+	if label not in _BY_LABEL:
	+	logger.error("Unknown step %r. Use --list.", args.source)
	+	sys.exit(1)
	+	mod, extra = _BY_LABEL[label]
	+	sys.exit(0 if _run_step(label, mod, extra) else 1)
	+	if args.once:
		_run_all()
	+	return
	+	parser.print_help()


		if __name__ == "__main__":

services/ingest/list_crec.py +0 -36

		@@ -1,36 +0,0 @@
	-	import zipfile, re, html as html_module
	-	zf = zipfile.ZipFile("/tmp/crec_2026.zip")
	-	names = zf.namelist()
	-	htm_names = [n for n in names if n.endswith('.htm')]
	-	print(f"HTM files: {len(htm_names)}, sample: {htm_names[:4]}")
	-
	-	patterns = {
	-	"voice_vote": r'ayes? (?:have\|has) it\|noes? (?:have\|has) it\|yeas? (?:have\|has) it\|nays? (?:have\|has) it',
	-	"unanimous_consent": r'without objection\|unanimous consent\|so ordered',
	-	"cloture": r'cloture\|invoke cloture',
	-	"suspension": r'suspend the rules\|suspension of the rules',
	-	"mtr": r'motion to recommit',
	-	}
	-
	-	counts = {k: 0 for k in patterns}
	-	for fname in htm_names:
	-	raw = zf.read(fname).decode("utf-8", errors="replace")
	-	text = re.sub(r'<[^>]+>', ' ', raw)
	-	text = html_module.unescape(text)
	-	for kind, pat in patterns.items():
	-	for m in re.finditer(pat, text, re.IGNORECASE):
	-	counts[kind] += 1
	-
	-	print("Pattern counts in CREC-2026-06-16:", counts)
	-
	-	# Show one sample match
	-	for fname in htm_names:
	-	raw = zf.read(fname).decode("utf-8", errors="replace")
	-	text = re.sub(r'<[^>]+>', ' ', raw)
	-	text = html_module.unescape(text)
	-	m = re.search(r'ayes? (?:have\|has) it\|noes? (?:have\|has) it', text, re.IGNORECASE)
	-	if m:
	-	ctx = text[max(0,m.start()-80):m.end()+100].replace('\n',' ')
	-	print(f"\nVoice vote sample from {fname}:")
	-	print(" ", ctx.strip()[:250])
	-	break

services/ingest/run_diff2_tmp.py +0 -44

		@@ -1,44 +0,0 @@
	-	import sqlite3, os, sys
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	from ingest.db import get_connection, init_schema
	-	from ingest.pollers.bills_bulk import ensure_schema
	-	from ingest.diff import diff_bill
	-	from ingest.rider_detect import scan_bill_diffs
	-
	-	conn = get_connection(db)
	-	init_schema(conn)
	-	ensure_schema(conn)
	-
	-	# H.R. 41 has ih → rh → eh (committee markup)
	-	row = conn.execute("SELECT id, title FROM bills WHERE number='H.R. 41' AND source='govinfo_billstatus'").fetchone()
	-	if not row:
	-	# Try H.J.Res. 25 which has ih→rh→eh (6 total)
	-	row = conn.execute("SELECT id, title FROM bills WHERE number='H.J.Res. 25' AND source='govinfo_billstatus'").fetchone()
	-
	-	bill_id = row["id"]
	-	print(f"Diffing: {row['title'][:70]}")
	-
	-	conn.execute("DELETE FROM bill_diffs WHERE bill_id=?", (bill_id,))
	-	conn.commit()
	-
	-	versions = conn.execute("SELECT version_code, version_date FROM bill_versions WHERE bill_id=? ORDER BY version_date", (bill_id,)).fetchall()
	-	print("Versions:", [(v["version_code"], v["version_date"]) for v in versions])
	-
	-	n = diff_bill(bill_id, conn)
	-	conn.commit()
	-	print(f"\nTotal diff rows: {n}")
	-
	-	print("\n=== BILL_DIFFS ===")
	-	for d in conn.execute("SELECT from_version, to_version, section_id, change, rider_flag, snippet FROM bill_diffs WHERE bill_id=? ORDER BY from_version, change", (bill_id,)).fetchall():
	-	print(f" {d['from_version']:5s}→{d['to_version']:5s} \| {d['change']:8s} \| rider={d['rider_flag']} \| {d['section_id']}")
	-	if d['snippet'] and d['change'] == 'amended':
	-	print(f" DIFF SNIPPET: {d['snippet'][:300]}")
	-
	-	flagged = scan_bill_diffs(bill_id, conn)
	-	conn.commit()
	-	print(f"\nRider flags set: {flagged}")
	-
	-	conn.close()

services/ingest/run_diff3_tmp.py +0 -59

		@@ -1,59 +0,0 @@
	-	import sqlite3, os, sys
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	from ingest.db import get_connection, init_schema
	-	from ingest.pollers.bills_bulk import ensure_schema
	-	from ingest.diff import diff_bill
	-	from ingest.rider_detect import scan_bill_diffs, scan_all_bills
	-
	-	conn = get_connection(db)
	-	init_schema(conn)
	-	ensure_schema(conn)
	-
	-	# Clear all existing diffs and re-run everything
	-	conn.execute("DELETE FROM bill_diffs")
	-	conn.commit()
	-
	-	# Find all bills with 2+ versions
	-	multi_ver = conn.execute("""
	-	SELECT b.id, b.number, b.title, COUNT(bv.id) nv
	-	FROM bills b JOIN bill_versions bv ON bv.bill_id=b.id
	-	WHERE b.source='govinfo_billstatus'
	-	GROUP BY b.id HAVING nv >= 2 ORDER BY nv DESC
	-	""").fetchall()
	-
	-	print(f"Bills with 2+ versions: {len(multi_ver)}")
	-
	-	total_diff_rows = 0
	-	for row in multi_ver:
	-	n = diff_bill(row["id"], conn)
	-	conn.commit()
	-	total_diff_rows += n
	-	if n > 0:
	-	print(f" {row['number']}: {n} diff rows")
	-
	-	print(f"\nTotal bill_diffs rows: {total_diff_rows}")
	-
	-	# Run rider detection
	-	print("\nRunning rider detection...")
	-	flagged = scan_all_bills(conn, congress=119)
	-	print(f"Rider flags: {flagged}")
	-
	-	# Show flagged rows
	-	print("\n=== RIDER FLAGGED DIFFS ===")
	-	for d in conn.execute("""
	-	SELECT b.number, bd.from_version, bd.to_version, bd.section_id, bd.snippet
	-	FROM bill_diffs bd JOIN bills b ON b.id=bd.bill_id
	-	WHERE bd.rider_flag=1 LIMIT 10
	-	""").fetchall():
	-	print(f" {d['number']} {d['from_version']}→{d['to_version']} \| {d['snippet'][:150]}")
	-
	-	# Summary
	-	print("\n=== FINAL COUNTS ===")
	-	for table in ["bills", "bill_versions", "bill_diffs", "amendments", "actions"]:
	-	r = conn.execute(f"SELECT COUNT(*) n FROM {table}").fetchone()
	-	print(f" {table}: {r['n']}")
	-
	-	conn.close()

services/ingest/run_diff_tmp.py +0 -40

		@@ -1,40 +0,0 @@
	-	import sqlite3, os, sys
	-
	-	db = os.environ.get("DOCKET_DB_PATH", "/opt/docket/data/docket.db")
	-	sys.path.insert(0, "/shared/projects/docket/services/ingest")
	-
	-	from ingest.db import get_connection, init_schema
	-	from ingest.pollers.bills_bulk import ensure_schema
	-	from ingest.diff import diff_bill
	-	from ingest.rider_detect import scan_bill_diffs
	-
	-	conn = get_connection(db)
	-	init_schema(conn)
	-	ensure_schema(conn)
	-
	-	# Get H.R. 21 bill_id
	-	row = conn.execute("SELECT id, title FROM bills WHERE number='H.R. 21' AND source='govinfo_billstatus'").fetchone()
	-	bill_id = row["id"]
	-	print(f"Diffing bill_id={bill_id}: {row['title']}")
	-
	-	# Clear any existing diffs first
	-	conn.execute("DELETE FROM bill_diffs WHERE bill_id=?", (bill_id,))
	-	conn.commit()
	-
	-	n = diff_bill(bill_id, conn)
	-	print(f"Diff rows: {n}")
	-
	-	# Show diffs
	-	print("\n=== BILL_DIFFS (H.R. 21) ===")
	-	for d in conn.execute("SELECT from_version, to_version, section_id, change, rider_flag, snippet FROM bill_diffs WHERE bill_id=? ORDER BY from_version, section_id", (bill_id,)).fetchall():
	-	print(f" {d['from_version']:4s}→{d['to_version']:4s} \| {d['change']:8s} \| section={d['section_id']}")
	-	if d['snippet']:
	-	print(f" SNIPPET: {d['snippet'][:200]}")
	-
	-	# Rider scan
	-	print("\n=== RIDER DETECTION ===")
	-	flagged = scan_bill_diffs(bill_id, conn)
	-	conn.commit()
	-	print(f" Flagged {flagged} diff rows as rider candidates")
	-
	-	conn.close()

services/ingest/tests/test_votes_house_parser.py +539 -0

		@@ -0,0 +1,539 @@
	+	"""
	+	Tests for votes_house XML parser and votes_enrich Senate parser.
	+
	+	Coverage:
	+	• parse_house_vote_xml() - position enum, bill linkage, aggregate counts
	+	• _normalize_position() - canonical mapping including edge cases
	+	• _parse_legis_num() - bill-type normalization
	+	• parse_senate_vote_positions() - lis_member_id extraction, position mapping
	+	• Integration: schema setup + position insertion via temp DB
	+	"""
	+
	+	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.votes_house import (
	+	_normalize_position,
	+	_parse_legis_num,
	+	parse_house_vote_xml,
	+	)
	+	from ingest.pollers.votes_enrich import parse_senate_vote_positions
	+
	+
	+	# ─── Fixtures ─────────────────────────────────────────────────────────────────
	+
	+	SAMPLE_HOUSE_XML = """\
	+	<?xml version="1.0" encoding="UTF-8"?>
	+	<rollcall-vote>
	+	<vote-metadata>
	+	<majority>Republican</majority>
	+	<congress>119</congress>
	+	<session>1</session>
	+	<chamber>U.S. House of Representatives</chamber>
	+	<rollcall-num>42</rollcall-num>
	+	<legis-num>H R 1000</legis-num>
	+	<vote-question>On Passage</vote-question>
	+	<vote-type>YEA-AND-NAY</vote-type>
	+	<vote-result>Passed</vote-result>
	+	<action-date>15-Mar-2025</action-date>
	+	<action-time time-etz="2:30 PM">2:30 PM</action-time>
	+	<vote-desc>Some bill</vote-desc>
	+	<vote-totals>
	+	<totals-by-party>
	+	<party>Republican</party>
	+	<yea-total>210</yea-total>
	+	<nay-total>5</nay-total>
	+	<present-total>0</present-total>
	+	<not-voting-total>2</not-voting-total>
	+	</totals-by-party>
	+	<totals-by-party>
	+	<party>Democratic</party>
	+	<yea-total>3</yea-total>
	+	<nay-total>195</nay-total>
	+	<present-total>1</present-total>
	+	<not-voting-total>1</not-voting-total>
	+	</totals-by-party>
	+	<totals-by-vote>
	+	<yea-total>213</yea-total>
	+	<nay-total>200</nay-total>
	+	<present-total>1</present-total>
	+	<not-voting-total>3</not-voting-total>
	+	</totals-by-vote>
	+	</vote-totals>
	+	</vote-metadata>
	+	<vote-data>
	+	<recorded-vote>
	+	<legislator name-id="A000055" sort-field="Aderholt" party="R" state="AL" role="Member">Aderholt</legislator>
	+	<vote>Yea</vote>
	+	</recorded-vote>
	+	<recorded-vote>
	+	<legislator name-id="B000574" sort-field="Blumenauer" party="D" state="OR" role="Member">Blumenauer</legislator>
	+	<vote>Nay</vote>
	+	</recorded-vote>
	+	<recorded-vote>
	+	<legislator name-id="C000537" sort-field="Clyburn" party="D" state="SC" role="Member">Clyburn</legislator>
	+	<vote>Present</vote>
	+	</recorded-vote>
	+	<recorded-vote>
	+	<legislator name-id="D000617" sort-field="DelBene" party="D" state="WA" role="Member">DelBene</legislator>
	+	<vote>Not Voting</vote>
	+	</recorded-vote>
	+	</vote-data>
	+	</rollcall-vote>
	+	"""
	+
	+	VOICE_VOTE_XML = """\
	+	<?xml version="1.0" encoding="UTF-8"?>
	+	<rollcall-vote>
	+	<vote-metadata>
	+	<congress>119</congress>
	+	<rollcall-num>10</rollcall-num>
	+	<vote-type>VOICE VOTE</vote-type>
	+	<vote-result>Passed</vote-result>
	+	<action-date>10-Jan-2025</action-date>
	+	<vote-totals>
	+	<totals-by-vote>
	+	<yea-total>0</yea-total>
	+	<nay-total>0</nay-total>
	+	<present-total>0</present-total>
	+	<not-voting-total>0</not-voting-total>
	+	</totals-by-vote>
	+	</vote-totals>
	+	</vote-metadata>
	+	</rollcall-vote>
	+	"""
	+
	+	SAMPLE_SENATE_XML = """\
	+	<?xml version="1.0" encoding="UTF-8"?>
	+	<roll_call_vote>
	+	<congress>119</congress>
	+	<session>1</session>
	+	<congress_year>2025</congress_year>
	+	<vote_number>100</vote_number>
	+	<vote_date>March 15, 2025, 02:00 PM</vote_date>
	+	<vote_question_text>On Passage of the Bill</vote_question_text>
	+	<vote_result_text>Passed</vote_result_text>
	+	<count>
	+	<yeas>51</yeas>
	+	<nays>49</nays>
	+	<abstains>0</abstains>
	+	<present>0</present>
	+	</count>
	+	<members>
	+	<member>
	+	<lis_member_id>S000148</lis_member_id>
	+	<member_full>Schumer, Charles E.</member_full>
	+	<last_name>Schumer</last_name>
	+	<first_name>Charles</first_name>
	+	<party>D</party>
	+	<state>NY</state>
	+	<vote_cast>Yea</vote_cast>
	+	</member>
	+	<member>
	+	<lis_member_id>S001217</lis_member_id>
	+	<member_full>Scott, Rick</member_full>
	+	<last_name>Scott</last_name>
	+	<first_name>Rick</first_name>
	+	<party>R</party>
	+	<state>FL</state>
	+	<vote_cast>Nay</vote_cast>
	+	</member>
	+	<member>
	+	<lis_member_id>S001181</lis_member_id>
	+	<member_full>Shaheen, Jeanne</member_full>
	+	<last_name>Shaheen</last_name>
	+	<first_name>Jeanne</first_name>
	+	<party>D</party>
	+	<state>NH</state>
	+	<vote_cast>Not Voting</vote_cast>
	+	</member>
	+	<member>
	+	<lis_member_id>S001184</lis_member_id>
	+	<member_full>Scott, Tim</member_full>
	+	<last_name>Scott</last_name>
	+	<first_name>Tim</first_name>
	+	<party>R</party>
	+	<state>SC</state>
	+	<vote_cast>Present</vote_cast>
	+	</member>
	+	</members>
	+	</roll_call_vote>
	+	"""
	+
	+
	+	# ─── _normalize_position ──────────────────────────────────────────────────────
	+
	+	class TestNormalizePosition:
	+	def test_yea_variants(self):
	+	assert _normalize_position("Yea") == "Yea"
	+	assert _normalize_position("YEA") == "Yea"
	+	assert _normalize_position("yea") == "Yea"
	+	assert _normalize_position("Aye") == "Yea"
	+	assert _normalize_position("aye") == "Yea"
	+
	+	def test_nay_variants(self):
	+	assert _normalize_position("Nay") == "Nay"
	+	assert _normalize_position("NAY") == "Nay"
	+	assert _normalize_position("No") == "Nay"
	+	assert _normalize_position("no") == "Nay"
	+
	+	def test_present(self):
	+	assert _normalize_position("Present") == "Present"
	+	assert _normalize_position("present") == "Present"
	+	assert _normalize_position("PRESENT") == "Present"
	+
	+	def test_not_voting(self):
	+	assert _normalize_position("Not Voting") == "Not Voting"
	+	assert _normalize_position("not voting") == "Not Voting"
	+	assert _normalize_position("not-voting") == "Not Voting"
	+
	+	def test_unknown_returns_none(self):
	+	assert _normalize_position("Abstain") is None
	+	assert _normalize_position("") is None
	+	assert _normalize_position("Pass") is None
	+
	+	def test_positions_are_in_valid_set(self):
	+	valid = {"Yea", "Nay", "Present", "Not Voting"}
	+	for raw in ["Yea", "Nay", "Present", "Not Voting", "Aye", "No"]:
	+	result = _normalize_position(raw)
	+	assert result in valid, f"Got unexpected position {result!r} for {raw!r}"
	+
	+
	+	# ─── _parse_legis_num ─────────────────────────────────────────────────────────
	+
	+	class TestParseLegisNum:
	+	def test_house_bill(self):
	+	bt, num, ext = _parse_legis_num("H R 1234", 119)
	+	assert bt == "hr"
	+	assert num == "1234"
	+	assert ext == "119-hr-1234"
	+
	+	def test_house_resolution(self):
	+	bt, num, ext = _parse_legis_num("H RES 50", 119)
	+	assert bt == "hres"
	+	assert num == "50"
	+	assert ext == "119-hres-50"
	+
	+	def test_house_joint_resolution(self):
	+	bt, num, ext = _parse_legis_num("H J RES 88", 119)
	+	assert bt == "hjres"
	+	assert num == "88"
	+	assert ext == "119-hjres-88"
	+
	+	def test_house_concurrent_resolution(self):
	+	bt, num, ext = _parse_legis_num("H CON RES 3", 119)
	+	assert bt == "hconres"
	+	assert num == "3"
	+	assert ext == "119-hconres-3"
	+
	+	def test_senate_bill(self):
	+	bt, num, ext = _parse_legis_num("S 500", 119)
	+	assert bt == "s"
	+	assert num == "500"
	+	assert ext == "119-s-500"
	+
	+	def test_senate_resolution(self):
	+	bt, num, ext = _parse_legis_num("S RES 22", 119)
	+	assert bt == "sres"
	+	assert num == "22"
	+	assert ext == "119-sres-22"
	+
	+	def test_unparseable_returns_none_triple(self):
	+	bt, num, ext = _parse_legis_num("QUORUM CALL", 119)
	+	assert bt is None and num is None and ext is None
	+
	+	def test_empty_returns_none_triple(self):
	+	bt, num, ext = _parse_legis_num("", 119)
	+	assert bt is None and num is None and ext is None
	+
	+	def test_case_insensitive_prefix(self):
	+	bt, num, ext = _parse_legis_num("h r 999", 119)
	+	# Prefix matching normalises to upper first
	+	assert bt == "hr"
	+	assert num == "999"
	+
	+
	+	# ─── parse_house_vote_xml ─────────────────────────────────────────────────────
	+
	+	class TestParseHouseVoteXml:
	+	SOURCE_URL = "https://clerk.house.gov/evs/2025/roll042.xml"
	+
	+	def test_basic_fields(self):
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	assert result["congress"] == 119
	+	assert result["session"] == 1
	+	assert result["roll_number"] == 42
	+	assert result["action_date"] == "2025-03-15"
	+	assert result["question"] == "On Passage"
	+	assert result["result"] == "Passed"
	+	assert result["source_url"] == self.SOURCE_URL
	+
	+	def test_aggregate_counts(self):
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	assert result["yea"] == 213
	+	assert result["nay"] == 200
	+	assert result["present"] == 1
	+	assert result["not_voting"] == 3
	+
	+	def test_bill_linkage_parsed(self):
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	assert result["bill_text_ref"] == "H R 1000"
	+	assert result["legis_external_id"] == "119-hr-1000"
	+
	+	def test_positions_extracted(self):
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	positions = result["positions"]
	+	assert len(positions) == 4
	+
	+	def test_position_bioguides(self):
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	pos_by_bio = {p["bioguide"]: p["position"] for p in result["positions"]}
	+	assert pos_by_bio["A000055"] == "Yea"
	+	assert pos_by_bio["B000574"] == "Nay"
	+	assert pos_by_bio["C000537"] == "Present"
	+	assert pos_by_bio["D000617"] == "Not Voting"
	+
	+	def test_all_positions_in_valid_enum(self):
	+	valid = {"Yea", "Nay", "Present", "Not Voting"}
	+	result = parse_house_vote_xml(SAMPLE_HOUSE_XML, self.SOURCE_URL)
	+	assert result is not None
	+	for p in result["positions"]:
	+	assert p["position"] in valid, f"Invalid position: {p['position']!r}"
	+
	+	def test_voice_vote_returns_none(self):
	+	"""Voice votes must be skipped - no individual record exists."""
	+	result = parse_house_vote_xml(VOICE_VOTE_XML, self.SOURCE_URL)
	+	assert result is None, "Voice vote should not produce position data"
	+
	+	def test_malformed_xml_returns_none(self):
	+	result = parse_house_vote_xml("<broken>", self.SOURCE_URL)
	+	assert result is None
	+
	+	def test_empty_xml_returns_none(self):
	+	result = parse_house_vote_xml("", self.SOURCE_URL)
	+	assert result is None
	+
	+
	+	# ─── parse_senate_vote_positions ─────────────────────────────────────────────
	+
	+	class TestParseSenateVotePositions:
	+	def test_member_count(self):
	+	positions = parse_senate_vote_positions(SAMPLE_SENATE_XML)
	+	assert len(positions) == 4
	+
	+	def test_all_positions_valid(self):
	+	valid = {"Yea", "Nay", "Present", "Not Voting"}
	+	for p in parse_senate_vote_positions(SAMPLE_SENATE_XML):
	+	assert p["position"] in valid
	+
	+	def test_yea_member(self):
	+	positions = parse_senate_vote_positions(SAMPLE_SENATE_XML)
	+	schumer = next(p for p in positions if p["lis_member_id"] == "S000148")
	+	assert schumer["position"] == "Yea"
	+	assert schumer["state"] == "NY"
	+
	+	def test_nay_member(self):
	+	positions = parse_senate_vote_positions(SAMPLE_SENATE_XML)
	+	scott = next(p for p in positions if p["lis_member_id"] == "S001217")
	+	assert scott["position"] == "Nay"
	+
	+	def test_not_voting_member(self):
	+	positions = parse_senate_vote_positions(SAMPLE_SENATE_XML)
	+	shaheen = next(p for p in positions if p["lis_member_id"] == "S001181")
	+	assert shaheen["position"] == "Not Voting"
	+
	+	def test_present_member(self):
	+	positions = parse_senate_vote_positions(SAMPLE_SENATE_XML)
	+	tim_scott = next(p for p in positions if p["lis_member_id"] == "S001184")
	+	assert tim_scott["position"] == "Present"
	+
	+	def test_malformed_xml_returns_empty(self):
	+	assert parse_senate_vote_positions("<broken>") == []
	+
	+	def test_empty_xml_returns_empty(self):
	+	assert parse_senate_vote_positions("") == []
	+
	+	def test_no_members_element_returns_empty(self):
	+	xml = "<roll_call_vote><congress>119</congress></roll_call_vote>"
	+	assert parse_senate_vote_positions(xml) == []
	+
	+
	+	# ─── Integration: schema + position insertion ─────────────────────────────────
	+
	+	class TestSchemaAndInsertion:
	+	"""Verify that the schema setup and actual DB writes work end-to-end."""
	+
	+	def _make_db(self):
	+	tmp = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(tmp)
	+	init_schema(conn)
	+	ensure_votes_schema(conn)
	+	return conn, tmp
	+
	+	def test_members_table_created(self):
	+	conn, tmp = self._make_db()
	+	tables = {
	+	row[0]
	+	for row in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='table'"
	+	)
	+	}
	+	assert "members" in tables
	+	assert "vote_positions" in tables
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_votes_extended_columns_present(self):
	+	conn, tmp = self._make_db()
	+	info = {
	+	row[1]
	+	for row in conn.execute("PRAGMA table_info(votes)")
	+	}
	+	for col in ["congress", "session", "roll_number", "question",
	+	"not_voting", "recorded", "vote_external_id", "bill_text_ref"]:
	+	assert col in info, f"Expected column {col!r} in votes"
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_vote_positions_indexes_exist(self):
	+	conn, tmp = self._make_db()
	+	indexes = {
	+	row[0]
	+	for row in conn.execute(
	+	"SELECT name FROM sqlite_master WHERE type='index'"
	+	)
	+	}
	+	assert "idx_vote_positions_vote_id" in indexes
	+	assert "idx_vote_positions_member_id" in indexes
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_insert_position_row(self):
	+	conn, tmp = self._make_db()
	+	# Insert a stub vote
	+	conn.execute(
	+	"INSERT INTO votes (chamber, date, yea, nay, present, result, source_url) "
	+	"VALUES (?, ?, ?, ?, ?, ?, ?)",
	+	("house", "2025-03-15", 213, 200, 1, "Passed",
	+	"https://clerk.house.gov/evs/2025/ROLL_042.xml"),
	+	)
	+	vote_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+
	+	# Insert a position
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position) "
	+	"VALUES (?, ?, ?, ?)",
	+	(vote_id, "A000055", None, "Yea"),
	+	)
	+	conn.commit()
	+
	+	row = conn.execute(
	+	"SELECT * FROM vote_positions WHERE vote_id=?", (vote_id,)
	+	).fetchone()
	+	assert row is not None
	+	assert row["member_bioguide"] == "A000055"
	+	assert row["position"] == "Yea"
	+
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_position_enum_enforced(self):
	+	"""SQLite CHECK constraint must reject invalid position values."""
	+	conn, tmp = self._make_db()
	+	conn.execute(
	+	"INSERT INTO votes (chamber, date, yea, nay, present, result, source_url) "
	+	"VALUES (?, ?, ?, ?, ?, ?, ?)",
	+	("house", "2025-03-15", 1, 0, 0, "Passed",
	+	"https://clerk.house.gov/evs/2025/roll001.xml"),
	+	)
	+	vote_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+	conn.execute("PRAGMA foreign_keys=ON")
	+
	+	with pytest.raises(sqlite3.IntegrityError):
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, position) "
	+	"VALUES (?, ?, ?)",
	+	(vote_id, "A000055", "Abstain"), # not in enum
	+	)
	+	conn.commit()
	+
	+	conn.close()
	+	os.unlink(tmp)
	+
	+	def test_both_pivots_queryable(self):
	+	"""
	+	Confirm that both query directions are supported:
	+	- all positions for a vote (vote_id → vote_positions → members)
	+	- all votes for a member (bioguide → vote_positions → votes)
	+	"""
	+	conn, tmp = self._make_db()
	+
	+	# Seed member
	+	conn.execute(
	+	"INSERT INTO members (bioguide_id, full_name, party, state, chamber, source_url) "
	+	"VALUES (?, ?, ?, ?, ?, ?)",
	+	("A000055", "Robert Aderholt", "Republican", "AL", "house",
	+	"https://bioguide.congress.gov/search/bio/A000055"),
	+	)
	+	member_id = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+
	+	# Seed two votes
	+	for roll in [42, 43]:
	+	conn.execute(
	+	"INSERT INTO votes (chamber, date, yea, nay, present, result, source_url, roll_number) "
	+	"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
	+	("house", "2025-03-15", 213, 200, 1, "Passed",
	+	f"https://clerk.house.gov/evs/2025/ROLL_{roll:03d}.xml", roll),
	+	)
	+	vid = conn.execute("SELECT last_insert_rowid() AS id").fetchone()["id"]
	+	conn.execute(
	+	"INSERT INTO vote_positions (vote_id, member_bioguide, member_id, position) "
	+	"VALUES (?, ?, ?, ?)",
	+	(vid, "A000055", member_id, "Yea"),
	+	)
	+
	+	conn.commit()
	+
	+	# Pivot 1: all positions for a given vote
	+	vote_id_42 = conn.execute(
	+	"SELECT id FROM votes WHERE roll_number=42"
	+	).fetchone()["id"]
	+	pos_rows = conn.execute(
	+	"""
	+	SELECT vp.position, m.full_name, m.state
	+	FROM vote_positions vp
	+	JOIN members m ON m.id = vp.member_id
	+	WHERE vp.vote_id = ?
	+	""",
	+	(vote_id_42,),
	+	).fetchall()
	+	assert len(pos_rows) == 1
	+	assert pos_rows[0]["full_name"] == "Robert Aderholt"
	+	assert pos_rows[0]["position"] == "Yea"
	+
	+	# Pivot 2: all votes for a given member
	+	member_votes = conn.execute(
	+	"""
	+	SELECT v.roll_number, vp.position
	+	FROM vote_positions vp
	+	JOIN votes v ON v.id = vp.vote_id
	+	WHERE vp.member_bioguide = 'A000055'
	+	ORDER BY v.roll_number
	+	""",
	+	).fetchall()
	+	assert len(member_votes) == 2
	+	assert member_votes[0]["roll_number"] == 42
	+	assert member_votes[1]["roll_number"] == 43
	+
	+	conn.close()
	+	os.unlink(tmp)