0c80114 · Docket

Ingestion strategy doc (efficiency/automation research); API contract; state layer (coverage + STATE_LEG_HOSTS)

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

_ops/glm-f1.txt +15 -0

		@@ -0,0 +1,15 @@
	+	You are GLM-5.2, frontend lead for Docket (neutral US government transparency app). Work in /opt/docket (local build dir - npm builds work here). Be RAG-first; read ONLY these specs: docs/API.md, docs/ACCESSIBILITY.md, docs/BRAND.md, docs/NEUTRALITY.md, docs/ARCHITECTURE.md. Look at the existing app/ and components/ you already built.
	+
	+	WAVE F1 - make it REAL + ACCESSIBLE:
	+
	+	1. READ-API: implement Next.js route handlers under app/api/ EXACTLY per docs/API.md (feed, bills, bills/[id], bills/[id]/diff, votes, spending/awards, spending/fiscal, spending/apportionments, executive, lobbying, states, members/[id], search). Read the SQLite DB at process.env.DOCKET_DB_PATH (default /opt/docket/data/docket.db) using better-sqlite3 (npm install better-sqlite3). BE DEFENSIVE: some tables do not exist yet (other agents are still building ingest) - wrap every query in try/catch and return {items:[],total:0} instead of crashing. Every returned item MUST include sourceUrl + href.
	+
	+	2. WIRE REAL DATA: replace the MOCK data on /register and /spending with fetches to your new /api routes. The DB ALREADY has real USAspending awards and real Senate votes - show them. Keep the live Register hero but feed it from /api/feed (fall back to a calm empty state if a category is empty).
	+
	+	3. ACCESSIBILITY (docs/ACCESSIBILITY.md): add the 3-mode display toggle - Dark (default brand) / Readable-Light / Plain (high-contrast, system font, >=20px) - persisted in localStorage, always-visible labeled control >=44x44px. Enforce contrast >=4.5:1, body >=18px dark / >=16px light, visible focus outlines, underlined links, NO hover-only interactions.
	+
	+	Keep the BRAND (docs/BRAND.md: brass rationed, custody-stamp, mono data) and the NEUTRALITY gate (no adjectives, a Source link on every fact, no comment sections, muted vote colors never vivid red/blue).
	+
	+	BUILD + VERIFY (do not skip): npm install; npm run build MUST pass. Then restart the server: kill whatever is on port 3000 (fuser -k 3000/tcp), then `DOCKET_DB_PATH=/opt/docket/data/docket.db nohup ./node_modules/.bin/next start -H 0.0.0.0 -p 3000 &`. curl http://localhost:3000/api/feed and http://localhost:3000/register to confirm REAL data renders (not mock).
	+
	+	Write FRONTEND_NOTES.md (what you built, the API routes, how to run). Definition of done: clean build, /register shows REAL awards + votes from the API, all /api routes respond (even if empty), 3-mode toggle works and persists.

docs/API.md +49 -0

		@@ -0,0 +1,49 @@
	+	# Docket - Internal API Contract
	+
	+	Single-box deployment (CT247): Next.js app + SQLite (local, `/opt/docket/data/docket.db`) populated by the Python ingest. Frontend reads its OWN Next.js route handlers (`app/api/*`) via `better-sqlite3`. Backend agents write to the SQLite schema below; frontend consumes these JSON shapes. This is the contract - build to it on both sides.
	+
	+	## Conventions
	+	- All responses JSON. Every item carries `sourceUrl` (allowlisted official `.gov`/state-leg host; aggregator → underlying official record).
	+	- Pagination: `?limit=` (default 50, max 200) `&offset=`. Responses: `{ items: [...], total, limit, offset }`.
	+	- Dates ISO-8601. Money in integer cents or a numeric `amount` + `amountText`.
	+	- Ranking endpoints obey neutrality rule #3 (action date, dollar magnitude, affected estimate, action type only).
	+
	+	## Endpoints
	+
	+	### `GET /api/feed`
	+	The Register. Mixed stream of official events, ranked (rule #3).
	+	`?type=bill\|vote\|award\|amendment\|rider\|procedural\|executive\|apportionment` (repeatable) `?state=US\|CA\|…` `?q=`
	+	Item: `{ id, kind, actionDate, chamber, title, summary?, actionType, amount?, amountText?, jurisdiction, sourceUrl, href }`
	+
	+	### `GET /api/bills` · `GET /api/bills/:id`
	+	List/lookup. `?q=&state=&status=&congress=&policyArea=&sort=updated\|action`
	+	Detail: `{ id, number, title, sponsor{name,party,state}, cosponsors, status, timeline[], policyArea, fullTextUrl, summaryPlain, versions[], diffs[], amendments[], votes[], sourceUrl }`
	+
	+	### `GET /api/bills/:id/diff?from=ih&to=rh` ← signature feature
	+	`{ from, to, changes:[{ sectionId, change:"added\|removed\|amended", snippet, riderFlag:bool }], note? }`
	+	`note` carries honesty text when a change entered via voice/UC/conference ("no individual record exists").
	+
	+	### `GET /api/votes` · `GET /api/votes/:id`
	+	`{ id, billId?, chamber, date, result, yea, nay, present, recorded:bool, positions:[{member,party,state,position}], sourceUrl }`
	+	`recorded:false` → voice/UC; `positions` empty; UI shows "no individual record exists."
	+
	+	### `GET /api/spending/*`
	+	`/api/spending/awards` (list), `/api/spending/fiscal` (debt/deficit/outlays), `/api/spending/apportionments` ← impound tracker: `{ agency, account, appropriated, apportioned, status:"released\|delayed\|withheld", date, sourceUrl }`
	+
	+	### `GET /api/executive` - Federal Register: EOs, rules, notices
	+	### `GET /api/lobbying?bill=` - LDA.gov filings
	+	### `GET /api/states` · `GET /api/states/:code`
	+	`/api/states` returns coverage + confidence per state: `{ code, name, confidence:"ok\|stale\|missing", lastScrape, missingTypes[] }`. Detail = state bills/votes.
	+	### `GET /api/members/:id` - a rep's record (votes, sponsorships, riders authored)
	+	### `GET /api/search?q=` - global lookup across bills/members/states/awards
	+
	+	## Backend → DB contract
	+	Ingest writes the tables in ARCHITECTURE.md "Data model" + "Full-spectrum additions". Each poller module is self-contained, does `CREATE TABLE IF NOT EXISTS`, is idempotent (dedupe on source+externalId+updateDate), and stores an allowlisted `source_url` on every row. Pollers do NOT modify `run.py` - the overseer wires registration.
	+
	+	## Keyless data paths (no api.data.gov key needed)
	+	- GovInfo BULK DATA `https://www.govinfo.gov/bulkdata/` - BILLS, BILLSTATUS, BILLSUM, CREC (no key). Primary path for bills, status, full text, versions (diff inputs), and Congressional Record.
	+	- Federal Register `federalregister.gov/api/v1` - no key.
	+	- OpenOMB `openomb.org` - no key (link source to underlying `.max.gov`).
	+	- USAspending / Treasury Fiscal - no key (already live).
	+	- House/Senate vote XML - no key (already live for Senate).
	+	Key-gated (build code now, run when key present): Congress.gov JSON, GovInfo JSON API (DOCKET_CONGRESS_KEY); OpenStates (DOCKET_OPENSTATES_KEY); LegiScan (DOCKET_LEGISCAN_KEY).

docs/INGESTION_STRATEGY.md +39 -0

		@@ -0,0 +1,39 @@
	+	# Docket - Ingestion Strategy (efficiency & automation)
	+
	+	Roadmap for ingest v2. Today's pollers are naive (full pull each run). This is how we make 50-state + federal ingestion fast, cheap, automated, and self-healing. Ranked by impact-to-effort.
	+
	+	## Top 10 (impact-to-effort)
	+	1. Watermark change-detection - GovInfo `collections/{col}/{lastModStart}/{lastModEnd}?offsetMark=`; Congress.gov `fromDateTime` + `updateDateIncludingText`* (catches text-only edits `updateDate` misses). Store `(source, last_run_at)` in `poller_watermarks`; update atomically only on successful upsert batch. ~10 lines, kills re-pulling.
	+	2. LegiScan weekly ZIP bulk - `getDatasetList`/`getDataset`: one download = all 50 states + Congress. Public API caps at 30k/mo (~one busy state). Bulk for backfill + weekly; API only for intra-week deltas.
	+	3. OpenStates / Plural Postgres dump (`open.pluralpolicy.com/downloads/`) - monthly all-50-state bootstrap, zero scraping; v3 API for incrementals.
	+	4. SHA-256 content-hash gate - hash raw payload before parse; skip parse/upsert if unchanged. `INSERT … ON CONFLICT(source,external_id) DO UPDATE … WHERE hash != EXCLUDED.hash`. Applies everywhere.
	+	5. RSS/Atom feed registry - ~20 states publish official feeds (KS, WA, HI, TX, WI, WV…). `feedparser` + last-seen GUID = sub-minute, zero-budget change detection.
	+	6. Session-calendar priority polling - `sessions` table (active/interim/dormant); poll active hourly, recess daily, dormant weekly. ~70% volume cut off-season.
	+	7. ARQ + aiometer async workers - Redis-backed asyncio queue (better fit than Celery here); `aiometer` for per-domain concurrency + rps caps; `tenacity` backoff on 429. Distribute across CT215 + CT220 via worker hash-ring.
	+	8. LLM selector self-healing (3060) - on field-extraction failure, queue raw HTML + schema → local `mistral-nemo:12b` proposes CSS/XPath → validate against live HTML → write to `scraper_selectors` table (no redeploy). ~98% accuracy across layout changes (2025 research).
	+	9. Embedding action-vocab normalization - 50 states use different action strings → embed with `nomic-embed-text` (Ollama/3060) → cluster (HDBSCAN) → map to one canonical taxonomy (INTRODUCED/REFERRED/PASSED/SIGNED/VETOED…). One-time batch, re-run on new strings.
	+	10. Near-duplicate bill clustering - embed bill summaries → store vectors in the existing Qdrant → cosine >0.92 flags model legislation (ALEC etc.) spreading across states. Unique Docket feature; zero marginal API cost.
	+
	+	## Recommended architecture
	+	- Source registry YAML - each source: `type` (api\|bulk\|rss\|scrape), `change_detection` (watermark\|rss\|hash\|head), `poll_interval_active/dormant`, `rate_limit_rps`, `js_required`, `priority`. Scheduler is data-driven; new/changed sources need no code.
	+	- Three-tier ingest: (1) bulk bootstrap once per session/congress; (2) delta API polling via watermarks (hourly active / daily dormant); (3) persistent RSS watchers emitting change events immediately.
	+	- ARQ worker pool (CT215+CT220), partitioned by source hash-ring, Redis broker; workers write raw payloads → `raw_ingest` staging.
	+	- Hash-gated parse stage - second worker computes `content_hash`, skips unchanged, else runs source parser → `*_staging` → idempotent upsert + emit `bill_versions` diff on text change.
	+	- Dead-letter + alerting - failures → `scrape_failures` (raw error + HTML snapshot + retry count); Discord webhook after 3; 6-hour retry worker.
	+	- LLM repair sidecar - separate process on the 3060 (Ollama), consumes a Redis `selector_repair_queue`; main scrapers read selectors from DB each run.
	+
	+	## 50-state scraping
	+	- Fork `openstates/openstates-scrapers` (~200 scrapers on `scrapelib` + `spatula` + `pupa`) - battle-tested session mgmt/robots/retry. Don't start from scratch.
	+	- JS-rendered states (~15: FL, IL…): `scrapy-playwright`. But probe first - ~40% of "JS" state sites call an undocumented JSON/GraphQL endpoint that's cleaner + more stable than the DOM. Catalog these as a priority tier.
	+	- LLM-as-parser only for the messy tail (PDFs, unstructured HTML) with `Outlines` constraint-decoded JSON + Pydantic validation. `lxml`/`trafilatura` beats LLM on structured HTML by 1000× - never put the LLM in the hot path.
	+
	+	## Creative / wildcards
	+	- Wayback CDX (`web.archive.org/cdx/search/cdx`) for historical backfill of states that purge old sessions.
	+	- GovTrack/`unitedstates/congress` git mirrors - bill versions as commits; `git log --diff-filter=M` gives changed-bills-since-last-sync without API.
	+	- Undocumented state JSON endpoints - check the network tab before writing any scraper.
	+
	+	## 3060 LLM: value vs trap
	+	- Use: selector self-healing, action-vocab embedding/normalization, PDF/messy-HTML extraction (long tail), near-dup clustering. All batch/exception-path.
	+	- Trap: LLM as primary parser for structured HTML; LLM in the real-time hot path (creates permanent backlog at 100k+ bills/session); trusting LLM output without schema validation; re-parsing the Congressional Record (use `unitedstates/congressional-record`).
	+
	+	Adopt incrementally: watermarks + hash-gate + bulk-first first (items 1-4) - biggest win, least effort - then the registry + workers, then the LLM sidecar.

services/ingest/STATE_NOTES.md +213 -0

		@@ -0,0 +1,213 @@
	+	# State-Level Ingest - Notes
	+
	+	## What Was Built
	+
	+	State-level granular coverage with honest confidence scoring for all 52 US
	+	jurisdictions (50 states + DC + PR).
	+
	+	### Files created / modified
	+
	+	\| File \| Change \|
	+	\|------\|--------\|
	+	\| `ingest/source_validator.py` \| Added `STATE_LEG_HOSTS` + `_is_state_leg_pattern()` + updated `is_allowlisted()` \|
	+	\| `ingest/pollers/openstates.py` \| Full rewrite - was a stub; now fetches bills + votes + sponsors for all 52 jurisdictions \|
	+	\| `ingest/pollers/legiscan.py` \| New - LegiScan API poller (completeness supplement) \|
	+	\| `ingest/state_coverage.py` \| New - honesty layer; computes + writes state_coverage rows \|
	+	\| `tests/test_state_coverage.py` \| New - 55 tests for coverage logic + state-host allowlist \|
	+
	+	---
	+
	+	## Running
	+
	+	### Compute state coverage (works without any API keys)
	+
	+	```bash
	+	# On CT247 with the production DB
	+	cd /shared/projects/docket/services/ingest
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.state_coverage
	+
	+	# JSON output
	+	DOCKET_DB_PATH=/opt/docket/data/docket.db \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.state_coverage --json
	+	```
	+
	+	Expected output (no keys, empty state DB):
	+	```
	+	Jurisdictions tracked : 52 (50 states + DC + PR)
	+	Total coverage rows : 312
	+	ok (fresh) : 0
	+	stale (>48h) : 0
	+	missing (never seen) : 52
	+	```
	+
	+	### Run OpenStates poller (requires key)
	+
	+	```bash
	+	export DOCKET_OPENSTATES_KEY=<your_key>
	+	# MVP subset (12 jurisdictions - conserves 500/day free quota)
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.run --source openstates
	+
	+	# All 52 jurisdictions (needs paid tier)
	+	DOCKET_OPENSTATES_ALL=1 \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.run --source openstates
	+	```
	+
	+	### Run LegiScan poller (requires key)
	+
	+	```bash
	+	export DOCKET_LEGISCAN_KEY=<your_key>
	+	# MVP subset (12 states - ~200 API calls)
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.run --source legiscan
	+
	+	# All 52 jurisdictions (needs ~2k API calls - monitor monthly quota)
	+	DOCKET_LEGISCAN_ALL=1 \
	+	/opt/docket/services/ingest/.venv/bin/python -m ingest.run --source legiscan
	+	```
	+
	+	### Run tests
	+
	+	```bash
	+	cd /shared/projects/docket/services/ingest
	+	/opt/docket/services/ingest/.venv/bin/python -m pytest tests/ -v
	+	# Expected: 190 passed
	+	```
	+
	+	---
	+
	+	## Getting API Keys (free)
	+
	+	### OpenStates
	+	- URL: https://openstates.org/accounts/login/
	+	- Free tier: 500 requests/day, 10 requests/second
	+	- Env var: `DOCKET_OPENSTATES_KEY`
	+	- Caveat: 500 req/day is too tight for all 52 jurisdictions. MVP polls 12
	+	jurisdictions (the 10 most-populated states + DC + PR) to stay within quota.
	+	Production requires Bronze (5k/day) or Silver (50k/day) - contact
	+	openstates@openstates.org for pricing.
	+
	+	### LegiScan
	+	- URL: https://legiscan.com/legiscan
	+	- Free tier: ~30,000 requests/month (~1,000/day)
	+	- Env var: `DOCKET_LEGISCAN_KEY`
	+	- Efficiency: The `getMasterListRaw` sync strategy minimises API calls -
	+	only bills whose `change_hash` changed are fetched via `getBill`. Cap:
	+	`MAX_BILL_FETCH_PER_STATE=50` per state per run (configurable in code).
	+
	+	---
	+
	+	## Source URL Allowlist Extensions
	+
	+	State legislatures that don't use a `.gov` TLD require explicit allowlisting.
	+	The following were added to `source_validator.py`:
	+
	+	### STATE_LEG_HOSTS (explicit non-.gov hosts)
	+
	+	\| State \| Domain \|
	+	\|-------\|--------\|
	+	\| Pennsylvania \| `www.legis.state.pa.us` \|
	+	\| Texas \| `capitol.state.tx.us`, `legis.state.tx.us` \|
	+	\| Minnesota \| `www.leg.state.mn.us`, `www.house.leg.state.mn.us`, `www.senate.leg.state.mn.us` \|
	+	\| Mississippi \| `billstatus.ls.state.ms.us`, `ls.state.ms.us` \|
	+	\| Nevada \| `www.leg.state.nv.us` \|
	+	\| New Hampshire \| `www.gencourt.state.nh.us` \|
	+	\| New Jersey \| `www.njleg.state.nj.us` \|
	+	\| Alabama \| `alison.legislature.state.al.us` \|
	+	\| Alaska \| `www.legis.state.ak.us` \|
	+	\| Arkansas \| `www.arkleg.state.ar.us` \|
	+	\| Kansas \| `www.kslegislature.org` ← unique .org \|
	+	\| Puerto Rico \| `www.oslpr.org`, `www.camaraderepresentantes.org`, `www.senadopr.us` \|
	+	\| DC Council \| `dccouncil.us` \|
	+
	+	### Pattern rule: `*.state.<XX>.us`
	+
	+	Any host matching `<anything>.state.<2-letter-code>.us` is automatically
	+	allowlisted via `_is_state_leg_pattern()`. This covers:
	+	- `legis.state.pa.us` (Pennsylvania)
	+	- `njleg.state.nj.us` (New Jersey)
	+	- `billstatus.ls.state.ms.us` (Mississippi)
	+	- Any new state portal added at a `*.state.XX.us` subdomain without code change
	+
	+	This pattern does NOT match `govtrack.us`, `example.us`, or similar non-government `.us` domains.
	+
	+	---
	+
	+	## Architecture
	+
	+	### state_coverage table
	+
	+	```sql
	+	CREATE TABLE IF NOT EXISTS state_coverage (
	+	state TEXT NOT NULL,
	+	source TEXT NOT NULL, -- 'openstates' or 'legiscan'
	+	data_type TEXT NOT NULL, -- 'bills', 'votes', 'sponsors'
	+	last_scrape TEXT, -- ISO UTC timestamp of last successful scrape
	+	confidence TEXT NOT NULL DEFAULT 'missing', -- 'ok' \| 'stale' \| 'missing'
	+	PRIMARY KEY (state, source, data_type)
	+	);
	+	```
	+
	+	### scrape_log table
	+
	+	```sql
	+	CREATE TABLE IF NOT EXISTS scrape_log (
	+	source TEXT NOT NULL,
	+	state TEXT NOT NULL,
	+	data_type TEXT NOT NULL,
	+	scraped_at TEXT NOT NULL, -- ISO UTC timestamp
	+	row_count INTEGER NOT NULL DEFAULT 0,
	+	PRIMARY KEY (source, state, data_type)
	+	);
	+	```
	+
	+	### Confidence logic
	+
	+	- `ok`: `scraped_at` within the last 48 hours
	+	- `stale`: `scraped_at` exists but older than 48 hours
	+	- `missing`: no scrape record for this (state, source, data_type) combination
	+
	+	### Poller → coverage flow
	+
	+	```
	+	openstates.poll() / legiscan.poll()
	+	└─ for each jurisdiction:
	+	├─ fetch + store bills + votes
	+	└─ log_scrape(conn, source, state, 'bills', count)
	+	log_scrape(conn, source, state, 'votes', count)
	+	log_scrape(conn, source, state, 'sponsors', count)
	+
	+	state_coverage.compute()
	+	├─ reads scrape_log
	+	├─ fallback: infers from raw_events (OCD/legiscan external_id parsing)
	+	└─ writes 312 rows to state_coverage (52 × 2 × 3)
	+	```
	+
	+	### API surface (GET /api/states)
	+
	+	The `state_coverage` table directly backs the `/api/states` endpoint:
	+	```json
	+	{
	+	"code": "CA",
	+	"name": "California",
	+	"confidence": "ok",
	+	"lastScrape": "2026-06-26T12:00:00Z",
	+	"missingTypes": []
	+	}
	+	```
	+
	+	---
	+
	+	## Known Gaps (surface in-product - never hide)
	+
	+	1. Free quota insufficient for prod: OpenStates 500 req/day covers ~12 states/day;
	+	LegiScan 30k/month covers all states if running daily (tight).
	+	2. Historical coverage: pollers default to `days_back=1`. First run on a state
	+	won't have historical bill history - only recent changes.
	+	3. Worst-lagging states: MA, NE, KY, AL historically 24-48h behind in both
	+	OpenStates and LegiScan (known in DATA_SOURCES.md).
	+	4. Ballot measures / interim committees: poorly covered by both APIs.
	+	5. Amendment text: LegiScan has better amendment completeness than OpenStates,
	+	but pre-2021 coverage is spotty in both.
	+	6. Vote positions (`vote_positions` table): not yet populated - yea/nay counts
	+	only. Individual member positions require `getVote` (LegiScan) per vote - costs
	+	extra API calls. Phase 2 work.

services/ingest/check_db_tmp.py +29 -0

		@@ -0,0 +1,29 @@
	+	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 +38 -0

		@@ -0,0 +1,38 @@
	+	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 +53 -0

		@@ -0,0 +1,53 @@
	+	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 +38 -0

		@@ -0,0 +1,38 @@
	+	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 +322 -0

		@@ -0,0 +1,322 @@
	+	"""
	+	Amendments poller - populate the `amendments` table.
	+
	+	Data strategy (bulk / no key):
	+	1. Primary: parse amendment references from BILLSTATUS XML already stored
	+	in raw_events (where the bills_bulk poller stored the full bill payload).
	+	2. Supplementary: attempt to fetch the amendment list from Congress.gov
	+	bulk URLs if DOCKET_CONGRESS_KEY is set (gracefully skips if not).
	+	3. Honest gaps:
	+	- Senate amendment TEXT: machine-readable only for 117th Congress (2021)+.
	+	- House amendments: text lives in Congressional Record, not discrete objects.
	+	- Manager's / en-bloc packages: not public as discrete objects.
	+	These are recorded as text=NULL with appropriate agreed_method.
	+
	+	Schema (created by bills_bulk.ensure_schema):
	+	amendments (id, congress, type, number, bill_id, sponsor_id, status,
	+	agreed_method, text, source_url)
	+
	+	agreed_method values:
	+	'recorded' - roll-call vote (individual record exists)
	+	'voice' - voice vote (no individual record)
	+	'uc' - unanimous consent
	+	NULL - unknown / not public
	+	"""
	+
	+	import argparse
	+	import json
	+	import logging
	+	import os
	+	import sqlite3
	+	import sys
	+	import xml.etree.ElementTree as ET
	+	from datetime import datetime, timezone
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.bills_bulk import ensure_schema, SOURCE as BILL_SOURCE
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "govinfo_amendments"
	+	CONGRESS_GOV_BASE = "https://www.congress.gov"
	+
	+	# ---------------------------------------------------------------------------
	+	# Helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _congress_amdt_url(congress: int, amdt_type: str, number: str) -> str:
	+	"""Return the congress.gov URL for an amendment."""
	+	slug_map = {
	+	"HAMDT": "house-amendment",
	+	"SAMDT": "senate-amendment",
	+	}
	+	slug = slug_map.get(amdt_type.upper(), amdt_type.lower())
	+	return f"{CONGRESS_GOV_BASE}/amendment/{congress}/{slug}/{number}"
	+
	+
	+	def _txt(el: Optional[ET.Element], tag: str, default: str = "") -> str:
	+	if el is None:
	+	return default
	+	child = el.find(tag)
	+	return (child.text or "").strip() if child is not None and child.text else default
	+
	+
	+	def _classify_agreed_method(description: str) -> Optional[str]:
	+	"""
	+	Classify how an amendment was agreed to, based on action text.
	+	Returns 'recorded', 'voice', 'uc', or None.
	+	"""
	+	d = (description or "").lower()
	+	if any(k in d for k in ("roll call", "yeas and nays", "recorded vote")):
	+	return "recorded"
	+	if "voice vote" in d:
	+	return "voice"
	+	if "unanimous consent" in d or "u.c." in d:
	+	return "uc"
	+	return None
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Parse amendments from BILLSTATUS raw payload
	+	# ---------------------------------------------------------------------------
	+
	+	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.
	+	"""
	+	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 = ?
	+	""",
	+	(BILL_SOURCE, congress),
	+	).fetchall()
	+
	+	stored = 0
	+	for row in rows:
	+	try:
	+	payload = json.loads(row["payload"])
	+	except (json.JSONDecodeError, TypeError):
	+	continue
	+
	+	# 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')
	+	amdt_actions = conn.execute(
	+	"""
	+	SELECT action_date, text, action_type, chamber
	+	FROM actions
	+	WHERE bill_id = ? AND (
	+	LOWER(action_type) LIKE '%amendment%'
	+	OR LOWER(text) LIKE '%amendment%'
	+	)
	+	ORDER BY action_date
	+	""",
	+	(bill_id,),
	+	).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)
	+	try:
	+	validate_source_url(source_url)
	+	except Exception:
	+	continue
	+	agreed = _classify_agreed_method(atext)
	+	conn.execute(
	+	"""
	+	INSERT INTO amendments
	+	(congress, type, number, bill_id, status,
	+	agreed_method, text, source_url)
	+	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),
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	congress, amdt_type, amdt_num, bill_id,
	+	"enacted" if "enacted" in atext.lower() else "proposed",
	+	agreed, None, source_url,
	+	),
	+	)
	+	stored += 1
	+
	+	conn.commit()
	+	return stored
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Congress.gov API amendments (key-gated, graceful skip)
	+	# ---------------------------------------------------------------------------
	+
	+	def _ingest_from_congress_api(
	+	conn: sqlite3.Connection,
	+	congress: int,
	+	key: str,
	+	) -> int:
	+	"""
	+	Pull amendments from the Congress.gov API (requires key).
	+	Only called when DOCKET_CONGRESS_KEY is set.
	+	Covers HAMDT and SAMDT for the given congress.
	+	"""
	+	BASE = "https://api.congress.gov/v3"
	+	stored = 0
	+
	+	for amdt_type in ("hamdt", "samdt"):
	+	url = f"{BASE}/amendment/{congress}/{amdt_type}"
	+	params = {
	+	"api_key": key,
	+	"limit": 250,
	+	"sort": "updateDate+desc",
	+	}
	+	page = 0
	+	while True:
	+	params["offset"] = page * 250
	+	try:
	+	r = requests.get(url, params=params, timeout=30)
	+	if r.status_code == 403:
	+	logger.warning("Congress.gov: 403 for %s amendments", amdt_type)
	+	break
	+	r.raise_for_status()
	+	except requests.RequestException as exc:
	+	logger.error("Congress.gov amendments fetch failed: %s", exc)
	+	break
	+
	+	data = r.json()
	+	items = data.get("amendments", [])
	+	if not items:
	+	break
	+
	+	for item in items:
	+	num = str(item.get("number", ""))
	+	sponsor = (item.get("sponsors") or [{}])[0].get("bioguideId", "")
	+	status = item.get("latestAction", {}).get("text", "")
	+	agreed = _classify_agreed_method(status)
	+	# Try to find the linked bill
	+	bill_id = None
	+	amended = item.get("amendedBill", {})
	+	if amended:
	+	b_congress = amended.get("congress", congress)
	+	b_type = (amended.get("type") or "").lower()
	+	b_num = str(amended.get("number", ""))
	+	if b_type and b_num:
	+	brow = conn.execute(
	+	"SELECT id FROM bills WHERE source=? AND external_id=?",
	+	(BILL_SOURCE, f"{b_congress}-{b_type}-{b_num}"),
	+	).fetchone()
	+	if brow:
	+	bill_id = brow["id"]
	+
	+	source_url = _congress_amdt_url(
	+	congress, amdt_type.upper(), num
	+	)
	+	try:
	+	validate_source_url(source_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, bill_id),
	+	sponsor_id = excluded.sponsor_id,
	+	status = excluded.status,
	+	agreed_method = COALESCE(excluded.agreed_method, agreed_method),
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	congress, amdt_type.upper(), num, bill_id,
	+	sponsor, status, agreed, None, source_url,
	+	),
	+	)
	+	stored += 1
	+
	+	if len(items) < 250:
	+	break
	+	page += 1
	+
	+	conn.commit()
	+	return stored
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Main poll
	+	# ---------------------------------------------------------------------------
	+
	+	def poll(congress: int = 119, db_path: Optional[str] = None) -> int:
	+	"""
	+	Populate the amendments table.
	+	Returns total number of rows stored (new + updated).
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	total = 0
	+
	+	# Path 1: parse from already-stored bill action rows
	+	n = _ingest_from_billstatus(conn, congress)
	+	logger.info("amendments: %d rows from BILLSTATUS action parsing", n)
	+	total += n
	+
	+	# Path 2: Congress.gov API (if key present)
	+	key = os.environ.get("DOCKET_CONGRESS_KEY", "")
	+	if key:
	+	n = _ingest_from_congress_api(conn, congress, key)
	+	logger.info("amendments: %d rows from Congress.gov API", n)
	+	total += n
	+	else:
	+	logger.info("amendments: DOCKET_CONGRESS_KEY not set - skipping API path")
	+
	+	conn.close()
	+	logger.info("amendments: %d total rows stored", total)
	+	return total
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	ap = argparse.ArgumentParser(description="Amendments ingest")
	+	ap.add_argument("--once", action="store_true")
	+	ap.add_argument("--congress", type=int, default=119)
	+	args = ap.parse_args()
	+
	+	n = poll(congress=args.congress)
	+	print(f"Done: {n} amendment rows stored")

services/ingest/ingest/diff.py +534 -0

		@@ -0,0 +1,534 @@
	+	"""
	+	Bill text diff engine - the "what got slipped in" feature.
	+
	+	Given two consecutive bill versions, this module:
	+	1. Fetches their XML from GovInfo (no key needed via /content/pkg/).
	+	2. Parses <section>/<subsection> elements by identifier.
	+	3. Aligns old↔new sections by (a) id attribute, then (b) fuzzy header match.
	+	4. Diffs aligned text with difflib; classifies added / removed / amended.
	+	5. Writes bill_diffs rows.
	+
	+	Entry point:
	+	diff_bill(bill_id, conn) - diffs the consecutive version chain for a bill.
	+	diff_versions(bill_id, from_code, to_code, conn) - diffs two specific versions.
	+	"""
	+
	+	import difflib
	+	import logging
	+	import re
	+	import sqlite3
	+	from dataclasses import dataclass, field
	+	from typing import Dict, List, Optional, Tuple
	+
	+	import requests
	+
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	GOVINFO_PKG_BASE = "https://www.govinfo.gov/content/pkg"
	+
	+	# Ordered version chains (most to least common transitions)
	+	HOUSE_CHAIN = ["ih", "rh", "rh2", "eh", "eah", "eas", "enr"]
	+	SENATE_CHAIN = ["is", "rs", "pcs", "es", "eas", "enr"]
	+
	+	# Snippet length (chars) stored in bill_diffs.snippet
	+	SNIPPET_LEN = 500
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Data structures
	+	# ---------------------------------------------------------------------------
	+
	+	@dataclass
	+	class Section:
	+	"""One parsed section or subsection from bill XML."""
	+	sec_id: str # from id attribute (may be empty)
	+	enum: str # section number, e.g. "1." or "(a)"
	+	header: str # heading text
	+	text: str # full text content (flat, all nested text joined)
	+	depth: int = 0 # 0 = section, 1 = subsection, 2 = paragraph, …
	+
	+
	+	@dataclass
	+	class DiffChange:
	+	"""One classified section change."""
	+	section_id: str
	+	change: str # 'added' \| 'removed' \| 'amended'
	+	snippet: str
	+	rider_flag: bool = False
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# XML fetching
	+	# ---------------------------------------------------------------------------
	+
	+	def fetch_version_xml(text_ref: str) -> Optional[str]:
	+	"""
	+	Fetch the XML text for a bill version from GovInfo.
	+	text_ref is the URL stored in bill_versions.text_ref.
	+	Returns None on failure.
	+	"""
	+	try:
	+	r = requests.get(text_ref, timeout=60)
	+	if r.status_code == 200:
	+	return r.text
	+	logger.warning("fetch_version_xml: HTTP %s for %s", r.status_code, text_ref)
	+	except requests.RequestException as exc:
	+	logger.error("fetch_version_xml: %s - %s", text_ref, exc)
	+	return None
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# XML parsing
	+	# ---------------------------------------------------------------------------
	+
	+	# Namespace variants used in GovInfo bill XML
	+	_NS_PREFIXES = [
	+	"", # no namespace
	+	"{http://xml.house.gov/schemas/uslm/1.0}",
	+	"{http://schemas.gpo.gov/xml/uslm}",
	+	]
	+
	+
	+	def _strip_ns(tag: str) -> str:
	+	"""Remove XML namespace prefix from a tag name."""
	+	if tag.startswith("{"):
	+	return tag.split("}", 1)[1]
	+	return tag
	+
	+
	+	def _all_text(el) -> str:
	+	"""Recursively collect all text inside an element."""
	+	parts = []
	+	if el.text:
	+	parts.append(el.text.strip())
	+	for child in el:
	+	parts.append(_all_text(child))
	+	if child.tail:
	+	parts.append(child.tail.strip())
	+	return " ".join(p for p in parts if p)
	+
	+
	+	def _parse_section_el(el, depth: int = 0) -> Section:
	+	"""Parse a single <section> or <subsection> element into a Section."""
	+	sec_id = el.get("id", "") or el.get("identifier", "")
	+	enum = ""
	+	header = ""
	+	text_parts: List[str] = []
	+
	+	for child in el:
	+	tag = _strip_ns(child.tag)
	+	if tag == "enum":
	+	enum = (child.text or "").strip()
	+	elif tag == "header":
	+	header = _all_text(child)
	+	elif tag in ("text", "chapeau", "continuation"):
	+	text_parts.append(_all_text(child))
	+	elif tag in ("subsection", "paragraph", "subparagraph", "clause"):
	+	# Recurse but flatten into parent text for diffing
	+	sub = _parse_section_el(child, depth + 1)
	+	text_parts.append(f"({sub.enum}) {sub.header} {sub.text}".strip())
	+
	+	return Section(
	+	sec_id=sec_id,
	+	enum=enum,
	+	header=header,
	+	text=" ".join(text_parts),
	+	depth=depth,
	+	)
	+
	+
	+	def _find_sections(root) -> List:
	+	"""Find all <section> elements regardless of namespace."""
	+	sections = []
	+	# Try namespace variants
	+	for ns in _NS_PREFIXES:
	+	els = root.findall(f".//{ns}section")
	+	if els:
	+	sections = els
	+	break
	+	# Also try direct bill/body/section
	+	if not sections:
	+	for ns in _NS_PREFIXES:
	+	els = root.findall(f".//{ns}body/{ns}section")
	+	if els:
	+	sections = els
	+	break
	+	return sections
	+
	+
	+	def parse_sections(xml_text: str) -> List[Section]:
	+	"""
	+	Parse a GovInfo bill XML into a list of Section objects.
	+	Returns an empty list if parsing fails.
	+	"""
	+	import xml.etree.ElementTree as ET
	+	try:
	+	root = ET.fromstring(xml_text)
	+	except ET.ParseError as exc:
	+	logger.warning("parse_sections: XML parse error: %s", exc)
	+	return []
	+
	+	section_els = _find_sections(root)
	+	if not section_els:
	+	logger.debug("parse_sections: no <section> elements found")
	+	return []
	+
	+	return [_parse_section_el(el) for el in section_els]
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Section alignment
	+	# ---------------------------------------------------------------------------
	+
	+	def _fuzzy_score(a: str, b: str) -> float:
	+	"""SequenceMatcher ratio for two strings (0-1)."""
	+	return difflib.SequenceMatcher(None, a.lower(), b.lower()).ratio()
	+
	+
	+	def align_sections(
	+	old: List[Section],
	+	new: List[Section],
	+	) -> List[Tuple[Optional[Section], Optional[Section]]]:
	+	"""
	+	Align old and new section lists.
	+
	+	Strategy:
	+	1. Match by id attribute (exact).
	+	2. For unmatched sections, fuzzy-match by (enum + header) with threshold 0.6.
	+	3. Remaining new sections are added; remaining old sections are removed.
	+
	+	Returns a list of (old_section \| None, new_section \| None) pairs.
	+	"""
	+	pairs: List[Tuple[Optional[Section], Optional[Section]]] = []
	+
	+	old_by_id: Dict[str, Section] = {s.sec_id: s for s in old if s.sec_id}
	+	new_by_id: Dict[str, Section] = {s.sec_id: s for s in new if s.sec_id}
	+
	+	matched_old: set = set()
	+	matched_new: set = set()
	+
	+	# Phase 1: exact id match
	+	for sid, new_sec in new_by_id.items():
	+	if sid in old_by_id:
	+	pairs.append((old_by_id[sid], new_sec))
	+	matched_old.add(id(old_by_id[sid]))
	+	matched_new.add(id(new_sec))
	+
	+	# Phase 2: fuzzy match on (enum + header)
	+	unmatched_old = [s for s in old if id(s) not in matched_old]
	+	unmatched_new = [s for s in new if id(s) not in matched_new]
	+
	+	used_old: set = set()
	+	used_new: set = set()
	+
	+	for n_sec in unmatched_new:
	+	n_key = f"{n_sec.enum} {n_sec.header}".strip()
	+	best_score = 0.0
	+	best_old = None
	+	for o_sec in unmatched_old:
	+	if id(o_sec) in used_old:
	+	continue
	+	o_key = f"{o_sec.enum} {o_sec.header}".strip()
	+	score = _fuzzy_score(o_key, n_key)
	+	if score > best_score:
	+	best_score = score
	+	best_old = o_sec
	+	if best_old is not None and best_score >= 0.6:
	+	pairs.append((best_old, n_sec))
	+	used_old.add(id(best_old))
	+	used_new.add(id(n_sec))
	+
	+	# Phase 3: leftovers
	+	for o_sec in unmatched_old:
	+	if id(o_sec) not in used_old:
	+	pairs.append((o_sec, None)) # removed
	+
	+	for n_sec in unmatched_new:
	+	if id(n_sec) not in used_new:
	+	pairs.append((None, n_sec)) # added
	+
	+	return pairs
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Diff classification
	+	# ---------------------------------------------------------------------------
	+
	+	def classify_change(old_text: Optional[str], new_text: Optional[str]) -> Optional[str]:
	+	"""
	+	Classify the change between two section texts.
	+
	+	Returns 'added', 'removed', or 'amended', or None if no change.
	+	"""
	+	old = (old_text or "").strip()
	+	new = (new_text or "").strip()
	+
	+	if not old and new:
	+	return "added"
	+	if old and not new:
	+	return "removed"
	+	if old == new:
	+	return None
	+	return "amended"
	+
	+
	+	def _make_snippet(old_text: Optional[str], new_text: Optional[str], change: str) -> str:
	+	"""
	+	Build a concise snippet showing what changed.
	+	For 'amended': show a unified-diff excerpt.
	+	For 'added'/'removed': truncate the text.
	+	"""
	+	if change == "added":
	+	text = (new_text or "")
	+	return text[:SNIPPET_LEN] + ("…" if len(text) > SNIPPET_LEN else "")
	+	if change == "removed":
	+	text = (old_text or "")
	+	return text[:SNIPPET_LEN] + ("…" if len(text) > SNIPPET_LEN else "")
	+	# amended: show diff lines
	+	old_lines = (old_text or "").splitlines(keepends=True)
	+	new_lines = (new_text or "").splitlines(keepends=True)
	+	diff = list(difflib.unified_diff(old_lines, new_lines, n=2, lineterm=""))
	+	snippet = "".join(diff[:30]) # first 30 diff lines
	+	return snippet[:SNIPPET_LEN] + ("…" if len(snippet) > SNIPPET_LEN else "")
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# DB writes
	+	# ---------------------------------------------------------------------------
	+
	+	def _store_diff(
	+	conn: sqlite3.Connection,
	+	bill_id: int,
	+	from_version: str,
	+	to_version: str,
	+	section_id: str,
	+	change: str,
	+	snippet: str,
	+	rider_flag: bool,
	+	source_url: str,
	+	) -> int:
	+	"""Insert a bill_diffs row. Returns rowid."""
	+	cur = conn.execute(
	+	"""
	+	INSERT INTO bill_diffs
	+	(bill_id, from_version, to_version, section_id,
	+	change, snippet, rider_flag, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(
	+	bill_id, from_version, to_version, section_id,
	+	change, snippet, 1 if rider_flag else 0, source_url,
	+	),
	+	)
	+	return cur.lastrowid
	+
	+
	+	def _clear_diffs(conn: sqlite3.Connection, bill_id: int,
	+	from_ver: str, to_ver: str) -> None:
	+	"""Remove existing diff rows for this pair (idempotent re-run)."""
	+	conn.execute(
	+	"DELETE FROM bill_diffs WHERE bill_id=? AND from_version=? AND to_version=?",
	+	(bill_id, from_ver, to_ver),
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# High-level diff functions
	+	# ---------------------------------------------------------------------------
	+
	+	def diff_versions(
	+	bill_id: int,
	+	from_code: str,
	+	to_code: str,
	+	conn: sqlite3.Connection,
	+	source_url: str,
	+	) -> List[DiffChange]:
	+	"""
	+	Diff two specific bill versions. Fetches XML, parses, aligns, classifies.
	+	Writes results to bill_diffs and returns the list of DiffChange objects.
	+	"""
	+	def _get_text_ref(vcode: str) -> Optional[str]:
	+	row = conn.execute(
	+	"SELECT text_ref FROM bill_versions WHERE bill_id=? AND version_code=?",
	+	(bill_id, vcode),
	+	).fetchone()
	+	return row["text_ref"] if row else None
	+
	+	from_ref = _get_text_ref(from_code)
	+	to_ref = _get_text_ref(to_code)
	+
	+	if not from_ref or not to_ref:
	+	logger.warning("diff_versions: missing text_ref for bill_id=%s %s→%s",
	+	bill_id, from_code, to_code)
	+	return []
	+
	+	logger.info("Fetching %s → %s for bill_id=%s", from_code, to_code, bill_id)
	+	old_xml = fetch_version_xml(from_ref)
	+	new_xml = fetch_version_xml(to_ref)
	+
	+	if not old_xml or not new_xml:
	+	logger.warning("diff_versions: could not fetch XML for bill_id=%s", bill_id)
	+	return []
	+
	+	old_sections = parse_sections(old_xml)
	+	new_sections = parse_sections(new_xml)
	+
	+	if not old_sections and not new_sections:
	+	logger.warning("diff_versions: no sections parsed for bill_id=%s", bill_id)
	+	return []
	+
	+	pairs = align_sections(old_sections, new_sections)
	+
	+	_clear_diffs(conn, bill_id, from_code, to_code)
	+
	+	changes: List[DiffChange] = []
	+	for old_sec, new_sec in pairs:
	+	old_text = old_sec.text if old_sec else None
	+	new_text = new_sec.text if new_sec else None
	+	change = classify_change(old_text, new_text)
	+	if not change:
	+	continue
	+
	+	sec_id = (new_sec or old_sec).sec_id or \
	+	f"{(new_sec or old_sec).enum}_{(new_sec or old_sec).header[:20]}"
	+
	+	snippet = _make_snippet(old_text, new_text, change)
	+	dc = DiffChange(section_id=sec_id, change=change, snippet=snippet)
	+	_store_diff(conn, bill_id, from_code, to_code, sec_id,
	+	change, snippet, False, source_url)
	+	changes.append(dc)
	+
	+	logger.info("diff_versions: bill_id=%s %s→%s → %d changes",
	+	bill_id, from_code, to_code, len(changes))
	+	return changes
	+
	+
	+	def diff_bill(bill_id: int, conn: sqlite3.Connection) -> int:
	+	"""
	+	Diff the consecutive version chain for a bill.
	+	Determines the chain order, diffs each consecutive pair.
	+	Returns total number of diff rows written.
	+	"""
	+	# Load versions
	+	rows = conn.execute(
	+	"SELECT version_code, text_ref, source_url FROM bill_versions WHERE bill_id=? ORDER BY version_date",
	+	(bill_id,),
	+	).fetchall()
	+
	+	if len(rows) < 2:
	+	logger.info("diff_bill: bill_id=%s has fewer than 2 versions, skipping", bill_id)
	+	return 0
	+
	+	# Determine bill type to pick chain
	+	be_row = conn.execute(
	+	"SELECT bill_type FROM bills_ext WHERE bill_id=?", (bill_id,)
	+	).fetchone()
	+	btype = (be_row["bill_type"] or "").lower() if be_row else ""
	+
	+	if btype in ("s", "sres", "sjres", "sconres"):
	+	chain_order = SENATE_CHAIN
	+	else:
	+	chain_order = HOUSE_CHAIN
	+
	+	# Cross-chamber codes appear in both House and Senate chains
	+	CROSS_CHAIN = ["pcs", "pch", "rds", "rih", "ris"]
	+
	+	def _rank(vcode: str) -> int:
	+	# Known chain position
	+	try:
	+	return chain_order.index(vcode)
	+	except ValueError:
	+	pass
	+	# Cross-chamber: place after the typical penultimate engrossed position
	+	if vcode in CROSS_CHAIN:
	+	return 5 # after eh/es, before enr
	+	# Public law codes go last
	+	if vcode.startswith("-") or "publ" in vcode.lower():
	+	return 998
	+	return 999
	+
	+	def _sort_key(row) -> tuple:
	+	"""Sort by (chain_rank, version_date).
	+	Versions without a date are placed within their rank group at the end."""
	+	vdate = row["version_code"] # use version_code as fallback
	+	vdate_raw = row["version_date"] if "version_date" in row.keys() else ""
	+	return (_rank(row["version_code"]), vdate_raw or "9999")
	+
	+	# Load version_date into the rows
	+	rows_with_date = conn.execute(
	+	"""SELECT version_code, version_date, text_ref
	+	FROM bill_versions WHERE bill_id=? ORDER BY version_date""",
	+	(bill_id,),
	+	).fetchall()
	+
	+	version_rows = sorted(rows_with_date, key=_sort_key)
	+
	+	# Get source_url from the bills table
	+	bill_row = conn.execute(
	+	"SELECT source_url FROM bills WHERE id=?", (bill_id,)
	+	).fetchone()
	+	source_url = bill_row["source_url"] if bill_row else "https://www.congress.gov"
	+
	+	total = 0
	+	for i in range(len(version_rows) - 1):
	+	from_v = version_rows[i]["version_code"]
	+	to_v = version_rows[i + 1]["version_code"]
	+	changes = diff_versions(bill_id, from_v, to_v, conn, source_url)
	+	conn.commit()
	+	total += len(changes)
	+
	+	return total
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	if __name__ == "__main__":
	+	import argparse
	+	import sys
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.bills_bulk import ensure_schema
	+
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	ap = argparse.ArgumentParser(description="Bill diff engine")
	+	ap.add_argument("--once", action="store_true")
	+	ap.add_argument("--bill-id", type=int, help="Diff a specific bill by DB id")
	+	ap.add_argument("--all", action="store_true",
	+	help="Diff all bills with 2+ versions")
	+	ap.add_argument("--congress", type=int, default=119)
	+	args = ap.parse_args()
	+
	+	conn = get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	if args.bill_id:
	+	n = diff_bill(args.bill_id, conn)
	+	print(f"bill_id={args.bill_id}: {n} diff rows written")
	+	elif args.all:
	+	# Find bills with 2+ versions
	+	rows = conn.execute(
	+	"""
	+	SELECT DISTINCT bv.bill_id
	+	FROM bill_versions bv
	+	JOIN bills b ON b.id = bv.bill_id
	+	WHERE b.congress = ?
	+	GROUP BY bv.bill_id HAVING COUNT(*) >= 2
	+	""",
	+	(args.congress,),
	+	).fetchall()
	+	total = 0
	+	for row in rows:
	+	total += diff_bill(row["bill_id"], conn)
	+	print(f"Total diff rows: {total} across {len(rows)} bills")
	+	else:
	+	ap.print_help()
	+
	+	conn.close()

services/ingest/ingest/pollers/bill_versions.py +199 -0

		@@ -0,0 +1,199 @@
	+	"""
	+	GovInfo BILLS bulk poller - supplementary text-version discovery.
	+
	+	For bills already in the DB that have fewer than the expected versions,
	+	this poller probes GovInfo's /content/pkg/ content path (no key) for
	+	common version codes and stores any new bill_versions rows.
	+
	+	Version code order (House): ih → rh → eh → eas → enr
	+	Version code order (Senate): is → rs → es → eas → enr
	+	Additional common codes: pcs pch ph ps pp eah rih ris
	+
	+	Bulk text URL pattern (no key):
	+	https://www.govinfo.gov/content/pkg/BILLS-{congress}{type}{number}{version}/xml/BILLS-...xml
	+	"""
	+
	+	import argparse
	+	import logging
	+	import sys
	+	from datetime import datetime, timezone
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.bills_bulk import ensure_schema, BILL_TYPE_ABBR
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "govinfo_bills"
	+	GOVINFO_PKG_BASE = "https://www.govinfo.gov/content/pkg"
	+
	+	# Ordered version chains: try these in order for each bill
	+	HOUSE_VERSION_CHAIN = ["ih", "rh", "rh2", "eh", "eah", "eas", "enr"]
	+	SENATE_VERSION_CHAIN = ["is", "rs", "pcs", "es", "eas", "enr"]
	+	EXTRA_CODES = ["pch", "ph", "ps", "pp", "rih", "ris", "ath", "ats"]
	+
	+	ALL_VERSION_CODES = list(dict.fromkeys(
	+	HOUSE_VERSION_CHAIN + SENATE_VERSION_CHAIN + EXTRA_CODES
	+	))
	+
	+	# How many HEAD requests to send per bill max (cap to avoid hammering GovInfo)
	+	MAX_PROBES_PER_BILL = 12
	+
	+
	+	def _text_ref(congress: int, bill_type: str, number: str, vcode: str) -> str:
	+	pkg = f"BILLS-{congress}{bill_type}{number}{vcode}"
	+	return f"{GOVINFO_PKG_BASE}/{pkg}/xml/{pkg}.xml"
	+
	+
	+	def _head_ok(url: str, session: requests.Session) -> bool:
	+	"""Return True if GovInfo returns 200 for a HEAD request."""
	+	try:
	+	r = session.head(url, timeout=15, allow_redirects=True)
	+	return r.status_code == 200
	+	except requests.RequestException:
	+	return False
	+
	+
	+	def _bills_for_congress(conn, congress: int) -> list:
	+	"""Return list of (id, number, bill_type) for bills of this congress."""
	+	rows = 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.congress = ? AND b.source = 'govinfo_billstatus'
	+	""",
	+	(congress,),
	+	).fetchall()
	+	return rows
	+
	+
	+	def _existing_vcodes(conn, bill_id: int) -> set:
	+	rows = conn.execute(
	+	"SELECT version_code FROM bill_versions WHERE bill_id=?",
	+	(bill_id,),
	+	).fetchall()
	+	return {r["version_code"] for r in rows}
	+
	+
	+	def _store_version(
	+	conn,
	+	bill_id: int,
	+	vcode: str,
	+	pkg_id: str,
	+	text_ref: str,
	+	source_url: str,
	+	) -> None:
	+	conn.execute(
	+	"""
	+	INSERT INTO bill_versions
	+	(bill_id, version_code, govinfo_package_id, version_date,
	+	fetched_at, text_ref, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(bill_id, version_code) DO NOTHING
	+	""",
	+	(
	+	bill_id, vcode, pkg_id, None,
	+	datetime.now(timezone.utc).isoformat(),
	+	text_ref, source_url,
	+	),
	+	)
	+
	+
	+	def poll(
	+	congress: int = 119,
	+	db_path: Optional[str] = None,
	+	max_bills: int = 200,
	+	) -> int:
	+	"""
	+	Probe GovInfo for bill text versions not yet in bill_versions.
	+
	+	Parameters
	+	----------
	+	congress : Target congress (default 119).
	+	db_path : Override SQLite path.
	+	max_bills : Cap on bills to probe per run (avoids runaway HTTP calls).
	+
	+	Returns
	+	-------
	+	Number of new version rows stored.
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	bills = _bills_for_congress(conn, congress)
	+	logger.info("bill_versions: %d bills in DB for congress %s", len(bills), congress)
	+
	+	session = requests.Session()
	+	session.headers["User-Agent"] = "Docket-ingest/1.0 (transparency research)"
	+
	+	new_rows = 0
	+	probed = 0
	+
	+	for row in bills[:max_bills]:
	+	bill_id = row["id"]
	+	raw_number = row["number"] or "" # e.g. "H.R. 1" or just "1"
	+	bill_type = (row["bill_type"] or "").lower()
	+	source_url = row["source_url"] or ""
	+
	+	# Extract numeric part from formatted number
	+	num = "".join(c for c in raw_number if c.isdigit())
	+	if not num or not bill_type:
	+	continue
	+
	+	existing = _existing_vcodes(conn, bill_id)
	+
	+	# Determine which chain to try based on bill type
	+	if bill_type in ("s", "sres", "sjres", "sconres"):
	+	chain = SENATE_VERSION_CHAIN + EXTRA_CODES
	+	else:
	+	chain = HOUSE_VERSION_CHAIN + EXTRA_CODES
	+
	+	probes = 0
	+	for vcode in chain:
	+	if vcode in existing:
	+	continue
	+	if probes >= MAX_PROBES_PER_BILL:
	+	break
	+
	+	url = _text_ref(congress, bill_type, num, vcode)
	+	probes += 1
	+
	+	if _head_ok(url, session):
	+	pkg_id = f"BILLS-{congress}{bill_type}{num}{vcode}"
	+	try:
	+	validate_source_url(url)
	+	except Exception:
	+	continue
	+	_store_version(conn, bill_id, vcode, pkg_id, url, url)
	+	existing.add(vcode)
	+	new_rows += 1
	+	logger.info("Found version %s for bill_id=%s (%s %s)",
	+	vcode, bill_id, bill_type.upper(), num)
	+
	+	probed += 1
	+
	+	conn.commit()
	+	conn.close()
	+	logger.info("bill_versions: %d new version rows (probed %d bills)", new_rows, probed)
	+	return new_rows
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	ap = argparse.ArgumentParser(description="GovInfo BILLS version discovery")
	+	ap.add_argument("--once", action="store_true")
	+	ap.add_argument("--congress", type=int, default=119)
	+	ap.add_argument("--max-bills", type=int, default=200)
	+	args = ap.parse_args()
	+
	+	n = poll(congress=args.congress, max_bills=args.max_bills)
	+	print(f"Done: {n} new version rows")

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

		@@ -0,0 +1,578 @@
	+	"""
	+	GovInfo BULKDATA BILLSTATUS poller - federal bills, status, sponsors, actions.
	+
	+	Enumerates bills via the GovInfo sitemap (no API key required), fetches
	+	individual BILLSTATUS XML files, and stores results in:
	+	bills, bills_ext, actions, bill_versions
	+
	+	Also creates bill_diffs and amendments tables (populated by diff.py /
	+	amendments.py later).
	+
	+	Sitemap URL (no key):
	+	https://www.govinfo.gov/sitemap/bulkdata/BILLSTATUS/{congress}{type}/sitemap.xml
	+
	+	Individual XML URL:
	+	https://www.govinfo.gov/bulkdata/BILLSTATUS/{congress}/{type}/BILLSTATUS-{congress}{type}{num}.xml
	+
	+	Bill types: hr s hjres sjres hconres sconres hres sres
	+	"""
	+
	+	import argparse
	+	import logging
	+	import sqlite3
	+	import sys
	+	import time
	+	import xml.etree.ElementTree as ET
	+	from datetime import datetime, timezone
	+	from typing import Iterator, List, Optional, Tuple
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "govinfo_billstatus"
	+	SITEMAP_BASE = "https://www.govinfo.gov/sitemap/bulkdata/BILLSTATUS"
	+	CONGRESS_GOV_BASE = "https://www.congress.gov/bill"
	+
	+	BILL_TYPES = ["hr", "s", "hjres", "sjres", "hconres", "sconres", "hres", "sres"]
	+
	+	BILL_TYPE_SLUG = {
	+	"hr": "house-bill",
	+	"s": "senate-bill",
	+	"hjres": "house-joint-resolution",
	+	"sjres": "senate-joint-resolution",
	+	"hconres": "house-concurrent-resolution",
	+	"sconres": "senate-concurrent-resolution",
	+	"hres": "house-resolution",
	+	"sres": "senate-resolution",
	+	}
	+
	+	BILL_TYPE_ABBR = {
	+	"hr": "H.R.",
	+	"s": "S.",
	+	"hjres": "H.J.Res.",
	+	"sjres": "S.J.Res.",
	+	"hconres": "H.Con.Res.",
	+	"sconres": "S.Con.Res.",
	+	"hres": "H.Res.",
	+	"sres": "S.Res.",
	+	}
	+
	+	VERSION_LABEL_MAP = {
	+	"Introduced in House": "ih",
	+	"Introduced in Senate": "is",
	+	"Reported in House": "rh",
	+	"Reported in Senate": "rs",
	+	"Passed House": "ph",
	+	"Passed Senate": "ps",
	+	"Engrossed in House": "eh",
	+	"Engrossed in Senate": "es",
	+	"Engrossed Amendment House": "eah",
	+	"Engrossed Amendment Senate": "eas",
	+	"Enrolled Bill": "enr",
	+	"Placed on Calendar House": "pch",
	+	"Placed on Calendar Senate": "pcs",
	+	"Public Print": "pp",
	+	"Received in the House": "rih",
	+	"Received in the Senate": "ris",
	+	"Amendment in the Nature of a Substitute": "ans",
	+	"Held at the Desk": "hds",
	+	"Referred in Senate": "ris",
	+	"Referred in House": "rih",
	+	}
	+
	+	# Sitemap XML namespace
	+	_SM_NS = "http://www.sitemaps.org/schemas/sitemap/0.9"
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Schema
	+	# ---------------------------------------------------------------------------
	+
	+	def ensure_schema(conn: sqlite3.Connection) -> None:
	+	"""Create all tables needed by the bills pipeline."""
	+	conn.executescript("""
	+	CREATE TABLE IF NOT EXISTS bills_ext (
	+	bill_id INTEGER PRIMARY KEY REFERENCES bills(id),
	+	bill_type TEXT,
	+	full_number TEXT,
	+	sponsor_name TEXT,
	+	sponsor_party TEXT,
	+	sponsor_state TEXT
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS bill_versions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bill_id INTEGER NOT NULL REFERENCES bills(id),
	+	version_code TEXT NOT NULL,
	+	govinfo_package_id TEXT NOT NULL,
	+	version_label TEXT,
	+	version_date TEXT,
	+	fetched_at TEXT,
	+	text_ref TEXT NOT NULL,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (bill_id, version_code)
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS bill_diffs (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	bill_id INTEGER NOT NULL REFERENCES bills(id),
	+	from_version TEXT NOT NULL,
	+	to_version TEXT NOT NULL,
	+	section_id TEXT,
	+	change TEXT NOT NULL CHECK(change IN ('added','removed','amended')),
	+	snippet TEXT,
	+	rider_flag INTEGER NOT NULL DEFAULT 0,
	+	source_url TEXT NOT NULL
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS amendments (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	congress INTEGER NOT NULL,
	+	type TEXT NOT NULL,
	+	number TEXT NOT NULL,
	+	bill_id INTEGER REFERENCES bills(id),
	+	sponsor_id TEXT,
	+	status TEXT,
	+	agreed_method TEXT,
	+	text TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (congress, type, number)
	+	);
	+	""")
	+	conn.commit()
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Helpers
	+	# ---------------------------------------------------------------------------
	+
	+	def _ordinal(n: int) -> str:
	+	if n == 1: return "1st"
	+	if n == 2: return "2nd"
	+	if n == 3: return "3rd"
	+	return f"{n}th"
	+
	+
	+	def _congress_gov_url(congress: int, bill_type: str, number: str) -> str:
	+	slug = BILL_TYPE_SLUG.get(bill_type.lower(), bill_type)
	+	return f"{CONGRESS_GOV_BASE}/{_ordinal(congress)}-congress/{slug}/{number}"
	+
	+
	+	def _txt(el: Optional[ET.Element], tag: str, default: str = "") -> str:
	+	if el is None:
	+	return default
	+	child = el.find(tag)
	+	if child is None or child.text is None:
	+	return default
	+	return child.text.strip()
	+
	+
	+	def _pkg_id_from_url(url: str) -> Optional[str]:
	+	"""Extract GovInfo package ID from a /content/pkg/ URL."""
	+	parts = url.split("/")
	+	try:
	+	return parts[parts.index("pkg") + 1]
	+	except (ValueError, IndexError):
	+	return None
	+
	+
	+	def _version_code_from_pkg(pkg_id: str) -> str:
	+	"""
	+	Extract the trailing version code from a package ID.
	+	BILLS-119hr1ih → ih \| BILLS-117s2007eas → eas
	+	"""
	+	s = pkg_id
	+	if s.upper().startswith("BILLS-"):
	+	s = s[6:]
	+	i = 0
	+	while i < len(s) and s[i].isdigit():
	+	i += 1
	+	while i < len(s) and s[i].isalpha():
	+	i += 1
	+	while i < len(s) and s[i].isdigit():
	+	i += 1
	+	return s[i:] if i < len(s) else "unk"
	+
	+
	+	def _label_to_code(label: str) -> str:
	+	return VERSION_LABEL_MAP.get(label, label.lower().replace(" ", "_")[:10])
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Sitemap fetch
	+	# ---------------------------------------------------------------------------
	+
	+	def _fetch_sitemap(congress: int, bill_type: str) -> List[Tuple[str, str]]:
	+	"""
	+	Fetch the GovInfo sitemap for this congress+bill_type.
	+	Returns list of (url, lastmod) pairs.
	+	"""
	+	sitemap_url = f"{SITEMAP_BASE}/{congress}{bill_type}/sitemap.xml"
	+	logger.info("Fetching sitemap: %s", sitemap_url)
	+	try:
	+	r = requests.get(sitemap_url, timeout=60)
	+	if r.status_code == 404:
	+	logger.warning("No sitemap for %s%s (404)", congress, bill_type)
	+	return []
	+	r.raise_for_status()
	+	except requests.RequestException as exc:
	+	logger.error("Sitemap fetch failed: %s", exc)
	+	return []
	+
	+	try:
	+	root = ET.fromstring(r.content)
	+	except ET.ParseError as exc:
	+	logger.error("Sitemap XML parse error: %s", exc)
	+	return []
	+
	+	entries = []
	+	for url_el in root.findall(f"{{{_SM_NS}}}url"):
	+	loc_el = url_el.find(f"{{{_SM_NS}}}loc")
	+	mod_el = url_el.find(f"{{{_SM_NS}}}lastmod")
	+	loc = loc_el.text.strip() if loc_el is not None and loc_el.text else ""
	+	lastmod = mod_el.text.strip() if mod_el is not None and mod_el.text else ""
	+	if loc:
	+	entries.append((loc, lastmod))
	+
	+	logger.info("Sitemap %s%s: %d entries", congress, bill_type, len(entries))
	+	return entries
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Individual XML fetch + parse
	+	# ---------------------------------------------------------------------------
	+
	+	def _fetch_bill_xml(url: str, session: requests.Session) -> Optional[bytes]:
	+	"""Fetch a single BILLSTATUS XML file. Returns None on error."""
	+	try:
	+	r = session.get(url, timeout=30)
	+	if r.status_code == 200:
	+	return r.content
	+	logger.debug("HTTP %s for %s", r.status_code, url)
	+	except requests.RequestException as exc:
	+	logger.warning("Fetch error %s: %s", url, exc)
	+	return None
	+
	+
	+	def _iter_bill_elements(xml_bytes: bytes) -> Iterator[ET.Element]:
	+	"""Yield the <bill> element(s) from a BILLSTATUS XML byte string."""
	+	try:
	+	root = ET.fromstring(xml_bytes)
	+	except ET.ParseError as exc:
	+	logger.warning("XML parse error: %s", exc)
	+	return
	+	# Root: <billStatus> or <BillStatus>
	+	bill = root.find("bill")
	+	if bill is not None:
	+	yield bill
	+	elif root.tag in ("bill", "Bill"):
	+	yield root
	+
	+
	+	def _store_bill(
	+	conn: sqlite3.Connection,
	+	bill_el: ET.Element,
	+	congress: int,
	+	bill_type: str,
	+	) -> Optional[int]:
	+	"""
	+	Parse a BILLSTATUS <bill> element and upsert rows into:
	+	bills, bills_ext, actions, bill_versions
	+	Returns bills.id on success, None on skip/error.
	+	"""
	+	number = _txt(bill_el, "number")
	+	title = (_txt(bill_el, "title") or _txt(bill_el, "officialTitle")).strip()
	+	updated = _txt(bill_el, "updateDate")
	+	# Prefer the canonical congress.gov URL embedded in the XML
	+	leg_url = _txt(bill_el, "legislationUrl")
	+
	+	if not number:
	+	return None
	+
	+	# Sponsor
	+	sponsor_bio = sponsor_name = sponsor_party = sponsor_state = ""
	+	sponsors_el = bill_el.find("sponsors")
	+	if sponsors_el is not None:
	+	first = sponsors_el.find("item")
	+	if first is not None:
	+	sponsor_bio = _txt(first, "bioguideId")
	+	sponsor_name = _txt(first, "fullName")
	+	sponsor_party = _txt(first, "party")
	+	sponsor_state = _txt(first, "state")
	+
	+	# Status from latestAction
	+	latest_el = bill_el.find("latestAction")
	+	status_text = _txt(latest_el, "text") if latest_el is not None else ""
	+
	+	# Policy area
	+	pa_el = bill_el.find("policyArea")
	+	policy_area = _txt(pa_el, "name") if pa_el is not None else ""
	+
	+	# Source URL: prefer embedded legislationUrl (already allowlisted), fall back
	+	source_url = leg_url or _congress_gov_url(congress, bill_type, number)
	+	try:
	+	validate_source_url(source_url)
	+	except Exception as exc:
	+	logger.error("Invalid source_url %s-%s-%s: %s", congress, bill_type, number, exc)
	+	return None
	+
	+	external_id = f"{congress}-{bill_type}-{number}"
	+
	+	# Upsert bills
	+	conn.execute(
	+	"""
	+	INSERT INTO bills
	+	(source, external_id, congress, number, title,
	+	sponsor_id, status, policy_area, updated_at, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	title = excluded.title,
	+	sponsor_id = excluded.sponsor_id,
	+	status = excluded.status,
	+	policy_area = excluded.policy_area,
	+	updated_at = excluded.updated_at,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	SOURCE, external_id, congress,
	+	f"{BILL_TYPE_ABBR.get(bill_type, bill_type.upper())} {number}",
	+	title, sponsor_bio or sponsor_name,
	+	status_text, policy_area, updated, source_url,
	+	),
	+	)
	+	row = conn.execute(
	+	"SELECT id FROM bills WHERE source=? AND external_id=?",
	+	(SOURCE, external_id),
	+	).fetchone()
	+	if row is None:
	+	return None
	+	bill_id = row["id"]
	+
	+	# bills_ext
	+	conn.execute(
	+	"""
	+	INSERT INTO bills_ext
	+	(bill_id, bill_type, full_number, sponsor_name, sponsor_party, sponsor_state)
	+	VALUES (?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(bill_id) DO UPDATE SET
	+	bill_type = excluded.bill_type,
	+	full_number = excluded.full_number,
	+	sponsor_name = excluded.sponsor_name,
	+	sponsor_party = excluded.sponsor_party,
	+	sponsor_state = excluded.sponsor_state
	+	""",
	+	(
	+	bill_id, bill_type.upper(),
	+	f"{BILL_TYPE_ABBR.get(bill_type, bill_type.upper())} {number}",
	+	sponsor_name, sponsor_party, sponsor_state,
	+	),
	+	)
	+
	+	# Actions
	+	actions_el = bill_el.find("actions")
	+	if actions_el is not None:
	+	for item in actions_el.findall("item"):
	+	adate = _txt(item, "actionDate")
	+	atext = _txt(item, "text")
	+	atype = _txt(item, "type") or "BillAction"
	+	chamber = ""
	+	ss_el = item.find("sourceSystem")
	+	if ss_el is not None:
	+	sname = _txt(ss_el, "name")
	+	if "House" in sname:
	+	chamber = "House"
	+	elif "Senate" in sname:
	+	chamber = "Senate"
	+	if adate and atext:
	+	conn.execute(
	+	"""
	+	INSERT INTO actions
	+	(bill_id, source, action_type, action_date,
	+	chamber, text, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(bill_id, SOURCE, atype, adate, chamber, atext, source_url),
	+	)
	+
	+	# Text versions → bill_versions
	+	tv_el = bill_el.find("textVersions")
	+	if tv_el is not None:
	+	for tv in tv_el.findall("item"):
	+	label = _txt(tv, "type")
	+	vdate = _txt(tv, "date")
	+
	+	xml_url = ""
	+	fmts_el = tv.find("formats")
	+	if fmts_el is not None:
	+	for fmt in fmts_el.findall("item"):
	+	if _txt(fmt, "type") == "XML":
	+	xml_url = _txt(fmt, "url")
	+	break
	+	if not xml_url:
	+	first_fmt = fmts_el.find("item")
	+	if first_fmt is not None:
	+	xml_url = _txt(first_fmt, "url")
	+
	+	if not xml_url:
	+	continue
	+
	+	pkg_id = _pkg_id_from_url(xml_url)
	+	if not pkg_id:
	+	vcode = _label_to_code(label)
	+	pkg_id = f"BILLS-{congress}{bill_type}{number}{vcode}"
	+	else:
	+	vcode = _version_code_from_pkg(pkg_id) or _label_to_code(label)
	+
	+	text_ref = xml_url
	+	if not text_ref.startswith("https://www.govinfo.gov"):
	+	text_ref = (
	+	f"https://www.govinfo.gov/content/pkg/{pkg_id}/xml/{pkg_id}.xml"
	+	)
	+	try:
	+	validate_source_url(text_ref)
	+	except Exception:
	+	text_ref = (
	+	f"https://www.govinfo.gov/content/pkg/{pkg_id}/xml/{pkg_id}.xml"
	+	)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO bill_versions
	+	(bill_id, version_code, govinfo_package_id, version_label,
	+	version_date, text_ref, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(bill_id, version_code) DO UPDATE SET
	+	govinfo_package_id = excluded.govinfo_package_id,
	+	version_label = excluded.version_label,
	+	version_date = excluded.version_date,
	+	text_ref = excluded.text_ref,
	+	source_url = excluded.source_url
	+	""",
	+	(bill_id, vcode, pkg_id, label, vdate, text_ref, text_ref),
	+	)
	+
	+	return bill_id
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Main poll
	+	# ---------------------------------------------------------------------------
	+
	+	def poll(
	+	congress: int = 119,
	+	bill_types: Optional[List[str]] = None,
	+	db_path: Optional[str] = None,
	+	limit: Optional[int] = None,
	+	delay: float = 0.05,
	+	) -> int:
	+	"""
	+	Enumerate bills via GovInfo sitemap, download individual BILLSTATUS XMLs,
	+	and store into the DB.
	+
	+	Parameters
	+	----------
	+	congress : Congress number (default 119).
	+	bill_types : List of bill types to ingest; default = all BILL_TYPES.
	+	db_path : Override SQLite path.
	+	limit : Maximum number of bills to fetch per bill type (None = all).
	+	delay : Seconds to sleep between requests (polite crawling).
	+
	+	Returns
	+	-------
	+	Total number of new or updated bills stored.
	+	"""
	+	if bill_types is None:
	+	bill_types = BILL_TYPES
	+
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	session = requests.Session()
	+	session.headers["User-Agent"] = "Docket-ingest/1.0 (public transparency research)"
	+
	+	total = 0
	+
	+	for bill_type in bill_types:
	+	entries = _fetch_sitemap(congress, bill_type)
	+	if not entries:
	+	continue
	+
	+	if limit is not None:
	+	entries = entries[:limit]
	+
	+	batch = 0
	+	for xml_url, lastmod in entries:
	+	# Use (source, url, lastmod) as dedupe key so we re-fetch on update
	+	sitemap_key = f"sitemap:{xml_url}"
	+	if not is_new_event(conn, SOURCE + "_sitemap", xml_url, lastmod):
	+	continue
	+
	+	xml_bytes = _fetch_bill_xml(xml_url, session)
	+	if xml_bytes is None:
	+	continue
	+
	+	for bill_el in _iter_bill_elements(xml_bytes):
	+	number = _txt(bill_el, "number")
	+	updated = _txt(bill_el, "updateDate")
	+	ext_id = f"{congress}-{bill_type}-{number}"
	+
	+	if not is_new_event(conn, SOURCE, ext_id, updated):
	+	mark_seen(conn, SOURCE + "_sitemap", xml_url, lastmod)
	+	continue
	+
	+	bill_id = _store_bill(conn, bill_el, congress, bill_type)
	+	if bill_id is not None:
	+	mark_seen(conn, SOURCE, ext_id, updated)
	+	mark_seen(conn, SOURCE + "_sitemap", xml_url, lastmod)
	+	batch += 1
	+
	+	if delay > 0:
	+	time.sleep(delay)
	+
	+	if batch % 100 == 0 and batch > 0:
	+	conn.commit()
	+	logger.info("%s/%s: %d bills committed so far", congress, bill_type, batch)
	+
	+	conn.commit()
	+	logger.info("%s/%s: %d new/updated bills", congress, bill_type, batch)
	+	total += batch
	+
	+	conn.close()
	+	logger.info("bills_bulk total: %d bills", total)
	+	return total
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	ap = argparse.ArgumentParser(description="GovInfo BILLSTATUS bulk ingest")
	+	ap.add_argument("--once", action="store_true", help="Run once and exit")
	+	ap.add_argument("--congress", type=int, default=119)
	+	ap.add_argument("--types", nargs="+", default=None,
	+	help="Bill types to ingest (hr s hjres …)")
	+	ap.add_argument("--limit", type=int, default=None,
	+	help="Max bills per type (for testing)")
	+	ap.add_argument("--delay", type=float, default=0.05,
	+	help="Seconds between requests (default 0.05)")
	+	args = ap.parse_args()
	+
	+	n = poll(
	+	congress=args.congress,
	+	bill_types=args.types,
	+	limit=args.limit,
	+	delay=args.delay,
	+	)
	+	print(f"Done: {n} bills stored")

services/ingest/ingest/pollers/crec.py +401 -0

		@@ -0,0 +1,401 @@
	+	"""
	+	Congressional Record (CREC) bulk-data poller - procedural event detector.
	+
	+	Source: GovInfo Bulk Data - https://www.govinfo.gov/bulkdata/CREC/
	+	Auth: none (bulk data is public)
	+
	+	Strategy:
	+	1. Fetch the daily ZIP for the target date from govinfo.gov/bulkdata/CREC/.
	+	2. Extract all XML files from the ZIP in-memory.
	+	3. Parse text content from each XML section.
	+	4. Apply regex patterns to detect procedural events:
	+	voice_vote - "the ayes have it" / "the noes have it"
	+	unanimous_consent - "without objection" / "so ordered"
	+	cloture - "cloture motion", "invoke cloture"
	+	suspension - "suspend the rules"
	+	mtr - "motion to recommit"
	+	discharge - "discharge petition"
	+	5. Write detected events to procedural_events.
	+
	+	Honest gaps (per DATA_SOURCES.md + NEUTRALITY.md):
	+	- Voice votes and UC agreements have NO individual member record.
	+	The `members_jsonb` column is NULL for these rows; the UI must label them
	+	"no individual record exists" - the absence is itself information.
	+	- CREC text can be revised after delivery and is not verbatim; label as
	+	"Congressional Record" not "floor transcript."
	+
	+	Writes to: procedural_events table (created if absent - does not touch db.py core)
	+	"""
	+
	+	import argparse
	+	import html as html_module
	+	import io
	+	import json
	+	import logging
	+	import re
	+	import sys
	+	import xml.etree.ElementTree as ET
	+	import zipfile
	+	from datetime import datetime, timedelta
	+	from typing import Iterator, List, Optional, Tuple
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "crec"
	+	# GovInfo content/pkg endpoint - the correct path for the CREC ZIP.
	+	# The bulkdata URL path returns HTML, not ZIPs.
	+	CONTENT_BASE = "https://www.govinfo.gov/content/pkg"
	+
	+	# ── Procedural-event regex patterns ──────────────────────────────────────────
	+
	+	# Voice vote: presiding officer declares the result
	+	_VOICE_VOTE = re.compile(
	+	r"the\s+(?:ayes?\|yeas?)\s+(?:have\|has)\s+it"
	+	r"\|the\s+(?:noes?\|nays?)\s+(?:have\|has)\s+it"
	+	r"\|(?:ayes?\|yeas?)\s+have\s+it[,;.]?\s*the\s+(?:ayes?\|yeas?)\s+have\s+it",
	+	re.IGNORECASE,
	+	)
	+
	+	# Unanimous consent
	+	_UC = re.compile(
	+	r"without\s+objection\b"
	+	r"\|so\s+ordered\b"
	+	r"\|unanimous\s+consent\b",
	+	re.IGNORECASE,
	+	)
	+
	+	# Cloture
	+	_CLOTURE = re.compile(
	+	r"cloture\s+(?:motion\|vote\|petition\|agreement)"
	+	r"\|(?:motion\|vote)\s+(?:on\s+(?:the\s+)?)?(?:to\s+(?:invoke\s+)?)?cloture"
	+	r"\|invoke\s+cloture"
	+	r"\|cloture\s+was\s+(?:invoked\|agreed\s+to)"
	+	r"\|vote\s+on\s+(?:the\s+)?cloture",
	+	re.IGNORECASE,
	+	)
	+
	+	# Suspension of the rules (House)
	+	_SUSPENSION = re.compile(
	+	r"(?:motion\s+to\s+)?suspend\s+the\s+rules"
	+	r"\|suspension\s+of\s+the\s+rules"
	+	r"\|on\s+motion\s+to\s+suspend",
	+	re.IGNORECASE,
	+	)
	+
	+	# Motion to recommit (House)
	+	_MTR = re.compile(
	+	r"motion\s+to\s+recommit",
	+	re.IGNORECASE,
	+	)
	+
	+	# Discharge petition (House)
	+	_DISCHARGE = re.compile(
	+	r"discharge\s+petition"
	+	r"\|discharged?\s+from\s+(?:further\s+)?committee"
	+	r"\|discharge\s+calendar",
	+	re.IGNORECASE,
	+	)
	+
	+	# Map pattern → (kind, label for detail)
	+	_PATTERNS: List[Tuple[str, re.Pattern, str]] = [
	+	("voice_vote", _VOICE_VOTE, "Voice vote detected"),
	+	("unanimous_consent", _UC, "Unanimous consent / without objection"),
	+	("cloture", _CLOTURE, "Cloture action"),
	+	("suspension", _SUSPENSION, "Suspension of the rules"),
	+	("mtr", _MTR, "Motion to recommit"),
	+	("discharge", _DISCHARGE, "Discharge petition"),
	+	]
	+
	+	# Heuristic chamber detection from filename
	+	_HOUSE_RE = re.compile(r"PgH\d\|CREC-.*-H\d", re.IGNORECASE)
	+	_SENATE_RE = re.compile(r"PgS\d\|CREC-.*-S\d", re.IGNORECASE)
	+	_DIGEST_RE = re.compile(r"PgD\d\|CREC-.*-D\d\|DIGEST", re.IGNORECASE)
	+
	+
	+	def _ensure_tables(conn) -> None:
	+	"""Create procedural_events table if it does not exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS procedural_events (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'crec',
	+	external_id TEXT NOT NULL,
	+	kind TEXT NOT NULL
	+	CHECK(kind IN ('voice_vote','unanimous_consent',
	+	'cloture','suspension','mtr','discharge')),
	+	date TEXT NOT NULL,
	+	chamber TEXT,
	+	detail TEXT,
	+	members_jsonb TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (source, external_id)
	+	);
	+	"""
	+	)
	+	conn.commit()
	+
	+
	+	def _govinfo_url(date: datetime) -> str:
	+	"""
	+	Return the GovInfo content/pkg ZIP URL for the CREC on a given date.
	+
	+	URL format: https://www.govinfo.gov/content/pkg/CREC-YYYY-MM-DD/zip/CREC-YYYY-MM-DD.zip
	+	Note: the bulkdata/ path returns HTML pages, not ZIPs.
	+	"""
	+	slug = f"CREC-{date.year}-{date.month:02d}-{date.day:02d}"
	+	return f"{CONTENT_BASE}/{slug}/zip/{slug}.zip"
	+
	+
	+	def _govinfo_page_url(date: datetime) -> str:
	+	"""Return a citable GovInfo browse URL for the CREC on a given date."""
	+	return (
	+	f"https://www.govinfo.gov/app/details/CREC-{date.year}-{date.month:02d}-{date.day:02d}"
	+	)
	+
	+
	+	def _infer_chamber(filename: str) -> str:
	+	"""Guess the chamber from the CREC XML filename."""
	+	if _HOUSE_RE.search(filename):
	+	return "House"
	+	if _SENATE_RE.search(filename):
	+	return "Senate"
	+	if _DIGEST_RE.search(filename):
	+	return "Both"
	+	return "Unknown"
	+
	+
	+	_HTML_TAG_RE = re.compile(r"<[^>]+>")
	+
	+
	+	def _extract_text_from_xml(xml_bytes: bytes) -> str:
	+	"""
	+	Extract plain text from a CREC HTML/XML file.
	+
	+	GovInfo CREC ZIPs contain .htm files (not standalone XML).
	+	We strip HTML tags with regex and unescape entities; this is intentionally
	+	simple - procedural-phrase detection needs only the word sequence, not
	+	structure. Falls back gracefully on parse errors.
	+	"""
	+	try:
	+	raw = xml_bytes.decode("utf-8", errors="replace")
	+	except Exception:
	+	return ""
	+	# Strip all markup
	+	text = _HTML_TAG_RE.sub(" ", raw)
	+	# Unescape HTML entities (& <   etc.)
	+	text = html_module.unescape(text)
	+	# Collapse whitespace
	+	text = re.sub(r"\s+", " ", text).strip()
	+	return text
	+
	+
	+	def _context_snippet(text: str, match: re.Match, window: int = 120) -> str:
	+	"""Return up to `window` characters around a regex match as a detail string."""
	+	start = max(0, match.start() - window // 2)
	+	end = min(len(text), match.end() + window // 2)
	+	snippet = text[start:end].replace("\n", " ").strip()
	+	return snippet
	+
	+
	+	def _detect_events(
	+	text: str,
	+	chamber: str,
	+	date_iso: str,
	+	source_url: str,
	+	filename: str,
	+	) -> Iterator[dict]:
	+	"""
	+	Yield one dict per detected procedural event in `text`.
	+
	+	Each dict has: kind, date, chamber, detail, members_jsonb, source_url, external_id.
	+	"""
	+	for kind, pattern, label in _PATTERNS:
	+	for i, m in enumerate(pattern.finditer(text)):
	+	snippet = _context_snippet(text, m)
	+	# external_id: filename + kind + match index (stable across re-runs)
	+	ext_id = f"{date_iso}_{filename}_{kind}_{i}"
	+	yield {
	+	"kind": kind,
	+	"date": date_iso,
	+	"chamber": chamber,
	+	"detail": f"{label}: …{snippet}…",
	+	"members_jsonb": None, # no individual record for voice votes / UC
	+	"source_url": source_url,
	+	"external_id": ext_id,
	+	"update_date": date_iso,
	+	}
	+
	+
	+	def _fetch_crec_zip(date: datetime) -> Optional[bytes]:
	+	"""Download the CREC ZIP for a date; return raw bytes or None."""
	+	url = _govinfo_url(date)
	+	logger.info("crec: fetching %s", url)
	+	try:
	+	resp = requests.get(url, timeout=60, stream=True)
	+	if resp.status_code == 404:
	+	logger.info("crec: no CREC for %s (404 - non-session day?)", date.date())
	+	return None
	+	resp.raise_for_status()
	+	return resp.content
	+	except requests.RequestException as exc:
	+	logger.error("crec: download failed for %s: %s", date.date(), exc)
	+	return None
	+
	+
	+	def poll_date(date: datetime, conn, source_url: str) -> int:
	+	"""Process a single CREC date. Returns number of new events stored."""
	+	date_iso = date.strftime("%Y-%m-%d")
	+	zip_bytes = _fetch_crec_zip(date)
	+	if not zip_bytes:
	+	return 0
	+
	+	try:
	+	zf = zipfile.ZipFile(io.BytesIO(zip_bytes))
	+	except zipfile.BadZipFile as exc:
	+	logger.error("crec: bad ZIP for %s: %s", date_iso, exc)
	+	return 0
	+
	+	# CREC content/pkg ZIPs contain .htm files (not standalone XML).
	+	# We skip -2.htm, -3.htm etc. (alternate formats of the same page)
	+	# and process only the canonical page files.
	+	htm_names = [
	+	n for n in zf.namelist()
	+	if n.lower().endswith(".htm")
	+	and not re.search(r"-\d+\.htm$", n) # skip alternate-format duplicates
	+	]
	+	logger.info("crec: %s - %d HTML page(s) in ZIP", date_iso, len(htm_names))
	+
	+	stored = 0
	+	for xml_name in htm_names:
	+	chamber = _infer_chamber(xml_name)
	+	try:
	+	xml_bytes = zf.read(xml_name)
	+	except Exception as exc:
	+	logger.warning("crec: could not read %s: %s", xml_name, exc)
	+	continue
	+
	+	text = _extract_text_from_xml(xml_bytes)
	+	if not text:
	+	continue
	+
	+	for evt in _detect_events(text, chamber, date_iso, source_url, xml_name):
	+	if not is_new_event(conn, SOURCE, evt["external_id"], evt["update_date"]):
	+	continue
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO procedural_events
	+	(source, external_id, kind, date, chamber, detail,
	+	members_jsonb, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	kind = excluded.kind,
	+	date = excluded.date,
	+	chamber = excluded.chamber,
	+	detail = excluded.detail,
	+	members_jsonb = excluded.members_jsonb,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	SOURCE,
	+	evt["external_id"],
	+	evt["kind"],
	+	evt["date"],
	+	evt["chamber"],
	+	evt["detail"],
	+	evt["members_jsonb"],
	+	evt["source_url"],
	+	),
	+	)
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(
	+	SOURCE,
	+	evt["external_id"],
	+	json.dumps({
	+	"kind": evt["kind"],
	+	"date": evt["date"],
	+	"chamber": evt["chamber"],
	+	"detail": evt["detail"],
	+	"xml_file": xml_name,
	+	}),
	+	datetime.utcnow().isoformat(),
	+	),
	+	)
	+	mark_seen(conn, SOURCE, evt["external_id"], evt["update_date"])
	+	stored += 1
	+
	+	conn.commit()
	+	return stored
	+
	+
	+	def poll(days_back: int = 14, db_path: Optional[str] = None) -> int:
	+	"""
	+	Scan the last `days_back` days of CREC for procedural events.
	+
	+	Congressional Record has a ~1-day publishing lag; non-session days (weekends,
	+	recesses) return 200 HTML pages instead of ZIPs. We default to 14 days to
	+	ensure we catch at least a few session days.
	+
	+	Parameters
	+	----------
	+	days_back : Days of CREC history to fetch (default 5).
	+	db_path : Override default SQLite path.
	+
	+	Returns
	+	-------
	+	Total new procedural_events rows stored across all dates.
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	_ensure_tables(conn)
	+
	+	today = datetime.utcnow()
	+	total = 0
	+
	+	for offset in range(1, days_back + 1):
	+	date = today - timedelta(days=offset)
	+	source_url = _govinfo_page_url(date)
	+
	+	try:
	+	validate_source_url(source_url)
	+	except Exception as exc:
	+	logger.error("crec: source_url %r not allowlisted: %s", source_url, exc)
	+	continue
	+
	+	n = poll_date(date, conn, source_url)
	+	logger.info("crec: %s → %d new event(s)", date.strftime("%Y-%m-%d"), n)
	+	total += n
	+
	+	conn.close()
	+	logger.info("crec: total %d new procedural_events row(s)", total)
	+	return total
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	parser = argparse.ArgumentParser(description="CREC procedural-event poller")
	+	parser.add_argument("--once", action="store_true", help="Run once and exit")
	+	parser.add_argument("--days", type=int, default=14,
	+	help="Days of CREC history to scan (default 14)")
	+	args = parser.parse_args()
	+
	+	count = poll(days_back=args.days)
	+	print(f"Stored {count} new procedural_events row(s).")
	+	sys.exit(0)

services/ingest/ingest/pollers/federal_register.py +234 -0

		@@ -0,0 +1,234 @@
	+	"""
	+	Federal Register API poller - Executive Orders, proposed/final rules, notices.
	+
	+	Endpoint: GET https://www.federalregister.gov/api/v1/documents.json
	+	Auth: none (fully open)
	+	Change-detection: (source='federal_register', externalId=document_number, updateDate=publication_date)
	+	Source URL: html_url field from the API (federalregister.gov/.gov)
	+
	+	Types fetched:
	+	PRESDOCU - Presidential Documents (Executive Orders, proclamations, memos)
	+	RULE - Final rules
	+	PRORULE - Proposed rules (NPRM)
	+	NOTICE - Notices
	+
	+	Writes to: executive_actions table (created if absent - does not touch db.py core)
	+	"""
	+
	+	import argparse
	+	import json
	+	import logging
	+	import sys
	+	from datetime import datetime, timedelta
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "federal_register"
	+	BASE_URL = "https://www.federalregister.gov/api/v1/documents.json"
	+
	+	# FR document type → our normalized type
	+	_TYPE_MAP = {
	+	"Rule": "final_rule",
	+	"Proposed Rule": "nprm",
	+	"Notice": "notice",
	+	"Presidential Document": "eo",
	+	# Raw code forms returned in some API versions
	+	"RULE": "final_rule",
	+	"PRORULE": "nprm",
	+	"NOTICE": "notice",
	+	"PRESDOCU": "eo",
	+	}
	+
	+	_FR_TYPES = ["RULE", "PRORULE", "NOTICE", "PRESDOCU"]
	+
	+
	+	def _ensure_tables(conn) -> None:
	+	"""Create executive_actions table if it does not exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS executive_actions (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'federal_register',
	+	external_id TEXT NOT NULL,
	+	type TEXT NOT NULL,
	+	title TEXT,
	+	agency TEXT,
	+	fr_doc_number TEXT,
	+	date TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (source, external_id)
	+	);
	+	"""
	+	)
	+	conn.commit()
	+
	+
	+	def _fetch_page(start_date: str, page: int) -> dict:
	+	"""
	+	Fetch one page from the Federal Register API.
	+
	+	Uses a list of (key, value) tuples so that repeat keys (fields[], types[])
	+	are encoded correctly as array params rather than overwriting each other.
	+	"""
	+	params = []
	+	for field in ["document_number", "type", "title", "agencies",
	+	"publication_date", "html_url", "abstract"]:
	+	params.append(("fields[]", field))
	+	for doc_type in _FR_TYPES:
	+	params.append(("conditions[type][]", doc_type))
	+	params += [
	+	("conditions[publication_date][gte]", start_date),
	+	("per_page", 100),
	+	("order", "newest"),
	+	("page", page),
	+	]
	+	try:
	+	resp = requests.get(BASE_URL, params=params, timeout=30)
	+	resp.raise_for_status()
	+	return resp.json()
	+	except requests.RequestException as exc:
	+	logger.error("federal_register: request failed: %s", exc)
	+	return {}
	+
	+
	+	def poll(days_back: int = 7, db_path: Optional[str] = None) -> int:
	+	"""
	+	Fetch recent Executive Orders, rules, and notices from federalregister.gov.
	+
	+	Parameters
	+	----------
	+	days_back : Days of history to fetch (default 7).
	+	db_path : Override default SQLite path (useful in tests).
	+
	+	Returns
	+	-------
	+	Number of new rows stored.
	+	"""
	+	start_date = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
	+
	+	logger.info("federal_register: fetching documents since %s", start_date)
	+
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	_ensure_tables(conn)
	+
	+	stored = 0
	+	page = 1
	+
	+	while True:
	+	data = _fetch_page(start_date, page)
	+	if not data:
	+	break
	+
	+	results = data.get("results", [])
	+	if not results:
	+	logger.info("federal_register: no more results at page %d", page)
	+	break
	+
	+	logger.info("federal_register: page %d - %d document(s)", page, len(results))
	+
	+	for doc in results:
	+	doc_number = doc.get("document_number") or ""
	+	pub_date = doc.get("publication_date") or ""
	+	raw_type = doc.get("type") or ""
	+	title = doc.get("title") or ""
	+	html_url = doc.get("html_url") or ""
	+	agencies = doc.get("agencies") or []
	+
	+	if not doc_number or not html_url:
	+	logger.debug("federal_register: skipping doc with missing id/url")
	+	continue
	+
	+	if not is_new_event(conn, SOURCE, doc_number, pub_date):
	+	logger.debug("federal_register: already seen %s @ %s", doc_number, pub_date)
	+	continue
	+
	+	# Validate source URL - federalregister.gov is .gov ✓
	+	try:
	+	validate_source_url(html_url)
	+	except Exception as exc:
	+	logger.error("federal_register: invalid source_url %r: %s", html_url, exc)
	+	continue
	+
	+	# Normalize type
	+	norm_type = _TYPE_MAP.get(raw_type, raw_type.lower().replace(" ", "_"))
	+
	+	# Agency: join all agency names (typically one or two)
	+	agency_str = "; ".join(a.get("name", "") for a in agencies if a.get("name"))
	+
	+	# Upsert into executive_actions
	+	conn.execute(
	+	"""
	+	INSERT INTO executive_actions
	+	(source, external_id, type, title, agency, fr_doc_number, date, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	type = excluded.type,
	+	title = excluded.title,
	+	agency = excluded.agency,
	+	fr_doc_number = excluded.fr_doc_number,
	+	date = excluded.date,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	SOURCE,
	+	doc_number,
	+	norm_type,
	+	title,
	+	agency_str,
	+	doc_number,
	+	pub_date,
	+	html_url,
	+	),
	+	)
	+
	+	# Raw event
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(SOURCE, doc_number, json.dumps(doc), datetime.utcnow().isoformat()),
	+	)
	+
	+	mark_seen(conn, SOURCE, doc_number, pub_date)
	+	stored += 1
	+
	+	conn.commit()
	+
	+	# Check for next page
	+	total_pages = data.get("total_pages", 1)
	+	if page >= total_pages or page >= 5: # cap at 5 pages per poll
	+	break
	+	page += 1
	+
	+	conn.close()
	+	logger.info("federal_register: stored %d new executive-action row(s)", stored)
	+	return stored
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	parser = argparse.ArgumentParser(description="Federal Register poller")
	+	parser.add_argument("--once", action="store_true", help="Run once and exit")
	+	parser.add_argument("--days", type=int, default=7, help="Days of history to fetch")
	+	args = parser.parse_args()
	+
	+	if args.once or True: # always run immediately when invoked directly
	+	count = poll(days_back=args.days)
	+	print(f"Stored {count} new executive_actions row(s).")
	+	sys.exit(0)

services/ingest/ingest/pollers/legiscan.py +419 -0

		@@ -0,0 +1,419 @@
	+	"""
	+	LegiScan API poller - 50-state + DC + PR legislative data.
	+
	+	Base: https://api.legiscan.com/?key={KEY}&op={OPERATION}
	+	Auth: free API key required (set DOCKET_LEGISCAN_KEY).
	+	Free tier: ~30k requests/month (~1k/day).
	+	If absent, logs clearly and returns 0 - no crash.
	+	Free key signup: https://legiscan.com/legiscan
	+
	+	Per DATA_SOURCES.md: LegiScan is the completeness supplement - richer bill-text
	+	and amendment coverage than OpenStates. The user-facing source_url MUST point
	+	to the official state legislature record (bill.state_link), NOT legiscan.com.
	+
	+	Sync strategy (efficient for free tier):
	+	1. getMasterListRaw per state → get change_hash per bill.
	+	2. Compare change_hash against seen table (stored as update_date field).
	+	3. Call getBill only for changed/new bills - minimises request count.
	+
	+	State coverage: writes scrape_log rows via ingest.state_coverage.log_scrape().
	+	"""
	+
	+	import json
	+	import logging
	+	import os
	+	import time
	+	from datetime import datetime
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import is_allowlisted, validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "legiscan"
	+	BASE_URL = "https://api.legiscan.com/"
	+	KEY_ENV = "DOCKET_LEGISCAN_KEY"
	+
	+	# LegiScan 2-letter state codes for all 50 states + DC + PR
	+	ALL_STATES = [
	+	"AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA",
	+	"HI", "ID", "IL", "IN", "IA", "KS", "KY", "LA", "ME", "MD",
	+	"MA", "MI", "MN", "MS", "MO", "MT", "NE", "NV", "NH", "NJ",
	+	"NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI", "SC",
	+	"SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY",
	+	"DC", "PR",
	+	]
	+
	+	# MVP subset: top states by population + DC/PR for speed on free tier
	+	MVP_STATES = [
	+	"CA", "TX", "FL", "NY", "PA", "IL", "OH", "GA", "NC", "MI",
	+	"DC", "PR",
	+	]
	+
	+	# Seconds between API calls to stay within 10 rps limit
	+	_RATE_SLEEP = 0.15
	+
	+	# Max bills to fetch via getBill per state per run (free-tier guard)
	+	MAX_BILL_FETCH_PER_STATE = 50
	+
	+
	+	def _get_api_key() -> Optional[str]:
	+	return os.environ.get(KEY_ENV)
	+
	+
	+	def _api(key: str, op: str, **params) -> Optional[dict]:
	+	"""
	+	Make one LegiScan API call. Returns parsed JSON on success, None on error.
	+	"""
	+	query = {"key": key, "op": op}
	+	query.update(params)
	+	try:
	+	time.sleep(_RATE_SLEEP)
	+	resp = requests.get(BASE_URL, params=query, timeout=30)
	+	resp.raise_for_status()
	+	except requests.RequestException as exc:
	+	logger.error("legiscan op=%s params=%s failed: %s", op, params, exc)
	+	return None
	+
	+	data = resp.json()
	+	if data.get("status") != "OK":
	+	logger.warning("legiscan op=%s returned status=%s", op, data.get("status"))
	+	return None
	+	return data
	+
	+
	+	def _get_current_session_id(key: str, state: str) -> Optional[int]:
	+	"""Return the current (most recent) session ID for a state."""
	+	data = _api(key, "getSessionList", state=state)
	+	if not data:
	+	return None
	+	sessions = data.get("sessions", [])
	+	if not sessions:
	+	return None
	+	# Sessions are returned newest-first; find the one marked special=0 (regular) or
	+	# just take the first one (most recent).
	+	for sess in sessions:
	+	if sess.get("special", 0) == 0:
	+	return sess.get("session_id")
	+	return sessions[0].get("session_id")
	+
	+
	+	def _get_master_list_raw(key: str, session_id: int) -> dict:
	+	"""
	+	Return {bill_id -> {number, change_hash, status_date, state_link?, ...}}
	+	for all bills in a session.
	+	"""
	+	data = _api(key, "getMasterListRaw", id=session_id)
	+	if not data:
	+	return {}
	+
	+	master = data.get("masterlist", {})
	+	bills = {}
	+	for key_str, item in master.items():
	+	if key_str == "session":
	+	continue
	+	if not isinstance(item, dict):
	+	continue
	+	bill_id = item.get("bill_id")
	+	if bill_id:
	+	bills[str(bill_id)] = item
	+	return bills
	+
	+
	+	def _external_id(state: str, bill_id) -> str:
	+	"""Canonical external_id for a LegiScan bill row: 'legiscan-{STATE}-{id}'."""
	+	return f"legiscan-{state.upper()}-{bill_id}"
	+
	+
	+	def _store_bill(conn, key: str, state: str, bill_id, change_hash: str) -> Optional[int]:
	+	"""
	+	Fetch full bill via getBill and upsert into bills + actions tables.
	+	Returns internal bills.id on success, None if skipped.
	+	"""
	+	ext_id = _external_id(state, bill_id)
	+
	+	# change_hash stored as update_date for dedupe
	+	if not is_new_event(conn, SOURCE, ext_id, change_hash):
	+	return None
	+
	+	data = _api(key, "getBill", id=bill_id)
	+	if not data:
	+	return None
	+
	+	bill = data.get("bill", {})
	+
	+	# Official state legislature URL is the required source_url.
	+	state_link = bill.get("state_link", "")
	+	if not state_link or not is_allowlisted(state_link):
	+	logger.debug(
	+	"legiscan %s bill %s: state_link %r not allowlisted - skipping (rule #1)",
	+	state, bill_id, state_link,
	+	)
	+	return None
	+
	+	number = bill.get("bill_number", "")
	+	title = bill.get("title", "")
	+	status_desc = bill.get("status_desc", "")
	+	status_date = (bill.get("status_date") or "")[:10]
	+	session = bill.get("session", {})
	+	session_id = session.get("session_id", 0) if isinstance(session, dict) else 0
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO bills
	+	(source, external_id, number, title, status, updated_at, source_url, jurisdiction)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	title = excluded.title,
	+	status = excluded.status,
	+	updated_at = excluded.updated_at,
	+	source_url = excluded.source_url,
	+	jurisdiction = excluded.jurisdiction
	+	""",
	+	(
	+	SOURCE, ext_id, number, title, status_desc,
	+	status_date, state_link, state.upper(),
	+	),
	+	)
	+
	+	row = conn.execute(
	+	"SELECT id FROM bills WHERE source=? AND external_id=?",
	+	(SOURCE, ext_id),
	+	).fetchone()
	+	bill_db_id = row["id"] if row else None
	+
	+	# History → actions
	+	for event in bill.get("history", []):
	+	action_date = (event.get("date") or "")[:10]
	+	action_text = event.get("action", "")
	+	chamber_code = event.get("chamber", "")
	+	chamber = "Senate" if chamber_code == "S" else "House" if chamber_code == "H" else chamber_code
	+	if bill_db_id and action_date:
	+	conn.execute(
	+	"""
	+	INSERT INTO actions
	+	(bill_id, source, action_type, action_date, chamber, text, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(bill_db_id, SOURCE, "legislative_action",
	+	action_date, chamber, action_text, state_link),
	+	)
	+
	+	# Raw event (full bill payload)
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(SOURCE, ext_id, json.dumps(bill), datetime.utcnow().isoformat()),
	+	)
	+
	+	mark_seen(conn, SOURCE, ext_id, change_hash)
	+	return bill_db_id
	+
	+
	+	def _store_votes(conn, bill_db_id: int, bill: dict, state_link: str,
	+	state: str) -> int:
	+	"""
	+	Insert vote summary rows for one bill; return count of new votes stored.
	+	"""
	+	stored = 0
	+	for vote_summary in bill.get("votes", []):
	+	vote_id = vote_summary.get("vote_id")
	+	vote_date = (vote_summary.get("date") or "")[:10]
	+	if not vote_id or not vote_date:
	+	continue
	+
	+	ext_vote_id = f"legiscan-vote-{vote_id}"
	+	if not is_new_event(conn, f"{SOURCE}_vote", ext_vote_id, vote_date):
	+	continue
	+
	+	desc = vote_summary.get("desc", "")
	+	yea = int(vote_summary.get("yea") or 0)
	+	nay = int(vote_summary.get("nay") or 0)
	+	nv = int(vote_summary.get("nv") or 0)
	+	absent = int(vote_summary.get("absent") or 0)
	+	present = nv + absent
	+
	+	# result: passed / failed
	+	passed = vote_summary.get("passed", 0)
	+	result = "pass" if passed else "fail"
	+
	+	# chamber from description heuristic
	+	desc_lower = desc.lower()
	+	if "senate" in desc_lower:
	+	chamber = "Senate"
	+	elif "house" in desc_lower or "assembly" in desc_lower:
	+	chamber = "House"
	+	else:
	+	chamber = ""
	+
	+	# Vote URL - LegiScan provides a url field in vote summary
	+	vote_url = vote_summary.get("url", "")
	+	vote_source_url = vote_url if (vote_url and is_allowlisted(vote_url)) else state_link
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO votes
	+	(bill_id, chamber, date, yea, nay, present, result, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(bill_db_id, chamber, vote_date, yea, nay, present, result, vote_source_url),
	+	)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(
	+	f"{SOURCE}_vote", ext_vote_id,
	+	json.dumps(vote_summary), datetime.utcnow().isoformat(),
	+	),
	+	)
	+
	+	mark_seen(conn, f"{SOURCE}_vote", ext_vote_id, vote_date)
	+	stored += 1
	+
	+	return stored
	+
	+
	+	def poll(
	+	db_path: Optional[str] = None,
	+	states: Optional[list] = None,
	+	) -> int:
	+	"""
	+	Fetch changed bills from LegiScan for the given state list.
	+
	+	Uses getMasterListRaw to detect changes cheaply, then calls getBill only
	+	for bills whose change_hash has changed - minimises API quota usage.
	+
	+	If DOCKET_LEGISCAN_KEY is not set, logs and returns 0 - no crash.
	+
	+	Parameters
	+	----------
	+	db_path : Override DB path (useful in tests).
	+	states : Override state list (tests / specific runs).
	+
	+	Returns
	+	-------
	+	Number of new bill rows stored.
	+	"""
	+	api_key = _get_api_key()
	+	if not api_key:
	+	logger.warning(
	+	"LegiScan poller: DOCKET_LEGISCAN_KEY is not set. "
	+	"Free key signup (30k req/month): https://legiscan.com/legiscan "
	+	"Skipping state ingest this run."
	+	)
	+	return 0
	+
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+
	+	# Add jurisdiction column if missing (same migration as openstates.py)
	+	try:
	+	conn.execute("ALTER TABLE bills ADD COLUMN jurisdiction TEXT")
	+	conn.commit()
	+	except Exception:
	+	pass
	+
	+	try:
	+	from ingest.state_coverage import log_scrape
	+	_log = log_scrape
	+	except ImportError:
	+	_log = None
	+
	+	if states is None:
	+	if os.environ.get("DOCKET_LEGISCAN_ALL"):
	+	states = ALL_STATES
	+	else:
	+	states = MVP_STATES
	+
	+	total_bills = 0
	+	total_votes = 0
	+
	+	for state in states:
	+	bills_this_state = 0
	+	votes_this_state = 0
	+
	+	# 1. Find current session
	+	session_id = _get_current_session_id(api_key, state)
	+	if not session_id:
	+	logger.warning("legiscan %s: could not get current session ID", state)
	+	continue
	+
	+	# 2. Get master list with change hashes
	+	master = _get_master_list_raw(api_key, session_id)
	+	if not master:
	+	logger.warning("legiscan %s: empty master list", state)
	+	continue
	+
	+	logger.info("legiscan %s: %d bills in master list", state, len(master))
	+
	+	# 3. Fetch only changed bills (up to MAX_BILL_FETCH_PER_STATE)
	+	fetched = 0
	+	for bill_id_str, item in master.items():
	+	if fetched >= MAX_BILL_FETCH_PER_STATE:
	+	logger.info(
	+	"legiscan %s: hit MAX_BILL_FETCH_PER_STATE=%d - "
	+	"remaining bills deferred to next run",
	+	state, MAX_BILL_FETCH_PER_STATE,
	+	)
	+	break
	+
	+	bill_id = item.get("bill_id", bill_id_str)
	+	change_hash = item.get("change_hash", "")
	+	ext_id = _external_id(state, bill_id)
	+
	+	if not is_new_event(conn, SOURCE, ext_id, change_hash):
	+	continue # Nothing changed
	+
	+	bill_db_id = _store_bill(conn, api_key, state, bill_id, change_hash)
	+	fetched += 1
	+
	+	if bill_db_id is not None:
	+	bills_this_state += 1
	+	# Fetch vote details from the bill that was just stored
	+	raw = conn.execute(
	+	"SELECT payload FROM raw_events WHERE source=? AND external_id=?",
	+	(SOURCE, ext_id),
	+	).fetchone()
	+	if raw:
	+	bill_data = json.loads(raw["payload"])
	+	state_link = bill_data.get("state_link", "")
	+	v = _store_votes(conn, bill_db_id, bill_data, state_link, state)
	+	votes_this_state += v
	+
	+	conn.commit()
	+
	+	logger.info(
	+	"legiscan %s: stored %d bill(s), %d vote(s)",
	+	state, bills_this_state, votes_this_state,
	+	)
	+	total_bills += bills_this_state
	+	total_votes += votes_this_state
	+
	+	if _log:
	+	now_iso = datetime.utcnow().isoformat()
	+	_log(conn, SOURCE, state, "bills", bills_this_state, now_iso)
	+	_log(conn, SOURCE, state, "votes", votes_this_state, now_iso)
	+	_log(conn, SOURCE, state, "sponsors", bills_this_state, now_iso)
	+
	+	conn.commit()
	+	conn.close()
	+	logger.info(
	+	"legiscan: stored %d new bill(s), %d new vote(s) total",
	+	total_bills, total_votes,
	+	)
	+	return total_bills

services/ingest/ingest/pollers/lobbying.py +282 -0

		@@ -0,0 +1,282 @@
	+	"""
	+	LDA.gov lobbying-disclosure poller (LD-1 / LD-2 filings).
	+
	+	Endpoint: https://lda.gov/api/v1/filings/
	+	Auth: Free token in env DOCKET_LDA_KEY.
	+	If absent, logs clear instructions and returns 0 (no crash).
	+
	+	Note: lda.senate.gov retires 2026-06-30 → this poller targets lda.gov/api only.
	+
	+	Each filing links to the official lda.gov record; source_url = lda.gov (✓.gov).
	+
	+	Writes to: lobbying table (created if absent - does not touch db.py core)
	+	"""
	+
	+	import argparse
	+	import json
	+	import logging
	+	import os
	+	import sys
	+	from datetime import datetime, timedelta
	+	from typing import Optional
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import validate_source_url
	+
	+	logger = logging.getLogger(__name__)
	+
	+	SOURCE = "lda"
	+	BASE_URL = "https://lda.gov/api/v1"
	+
	+	LDA_KEY_ENV = "DOCKET_LDA_KEY"
	+
	+	# Amount-range bucketing - LDA reports income/expenses as dollar figures;
	+	# we store as a human-readable range string for the lobbying table.
	+	_AMOUNT_RANGES = [
	+	( 500_000, "≥$500k"),
	+	( 100_000, "$100k-$500k"),
	+	( 50_000, "$50k-$100k"),
	+	( 10_000, "$10k-$50k"),
	+	( 0, "<$10k"),
	+	]
	+
	+
	+	def _ensure_tables(conn) -> None:
	+	"""Create lobbying table if it does not exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS lobbying (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'lda',
	+	external_id TEXT NOT NULL,
	+	registrant TEXT,
	+	client TEXT,
	+	covered_bill_id TEXT,
	+	amount_range TEXT,
	+	period TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (source, external_id)
	+	);
	+	"""
	+	)
	+	conn.commit()
	+
	+
	+	def _get_api_key() -> Optional[str]:
	+	return os.environ.get(LDA_KEY_ENV)
	+
	+
	+	def _amount_range(amount: Optional[float]) -> str:
	+	"""Convert a raw dollar amount to a display range string."""
	+	if amount is None:
	+	return "undisclosed"
	+	for threshold, label in _AMOUNT_RANGES:
	+	if amount >= threshold:
	+	return label
	+	return "<$10k"
	+
	+
	+	def _filing_source_url(filing_uuid: str) -> str:
	+	"""Construct the canonical lda.gov filing URL."""
	+	return f"https://lda.gov/filings/{filing_uuid}/"
	+
	+
	+	def _fetch_page(api_key: str, params: dict) -> dict:
	+	"""Fetch one page from the LDA filings API."""
	+	url = f"{BASE_URL}/filings/"
	+	headers = {"Authorization": f"Token {api_key}"}
	+	try:
	+	resp = requests.get(url, params=params, headers=headers, timeout=30)
	+	resp.raise_for_status()
	+	return resp.json()
	+	except requests.RequestException as exc:
	+	logger.error("lda: request failed: %s", exc)
	+	return {}
	+
	+
	+	def _extract_covered_bill(filing: dict) -> Optional[str]:
	+	"""
	+	Extract the first covered bill ID from a filing, if any.
	+
	+	LDA LD-2 filings include a `covered_documents` list; each entry may
	+	reference a bill by congress + bill_type + bill_number.
	+	Returns a formatted string like "119-HR-1234" or None.
	+	"""
	+	docs = filing.get("lobbying_activities", []) or []
	+	for activity in docs:
	+	for item in (activity.get("general_issue_code_display", []) or []):
	+	pass # general issue codes aren't bill IDs
	+	# covered_documents lives at top-level in some LDA versions
	+	covered = filing.get("covered_documents") or []
	+	for doc in covered:
	+	congress = doc.get("congress") or ""
	+	bill_type = doc.get("bill_type") or ""
	+	bill_number = doc.get("bill_number") or ""
	+	if congress and bill_type and bill_number:
	+	return f"{congress}-{bill_type.upper()}-{bill_number}"
	+	return None
	+
	+
	+	def poll(days_back: int = 90, db_path: Optional[str] = None) -> int:
	+	"""
	+	Fetch recent LDA lobbying filings.
	+
	+	LDA filings are quarterly; we default to 90 days to catch the most
	+	recent full quarter.
	+
	+	Parameters
	+	----------
	+	days_back : Days of filing history to request (default 90).
	+	db_path : Override default SQLite path.
	+
	+	Returns
	+	-------
	+	Number of new rows stored, or 0 if key is missing.
	+	"""
	+	api_key = _get_api_key()
	+	if not api_key:
	+	logger.warning(
	+	"LDA poller: DOCKET_LDA_KEY is not set. "
	+	"Register for a free key at https://lda.gov/api/ and export it as "
	+	"DOCKET_LDA_KEY=<your_key>. Skipping lobbying source."
	+	)
	+	return 0
	+
	+	dt_from = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
	+	logger.info("lda: fetching filings posted since %s", dt_from)
	+
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	_ensure_tables(conn)
	+
	+	stored = 0
	+	page = 1
	+
	+	while True:
	+	params = {
	+	"ordering": "-dt_posted",
	+	"dt_posted__gte": dt_from,
	+	"limit": 250,
	+	"offset": (page - 1) * 250,
	+	}
	+
	+	data = _fetch_page(api_key, params)
	+	if not data:
	+	break
	+
	+	results = data.get("results", [])
	+	if not results:
	+	logger.info("lda: no more results at page %d", page)
	+	break
	+
	+	logger.info("lda: page %d - %d filing(s)", page, len(results))
	+
	+	for filing in results:
	+	filing_uuid = filing.get("filing_uuid") or filing.get("id") or ""
	+	dt_posted = (filing.get("dt_posted") or "")[:10]
	+
	+	if not filing_uuid:
	+	continue
	+
	+	if not is_new_event(conn, SOURCE, filing_uuid, dt_posted):
	+	continue
	+
	+	source_url = _filing_source_url(filing_uuid)
	+	try:
	+	validate_source_url(source_url)
	+	except Exception as exc:
	+	logger.error("lda: invalid source_url %r: %s", source_url, exc)
	+	continue
	+
	+	registrant = (
	+	(filing.get("registrant") or {}).get("name")
	+	or filing.get("registrant_name")
	+	or ""
	+	)
	+	client = (
	+	(filing.get("client") or {}).get("name")
	+	or filing.get("client_name")
	+	or ""
	+	)
	+	period = filing.get("filing_period_display") or filing.get("filing_period") or ""
	+	income = filing.get("income") or filing.get("income_amount") or None
	+	expense = filing.get("expenses") or filing.get("expense_amount") or None
	+
	+	try:
	+	amount_f = float(income or expense or 0)
	+	except (TypeError, ValueError):
	+	amount_f = 0.0
	+
	+	amount_range = _amount_range(amount_f if amount_f else None)
	+	covered_bill = _extract_covered_bill(filing)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO lobbying
	+	(source, external_id, registrant, client, covered_bill_id,
	+	amount_range, period, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	registrant = excluded.registrant,
	+	client = excluded.client,
	+	covered_bill_id = excluded.covered_bill_id,
	+	amount_range = excluded.amount_range,
	+	period = excluded.period,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	SOURCE,
	+	filing_uuid,
	+	registrant,
	+	client,
	+	covered_bill,
	+	amount_range,
	+	period,
	+	source_url,
	+	),
	+	)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(SOURCE, filing_uuid, json.dumps(filing), datetime.utcnow().isoformat()),
	+	)
	+
	+	mark_seen(conn, SOURCE, filing_uuid, dt_posted)
	+	stored += 1
	+
	+	conn.commit()
	+
	+	# LDA pagination
	+	if not data.get("next") or page >= 10:
	+	break
	+	page += 1
	+
	+	conn.close()
	+	logger.info("lda: stored %d new lobbying row(s)", stored)
	+	return stored
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	parser = argparse.ArgumentParser(description="LDA lobbying poller")
	+	parser.add_argument("--once", action="store_true", help="Run once and exit")
	+	parser.add_argument("--days", type=int, default=90,
	+	help="Days of filing history (default 90)")
	+	args = parser.parse_args()
	+
	+	count = poll(days_back=args.days)
	+	print(f"Stored {count} new lobbying row(s).")
	+	sys.exit(0)

services/ingest/ingest/pollers/openomb.py +327 -0

		@@ -0,0 +1,327 @@
	+	"""
	+	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.
	+
	+	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.)
	+
	+	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.
	+
	+	Writes to: apportionments table (created if absent - does not touch db.py core)
	+	"""
	+
	+	import argparse
	+	import json
	+	import logging
	+	import sys
	+	from datetime import datetime, timedelta
	+	from typing import Optional
	+	from urllib.parse import urlencode
	+
	+	import requests
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+	from ingest.db import get_connection, init_schema
	+	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"
	+
	+	# Status normalization
	+	_STATUS_MAP = {
	+	"released": "released",
	+	"approved": "released",
	+	"normal": "released",
	+	"delayed": "delayed",
	+	"withheld": "withheld",
	+	"not apportioned": "withheld",
	+	}
	+
	+
	+	def _ensure_tables(conn) -> None:
	+	"""Create apportionments table if it does not exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS apportionments (
	+	id INTEGER PRIMARY KEY AUTOINCREMENT,
	+	source TEXT NOT NULL DEFAULT 'openomb',
	+	external_id TEXT NOT NULL,
	+	agency TEXT,
	+	account TEXT,
	+	appropriated REAL,
	+	apportioned REAL,
	+	status TEXT CHECK(status IN ('released','delayed','withheld')),
	+	date TEXT,
	+	source_url TEXT NOT NULL,
	+	UNIQUE (source, external_id)
	+	);
	+	"""
	+	)
	+	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,
	+	}
	+	try:
	+	resp = requests.get(url, params=params, 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 []
	+
	+
	+	def _fetch_from_openomb(fy: int) -> list:
	+	"""
	+	Fetch apportionment data from openomb.org JSON feed.
	+
	+	Returns list of raw record dicts or [] on failure.
	+	"""
	+	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:
	+	return "released"
	+	lower = raw.strip().lower()
	+	for key, val in _STATUS_MAP.items():
	+	if key in lower:
	+	return val
	+	return "released"
	+
	+
	+	def _extract_fields(rec: dict) -> Optional[dict]:
	+	"""
	+	Extract normalized fields from a raw apportionment record dict.
	+
	+	Handles various field-name conventions from max.gov and openomb.org.
	+	Returns 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 "")
	+
	+	if not agency and not account:
	+	return None
	+
	+	ext_id = f"{year_str}_{agency_code or agency[:20]}_{account_code or account[:20]}".replace(" ", "_")
	+
	+	try:
	+	appropriated_f = float(appropriated or 0)
	+	apportioned_f = float(apportioned or 0)
	+	except (ValueError, TypeError):
	+	appropriated_f = 0.0
	+	apportioned_f = 0.0
	+
	+	source_url = _max_gov_url(agency_code, account_code, year_str)
	+
	+	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"),
	+	}
	+
	+
	+	def poll(fy: Optional[int] = None, db_path: Optional[str] = None) -> int:
	+	"""
	+	Pull OMB apportionment records and write to the apportionments table.
	+
	+	Always links source_url to apportionment-public.max.gov, never to openomb.org.
	+
	+	Parameters
	+	----------
	+	fy : Fiscal year to fetch (default: current FY).
	+	db_path : Override default SQLite path.
	+
	+	Returns
	+	-------
	+	Number of new rows stored.
	+	"""
	+	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)
	+
	+	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:
	+	conn.close()
	+	return 0
	+
	+	stored = 0
	+	for rec in records:
	+	fields = _extract_fields(rec)
	+	if fields is None:
	+	continue
	+
	+	try:
	+	validate_source_url(fields["source_url"])
	+	except Exception as exc:
	+	logger.error(
	+	"openomb: source_url %r not allowlisted (%s) - skipping row",
	+	fields["source_url"], exc,
	+	)
	+	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"])
	+	continue
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO apportionments
	+	(source, external_id, agency, account, appropriated,
	+	apportioned, status, date, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	agency = excluded.agency,
	+	account = excluded.account,
	+	appropriated = excluded.appropriated,
	+	apportioned = excluded.apportioned,
	+	status = excluded.status,
	+	date = excluded.date,
	+	source_url = excluded.source_url
	+	""",
	+	(
	+	SOURCE,
	+	fields["external_id"],
	+	fields["agency"],
	+	fields["account"],
	+	fields["appropriated"],
	+	fields["apportioned"],
	+	fields["status"],
	+	fields["date"],
	+	fields["source_url"],
	+	),
	+	)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(SOURCE, fields["external_id"], json.dumps(rec), datetime.utcnow().isoformat()),
	+	)
	+
	+	mark_seen(conn, SOURCE, fields["external_id"], update_date)
	+	stored += 1
	+
	+	conn.commit()
	+	conn.close()
	+	logger.info("openomb: stored %d new apportionment row(s)", stored)
	+	return stored
	+
	+
	+	if __name__ == "__main__":
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	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)")
	+	args = parser.parse_args()
	+
	+	count = poll(fy=args.fy)
	+	print(f"Stored {count} new apportionments row(s).")
	+	sys.exit(0)

services/ingest/ingest/pollers/openstates.py +343 -101

		@@ -1,23 +1,25 @@
		"""
	-	OpenStates API v3 poller - 50-state + DC/PR legislative data.
	+	OpenStates API v3 poller - 50 states + DC + PR legislative data.

		Base: https://v3.openstates.org/
		Auth: API key required (set DOCKET_OPENSTATES_KEY).
	-	Tiers: Free 500/day (too tight for prod) → paid Bronze/Silver.
	-	Skips gracefully if key is not set.
	+	If absent, logs clearly and returns 0 - no crash.
	+	Tiers: Free 500/day (too tight for prod) → Bronze/Silver paid tiers.
	+	See https://openstates.org/data/ for pricing.

	-	Per PLAN.md: state ingest is in MVP scope but production needs the paid tier.
	-	An honest per-state coverage map must be maintained so the UI never implies
	-	completeness it doesn't have.
	+	Per DATA_SOURCES.md: aggregator source_url must point to the official state
	+	legislature record, NOT openstates.org. Bills lacking an allowlisted
	+	official URL are skipped (neutrality rule #1).

	-	Source URLs: bill detail URLs point to the state's official legislature (.gov).
	-	If the OpenStates record lacks an official URL, the bill is skipped
	-	(neutrality rule #1: no fact without an allowlisted .gov source_url).
	+	State coverage: after each successful jurisdiction poll this poller writes
	+	a scrape_log row via ingest.state_coverage.log_scrape(). state_coverage.py
	+	reads that table to compute per-state confidence.
		"""

		import json
		import logging
		import os
	+	import time
		from datetime import datetime, timedelta
		from typing import Optional

		@@ -25,31 +27,253 @@ import requests

		from ingest.change_detection import is_new_event, mark_seen
		from ingest.db import get_connection, init_schema
	-	from ingest.source_validator import validate_source_url
	+	from ingest.source_validator import is_allowlisted, validate_source_url

		logger = logging.getLogger(__name__)

	-	SOURCE = "openstates"
	-	BASE_URL = "https://v3.openstates.org"
	-	KEY_ENV = "DOCKET_OPENSTATES_KEY"
	+	SOURCE = "openstates"
	+	BASE_URL = "https://v3.openstates.org"
	+	KEY_ENV = "DOCKET_OPENSTATES_KEY"

	-	# States to poll in the initial MVP run (all 50 + DC can be added incrementally).
	+	# All 52 US jurisdictions OpenStates covers (50 states + DC + PR).
	+	# These are the jurisdiction codes accepted by the v3 /bills endpoint.
	+	ALL_JURISDICTIONS = [
	+	"al", "ak", "az", "ar", "ca", "co", "ct", "de", "fl", "ga",
	+	"hi", "id", "il", "in", "ia", "ks", "ky", "la", "me", "md",
	+	"ma", "mi", "mn", "ms", "mo", "mt", "ne", "nv", "nh", "nj",
	+	"nm", "ny", "nc", "nd", "oh", "ok", "or", "pa", "ri", "sc",
	+	"sd", "tn", "tx", "ut", "vt", "va", "wa", "wv", "wi", "wy",
	+	"dc", "pr",
	+	]
	+
	+	# Default: start with the top-10 most-populated states for MVP polling.
	+	# Set DOCKET_OPENSTATES_ALL=1 to poll all 52 jurisdictions (uses more quota).
		MVP_JURISDICTIONS = [
	-	"us", # federal (fallback - OpenStates also covers Congress)
	-	"ca", "tx", "fl", "ny", "il", "pa", "oh", "ga", "nc", "mi",
	+	"ca", "tx", "fl", "ny", "pa", "il", "oh", "ga", "nc", "mi",
	+	"dc", "pr",
		]

	+	# Requests per second - free tier allows 10 rps; stay well under it.
	+	_RATE_SLEEP = 0.15 # seconds between requests
	+

		def _get_api_key() -> Optional[str]:
		return os.environ.get(KEY_ENV)


	-	def poll(days_back: int = 1, db_path: Optional[str] = None) -> int:
	+	def _state_code_from_ocd(ocd_id: str) -> str:
	+	"""
	+	Extract uppercase 2-letter state code from an OCD jurisdiction or bill ID.
	+
	+	Examples:
	+	'ocd-jurisdiction/country:us/state:ca/government' -> 'CA'
	+	'ocd-jurisdiction/country:us/district:dc/government' -> 'DC'
	+	'ocd-jurisdiction/country:us/territory:pr/government'-> 'PR'
	+	'ocd-bill/country:us/state:ca/legislature/2025-2026/HB1' -> 'CA'
	+	"""
	+	for segment in ocd_id.split("/"):
	+	for prefix in ("state:", "district:", "territory:"):
	+	if segment.startswith(prefix):
	+	return segment[len(prefix):].upper()
	+	return ""
	+
	+
	+	def _pick_source_url(sources: list) -> str:
	+	"""
	+	Return the first source URL that passes the allowlist check.
	+	Returns empty string if none qualify (bill will be skipped).
	+	"""
	+	for s in sources:
	+	url = s.get("url", "") if isinstance(s, dict) else ""
	+	if url and is_allowlisted(url):
	+	return url
	+	return ""
	+
	+
	+	def _chamber(classification: str) -> str:
	+	mapping = {"upper": "Senate", "lower": "House", "legislature": "Legislature"}
	+	return mapping.get(classification, classification or "Unknown")
	+
	+
	+	def _init_bills_jurisdiction(conn) -> None:
	+	"""Add jurisdiction column to bills table if it doesn't exist yet."""
	+	try:
	+	conn.execute("ALTER TABLE bills ADD COLUMN jurisdiction TEXT")
	+	conn.commit()
	+	except Exception:
	+	pass # Column already exists - safe to ignore
	+
	+
	+	def _store_bill(conn, bill: dict, state_code: str, api_key: str) -> Optional[int]:
	+	"""
	+	Upsert one bill row; return the internal bills.id (or None if skipped).
	+
	+	Skips bills with no allowlisted official source URL (rule #1).
		"""
	-	Fetch recently updated bills from OpenStates for MVP_JURISDICTIONS.
	+	external_id = bill.get("id", "")
	+	update_date = (bill.get("updated_at") or "")[:10]
	+
	+	if not external_id:
	+	return None
	+
	+	if not is_new_event(conn, SOURCE, external_id, update_date):
	+	return None
	+
	+	sources = bill.get("sources", [])
	+	source_url = _pick_source_url(sources)
	+	if not source_url:
	+	logger.debug(
	+	"openstates %s bill %s: no allowlisted source_url - skipping (rule #1)",
	+	state_code, external_id,
	+	)
	+	return None
	+
	+	session = bill.get("legislative_session", {})
	+	session_str = session.get("identifier", "") if isinstance(session, dict) else str(session)
	+	number = bill.get("identifier", "")
	+	title = bill.get("title", "")
	+	status = bill.get("status", "")
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO bills
	+	(source, external_id, number, title, status, updated_at, source_url, jurisdiction)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	title = excluded.title,
	+	status = excluded.status,
	+	updated_at = excluded.updated_at,
	+	source_url = excluded.source_url,
	+	jurisdiction = excluded.jurisdiction
	+	""",
	+	(SOURCE, external_id, number, title, status, update_date, source_url, state_code),
	+	)
	+
	+	row = conn.execute(
	+	"SELECT id FROM bills WHERE source=? AND external_id=?",
	+	(SOURCE, external_id),
	+	).fetchone()
	+	bill_db_id = row["id"] if row else None
	+
	+	# Actions
	+	for action in bill.get("actions", []):
	+	action_date = (action.get("date") or "")[:10]
	+	action_text = action.get("description", "")
	+	org = action.get("organization", {})
	+	org_class = org.get("classification", "") if isinstance(org, dict) else ""
	+	chamber = _chamber(org_class)
	+	if bill_db_id and action_date:
	+	conn.execute(
	+	"""
	+	INSERT INTO actions
	+	(bill_id, source, action_type, action_date, chamber, text, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(bill_db_id, SOURCE, "legislative_action",
	+	action_date, chamber, action_text, source_url),
	+	)
	+
	+	# Raw event
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(SOURCE, external_id, json.dumps(bill), datetime.utcnow().isoformat()),
	+	)
	+
	+	mark_seen(conn, SOURCE, external_id, update_date)
	+	return bill_db_id
	+
	+
	+	def _store_votes(conn, bill_db_id: int, votes: list, bill_source_url: str,
	+	state_code: str) -> int:
	+	"""
	+	Insert vote summary rows for one bill; return count of new votes stored.
	+	"""
	+	stored = 0
	+	for vote in votes:
	+	vote_id = vote.get("id", "")
	+	vote_date = (vote.get("start_date") or "")[:10]
	+	if not vote_id or not vote_date:
	+	continue
	+
	+	if not is_new_event(conn, f"{SOURCE}_vote", vote_id, vote_date):
	+	continue
	+
	+	org = vote.get("organization", {})
	+	org_class = org.get("classification", "") if isinstance(org, dict) else ""
	+	chamber = _chamber(org_class)
	+
	+	result = vote.get("result", "unknown")
	+	counts = vote.get("counts", [])
	+	yea = nay = present = 0
	+	for c in counts:
	+	opt = c.get("option", "").lower()
	+	val = int(c.get("value") or 0)
	+	if opt in ("yes", "yea"):
	+	yea = val
	+	elif opt in ("no", "nay"):
	+	nay = val
	+	elif opt in ("other", "absent", "present", "not voting"):
	+	present += val
	+
	+	# Vote source URL: prefer official source from vote sources; fall back to bill URL.
	+	vote_sources = vote.get("sources", [])
	+	vote_source_url = _pick_source_url(vote_sources) or bill_source_url
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO votes (bill_id, chamber, date, yea, nay, present, result, source_url)
	+	VALUES (?, ?, ?, ?, ?, ?, ?, ?)
	+	""",
	+	(bill_db_id, chamber, vote_date, yea, nay, present, result, vote_source_url),
	+	)
	+
	+	# Raw event for the vote itself
	+	conn.execute(
	+	"""
	+	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	+	VALUES (?, ?, ?, ?)
	+	ON CONFLICT(source, external_id) DO UPDATE SET
	+	payload = excluded.payload,
	+	fetched_at = excluded.fetched_at
	+	""",
	+	(
	+	f"{SOURCE}_vote", vote_id,
	+	json.dumps(vote), datetime.utcnow().isoformat(),
	+	),
	+	)
	+
	+	mark_seen(conn, f"{SOURCE}_vote", vote_id, vote_date)
	+	stored += 1
	+
	+	return stored
	+
	+
	+	def poll(
	+	days_back: int = 1,
	+	db_path: Optional[str] = None,
	+	jurisdictions: Optional[list] = None,
	+	max_pages: int = 3,
	+	) -> int:
	+	"""
	+	Fetch recently updated bills (+ embedded votes + sponsorships) from OpenStates.

		If DOCKET_OPENSTATES_KEY is not set, logs a clear message and returns 0.
	-	NOTE: Free tier (500 req/day) is too tight for prod - a paid plan is required.
	+
	+	Set DOCKET_OPENSTATES_ALL=1 env var to poll all 52 jurisdictions instead
	+	of the MVP subset.
	+
	+	Parameters
	+	----------
	+	days_back : How many days back to query (default 1).
	+	db_path : Override DB path (useful in tests).
	+	jurisdictions: Override the jurisdiction list (tests / specific runs).
	+	max_pages : Max pagination pages per jurisdiction (caps API usage).

		Returns
		-------
		@@ -59,103 +283,121 @@ def poll(days_back: int = 1, db_path: Optional[str] = None) -> int:
		if not api_key:
		logger.warning(
		"OpenStates poller: DOCKET_OPENSTATES_KEY is not set. "
	-	"Free tier: 500 req/day (insufficient for prod). "
	-	"Contact OpenStates for Bronze/Silver paid tier pricing. "
	-	"Skipping state ingest this run."
	+	"Free key signup: https://openstates.org/accounts/login/ "
	+	"(free tier = 500 req/day, too tight for prod; contact OpenStates "
	+	"for Bronze/Silver paid tier). Skipping state ingest this run."
		)
		return 0

		conn = get_connection(db_path) if db_path else get_connection()
		init_schema(conn)
	-	stored = 0
	+	_init_bills_jurisdiction(conn)
	+
	+	# Import here to avoid circular import at module load time
	+	try:
	+	from ingest.state_coverage import log_scrape
	+	_log = log_scrape
	+	except ImportError:
	+	_log = None
	+
	+	if jurisdictions is None:
	+	if os.environ.get("DOCKET_OPENSTATES_ALL"):
	+	jurisdictions = ALL_JURISDICTIONS
	+	else:
	+	jurisdictions = MVP_JURISDICTIONS

		since = (datetime.utcnow() - timedelta(days=days_back)).strftime("%Y-%m-%d")
		headers = {"X-API-KEY": api_key}

	-	for jid in MVP_JURISDICTIONS:
	-	url = f"{BASE_URL}/bills"
	-	params = {
	-	"jurisdiction": jid,
	-	"updated_since": since,
	-	"sort": "-updated_at",
	-	"per_page": 20,
	-	"page": 1,
	-	}
	-	try:
	-	resp = requests.get(url, params=params, headers=headers, timeout=30)
	-	resp.raise_for_status()
	-	except requests.RequestException as exc:
	-	logger.error("OpenStates %s request failed: %s", jid, exc)
	-	continue
	+	total_bills = 0
	+	total_votes = 0
	+
	+	for jid in jurisdictions:
	+	state_code = jid.upper()
	+	bills_this_jid = 0
	+	votes_this_jid = 0

	-	bills = resp.json().get("results", [])
	-	logger.info("OpenStates %s: received %d bill(s)", jid, len(bills))
	-
	-	for bill in bills:
	-	bill_id = bill.get("id", "")
	-	update_date = bill.get("updated_at", "")[:10] if bill.get("updated_at") else ""
	-
	-	if not bill_id or not is_new_event(conn, SOURCE, bill_id, update_date):
	-	continue
	-
	-	# Require an official .gov source URL (neutrality rule #1).
	-	sources = bill.get("sources", [])
	-	source_url = ""
	-	for s in sources:
	-	url_candidate = s.get("url", "")
	-	if validate_source_url.__module__ and url_candidate:
	-	try:
	-	validate_source_url(url_candidate)
	-	source_url = url_candidate
	-	break
	-	except Exception:
	-	continue
	-
	-	if not source_url:
	-	logger.debug(
	-	"OpenStates %s bill %s: no allowlisted .gov source_url - skipping (rule #1)",
	-	jid,
	-	bill_id,
	+	page = 1
	+	while page <= max_pages:
	+	params = {
	+	"jurisdiction": jid,
	+	"updated_since": since,
	+	"sort": "-updated_at",
	+	"per_page": 20,
	+	"page": page,
	+	"include": ["votes", "sponsorships", "actions"],
	+	}
	+	try:
	+	time.sleep(_RATE_SLEEP)
	+	resp = requests.get(
	+	f"{BASE_URL}/bills",
	+	params=params,
	+	headers=headers,
	+	timeout=30,
		)
	-	continue
	+	resp.raise_for_status()
	+	except requests.RequestException as exc:
	+	logger.error("openstates %s page %d request failed: %s", jid, page, exc)
	+	break

	-	conn.execute(
	-	"""
	-	INSERT INTO bills
	-	(source, external_id, number, title, status, updated_at, source_url)
	-	VALUES (?, ?, ?, ?, ?, ?, ?)
	-	ON CONFLICT(source, external_id) DO UPDATE SET
	-	title = excluded.title,
	-	status = excluded.status,
	-	updated_at = excluded.updated_at,
	-	source_url = excluded.source_url
	-	""",
	-	(
	-	SOURCE,
	-	bill_id,
	-	bill.get("identifier", ""),
	-	bill.get("title", ""),
	-	bill.get("status", ""),
	-	update_date,
	-	source_url,
	-	),
	-	)
	+	data = resp.json()
	+	bills = data.get("results", [])
	+	pagination = data.get("pagination", {})

	-	conn.execute(
	-	"""
	-	INSERT INTO raw_events (source, external_id, payload, fetched_at)
	-	VALUES (?, ?, ?, ?)
	-	ON CONFLICT(source, external_id) DO UPDATE SET
	-	payload = excluded.payload,
	-	fetched_at = excluded.fetched_at
	-	""",
	-	(SOURCE, bill_id, json.dumps(bill), datetime.utcnow().isoformat()),
	+	logger.info(
	+	"openstates %s page %d: %d bill(s) (total=%s)",
	+	jid, page, len(bills),
	+	pagination.get("total_items", "?"),
		)

	-	mark_seen(conn, SOURCE, bill_id, update_date)
	-	stored += 1
	+	if not bills:
	+	break
	+
	+	for bill in bills:
	+	# Determine source URL from bill sources
	+	sources = bill.get("sources", [])
	+	source_url = _pick_source_url(sources)
	+
	+	bill_db_id = _store_bill(conn, bill, state_code, api_key)
	+	if bill_db_id is not None:
	+	bills_this_jid += 1
	+	# Store embedded votes
	+	if source_url:
	+	v = _store_votes(
	+	conn,
	+	bill_db_id,
	+	bill.get("votes", []),
	+	source_url,
	+	state_code,
	+	)
	+	votes_this_jid += v
	+
	+	conn.commit()
	+
	+	# Check if there are more pages
	+	max_page = pagination.get("max_page", 1)
	+	if page >= max_page:
	+	break
	+	page += 1
	+
	+	logger.info(
	+	"openstates %s: stored %d bill(s), %d vote(s)",
	+	jid, bills_this_jid, votes_this_jid,
	+	)
	+	total_bills += bills_this_jid
	+	total_votes += votes_this_jid
	+
	+	# Log scrape to state_coverage tracking
	+	if _log:
	+	now_iso = datetime.utcnow().isoformat()
	+	_log(conn, SOURCE, state_code, "bills", bills_this_jid, now_iso)
	+	_log(conn, SOURCE, state_code, "votes", votes_this_jid, now_iso)
	+	_log(conn, SOURCE, state_code, "sponsors", bills_this_jid, now_iso)

		conn.commit()
		conn.close()
	-	logger.info("OpenStates: stored %d new bill(s)", stored)
	-	return stored
	+	logger.info(
	+	"openstates: stored %d new bill(s), %d new vote(s) total",
	+	total_bills, total_votes,
	+	)
	+	return total_bills

services/ingest/ingest/rider_detect.py +209 -0

		@@ -0,0 +1,209 @@
	+	"""
	+	Rider / policy-rider detection - R Street keyword signal scan.
	+
	+	Runs over ADDED sections in bill_diffs. Sets rider_flag=1 on any row
	+	whose snippet contains the R Street keyword signals.
	+
	+	R Street signals (per DATA_SOURCES.md):
	+	Notwithstanding \| Is Amended \| Are Authorized \| Provided, \|
	+	Provided Further, \| In General \| Unless \| Until \| Determines
	+
	+	IMPORTANT: rider_flag is a CANDIDATE flag, not a verdict.
	+	The UI must label it "may contain a policy rider" - never "IS a rider."
	+
	+	Appropriations bills are the primary target (type: hjres, s/h appropriations).
	+	The scan also runs on other bill types but the signal weight is higher for
	+	appropriations.
	+	"""
	+
	+	import logging
	+	import re
	+	import sqlite3
	+	from typing import List, Optional
	+
	+	logger = logging.getLogger(__name__)
	+
	+	# ---------------------------------------------------------------------------
	+	# Rider signal patterns (R Street Institute keyword heuristics)
	+	# ---------------------------------------------------------------------------
	+
	+	_RIDER_PATTERNS: List[re.Pattern] = [
	+	re.compile(r"\bNotwithstanding\b", re.IGNORECASE),
	+	re.compile(r"\bIs\s+Amended\b", re.IGNORECASE),
	+	re.compile(r"\bAre\s+Authorized\b", re.IGNORECASE),
	+	re.compile(r"\bProvided,", re.IGNORECASE),
	+	re.compile(r"\bProvided\s+Further,", re.IGNORECASE),
	+	re.compile(r"\bIn\s+General\b", re.IGNORECASE),
	+	re.compile(r"\bUnless\b", re.IGNORECASE),
	+	re.compile(r"\bUntil\b", re.IGNORECASE),
	+	re.compile(r"\bDetermines\b", re.IGNORECASE),
	+	]
	+
	+	# Extra appropriations-specific patterns (§ amendment / authorization language)
	+	_APPROPRIATIONS_PATTERNS: List[re.Pattern] = [
	+	re.compile(r"\bno\s+funds?\s+(appropriated\|made\s+available)\b", re.IGNORECASE),
	+	re.compile(r"\bnone\s+of\s+the\s+funds?\b", re.IGNORECASE),
	+	re.compile(r"\bappropriation[s]?\s+herein\b", re.IGNORECASE),
	+	re.compile(r"\bsec(?:tion)?\s+\d+\s+of\s+the\b", re.IGNORECASE),
	+	]
	+
	+	_ALL_PATTERNS = _RIDER_PATTERNS + _APPROPRIATIONS_PATTERNS
	+
	+
	+	def is_rider_signal(text: str) -> bool:
	+	"""
	+	Return True if the text contains any R Street keyword signal.
	+	Checks base signals only (use is_rider_signal_strong for both sets).
	+	"""
	+	for pat in _RIDER_PATTERNS:
	+	if pat.search(text):
	+	return True
	+	return False
	+
	+
	+	def is_rider_signal_strong(text: str) -> bool:
	+	"""Return True if any rider or appropriations keyword matches."""
	+	for pat in _ALL_PATTERNS:
	+	if pat.search(text):
	+	return True
	+	return False
	+
	+
	+	def matched_signals(text: str) -> List[str]:
	+	"""Return list of pattern strings that matched (for debugging/logging)."""
	+	matched = []
	+	for pat in _ALL_PATTERNS:
	+	if pat.search(text):
	+	matched.append(pat.pattern)
	+	return matched
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Bill-type helpers
	+	# ---------------------------------------------------------------------------
	+
	+	_APPROPRIATIONS_TYPES = frozenset(["hjres", "sjres", "hconres", "sconres"])
	+	_APPROPRIATIONS_KEYWORDS = re.compile(
	+	r"\b(appropriat\|omnibus\|continuing\s+resolution\|supplemental)\b", re.IGNORECASE
	+	)
	+
	+
	+	def _is_appropriations_bill(conn: sqlite3.Connection, bill_id: int) -> bool:
	+	"""Heuristic: is this an appropriations or CR bill?"""
	+	row = conn.execute(
	+	"SELECT b.number, b.title, b.policy_area, be.bill_type "
	+	"FROM bills b LEFT JOIN bills_ext be ON be.bill_id=b.id "
	+	"WHERE b.id=?",
	+	(bill_id,),
	+	).fetchone()
	+	if row is None:
	+	return False
	+	bill_type = (row["bill_type"] or "").lower()
	+	title = row["title"] or ""
	+	policy = row["policy_area"] or ""
	+	if bill_type in _APPROPRIATIONS_TYPES:
	+	return True
	+	if _APPROPRIATIONS_KEYWORDS.search(title) or _APPROPRIATIONS_KEYWORDS.search(policy):
	+	return True
	+	return False
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Core scan
	+	# ---------------------------------------------------------------------------
	+
	+	def scan_bill_diffs(bill_id: int, conn: sqlite3.Connection) -> int:
	+	"""
	+	Scan ADDED diff sections for this bill and set rider_flag where signals match.
	+	Returns the number of rows flagged.
	+	"""
	+	is_approp = _is_appropriations_bill(conn, bill_id)
	+
	+	rows = conn.execute(
	+	"SELECT id, snippet FROM bill_diffs WHERE bill_id=? AND change='added'",
	+	(bill_id,),
	+	).fetchall()
	+
	+	flagged = 0
	+	for row in rows:
	+	snippet = row["snippet"] or ""
	+	# Use strong check for appropriations bills, base check otherwise
	+	if is_approp:
	+	hit = is_rider_signal_strong(snippet)
	+	else:
	+	hit = is_rider_signal(snippet)
	+
	+	if hit:
	+	signals = matched_signals(snippet)
	+	logger.debug(
	+	"Rider flag: bill_id=%s diff_id=%s signals=%s",
	+	bill_id, row["id"], signals,
	+	)
	+	conn.execute(
	+	"UPDATE bill_diffs SET rider_flag=1 WHERE id=?", (row["id"],)
	+	)
	+	flagged += 1
	+
	+	return flagged
	+
	+
	+	def scan_all_bills(conn: sqlite3.Connection, congress: int = 119) -> int:
	+	"""
	+	Run rider detection over all bills with unscanned added diffs.
	+	Returns total number of flagged rows.
	+	"""
	+	# Bills that have at least one added diff row (rider_flag may already be set)
	+	rows = conn.execute(
	+	"""
	+	SELECT DISTINCT bd.bill_id
	+	FROM bill_diffs bd
	+	JOIN bills b ON b.id = bd.bill_id
	+	WHERE b.congress=? AND bd.change='added'
	+	""",
	+	(congress,),
	+	).fetchall()
	+
	+	total = 0
	+	for row in rows:
	+	n = scan_bill_diffs(row["bill_id"], conn)
	+	total += n
	+
	+	conn.commit()
	+	logger.info("rider_detect: %d rider flags set across %d bills", total, len(rows))
	+	return total
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# CLI
	+	# ---------------------------------------------------------------------------
	+
	+	if __name__ == "__main__":
	+	import argparse
	+	import sys
	+	from ingest.db import get_connection, init_schema
	+	from ingest.pollers.bills_bulk import ensure_schema
	+
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+	ap = argparse.ArgumentParser(description="Rider detection over bill_diffs")
	+	ap.add_argument("--once", action="store_true")
	+	ap.add_argument("--bill-id", type=int, help="Scan a specific bill")
	+	ap.add_argument("--congress", type=int, default=119)
	+	args = ap.parse_args()
	+
	+	conn = get_connection()
	+	init_schema(conn)
	+	ensure_schema(conn)
	+
	+	if args.bill_id:
	+	n = scan_bill_diffs(args.bill_id, conn)
	+	conn.commit()
	+	print(f"Flagged {n} diff rows for bill_id={args.bill_id}")
	+	else:
	+	n = scan_all_bills(conn, congress=args.congress)
	+	print(f"Total flagged: {n}")
	+
	+	conn.close()

services/ingest/ingest/source_validator.py +109 -5

		@@ -4,6 +4,12 @@ Source URL validation - neutrality rule #1.
		Every stored fact MUST carry a source_url whose host is an allowlisted .gov
		domain. This module is the single enforcement point; call validate_source_url()
		before writing any row.
	+
	+	State legislature portals: many states use .gov subdomains (already covered by
	+	the *.gov wildcard), but a minority use .us or .org TLDs. STATE_LEG_HOSTS
	+	enumerates those non-.gov hosts explicitly, and a pattern rule catches the
	+	common *.state.<XX>.us convention. Aggregator sources (OpenStates, LegiScan)
	+	must resolve to the underlying official state record - not the aggregator.
		"""

		from urllib.parse import urlparse
		@@ -39,6 +45,92 @@ ALLOWLISTED_HOSTS = frozenset(
		}
		)

	+	# State legislature portals that do NOT use a .gov TLD.
	+	# Sourced from each state's official legislature website.
	+	# Pattern *.state.<XX>.us is also handled in _is_state_leg_pattern().
	+	STATE_LEG_HOSTS = frozenset(
	+	{
	+	# Pennsylvania - legis.state.pa.us
	+	"legis.state.pa.us",
	+	"www.legis.state.pa.us",
	+	# Texas - capitol.state.tx.us and legis.state.tx.us
	+	"capitol.state.tx.us",
	+	"www.capitol.state.tx.us",
	+	"legis.state.tx.us",
	+	"www.legis.state.tx.us",
	+	# Minnesota - leg.state.mn.us
	+	"leg.state.mn.us",
	+	"www.leg.state.mn.us",
	+	"house.leg.state.mn.us",
	+	"www.house.leg.state.mn.us",
	+	"senate.leg.state.mn.us",
	+	"www.senate.leg.state.mn.us",
	+	# Mississippi - ls.state.ms.us
	+	"ls.state.ms.us",
	+	"www.ls.state.ms.us",
	+	"billstatus.ls.state.ms.us",
	+	# Nevada - leg.state.nv.us
	+	"leg.state.nv.us",
	+	"www.leg.state.nv.us",
	+	# New Hampshire - gencourt.state.nh.us
	+	"gencourt.state.nh.us",
	+	"www.gencourt.state.nh.us",
	+	# New Jersey - njleg.state.nj.us
	+	"njleg.state.nj.us",
	+	"www.njleg.state.nj.us",
	+	# Alabama - alison.legislature.state.al.us
	+	"alison.legislature.state.al.us",
	+	"legislature.state.al.us",
	+	"www.legislature.state.al.us",
	+	# Alaska - legis.state.ak.us (also legis.alaska.gov which is .gov)
	+	"legis.state.ak.us",
	+	"www.legis.state.ak.us",
	+	# Arkansas - arkleg.state.ar.us
	+	"arkleg.state.ar.us",
	+	"www.arkleg.state.ar.us",
	+	# Kansas - kslegislature.org (unique .org)
	+	"kslegislature.org",
	+	"www.kslegislature.org",
	+	# Puerto Rico legislature (not .gov)
	+	"www.oslpr.org",
	+	"oslpr.org",
	+	"www.camaraderepresentantes.org",
	+	"camaraderepresentantes.org",
	+	"www.senadopr.us",
	+	"senadopr.us",
	+	# DC Council (dccouncil.gov is .gov but dccouncil.us exists too)
	+	"dccouncil.us",
	+	"www.dccouncil.us",
	+	"lims.dccouncil.us",
	+	}
	+	)
	+
	+
	+	def _is_state_leg_pattern(host: str) -> bool:
	+	"""
	+	Return True for the *.state.<XX>.us pattern used by many state legislatures.
	+
	+	Examples that match:
	+	legis.state.pa.us
	+	www.leg.state.mn.us
	+	njleg.state.nj.us
	+	billstatus.ls.state.ms.us (depth > 4 - still matches)
	+
	+	Examples that do NOT match:
	+	govtrack.us (only 2 parts)
	+	example.state.co (wrong TLD)
	+	"""
	+	parts = host.split(".")
	+	# Need at least 4 segments: <sub>.state.<XX>.us
	+	if len(parts) < 4:
	+	return False
	+	if parts[-1] != "us":
	+	return False
	+	# The "state" segment must be 3rd from the end (for 4-part) or somewhere inside
	+	# for deeper hostnames. Scan from the right: ...<state>.<XX>.us
	+	# Canonical form: anything.state.<2-letter-code>.us
	+	return "state" in parts[:-2]
	+

		class SourceURLError(ValueError):
		"""Raised when a source_url host is not allowlisted."""
		@@ -46,8 +138,13 @@ class SourceURLError(ValueError):

		def is_allowlisted(url: str) -> bool:
		"""
	-	Return True if the URL's host is on the allowlist OR ends with .gov.
	-	The *.gov wildcard in the spec covers all official government domains.
	+	Return True if the URL's host passes the source allowlist.
	+
	+	Allowlist rules (in priority order):
	+	1. Host is in ALLOWLISTED_HOSTS (explicit federal list).
	+	2. Host ends with .gov (wildcard - covers all official .gov domains).
	+	3. Host is in STATE_LEG_HOSTS (explicit non-.gov state legislature list).
	+	4. Host matches the *.state.<XX>.us pattern (implicit state legislature rule).
		"""
		if not url:
		return False
		@@ -55,10 +152,17 @@ def is_allowlisted(url: str) -> bool:
		host = urlparse(url).hostname or ""
		except Exception:
		return False
	+	if not host:
	+	return False
		if host in ALLOWLISTED_HOSTS:
		return True
	-	# wildcard *.gov - covers state legislatures linked from OpenStates
	-	return host.endswith(".gov")
	+	if host.endswith(".gov"):
	+	return True
	+	if host in STATE_LEG_HOSTS:
	+	return True
	+	if _is_state_leg_pattern(host):
	+	return True
	+	return False


		def validate_source_url(url: str) -> None:
		@@ -69,6 +173,6 @@ def validate_source_url(url: str) -> None:
		if not is_allowlisted(url):
		host = urlparse(url).hostname if url else "<empty>"
		raise SourceURLError(
	-	f"source_url host '{host}' is not an allowlisted .gov domain. "
	+	f"source_url host '{host}' is not an allowlisted .gov / state-leg domain. "
		f"url={url!r}"
		)

services/ingest/ingest/state_coverage.py +398 -0

		@@ -0,0 +1,398 @@
	+	"""
	+	State-level coverage honesty layer.
	+
	+	Computes and writes state_coverage rows for every US jurisdiction (50 states +
	+	DC + PR = 52 total) covering both OpenStates and LegiScan sources.
	+
	+	Confidence levels:
	+	ok - data scraped within the last STALE_HOURS hours
	+	stale - data present but older than STALE_HOURS
	+	missing - no data has ever been ingested for this (state, source, data_type)
	+
	+	Also provides log_scrape() - called by pollers after each successful run to
	+	record scrape timestamps without requiring a second full DB pass.
	+
	+	Run standalone:
	+	python -m ingest.state_coverage # computes + prints summary
	+	python -m ingest.state_coverage --json # JSON output
	+	"""
	+
	+	import json
	+	import logging
	+	import os
	+	import sys
	+	from datetime import datetime, timedelta, timezone
	+	from typing import Optional
	+
	+	from ingest.db import get_connection, init_schema
	+
	+	logger = logging.getLogger(__name__)
	+
	+	# Number of hours before a scrape is considered "stale"
	+	STALE_HOURS = 48
	+
	+	# All US jurisdictions tracked: 50 states + DC + PR
	+	ALL_STATES: list = [
	+	"AL", "AK", "AZ", "AR", "CA", "CO", "CT", "DE", "FL", "GA",
	+	"HI", "ID", "IL", "IN", "IA", "KS", "KY", "LA", "ME", "MD",
	+	"MA", "MI", "MN", "MS", "MO", "MT", "NE", "NV", "NH", "NJ",
	+	"NM", "NY", "NC", "ND", "OH", "OK", "OR", "PA", "RI", "SC",
	+	"SD", "TN", "TX", "UT", "VT", "VA", "WA", "WV", "WI", "WY",
	+	"DC", "PR",
	+	] # 52 entries
	+
	+	SOURCES = ["openstates", "legiscan"]
	+	DATA_TYPES = ["bills", "votes", "sponsors"]
	+
	+
	+	# ── Schema helpers ─────────────────────────────────────────────────────────────
	+
	+	def _init_coverage_schema(conn) -> None:
	+	"""Create state_coverage and scrape_log tables if they don't exist."""
	+	conn.executescript(
	+	"""
	+	CREATE TABLE IF NOT EXISTS state_coverage (
	+	state TEXT NOT NULL,
	+	source TEXT NOT NULL,
	+	data_type TEXT NOT NULL,
	+	last_scrape TEXT,
	+	confidence TEXT NOT NULL DEFAULT 'missing',
	+	PRIMARY KEY (state, source, data_type)
	+	);
	+
	+	CREATE TABLE IF NOT EXISTS scrape_log (
	+	source TEXT NOT NULL,
	+	state TEXT NOT NULL,
	+	data_type TEXT NOT NULL,
	+	scraped_at TEXT NOT NULL,
	+	row_count INTEGER NOT NULL DEFAULT 0,
	+	PRIMARY KEY (source, state, data_type)
	+	);
	+	"""
	+	)
	+	conn.commit()
	+
	+
	+	# ── scrape_log writer (called by pollers) ─────────────────────────────────────
	+
	+	def log_scrape(
	+	conn,
	+	source: str,
	+	state: str,
	+	data_type: str,
	+	row_count: int,
	+	scraped_at: Optional[str] = None,
	+	) -> None:
	+	"""
	+	Record a successful scrape in scrape_log.
	+
	+	Called by pollers (openstates.py, legiscan.py) after each jurisdiction run.
	+	Creates the scrape_log table if it doesn't exist yet.
	+
	+	Parameters
	+	----------
	+	conn : Open SQLite connection (poller manages its own lifecycle).
	+	source : 'openstates' or 'legiscan'
	+	state : Uppercase 2-letter state code (e.g. 'CA')
	+	data_type : 'bills', 'votes', or 'sponsors'
	+	row_count : Number of rows stored in this run (0 is valid - means scraped but empty)
	+	scraped_at: ISO datetime string; defaults to now UTC.
	+	"""
	+	_init_coverage_schema(conn)
	+	if scraped_at is None:
	+	scraped_at = datetime.now(timezone.utc).isoformat()
	+	conn.execute(
	+	"""
	+	INSERT INTO scrape_log (source, state, data_type, scraped_at, row_count)
	+	VALUES (?, ?, ?, ?, ?)
	+	ON CONFLICT(source, state, data_type) DO UPDATE SET
	+	scraped_at = excluded.scraped_at,
	+	row_count = excluded.row_count
	+	""",
	+	(source, state.upper(), data_type, scraped_at, row_count),
	+	)
	+	# Don't commit here - caller (poller) commits its own transaction.
	+
	+
	+	# ── Confidence computation ─────────────────────────────────────────────────────
	+
	+	def _confidence(last_scrape: Optional[str]) -> str:
	+	"""
	+	Compute confidence from a last_scrape ISO timestamp.
	+
	+	Returns 'missing' if last_scrape is None/empty,
	+	'ok' if within STALE_HOURS,
	+	'stale' otherwise.
	+	"""
	+	if not last_scrape:
	+	return "missing"
	+	try:
	+	# Handle both Z-suffix and +00:00 style
	+	ts_str = last_scrape.replace("Z", "+00:00")
	+	# If no timezone info, treat as UTC
	+	if "+" not in ts_str and ts_str.count("-") < 3:
	+	ts = datetime.fromisoformat(ts_str).replace(tzinfo=timezone.utc)
	+	else:
	+	ts = datetime.fromisoformat(ts_str)
	+	now = datetime.now(timezone.utc)
	+	if ts.tzinfo is None:
	+	ts = ts.replace(tzinfo=timezone.utc)
	+	age_hours = (now - ts).total_seconds() / 3600
	+	return "ok" if age_hours <= STALE_HOURS else "stale"
	+	except Exception:
	+	return "missing"
	+
	+
	+	def _build_scrape_index(conn) -> dict:
	+	"""
	+	Return {(source, state, data_type) -> scraped_at} from scrape_log.
	+	Falls back to inferring from raw_events if scrape_log has no entries.
	+	"""
	+	index = {}
	+
	+	try:
	+	rows = conn.execute(
	+	"SELECT source, state, data_type, scraped_at FROM scrape_log"
	+	).fetchall()
	+	for row in rows:
	+	key = (row["source"], row["state"].upper(), row["data_type"])
	+	index[key] = row["scraped_at"]
	+	except Exception:
	+	pass # Table might not exist yet
	+
	+	# Fallback: infer from raw_events for 'bills' data_type
	+	# OpenStates: external_id like 'ocd-bill/country:us/state:ca/...'
	+	# LegiScan: external_id like 'legiscan-CA-12345'
	+	if not index:
	+	index.update(_infer_from_raw_events(conn))
	+
	+	return index
	+
	+
	+	def _infer_state_from_ocd(ext_id: str) -> str:
	+	"""
	+	Extract 2-letter uppercase state code from an OCD bill ID.
	+	Returns '' if extraction fails.
	+	"""
	+	for segment in ext_id.split("/"):
	+	for prefix in ("state:", "district:", "territory:"):
	+	if segment.startswith(prefix):
	+	return segment[len(prefix):].upper()
	+	return ""
	+
	+
	+	def _infer_state_from_legiscan_id(ext_id: str) -> str:
	+	"""
	+	Extract 2-letter uppercase state code from 'legiscan-CA-12345'.
	+	Returns '' if extraction fails.
	+	"""
	+	parts = ext_id.split("-")
	+	if len(parts) >= 3 and parts[0] == "legiscan":
	+	candidate = parts[1].upper()
	+	if len(candidate) == 2 and candidate.isalpha():
	+	return candidate
	+	return ""
	+
	+
	+	def _infer_from_raw_events(conn) -> dict:
	+	"""
	+	Infer last-scrape timestamps from raw_events when scrape_log is empty.
	+	Returns {(source, state, 'bills') -> fetched_at}.
	+	"""
	+	index = {}
	+	try:
	+	rows = conn.execute(
	+	"SELECT source, external_id, fetched_at FROM raw_events "
	+	"WHERE source IN ('openstates', 'legiscan')"
	+	).fetchall()
	+	except Exception:
	+	return index
	+
	+	for row in rows:
	+	source = row["source"]
	+	ext_id = row["external_id"]
	+	fetched = row["fetched_at"]
	+
	+	if source == "openstates":
	+	state = _infer_state_from_ocd(ext_id)
	+	elif source == "legiscan":
	+	state = _infer_state_from_legiscan_id(ext_id)
	+	else:
	+	continue
	+
	+	if not state:
	+	continue
	+
	+	key = (source, state, "bills")
	+	existing = index.get(key)
	+	# Keep the most recent timestamp
	+	if existing is None or fetched > existing:
	+	index[key] = fetched
	+
	+	return index
	+
	+
	+	# ── Main compute function ──────────────────────────────────────────────────────
	+
	+	def compute(db_path: Optional[str] = None) -> list:
	+	"""
	+	Compute coverage for all 52 US jurisdictions and write state_coverage rows.
	+
	+	Works entirely from DB state - no API calls. All states start as 'missing'
	+	until a poller has actually ingested data for them.
	+
	+	Returns
	+	-------
	+	List of dicts: {state, source, data_type, last_scrape, confidence}
	+	"""
	+	conn = get_connection(db_path) if db_path else get_connection()
	+	init_schema(conn)
	+	_init_coverage_schema(conn)
	+
	+	scrape_index = _build_scrape_index(conn)
	+
	+	rows_out = []
	+
	+	for state in ALL_STATES:
	+	for source in SOURCES:
	+	for data_type in DATA_TYPES:
	+	key = (source, state, data_type)
	+	last_scrape = scrape_index.get(key)
	+	confidence = _confidence(last_scrape)
	+
	+	conn.execute(
	+	"""
	+	INSERT INTO state_coverage
	+	(state, source, data_type, last_scrape, confidence)
	+	VALUES (?, ?, ?, ?, ?)
	+	ON CONFLICT(state, source, data_type) DO UPDATE SET
	+	last_scrape = excluded.last_scrape,
	+	confidence = excluded.confidence
	+	""",
	+	(state, source, data_type, last_scrape, confidence),
	+	)
	+
	+	rows_out.append({
	+	"state": state,
	+	"source": source,
	+	"data_type": data_type,
	+	"last_scrape": last_scrape,
	+	"confidence": confidence,
	+	})
	+
	+	conn.commit()
	+
	+	# Summary log
	+	ok_count = sum(1 for r in rows_out if r["confidence"] == "ok")
	+	stale_count = sum(1 for r in rows_out if r["confidence"] == "stale")
	+	missing_count = sum(1 for r in rows_out if r["confidence"] == "missing")
	+	logger.info(
	+	"state_coverage: %d rows written - ok=%d stale=%d missing=%d",
	+	len(rows_out), ok_count, stale_count, missing_count,
	+	)
	+
	+	conn.close()
	+	return rows_out
	+
	+
	+	def summary_by_state(db_path: Optional[str] = None) -> list:
	+	"""
	+	Return per-state rollup: {state, confidence, missing_types[]}.
	+
	+	Confidence is the worst confidence across all (source, data_type) combos
	+	for a given state: missing > stale > ok.
	+	"""
	+	rows = compute(db_path)
	+
	+	by_state = {}
	+	for r in rows:
	+	state = r["state"]
	+	if state not in by_state:
	+	by_state[state] = {"state": state, "confidence": "ok", "missing_types": []}
	+
	+	# Worst-case escalation
	+	conf = r["confidence"]
	+	current_worst = by_state[state]["confidence"]
	+	rank = {"ok": 0, "stale": 1, "missing": 2}
	+	if rank.get(conf, 0) > rank.get(current_worst, 0):
	+	by_state[state]["confidence"] = conf
	+
	+	if conf == "missing":
	+	label = f"{r['source']}:{r['data_type']}"
	+	by_state[state]["missing_types"].append(label)
	+
	+	return list(by_state.values())
	+
	+
	+	# ── CLI entry point ────────────────────────────────────────────────────────────
	+
	+	def main(argv: Optional[list] = None) -> int:
	+	"""
	+	Compute state coverage and print a summary.
	+
	+	Usage:
	+	python -m ingest.state_coverage # human-readable table
	+	python -m ingest.state_coverage --json # JSON array
	+	"""
	+	import argparse
	+
	+	logging.basicConfig(
	+	level=logging.INFO,
	+	format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
	+	stream=sys.stdout,
	+	)
	+
	+	parser = argparse.ArgumentParser(description="Compute Docket state coverage")
	+	parser.add_argument(
	+	"--json", action="store_true",
	+	help="Output full coverage rows as JSON instead of summary table",
	+	)
	+	parser.add_argument(
	+	"--db", metavar="PATH",
	+	help="Override DB path (default: DOCKET_DB_PATH env or /var/tmp/docket/docket.db)",
	+	)
	+	args = parser.parse_args(argv)
	+
	+	db_path = args.db or os.environ.get("DOCKET_DB_PATH")
	+	rows = compute(db_path)
	+
	+	if args.json:
	+	print(json.dumps(rows, indent=2))
	+	else:
	+	# Human-readable summary table
	+	rollup = summary_by_state(db_path)
	+	ok_states = [r["state"] for r in rollup if r["confidence"] == "ok"]
	+	stale_states = [r["state"] for r in rollup if r["confidence"] == "stale"]
	+	missing_states = [r["state"] for r in rollup if r["confidence"] == "missing"]
	+
	+	print(f"\n{'='*60}")
	+	print(f" Docket State Coverage - {datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M UTC')}")
	+	print(f"{'='*60}")
	+	print(f" Jurisdictions tracked : {len(ALL_STATES):>3} (50 states + DC + PR)")
	+	print(f" Total coverage rows : {len(rows):>3}")
	+	print(f" ok (fresh) : {len(ok_states):>3}")
	+	print(f" stale (>48h) : {len(stale_states):>3}")
	+	print(f" missing (never seen) : {len(missing_states):>3}")
	+	print(f"{'='*60}")
	+
	+	if ok_states:
	+	print(f"\n OK states: {', '.join(sorted(ok_states))}")
	+	if stale_states:
	+	print(f"\n Stale states: {', '.join(sorted(stale_states))}")
	+	if missing_states:
	+	print(f"\n Missing states ({len(missing_states)}): {', '.join(sorted(missing_states))}")
	+
	+	# Show any states with partial coverage
	+	partial = [r for r in rollup if r["missing_types"] and r["confidence"] != "missing"]
	+	if partial:
	+	print(f"\n Partial coverage ({len(partial)} states):")
	+	for r in sorted(partial, key=lambda x: x["state"]):
	+	print(f" {r['state']}: missing {', '.join(r['missing_types'])}")
	+
	+	print()
	+
	+	return 0
	+
	+
	+	if __name__ == "__main__":
	+	sys.exit(main())

services/ingest/list_crec.py +36 -0

		@@ -0,0 +1,36 @@
	+	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 +44 -0

		@@ -0,0 +1,44 @@
	+	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 +59 -0

		@@ -0,0 +1,59 @@
	+	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 +40 -0

		@@ -0,0 +1,40 @@
	+	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_bills_diff.py +404 -0

		@@ -0,0 +1,404 @@
	+	"""
	+	Unit tests for the bill diff classifier and rider detection regex.
	+
	+	All tests are pure-Python / no network / no DB.
	+	"""
	+
	+	import pytest
	+
	+	from ingest.diff import (
	+	Section,
	+	classify_change,
	+	align_sections,
	+	_fuzzy_score,
	+	parse_sections,
	+	_make_snippet,
	+	)
	+	from ingest.rider_detect import (
	+	is_rider_signal,
	+	is_rider_signal_strong,
	+	matched_signals,
	+	)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# classify_change tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestClassifyChange:
	+	def test_added_when_old_empty(self):
	+	assert classify_change(None, "new text here") == "added"
	+
	+	def test_added_when_old_blank(self):
	+	assert classify_change("", "new text here") == "added"
	+
	+	def test_removed_when_new_empty(self):
	+	assert classify_change("old text", None) == "removed"
	+
	+	def test_removed_when_new_blank(self):
	+	assert classify_change("old text", "") == "removed"
	+
	+	def test_amended_when_both_differ(self):
	+	assert classify_change("original language", "revised language") == "amended"
	+
	+	def test_no_change_when_identical(self):
	+	assert classify_change("same text", "same text") is None
	+
	+	def test_no_change_whitespace_stripped(self):
	+	assert classify_change(" same ", "same") is None
	+
	+	def test_added_short(self):
	+	assert classify_change("", "x") == "added"
	+
	+	def test_removed_short(self):
	+	assert classify_change("x", "") == "removed"
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# align_sections tests
	+	# ---------------------------------------------------------------------------
	+
	+	def _sec(sec_id="", enum="", header="", text=""):
	+	return Section(sec_id=sec_id, enum=enum, header=header, text=text)
	+
	+
	+	class TestAlignSections:
	+	def test_exact_id_match(self):
	+	old = [_sec(sec_id="H001", header="Short Title", text="old")]
	+	new = [_sec(sec_id="H001", header="Short Title", text="new")]
	+	pairs = align_sections(old, new)
	+	assert len(pairs) == 1
	+	assert pairs[0][0] is old[0]
	+	assert pairs[0][1] is new[0]
	+
	+	def test_added_section_no_id(self):
	+	old = [_sec(sec_id="H001", header="Definitions", text="some defs")]
	+	new = [
	+	_sec(sec_id="H001", header="Definitions", text="some defs"),
	+	_sec(sec_id="", header="New Provision", text="brand new"),
	+	]
	+	pairs = align_sections(old, new)
	+	# First pair: matched by id
	+	matched = [(o, n) for o, n in pairs if o and n]
	+	added = [(o, n) for o, n in pairs if o is None]
	+	assert len(matched) == 1
	+	assert len(added) == 1
	+
	+	def test_removed_section(self):
	+	old = [
	+	_sec(sec_id="H001", header="Authorization", text="text"),
	+	_sec(sec_id="H002", header="Definitions", text="defs"),
	+	]
	+	new = [_sec(sec_id="H001", header="Authorization", text="text")]
	+	pairs = align_sections(old, new)
	+	removed = [(o, n) for o, n in pairs if n is None]
	+	assert len(removed) == 1
	+	assert removed[0][0].sec_id == "H002"
	+
	+	def test_fuzzy_header_match(self):
	+	old = [_sec(sec_id="", enum="1.", header="Authorization of Appropriations")]
	+	new = [_sec(sec_id="", enum="1.", header="Authorization of Appropriationz")]
	+	pairs = align_sections(old, new)
	+	matched = [(o, n) for o, n in pairs if o and n]
	+	assert len(matched) == 1
	+
	+	def test_fuzzy_below_threshold_not_matched(self):
	+	old = [_sec(sec_id="", enum="1.", header="Authorization of Appropriations")]
	+	new = [_sec(sec_id="", enum="9.", header="Entirely Different Section")]
	+	pairs = align_sections(old, new)
	+	# Should not fuzzy-match; one added, one removed
	+	added = [(o, n) for o, n in pairs if o is None]
	+	removed = [(o, n) for o, n in pairs if n is None]
	+	assert len(added) == 1
	+	assert len(removed) == 1
	+
	+	def test_empty_old(self):
	+	new = [_sec(sec_id="H001", text="all new")]
	+	pairs = align_sections([], new)
	+	added = [(o, n) for o, n in pairs if o is None]
	+	assert len(added) == 1
	+
	+	def test_empty_new(self):
	+	old = [_sec(sec_id="H001", text="was here")]
	+	pairs = align_sections(old, [])
	+	removed = [(o, n) for o, n in pairs if n is None]
	+	assert len(removed) == 1
	+
	+	def test_both_empty(self):
	+	assert align_sections([], []) == []
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# _make_snippet tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestMakeSnippet:
	+	def test_added_snippet_length(self):
	+	long_text = "X" * 1000
	+	s = _make_snippet(None, long_text, "added")
	+	assert len(s) <= 510 # 500 + "…"
	+	assert s.endswith("…")
	+
	+	def test_removed_snippet_short(self):
	+	s = _make_snippet("short", None, "removed")
	+	assert s == "short"
	+
	+	def test_amended_has_diff_markers(self):
	+	s = _make_snippet("line one\nline two", "line one\nline three", "amended")
	+	# Unified diff output has +/- markers
	+	assert "+" in s or "-" in s or "@@" in s
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# parse_sections XML tests
	+	# ---------------------------------------------------------------------------
	+
	+	SIMPLE_BILL_XML = """<?xml version="1.0"?>
	+	<bill>
	+	<body>
	+	<section id="H001">
	+	<enum>1.</enum>
	+	<header>Short Title</header>
	+	<text>This Act may be cited as the Test Act.</text>
	+	</section>
	+	<section id="H002">
	+	<enum>2.</enum>
	+	<header>Definitions</header>
	+	<text>In this Act-</text>
	+	<subsection>
	+	<enum>(a)</enum>
	+	<header>Term</header>
	+	<text>The term "widget" means any gizmo.</text>
	+	</subsection>
	+	</section>
	+	</body>
	+	</bill>
	+	"""
	+
	+	NAMESPACED_BILL_XML = """<?xml version="1.0"?>
	+	<bill xmlns:ns="http://xml.house.gov/schemas/uslm/1.0">
	+	<ns:body>
	+	<ns:section id="S001">
	+	<ns:enum>1.</ns:enum>
	+	<ns:header>Purpose</ns:header>
	+	<ns:text>The purpose of this Act is to test parsing.</ns:text>
	+	</ns:section>
	+	</ns:body>
	+	</bill>
	+	"""
	+
	+
	+	class TestParseSections:
	+	def test_simple_xml_two_sections(self):
	+	sections = parse_sections(SIMPLE_BILL_XML)
	+	assert len(sections) == 2
	+	assert sections[0].sec_id == "H001"
	+	assert "Short Title" in sections[0].header
	+	assert "Test Act" in sections[0].text
	+
	+	def test_section_enum_captured(self):
	+	sections = parse_sections(SIMPLE_BILL_XML)
	+	assert sections[1].enum == "2."
	+
	+	def test_subsection_text_flattened(self):
	+	sections = parse_sections(SIMPLE_BILL_XML)
	+	# Subsection content should appear in the parent section's text
	+	assert "widget" in sections[1].text or "widget" in sections[1].text
	+
	+	def test_invalid_xml_returns_empty(self):
	+	sections = parse_sections("<not valid xml <<>>")
	+	assert sections == []
	+
	+	def test_empty_xml_returns_empty(self):
	+	sections = parse_sections("")
	+	assert sections == []
	+
	+	def test_namespaced_xml_parsed(self):
	+	# Namespaced elements may not be found without namespace handling
	+	# This test documents the current behavior
	+	sections = parse_sections(NAMESPACED_BILL_XML)
	+	# Either found (if namespace stripping works) or empty (acceptable)
	+	assert isinstance(sections, list)
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Rider detection tests
	+	# ---------------------------------------------------------------------------
	+
	+	class TestRiderDetect:
	+	def test_notwithstanding_triggers(self):
	+	assert is_rider_signal(
	+	"Notwithstanding any other provision of law, the Secretary shall..."
	+	)
	+
	+	def test_provided_comma_triggers(self):
	+	assert is_rider_signal(
	+	"Provided, That no funds shall be used for..."
	+	)
	+
	+	def test_provided_further_triggers(self):
	+	assert is_rider_signal(
	+	"Provided Further, That the Director may waive..."
	+	)
	+
	+	def test_is_amended_triggers(self):
	+	assert is_rider_signal(
	+	"Section 401(a) of the XYZ Act Is Amended by inserting..."
	+	)
	+
	+	def test_are_authorized_triggers(self):
	+	assert is_rider_signal(
	+	"There Are Authorized to be appropriated $5,000,000 for each..."
	+	)
	+
	+	def test_unless_triggers(self):
	+	assert is_rider_signal(
	+	"Unless the Secretary determines otherwise, the grant shall..."
	+	)
	+
	+	def test_until_triggers(self):
	+	assert is_rider_signal(
	+	"No action shall be taken Until the report is submitted."
	+	)
	+
	+	def test_determines_triggers(self):
	+	assert is_rider_signal(
	+	"The Administrator Determines that the standard is met."
	+	)
	+
	+	def test_in_general_triggers(self):
	+	assert is_rider_signal("In General. - The Secretary shall...")
	+
	+	def test_clean_text_no_trigger(self):
	+	assert not is_rider_signal(
	+	"This Act establishes a new national park in Colorado."
	+	)
	+
	+	def test_case_insensitive_notwithstanding(self):
	+	assert is_rider_signal("NOTWITHSTANDING section 4 of the prior Act,")
	+
	+	def test_case_insensitive_provided(self):
	+	assert is_rider_signal("provided, that all funds shall be...")
	+
	+	def test_strong_check_no_funds(self):
	+	assert is_rider_signal_strong(
	+	"None of the funds appropriated under this Act shall be..."
	+	)
	+
	+	def test_strong_check_no_funds_variant(self):
	+	assert is_rider_signal_strong(
	+	"No funds appropriated herein may be used to..."
	+	)
	+
	+	def test_matched_signals_returns_list(self):
	+	signals = matched_signals(
	+	"Notwithstanding other provisions, Provided, That the Secretary..."
	+	)
	+	assert len(signals) >= 2 # both 'Notwithstanding' and 'Provided,'
	+
	+	def test_no_false_positive_normal_text(self):
	+	text = (
	+	"The Secretary of Defense shall submit a report to Congress "
	+	"within 90 days of the date of enactment of this Act "
	+	"describing the status of the program."
	+	)
	+	assert not is_rider_signal(text)
	+
	+	def test_notwithstanding_inline(self):
	+	# Real appropriations rider pattern
	+	text = (
	+	"Notwithstanding section 1341 of title 31, United States Code, "
	+	"the Secretary may transfer amounts between appropriation accounts."
	+	)
	+	assert is_rider_signal(text)
	+
	+	def test_provided_further_complex(self):
	+	text = (
	+	"Provided, That the amount made available shall be $10,000,000: "
	+	"Provided Further, That not less than $2,000,000 shall be for..."
	+	)
	+	assert is_rider_signal(text)
	+	signals = matched_signals(text)
	+	assert len(signals) >= 2
	+
	+	def test_empty_text_no_trigger(self):
	+	assert not is_rider_signal("")
	+	assert not is_rider_signal_strong("")
	+
	+
	+	# ---------------------------------------------------------------------------
	+	# Integration: full diff pipeline on synthetic XML
	+	# ---------------------------------------------------------------------------
	+
	+	OLD_BILL_XML = """<?xml version="1.0"?>
	+	<bill>
	+	<body>
	+	<section id="H001">
	+	<enum>1.</enum>
	+	<header>Short Title</header>
	+	<text>This Act may be cited as the Widget Act of 2024.</text>
	+	</section>
	+	<section id="H002">
	+	<enum>2.</enum>
	+	<header>Authorization</header>
	+	<text>There is authorized to be appropriated $5,000,000.</text>
	+	</section>
	+	</body>
	+	</bill>
	+	"""
	+
	+	NEW_BILL_XML = """<?xml version="1.0"?>
	+	<bill>
	+	<body>
	+	<section id="H001">
	+	<enum>1.</enum>
	+	<header>Short Title</header>
	+	<text>This Act may be cited as the Widget Act of 2025.</text>
	+	</section>
	+	<section id="H002">
	+	<enum>2.</enum>
	+	<header>Authorization</header>
	+	<text>There is authorized to be appropriated $10,000,000.</text>
	+	</section>
	+	<section id="H003">
	+	<enum>3.</enum>
	+	<header>Notwithstanding Provision</header>
	+	<text>Notwithstanding any other provision of law, no funds appropriated
	+	under this Act shall be used to implement section 404.</text>
	+	</section>
	+	</body>
	+	</bill>
	+	"""
	+
	+
	+	class TestFullDiffPipeline:
	+	def test_changed_section_is_amended(self):
	+	old_secs = parse_sections(OLD_BILL_XML)
	+	new_secs = parse_sections(NEW_BILL_XML)
	+	pairs = align_sections(old_secs, new_secs)
	+
	+	changes = {}
	+	for o, n in pairs:
	+	change = classify_change(
	+	o.text if o else None,
	+	n.text if n else None,
	+	)
	+	sec = (n or o)
	+	if change:
	+	changes[sec.sec_id] = change
	+
	+	assert changes.get("H001") == "amended" # year changed
	+	assert changes.get("H002") == "amended" # dollar amount changed
	+	assert changes.get("H003") == "added" # new section
	+
	+	def test_added_section_rider_flag(self):
	+	old_secs = parse_sections(OLD_BILL_XML)
	+	new_secs = parse_sections(NEW_BILL_XML)
	+	pairs = align_sections(old_secs, new_secs)
	+
	+	for o, n in pairs:
	+	if o is None and n is not None:
	+	# This is the added section
	+	assert is_rider_signal(n.text), (
	+	f"Expected rider signal in added section: {n.text!r}"
	+	)

services/ingest/tests/test_crec_voice_vote.py +404 -0

		@@ -0,0 +1,404 @@
	+	"""
	+	Tests for the CREC procedural-event detector (crec.py).
	+
	+	Covers:
	+	- _VOICE_VOTE regex fires on canonical floor phrases
	+	- _UC regex fires on unanimous-consent phrases
	+	- _CLOTURE regex fires on cloture phrases
	+	- _SUSPENSION, _MTR, _DISCHARGE patterns
	+	- _detect_events() yields correct kind/chamber/date
	+	- members_jsonb is always None (no individual record for voice votes)
	+	- source_url is always a .gov URL
	+	- _extract_text_from_xml() handles valid and malformed XML
	+	- _infer_chamber() maps filenames correctly
	+	- _context_snippet() extracts reasonable surrounding text
	+	- Idempotency: re-processing the same text does not double-insert
	+	"""
	+
	+	import io
	+	import re
	+	import sqlite3
	+	import tempfile
	+	import xml.etree.ElementTree as ET
	+
	+	import pytest
	+
	+	from ingest.pollers.crec import (
	+	_CLOTURE,
	+	_DISCHARGE,
	+	_MTR,
	+	_SUSPENSION,
	+	_UC,
	+	_VOICE_VOTE,
	+	_context_snippet,
	+	_detect_events,
	+	_ensure_tables,
	+	_extract_text_from_xml,
	+	_infer_chamber,
	+	poll_date,
	+	)
	+	from ingest.db import get_connection, init_schema
	+	from ingest.source_validator import is_allowlisted
	+
	+
	+	# ── Fixtures ──────────────────────────────────────────────────────────────────
	+
	+	@pytest.fixture()
	+	def temp_db():
	+	"""In-memory (temp-file) SQLite DB with schema + procedural_events table."""
	+	tmp = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(tmp)
	+	init_schema(conn)
	+	_ensure_tables(conn)
	+	yield conn
	+	conn.close()
	+	import os; os.unlink(tmp)
	+
	+
	+	# ── Voice vote regex ──────────────────────────────────────────────────────────
	+
	+	class TestVoiceVotePattern:
	+	"""The PRESIDING OFFICER uses specific floor phrases to declare voice-vote results."""
	+
	+	def test_ayes_have_it_lowercase(self):
	+	assert _VOICE_VOTE.search("the ayes have it")
	+
	+	def test_ayes_have_it_titlecase(self):
	+	assert _VOICE_VOTE.search("The Ayes Have It")
	+
	+	def test_yeas_have_it(self):
	+	assert _VOICE_VOTE.search("the yeas have it")
	+
	+	def test_noes_have_it(self):
	+	assert _VOICE_VOTE.search("the noes have it")
	+
	+	def test_nays_have_it(self):
	+	assert _VOICE_VOTE.search("the nays have it")
	+
	+	def test_double_declaration(self):
	+	# Common floor pattern: "The ayes have it. The ayes have it."
	+	assert _VOICE_VOTE.search("the ayes have it, the ayes have it")
	+
	+	def test_embedded_in_sentence(self):
	+	text = (
	+	"The PRESIDING OFFICER. The question is on the amendment. "
	+	"The ayes have it. The amendment is agreed to."
	+	)
	+	assert _VOICE_VOTE.search(text)
	+
	+	def test_no_false_positive_on_roll_call(self):
	+	# A recorded vote announcement should NOT match voice-vote pattern
	+	text = "The yeas are 67, nays 33. The motion is agreed to."
	+	# This does NOT match the "have it" pattern
	+	assert not _VOICE_VOTE.search(text)
	+
	+	def test_aye_singular(self):
	+	assert _VOICE_VOTE.search("the aye has it")
	+
	+	def test_partial_words_no_match(self):
	+	# "ayesha" should not match
	+	assert not _VOICE_VOTE.search("ayesha")
	+
	+
	+	# ── Unanimous consent regex ───────────────────────────────────────────────────
	+
	+	class TestUCPattern:
	+	def test_without_objection(self):
	+	assert _UC.search("Without objection, so ordered.")
	+
	+	def test_so_ordered(self):
	+	assert _UC.search("So ordered.")
	+
	+	def test_unanimous_consent(self):
	+	assert _UC.search("Mr. Speaker, I ask unanimous consent to revise and extend")
	+
	+	def test_case_insensitive(self):
	+	assert _UC.search("WITHOUT OBJECTION")
	+	assert _UC.search("unanimous CONSENT")
	+
	+	def test_embedded(self):
	+	text = "The Senator from Ohio asks unanimous consent that the Senate proceed."
	+	assert _UC.search(text)
	+
	+	def test_no_match_on_unrelated(self):
	+	assert not _UC.search("The bill passed by a recorded vote of 52 to 48.")
	+
	+
	+	# ── Cloture regex ─────────────────────────────────────────────────────────────
	+
	+	class TestCloturePattern:
	+	def test_cloture_motion(self):
	+	assert _CLOTURE.search("A cloture motion was filed on the nomination.")
	+
	+	def test_invoke_cloture(self):
	+	assert _CLOTURE.search("The Senate voted to invoke cloture.")
	+
	+	def test_vote_on_cloture(self):
	+	assert _CLOTURE.search("A vote on cloture is scheduled for Monday.")
	+
	+	def test_cloture_was_invoked(self):
	+	assert _CLOTURE.search("Cloture was invoked by a vote of 60-38.")
	+
	+	def test_motion_to_invoke_cloture(self):
	+	assert _CLOTURE.search("A motion to invoke cloture on the bill was filed.")
	+
	+	def test_no_false_positive(self):
	+	assert not _CLOTURE.search("The committee voted on the closure of the hearing room.")
	+
	+
	+	# ── Suspension regex ──────────────────────────────────────────────────────────
	+
	+	class TestSuspensionPattern:
	+	def test_suspend_the_rules(self):
	+	assert _SUSPENSION.search("On motion to suspend the rules and pass the bill")
	+
	+	def test_suspension_of_the_rules(self):
	+	assert _SUSPENSION.search("Suspension of the rules - H.R. 1234")
	+
	+	def test_case_insensitive(self):
	+	assert _SUSPENSION.search("SUSPEND THE RULES")
	+
	+	def test_no_false_positive_on_other_suspend(self):
	+	# "suspended from office" is not a rules-suspension
	+	# The pattern requires "rules" nearby, so this should not match
	+	text = "The officer was suspended for misconduct."
	+	assert not _SUSPENSION.search(text)
	+
	+
	+	# ── MTR regex ─────────────────────────────────────────────────────────────────
	+
	+	class TestMTRPattern:
	+	def test_basic(self):
	+	assert _MTR.search("The gentleman offers a motion to recommit the bill.")
	+
	+	def test_case_insensitive(self):
	+	assert _MTR.search("MOTION TO RECOMMIT")
	+
	+	def test_no_false_positive(self):
	+	assert not _MTR.search("The motion to table failed.")
	+
	+
	+	# ── Discharge regex ───────────────────────────────────────────────────────────
	+
	+	class TestDischargePattern:
	+	def test_discharge_petition(self):
	+	assert _DISCHARGE.search("A discharge petition was filed for H.R. 10.")
	+
	+	def test_discharged_from_committee(self):
	+	assert _DISCHARGE.search("The bill was discharged from committee.")
	+
	+	def test_discharge_calendar(self):
	+	assert _DISCHARGE.search("The measure was placed on the discharge calendar.")
	+
	+	def test_no_false_positive(self):
	+	assert not _DISCHARGE.search("The soldier was honorably discharged.")
	+
	+
	+	# ── _extract_text_from_xml() ──────────────────────────────────────────────────
	+
	+	class TestExtractTextFromXml:
	+	def test_basic_xml(self):
	+	xml = b"<root><p>The ayes have it.</p><p>So ordered.</p></root>"
	+	text = _extract_text_from_xml(xml)
	+	assert "ayes have it" in text
	+	assert "So ordered" in text
	+
	+	def test_nested_elements(self):
	+	xml = b"<a><b><c>deep text</c></b></a>"
	+	assert "deep text" in _extract_text_from_xml(xml)
	+
	+	def test_html_tags_stripped(self):
	+	# CREC files are .htm - tags must be stripped
	+	htm = b"<p><b>Mr. JONES.</b> Without objection, so ordered.</p>"
	+	text = _extract_text_from_xml(htm)
	+	assert "without objection" in text.lower()
	+	assert "<p>" not in text
	+	assert "<b>" not in text
	+
	+	def test_html_entities_unescaped(self):
	+	htm = b"<p>The ayes have it.  The amendment is agreed to.</p>"
	+	text = _extract_text_from_xml(htm)
	+	assert "ayes have it" in text
	+
	+	def test_empty_bytes(self):
	+	assert _extract_text_from_xml(b"") == ""
	+
	+	def test_malformed_html_decoded(self):
	+	# Not valid XML - should still extract text
	+	raw = b"<broken>The noes have it.<br>"
	+	text = _extract_text_from_xml(raw)
	+	assert "noes have it" in text
	+
	+
	+	# ── _infer_chamber() ─────────────────────────────────────────────────────────
	+
	+	class TestInferChamber:
	+	def test_house_filename(self):
	+	assert _infer_chamber("CREC-2026-06-24-pt1-PgH1.xml") == "House"
	+
	+	def test_senate_filename(self):
	+	assert _infer_chamber("CREC-2026-06-24-pt1-PgS100.xml") == "Senate"
	+
	+	def test_digest_filename(self):
	+	assert _infer_chamber("CREC-2026-06-24-pt1-PgD1.xml") == "Both"
	+
	+	def test_unknown_filename(self):
	+	assert _infer_chamber("CREC-2026-06-24-extensions.xml") == "Unknown"
	+
	+
	+	# ── _context_snippet() ───────────────────────────────────────────────────────
	+
	+	class TestContextSnippet:
	+	def test_returns_text_around_match(self):
	+	text = "The Presiding Officer called for order. The ayes have it. The amendment is agreed to."
	+	m = _VOICE_VOTE.search(text)
	+	snippet = _context_snippet(text, m, window=40)
	+	assert "ayes have it" in snippet
	+
	+	def test_does_not_exceed_text_bounds(self):
	+	text = "The ayes have it." # match at position 4
	+	m = _VOICE_VOTE.search(text)
	+	snippet = _context_snippet(text, m, window=200)
	+	# Should not raise; should be a subset of the original text
	+	assert snippet in text or set(snippet) <= set(text + " ")
	+
	+	def test_strips_newlines(self):
	+	text = "Motion to recommit\nwas offered."
	+	m = _MTR.search(text)
	+	snippet = _context_snippet(text, m, window=60)
	+	assert "\n" not in snippet
	+
	+
	+	# ── _detect_events() ─────────────────────────────────────────────────────────
	+
	+	class TestDetectEvents:
	+	BASE_URL = "https://www.govinfo.gov/app/details/CREC-2026-06-24"
	+
	+	def _events(self, text, chamber="Senate"):
	+	return list(
	+	_detect_events(
	+	text,
	+	chamber=chamber,
	+	date_iso="2026-06-24",
	+	source_url=self.BASE_URL,
	+	filename="test-section.xml",
	+	)
	+	)
	+
	+	def test_voice_vote_detected(self):
	+	events = self._events("The PRESIDING OFFICER. The ayes have it.")
	+	assert any(e["kind"] == "voice_vote" for e in events)
	+
	+	def test_uc_detected(self):
	+	events = self._events("Without objection, so ordered.")
	+	assert any(e["kind"] == "unanimous_consent" for e in events)
	+
	+	def test_cloture_detected(self):
	+	events = self._events("Cloture was invoked on the nomination by 62-38.")
	+	assert any(e["kind"] == "cloture" for e in events)
	+
	+	def test_suspension_detected(self):
	+	events = self._events("On motion to suspend the rules and pass H.R. 99.")
	+	assert any(e["kind"] == "suspension" for e in events)
	+
	+	def test_mtr_detected(self):
	+	events = self._events("Mr. SMITH offered a motion to recommit with instructions.")
	+	assert any(e["kind"] == "mtr" for e in events)
	+
	+	def test_discharge_detected(self):
	+	events = self._events("A discharge petition was filed for H.R. 10.")
	+	assert any(e["kind"] == "discharge" for e in events)
	+
	+	def test_members_jsonb_is_none(self):
	+	"""Voice votes have no individual member record - members_jsonb must be null."""
	+	events = self._events("The ayes have it. The resolution is agreed to.")
	+	vv = [e for e in events if e["kind"] == "voice_vote"]
	+	assert len(vv) >= 1
	+	for e in vv:
	+	assert e["members_jsonb"] is None, (
	+	"Voice-vote events must have members_jsonb=None "
	+	"(no individual record exists - that absence is itself information)"
	+	)
	+
	+	def test_source_url_is_gov(self):
	+	events = self._events("The ayes have it.")
	+	for e in events:
	+	assert is_allowlisted(e["source_url"]), (
	+	f"source_url {e['source_url']!r} is not a .gov URL"
	+	)
	+
	+	def test_chamber_propagated(self):
	+	events = self._events("The ayes have it.", chamber="House")
	+	vv = [e for e in events if e["kind"] == "voice_vote"]
	+	assert all(e["chamber"] == "House" for e in vv)
	+
	+	def test_date_propagated(self):
	+	events = self._events("The noes have it.")
	+	for e in events:
	+	assert e["date"] == "2026-06-24"
	+
	+	def test_external_id_unique_per_match(self):
	+	# Multiple voice-vote matches in the same text should have unique external_ids
	+	text = "The ayes have it. The ayes have it. The noes have it."
	+	events = self._events(text)
	+	vv_ids = [e["external_id"] for e in events if e["kind"] == "voice_vote"]
	+	assert len(vv_ids) == len(set(vv_ids)), "external_ids must be unique"
	+
	+	def test_empty_text_yields_no_events(self):
	+	assert self._events("") == []
	+
	+	def test_unrelated_text_yields_no_events(self):
	+	text = "Mr. SMITH. I yield back the balance of my time."
	+	assert self._events(text) == []
	+
	+
	+	# ── Source URL integrity ──────────────────────────────────────────────────────
	+
	+	class TestSourceUrlIntegrity:
	+	"""All source_urls produced by the CREC poller must be .gov allowlisted."""
	+
	+	def test_govinfo_page_url_is_gov(self):
	+	from ingest.pollers.crec import _govinfo_page_url
	+	from datetime import datetime
	+	url = _govinfo_page_url(datetime(2026, 6, 24))
	+	assert "govinfo.gov" in url
	+	assert is_allowlisted(url)
	+
	+	def test_govinfo_zip_url_is_gov(self):
	+	from ingest.pollers.crec import _govinfo_url
	+	from datetime import datetime
	+	url = _govinfo_url(datetime(2026, 6, 24))
	+	assert "govinfo.gov" in url
	+	assert is_allowlisted(url)
	+
	+
	+	# ── Idempotency / DB integration ──────────────────────────────────────────────
	+
	+	class TestIdempotency:
	+	"""Re-processing the same events must not create duplicate rows."""
	+
	+	SOURCE_URL = "https://www.govinfo.gov/app/details/CREC-2026-06-24"
	+
	+	def test_no_duplicate_rows_on_second_pass(self, temp_db):
	+	"""
	+	Call _detect_events twice with the same text and simulate two DB
	+	insert passes; the second pass must yield 0 new stored rows.
	+	"""
	+	text = "The ayes have it. Without objection, so ordered."
	+	events = list(
	+	_detect_events(text, "Senate", "2026-06-24", self.SOURCE_URL, "test.xml")
	+	)
	+	assert len(events) > 0
	+
	+	from ingest.change_detection import is_new_event, mark_seen
	+
	+	# First pass - mark all as seen
	+	for evt in events:
	+	mark_seen(temp_db, "crec", evt["external_id"], evt["update_date"])
	+
	+	# Second pass - all should be "already seen"
	+	new = [
	+	e for e in events
	+	if is_new_event(temp_db, "crec", e["external_id"], e["update_date"])
	+	]
	+	assert new == [], f"Second pass should be empty; got {new}"

services/ingest/tests/test_state_coverage.py +342 -0

		@@ -0,0 +1,342 @@
	+	"""
	+	Tests for state coverage and the STATE_LEG_HOSTS allowlist.
	+
	+	Covers:
	+	1. STATE_LEG_HOSTS - explicit non-.gov state legislature hosts
	+	2. *.state.<XX>.us pattern matching
	+	3. state_coverage.compute() populates all 52 jurisdictions as "missing"
	+	when the DB is empty (keys absent scenario)
	+	4. state_coverage.compute() flips to "ok" after log_scrape() is called
	+	5. state_coverage.compute() correctly marks "stale" for old timestamps
	+	6. Helper functions: _confidence(), _infer_state_from_ocd(),
	+	_infer_state_from_legiscan_id()
	+	"""
	+
	+	import os
	+	import sqlite3
	+	import tempfile
	+	from datetime import datetime, timedelta, timezone
	+
	+	import pytest
	+
	+	from ingest.source_validator import (
	+	STATE_LEG_HOSTS,
	+	_is_state_leg_pattern,
	+	is_allowlisted,
	+	validate_source_url,
	+	SourceURLError,
	+	)
	+	from ingest.state_coverage import (
	+	ALL_STATES,
	+	SOURCES,
	+	DATA_TYPES,
	+	STALE_HOURS,
	+	_confidence,
	+	_infer_state_from_ocd,
	+	_infer_state_from_legiscan_id,
	+	compute,
	+	log_scrape,
	+	)
	+	from ingest.db import get_connection, init_schema
	+
	+
	+	# ── Helpers ────────────────────────────────────────────────────────────────────
	+
	+	def _make_db():
	+	tmp = tempfile.mktemp(suffix=".db")
	+	conn = get_connection(tmp)
	+	init_schema(conn)
	+	return conn, tmp
	+
	+
	+	# ── STATE_LEG_HOSTS allowlist ──────────────────────────────────────────────────
	+
	+	class TestStateLegHosts:
	+	def test_state_leg_hosts_is_non_empty(self):
	+	assert len(STATE_LEG_HOSTS) > 0
	+
	+	def test_explicit_pa_host_is_allowed(self):
	+	assert is_allowlisted("https://www.legis.state.pa.us/CFDOCS/billInfo/billInfo.cfm?sYear=2025&sInd=0&body=H&type=B&bn=1")
	+
	+	def test_explicit_nj_host_is_allowed(self):
	+	assert is_allowlisted("https://www.njleg.state.nj.us/Bills/2025/A0500/499_I1.HTM")
	+
	+	def test_ks_org_host_is_allowed(self):
	+	assert is_allowlisted("https://www.kslegislature.org/li/b2025_26/measures/hb2001/")
	+
	+	def test_tx_capitol_host_is_allowed(self):
	+	assert is_allowlisted("https://capitol.state.tx.us/BillLookup/History.aspx?LegSess=89R&Bill=HB1")
	+
	+	def test_mn_leg_host_is_allowed(self):
	+	assert is_allowlisted("https://www.leg.state.mn.us/cgi-bin/bi?mode=1&session=91")
	+
	+	def test_ms_billstatus_host_is_allowed(self):
	+	assert is_allowlisted("https://billstatus.ls.state.ms.us/2025/pdf/history/SB/SB2001.xml")
	+
	+	def test_nh_gencourt_host_is_allowed(self):
	+	assert is_allowlisted("https://www.gencourt.state.nh.us/legislation/2025/SB0001.html")
	+
	+	def test_ak_legis_host_is_allowed(self):
	+	assert is_allowlisted("https://www.legis.state.ak.us/basis/bill_status.asp?session=34&bill=SB1")
	+
	+	def test_ar_arkleg_host_is_allowed(self):
	+	assert is_allowlisted("https://www.arkleg.state.ar.us/Bills/Detail?id=SB1&ddBienniumSession=2025%2F2025R")
	+
	+	def test_dotgov_still_passes(self):
	+	"""Existing .gov URLs must still pass after the STATE_LEG_HOSTS addition."""
	+	gov_urls = [
	+	"https://leginfo.legislature.ca.gov/faces/billStatusClient.xhtml?bill_id=202520260AB1",
	+	"https://leg.colorado.gov/bills/hb25-1001",
	+	"https://legislature.mi.gov/documents/2025-2026/billengrossed/House/htm/2025-HEB-0001.htm",
	+	"https://malegislature.gov/Bills/194/H1",
	+	"https://www.congress.gov/bill/119th-congress/house-bill/1",
	+	]
	+	for url in gov_urls:
	+	assert is_allowlisted(url), f"Expected .gov URL to pass: {url}"
	+
	+	def test_aggregator_hosts_rejected(self):
	+	"""OpenStates and LegiScan URLs must NOT be accepted as source_url."""
	+	bad_urls = [
	+	"https://openstates.org/ca/bills/20252026/AB1/",
	+	"https://legiscan.com/CA/bill/AB1/2025",
	+	"https://api.legiscan.com/",
	+	]
	+	for url in bad_urls:
	+	assert not is_allowlisted(url), f"Aggregator URL should be rejected: {url}"
	+
	+
	+	# ── *.state.<XX>.us pattern ────────────────────────────────────────────────────
	+
	+	class TestStateLegPattern:
	+	def test_pattern_matches_pa(self):
	+	assert _is_state_leg_pattern("legis.state.pa.us")
	+
	+	def test_pattern_matches_pa_with_www(self):
	+	assert _is_state_leg_pattern("www.legis.state.pa.us")
	+
	+	def test_pattern_matches_nj(self):
	+	assert _is_state_leg_pattern("njleg.state.nj.us")
	+
	+	def test_pattern_matches_mn_deep(self):
	+	assert _is_state_leg_pattern("www.house.leg.state.mn.us")
	+
	+	def test_pattern_matches_ms(self):
	+	assert _is_state_leg_pattern("billstatus.ls.state.ms.us")
	+
	+	def test_pattern_rejects_govtrack(self):
	+	assert not _is_state_leg_pattern("govtrack.us")
	+
	+	def test_pattern_rejects_short_us(self):
	+	assert not _is_state_leg_pattern("example.us")
	+
	+	def test_pattern_rejects_no_state_segment(self):
	+	assert not _is_state_leg_pattern("www.legislature.mi.gov")
	+
	+	def test_pattern_rejects_wrong_tld(self):
	+	assert not _is_state_leg_pattern("legis.state.ca.com")
	+
	+	def test_full_url_via_is_allowlisted(self):
	+	assert is_allowlisted("https://gencourt.state.nh.us/legislation/2025/SB0001.html")
	+
	+
	+	# ── _confidence() ──────────────────────────────────────────────────────────────
	+
	+	class TestConfidence:
	+	def test_none_is_missing(self):
	+	assert _confidence(None) == "missing"
	+
	+	def test_empty_string_is_missing(self):
	+	assert _confidence("") == "missing"
	+
	+	def test_recent_timestamp_is_ok(self):
	+	recent = (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat()
	+	assert _confidence(recent) == "ok"
	+
	+	def test_exactly_stale_threshold_is_stale(self):
	+	old = (datetime.now(timezone.utc) - timedelta(hours=STALE_HOURS + 1)).isoformat()
	+	assert _confidence(old) == "stale"
	+
	+	def test_very_old_is_stale(self):
	+	ancient = "2020-01-01T00:00:00+00:00"
	+	assert _confidence(ancient) == "stale"
	+
	+	def test_invalid_timestamp_is_missing(self):
	+	assert _confidence("not-a-timestamp") == "missing"
	+
	+	def test_utc_z_suffix_parses(self):
	+	# ISO 8601 with Z
	+	recent = (datetime.now(timezone.utc) - timedelta(hours=2)).strftime("%Y-%m-%dT%H:%M:%SZ")
	+	assert _confidence(recent) == "ok"
	+
	+
	+	# ── OCD / LegiScan ID parsing ──────────────────────────────────────────────────
	+
	+	class TestIdParsing:
	+	def test_ocd_state(self):
	+	assert _infer_state_from_ocd(
	+	"ocd-bill/country:us/state:ca/legislature/2025-2026/HB1"
	+	) == "CA"
	+
	+	def test_ocd_district_dc(self):
	+	assert _infer_state_from_ocd(
	+	"ocd-jurisdiction/country:us/district:dc/government"
	+	) == "DC"
	+
	+	def test_ocd_territory_pr(self):
	+	assert _infer_state_from_ocd(
	+	"ocd-jurisdiction/country:us/territory:pr/government"
	+	) == "PR"
	+
	+	def test_ocd_empty_returns_empty(self):
	+	assert _infer_state_from_ocd("ocd-bill/country:us/") == ""
	+
	+	def test_legiscan_id_ca(self):
	+	assert _infer_state_from_legiscan_id("legiscan-CA-12345") == "CA"
	+
	+	def test_legiscan_id_dc(self):
	+	assert _infer_state_from_legiscan_id("legiscan-DC-999") == "DC"
	+
	+	def test_legiscan_id_wrong_format_returns_empty(self):
	+	assert _infer_state_from_legiscan_id("openstates-CA-1") == ""
	+
	+	def test_legiscan_id_no_state_returns_empty(self):
	+	assert _infer_state_from_legiscan_id("legiscan-12345") == ""
	+
	+
	+	# ── state_coverage.compute() - 52 jurisdictions ────────────────────────────────
	+
	+	class TestStateCoverageCompute:
	+	"""
	+	Core honesty layer tests. These run without any API keys - they verify
	+	that compute() works off DB state alone.
	+	"""
	+
	+	def test_all_states_list_length(self):
	+	"""ALL_STATES must contain exactly 52 entries (50 + DC + PR)."""
	+	assert len(ALL_STATES) == 52
	+
	+	def test_all_states_has_dc_and_pr(self):
	+	assert "DC" in ALL_STATES
	+	assert "PR" in ALL_STATES
	+
	+	def test_all_states_are_uppercase_two_char(self):
	+	for s in ALL_STATES:
	+	assert len(s) == 2, f"State code not 2 chars: {s!r}"
	+	assert s.isupper(), f"State code not uppercase: {s!r}"
	+
	+	def test_empty_db_produces_all_missing(self):
	+	"""With no data in DB, every (state, source, data_type) is 'missing'."""
	+	conn, tmp = _make_db()
	+	conn.close()
	+	try:
	+	rows = compute(tmp)
	+	expected = len(ALL_STATES) * len(SOURCES) * len(DATA_TYPES)
	+	assert len(rows) == expected, (
	+	f"Expected {expected} rows, got {len(rows)}"
	+	)
	+	for row in rows:
	+	assert row["confidence"] == "missing", (
	+	f"Expected missing for {row['state']}/{row['source']}/{row['data_type']}, "
	+	f"got {row['confidence']!r}"
	+	)
	+	assert row["last_scrape"] is None
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_row_count_matches_52_times_sources_times_types(self):
	+	conn, tmp = _make_db()
	+	conn.close()
	+	try:
	+	rows = compute(tmp)
	+	assert len(rows) == 52 * 2 * 3 # 52 states × 2 sources × 3 types = 312
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_after_log_scrape_confidence_is_ok(self):
	+	"""After log_scrape(), the relevant row should be 'ok'."""
	+	conn, tmp = _make_db()
	+	try:
	+	log_scrape(conn, "openstates", "CA", "bills", 10)
	+	conn.commit()
	+	conn.close()
	+	rows = compute(tmp)
	+	# Find CA/openstates/bills
	+	target = [
	+	r for r in rows
	+	if r["state"] == "CA"
	+	and r["source"] == "openstates"
	+	and r["data_type"] == "bills"
	+	]
	+	assert len(target) == 1
	+	assert target[0]["confidence"] == "ok"
	+	# Other types for CA should still be missing
	+	missing = [
	+	r for r in rows
	+	if r["state"] == "CA"
	+	and r["source"] == "openstates"
	+	and r["data_type"] != "bills"
	+	]
	+	for r in missing:
	+	assert r["confidence"] == "missing"
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_stale_timestamp_marks_stale(self):
	+	"""A scrape older than STALE_HOURS should show as 'stale'."""
	+	conn, tmp = _make_db()
	+	try:
	+	old_ts = (
	+	datetime.now(timezone.utc) - timedelta(hours=STALE_HOURS + 2)
	+	).isoformat()
	+	log_scrape(conn, "legiscan", "TX", "bills", 5, old_ts)
	+	conn.commit()
	+	conn.close()
	+	rows = compute(tmp)
	+	target = [
	+	r for r in rows
	+	if r["state"] == "TX"
	+	and r["source"] == "legiscan"
	+	and r["data_type"] == "bills"
	+	]
	+	assert len(target) == 1
	+	assert target[0]["confidence"] == "stale"
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_compute_is_idempotent(self):
	+	"""Calling compute() twice should produce the same rows."""
	+	conn, tmp = _make_db()
	+	conn.close()
	+	try:
	+	rows1 = compute(tmp)
	+	rows2 = compute(tmp)
	+	assert len(rows1) == len(rows2)
	+	for r1, r2 in zip(rows1, rows2):
	+	assert r1["state"] == r2["state"]
	+	assert r1["confidence"] == r2["confidence"]
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_no_duplicate_rows_in_output(self):
	+	conn, tmp = _make_db()
	+	conn.close()
	+	try:
	+	rows = compute(tmp)
	+	seen_keys = set()
	+	for r in rows:
	+	key = (r["state"], r["source"], r["data_type"])
	+	assert key not in seen_keys, f"Duplicate key: {key}"
	+	seen_keys.add(key)
	+	finally:
	+	os.unlink(tmp)
	+
	+	def test_all_50_states_present_in_output(self):
	+	conn, tmp = _make_db()
	+	conn.close()
	+	try:
	+	rows = compute(tmp)
	+	states_in_output = {r["state"] for r in rows}
	+	for state in ALL_STATES:
	+	assert state in states_in_output, f"State {state} missing from output"
	+	finally:
	+	os.unlink(tmp)

		@@ -0,0 +1,39 @@
	+	# Docket - Ingestion Strategy (efficiency & automation)
	+
	+	Roadmap for ingest v2. Today's pollers are naive (full pull each run). This is how we make 50-state + federal ingestion fast, cheap, automated, and self-healing. Ranked by impact-to-effort.
	+
	+	## Top 10 (impact-to-effort)
	+	1. Watermark change-detection - GovInfo `collections/{col}/{lastModStart}/{lastModEnd}?offsetMark=`; Congress.gov `fromDateTime` + `updateDateIncludingText`* (catches text-only edits `updateDate` misses). Store `(source, last_run_at)` in `poller_watermarks`; update atomically only on successful upsert batch. ~10 lines, kills re-pulling.
	+	2. LegiScan weekly ZIP bulk - `getDatasetList`/`getDataset`: one download = all 50 states + Congress. Public API caps at 30k/mo (~one busy state). Bulk for backfill + weekly; API only for intra-week deltas.
	+	3. OpenStates / Plural Postgres dump (`open.pluralpolicy.com/downloads/`) - monthly all-50-state bootstrap, zero scraping; v3 API for incrementals.
	+	4. SHA-256 content-hash gate - hash raw payload before parse; skip parse/upsert if unchanged. `INSERT … ON CONFLICT(source,external_id) DO UPDATE … WHERE hash != EXCLUDED.hash`. Applies everywhere.
	+	5. RSS/Atom feed registry - ~20 states publish official feeds (KS, WA, HI, TX, WI, WV…). `feedparser` + last-seen GUID = sub-minute, zero-budget change detection.
	+	6. Session-calendar priority polling - `sessions` table (active/interim/dormant); poll active hourly, recess daily, dormant weekly. ~70% volume cut off-season.
	+	7. ARQ + aiometer async workers - Redis-backed asyncio queue (better fit than Celery here); `aiometer` for per-domain concurrency + rps caps; `tenacity` backoff on 429. Distribute across CT215 + CT220 via worker hash-ring.
	+	8. LLM selector self-healing (3060) - on field-extraction failure, queue raw HTML + schema → local `mistral-nemo:12b` proposes CSS/XPath → validate against live HTML → write to `scraper_selectors` table (no redeploy). ~98% accuracy across layout changes (2025 research).
	+	9. Embedding action-vocab normalization - 50 states use different action strings → embed with `nomic-embed-text` (Ollama/3060) → cluster (HDBSCAN) → map to one canonical taxonomy (INTRODUCED/REFERRED/PASSED/SIGNED/VETOED…). One-time batch, re-run on new strings.
	+	10. Near-duplicate bill clustering - embed bill summaries → store vectors in the existing Qdrant → cosine >0.92 flags model legislation (ALEC etc.) spreading across states. Unique Docket feature; zero marginal API cost.
	+
	+	## Recommended architecture
	+	- Source registry YAML - each source: `type` (api\|bulk\|rss\|scrape), `change_detection` (watermark\|rss\|hash\|head), `poll_interval_active/dormant`, `rate_limit_rps`, `js_required`, `priority`. Scheduler is data-driven; new/changed sources need no code.
	+	- Three-tier ingest: (1) bulk bootstrap once per session/congress; (2) delta API polling via watermarks (hourly active / daily dormant); (3) persistent RSS watchers emitting change events immediately.
	+	- ARQ worker pool (CT215+CT220), partitioned by source hash-ring, Redis broker; workers write raw payloads → `raw_ingest` staging.
	+	- Hash-gated parse stage - second worker computes `content_hash`, skips unchanged, else runs source parser → `*_staging` → idempotent upsert + emit `bill_versions` diff on text change.
	+	- Dead-letter + alerting - failures → `scrape_failures` (raw error + HTML snapshot + retry count); Discord webhook after 3; 6-hour retry worker.
	+	- LLM repair sidecar - separate process on the 3060 (Ollama), consumes a Redis `selector_repair_queue`; main scrapers read selectors from DB each run.
	+
	+	## 50-state scraping
	+	- Fork `openstates/openstates-scrapers` (~200 scrapers on `scrapelib` + `spatula` + `pupa`) - battle-tested session mgmt/robots/retry. Don't start from scratch.
	+	- JS-rendered states (~15: FL, IL…): `scrapy-playwright`. But probe first - ~40% of "JS" state sites call an undocumented JSON/GraphQL endpoint that's cleaner + more stable than the DOM. Catalog these as a priority tier.
	+	- LLM-as-parser only for the messy tail (PDFs, unstructured HTML) with `Outlines` constraint-decoded JSON + Pydantic validation. `lxml`/`trafilatura` beats LLM on structured HTML by 1000× - never put the LLM in the hot path.
	+
	+	## Creative / wildcards
	+	- Wayback CDX (`web.archive.org/cdx/search/cdx`) for historical backfill of states that purge old sessions.
	+	- GovTrack/`unitedstates/congress` git mirrors - bill versions as commits; `git log --diff-filter=M` gives changed-bills-since-last-sync without API.
	+	- Undocumented state JSON endpoints - check the network tab before writing any scraper.
	+
	+	## 3060 LLM: value vs trap
	+	- Use: selector self-healing, action-vocab embedding/normalization, PDF/messy-HTML extraction (long tail), near-dup clustering. All batch/exception-path.
	+	- Trap: LLM as primary parser for structured HTML; LLM in the real-time hot path (creates permanent backlog at 100k+ bills/session); trusting LLM output without schema validation; re-parsing the Congressional Record (use `unitedstates/congressional-record`).
	+
	+	Adopt incrementally: watermarks + hash-gate + bulk-first first (items 1-4) - biggest win, least effort - then the registry + workers, then the LLM sidecar.