← back to Unclaimed Property Platform
Cycle 1: scaffold national unclaimed-property platform (B2G, synthetic-only)
9819a89765b3f30efa3260bf0920f3a8df08dc76 · 2026-07-31 12:27:11 -0700 · Steve (yoloforever loop)
- Thesis + hard guardrails (no scraping, no real feeds, no autonomous contracting)
- 50-state + DC acquisition matrix (feed-first, records-request fallback, public handoff)
- Polyglot architecture + canonical schema (provenance/version non-destructive)
- Reference services: idempotent NAUPA/CSV ETL, Fellegi-Sunter matcher w/ phonetic key,
masked FastAPI search, claim state-machine + transactional outbox
- Stdlib-only end-to-end smoke test (ingest idempotency + masking + matching) PASSES at $0
TK-10097
Files touched
A .gitignoreA README.mdA data/sample/sample_state_feed.csvA db/schema.sqlA docs/00-thesis-and-guardrails.mdA docs/01-regulatory-state-acquisition-matrix.mdA docs/02-architecture.mdA requirements.txtA services/__init__.pyA services/claims/__init__.pyA services/claims/claim_workflow.pyA services/common/__init__.pyA services/common/normalize.pyA services/common/sqlite_repo.pyA services/ingestion/__init__.pyA services/ingestion/ingest.pyA services/matching/__init__.pyA services/matching/entity_match.pyA services/search/__init__.pyA services/search/search_api.pyA tests/__init__.pyA tests/test_ingest_and_match.py
Diff
commit 9819a89765b3f30efa3260bf0920f3a8df08dc76
Author: Steve (yoloforever loop) <steve@designerwallcoverings.com>
Date: Fri Jul 31 12:27:11 2026 -0700
Cycle 1: scaffold national unclaimed-property platform (B2G, synthetic-only)
- Thesis + hard guardrails (no scraping, no real feeds, no autonomous contracting)
- 50-state + DC acquisition matrix (feed-first, records-request fallback, public handoff)
- Polyglot architecture + canonical schema (provenance/version non-destructive)
- Reference services: idempotent NAUPA/CSV ETL, Fellegi-Sunter matcher w/ phonetic key,
masked FastAPI search, claim state-machine + transactional outbox
- Stdlib-only end-to-end smoke test (ingest idempotency + masking + matching) PASSES at $0
TK-10097
---
.gitignore | 30 ++++
README.md | 60 ++++++++
data/sample/sample_state_feed.csv | 9 ++
db/schema.sql | 166 ++++++++++++++++++++++
docs/00-thesis-and-guardrails.md | 70 ++++++++++
docs/01-regulatory-state-acquisition-matrix.md | 90 ++++++++++++
docs/02-architecture.md | 60 ++++++++
requirements.txt | 13 ++
services/__init__.py | 0
services/claims/__init__.py | 0
services/claims/claim_workflow.py | 119 ++++++++++++++++
services/common/__init__.py | 0
services/common/normalize.py | 118 ++++++++++++++++
services/common/sqlite_repo.py | 182 +++++++++++++++++++++++++
services/ingestion/__init__.py | 0
services/ingestion/ingest.py | 144 +++++++++++++++++++
services/matching/__init__.py | 0
services/matching/entity_match.py | 105 ++++++++++++++
services/search/__init__.py | 0
services/search/search_api.py | 106 ++++++++++++++
tests/__init__.py | 0
tests/test_ingest_and_match.py | 112 +++++++++++++++
22 files changed, 1384 insertions(+)
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..10ada57
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,30 @@
+# deps
+node_modules/
+__pycache__/
+*.pyc
+.venv/
+venv/
+
+# env & secrets — NEVER commit
+.env
+.env.*
+*.key
+*.pem
+
+# local artifacts
+tmp/
+*.log
+.DS_Store
+dist/
+build/
+.next/
+
+# local prototype databases (synthetic only, but keep them out of git)
+*.db
+*.sqlite
+*.sqlite3
+data/local/
+
+# NEVER commit real jurisdiction data — the repo is synthetic-only by policy
+data/real/
+raw/
diff --git a/README.md b/README.md
new file mode 100644
index 0000000..9abce8d
--- /dev/null
+++ b/README.md
@@ -0,0 +1,60 @@
+# National Multi-State Unclaimed Property Platform
+
+A defensible, **government-partnered** replication of the MissingMoney model:
+a business-to-government platform that ingests **authorized** state unclaimed-property
+feeds, normalizes them into a common NAUPA-aligned schema, provides free national
+consumer search with masked results, and routes claims back to the state of record.
+
+> **This repository is a local prototype + living planning docs.**
+> It runs entirely on **synthetic sample data**. It does **not** connect to any real
+> state feed, does **not** scrape, and does **not** contract with anyone. Those steps
+> are business/legal actions gated to a human — see `docs/00-thesis-and-guardrails.md`.
+
+## Why this model (not a scrape-first consumer clone)
+
+The durable business is selling **infrastructure and services to state governments**,
+not selling access to claimants. Consumer search and claim initiation are free.
+Revenue comes from state participation, implementation, white-label portals, and
+claim-workflow modules. The moat is contracted feed reliability, normalized history,
+search quality, and a security posture acceptable to public agencies — not the raw
+records (which are public and obtainable by competitors).
+
+See `docs/` for the full thesis, the state-acquisition matrix, and the architecture.
+
+## What's in here
+
+| Path | What it is |
+|---|---|
+| `docs/00-thesis-and-guardrails.md` | The business thesis + the hard "never autonomously" line |
+| `docs/01-regulatory-state-acquisition-matrix.md` | 50-state + DC data-acquisition method matrix |
+| `docs/02-architecture.md` | Polyglot architecture + data model + service boundaries |
+| `db/schema.sql` | Canonical relational schema (PostgreSQL dialect; SQLite subset for the prototype) |
+| `services/ingestion/` | Idempotent authorized-feed ETL (NAUPA/CSV adapters + provenance) |
+| `services/matching/` | Entity resolution (Fellegi-Sunter-style probabilistic linkage) |
+| `services/search/` | Masked public search API (FastAPI reference) |
+| `services/claims/` | Claim workflow state-machine with transactional outbox |
+| `data/sample/` | **Synthetic** sample state feed (safe to commit) |
+| `tests/` | End-to-end smoke test proving the prototype runs at $0 (SQLite) |
+
+## Run the prototype (local, $0, synthetic data only)
+
+```bash
+python -m pip install -r requirements.txt # optional; core smoke test is stdlib-only
+python -m tests.test_ingest_and_match # proves idempotent ETL + matcher end-to-end
+```
+
+## Guardrails (enforced by policy, not just convention)
+
+1. **No unauthorized scraping.** Ingestion accepts files/feeds only. Any crawler is
+ limited to availability monitoring or a state-approved integration.
+2. **No real jurisdiction data in this repo.** `.gitignore` blocks `data/real/` and `raw/`.
+3. **State is the adjudicator and payer of record.** The platform is a workflow
+ processor; it never decides entitlement or custodies claimant funds at launch.
+4. **Masked results only.** Anonymous search returns state-approved projections; never
+ SSNs, full addresses, exact amounts, or claim evidence.
+5. **Every displayed datum is provenance-traceable** to a source batch.
+
+## Status
+
+Cycle 1 of a `/yoloforever` build loop (ticket **TK-10097**). Scaffold + reference
+services + smoke test. Real data rights, deployment, and any spend remain Steve-gated.
diff --git a/data/sample/sample_state_feed.csv b/data/sample/sample_state_feed.csv
new file mode 100644
index 0000000..9e1c7cc
--- /dev/null
+++ b/data/sample/sample_state_feed.csv
@@ -0,0 +1,9 @@
+property_id,holder_name,owner_name,address,city,state,zip,property_type,amount
+SP-0001,First Synthetic Bank,Catherine O'Neil,100 Test St,Springfield,SAMPLE,00001,uncashed_check,42.50
+SP-0002,First Synthetic Bank,Kathryn ONeill,100 Test Street,Springfield,SAMPLE,00001,uncashed_check,42.50
+SP-0003,Placeholder Utility Co,Robert Q Public,200 Example Ave,Rivertown,SAMPLE,00002,utility_deposit,175.00
+SP-0004,Made-Up Securities LLC,Acme Widgets Inc,300 Nowhere Blvd,Rivertown,SAMPLE,00002,securities,5300.00
+SP-0005,Made-Up Securities LLC,Acme Widgets Incorporated,300 Nowhere Blvd,Rivertown,SAMPLE,00002,securities,5300.00
+SP-0006,Fictional Insurance,Maria Testcase,44 Imaginary Rd,Lakeside,SAMPLE,00003,insurance_proceeds,910.25
+SP-0007,Fictional Insurance,,44 Imaginary Rd,Lakeside,SAMPLE,00003,insurance_proceeds,12.00
+SP-0008,Sample Payroll Corp,Jonathan Doe,7 Dummy Ct,Springfield,SAMPLE,00001,wages,88.40
diff --git a/db/schema.sql b/db/schema.sql
new file mode 100644
index 0000000..71f2365
--- /dev/null
+++ b/db/schema.sql
@@ -0,0 +1,166 @@
+-- Canonical relational schema — PostgreSQL dialect.
+-- The prototype (services/common/sqlite_repo.py) uses a SQLite-compatible subset.
+-- Design principle: raw data is NEVER overwritten; canonical fields live alongside it;
+-- every displayed datum is traceable to a source batch (provenance).
+
+-- ---------------------------------------------------------------------------
+-- Jurisdiction configuration
+-- ---------------------------------------------------------------------------
+CREATE TABLE jurisdiction (
+ jurisdiction_id TEXT PRIMARY KEY, -- e.g. 'CA'
+ name TEXT NOT NULL,
+ status TEXT NOT NULL DEFAULT 'prospect' -- prospect|contracted|live|paused
+);
+
+CREATE TABLE jurisdiction_policy (
+ policy_id TEXT PRIMARY KEY,
+ jurisdiction_id TEXT NOT NULL REFERENCES jurisdiction(jurisdiction_id),
+ version INTEGER NOT NULL,
+ effective_from TIMESTAMP NOT NULL,
+ effective_to TIMESTAMP,
+ -- masking / claim tier / retention / required evidence / search limits (JSON)
+ config_json TEXT NOT NULL
+);
+
+-- ---------------------------------------------------------------------------
+-- Ingestion + provenance
+-- ---------------------------------------------------------------------------
+CREATE TABLE source_feed (
+ feed_id TEXT PRIMARY KEY,
+ jurisdiction_id TEXT NOT NULL REFERENCES jurisdiction(jurisdiction_id),
+ format_name TEXT NOT NULL, -- naupa2 | naupa3 | state_csv_v1 | ...
+ cadence TEXT,
+ transport TEXT
+);
+
+CREATE TABLE ingestion_batch (
+ batch_id TEXT PRIMARY KEY,
+ feed_id TEXT REFERENCES source_feed(feed_id),
+ jurisdiction_id TEXT NOT NULL,
+ source_uri TEXT,
+ checksum TEXT NOT NULL, -- sha256 of the source file
+ parser_version TEXT NOT NULL,
+ received_at TIMESTAMP NOT NULL,
+ accepted_count INTEGER NOT NULL DEFAULT 0,
+ rejected_count INTEGER NOT NULL DEFAULT 0,
+ status TEXT NOT NULL, -- running|completed|completed_with_errors|failed|duplicate
+ UNIQUE (jurisdiction_id, checksum) -- idempotency: same file never processed twice
+);
+
+CREATE TABLE raw_object (
+ raw_object_id TEXT PRIMARY KEY,
+ batch_id TEXT NOT NULL REFERENCES ingestion_batch(batch_id),
+ storage_uri TEXT NOT NULL,
+ checksum TEXT NOT NULL
+);
+
+-- ---------------------------------------------------------------------------
+-- Canonical property + owners (non-destructive history)
+-- ---------------------------------------------------------------------------
+CREATE TABLE property (
+ property_id TEXT PRIMARY KEY, -- surrogate
+ jurisdiction_id TEXT NOT NULL,
+ source_property_id TEXT NOT NULL, -- the state's record id
+ holder_name_raw TEXT,
+ property_type TEXT,
+ amount NUMERIC,
+ status TEXT NOT NULL DEFAULT 'active',
+ UNIQUE (jurisdiction_id, source_property_id)
+);
+
+CREATE TABLE property_version (
+ property_version_id TEXT PRIMARY KEY,
+ property_id TEXT NOT NULL REFERENCES property(property_id),
+ batch_id TEXT NOT NULL REFERENCES ingestion_batch(batch_id),
+ effective_from TIMESTAMP NOT NULL,
+ effective_to TIMESTAMP,
+ raw_payload TEXT NOT NULL, -- the exact source row (JSON)
+ raw_record_hash TEXT NOT NULL -- dedup / forensic compare
+);
+
+CREATE TABLE owner (
+ owner_id TEXT PRIMARY KEY,
+ property_id TEXT NOT NULL REFERENCES property(property_id),
+ owner_type TEXT NOT NULL DEFAULT 'person', -- person|business
+ owner_name_raw TEXT NOT NULL,
+ owner_name_normalized TEXT NOT NULL,
+ city_normalized TEXT,
+ region TEXT,
+ postal_code TEXT
+);
+
+-- Optional real-world entity clustering (non-destructive; never a hard merge)
+CREATE TABLE canonical_entity (
+ entity_id TEXT PRIMARY KEY,
+ entity_type TEXT NOT NULL,
+ status TEXT NOT NULL DEFAULT 'candidate'
+);
+
+CREATE TABLE entity_link (
+ entity_link_id TEXT PRIMARY KEY,
+ owner_id TEXT NOT NULL REFERENCES owner(owner_id),
+ entity_id TEXT NOT NULL REFERENCES canonical_entity(entity_id),
+ score REAL NOT NULL,
+ model_version TEXT NOT NULL,
+ review_status TEXT NOT NULL DEFAULT 'unreviewed' -- unreviewed|confirmed|rejected
+);
+
+-- ---------------------------------------------------------------------------
+-- Search publication control (what the public index is allowed to show)
+-- ---------------------------------------------------------------------------
+CREATE TABLE search_document_state (
+ property_id TEXT PRIMARY KEY REFERENCES property(property_id),
+ index_version INTEGER NOT NULL DEFAULT 0,
+ is_public BOOLEAN NOT NULL DEFAULT 1,
+ is_suppressed BOOLEAN NOT NULL DEFAULT 0,
+ owner_name_masked TEXT,
+ amount_band TEXT
+);
+
+-- ---------------------------------------------------------------------------
+-- Claims (state machine + transactional outbox)
+-- ---------------------------------------------------------------------------
+CREATE TABLE claim_case (
+ claim_id TEXT PRIMARY KEY,
+ jurisdiction_id TEXT NOT NULL,
+ public_property_reference TEXT NOT NULL,
+ claimant_id TEXT,
+ status TEXT NOT NULL,
+ version INTEGER NOT NULL DEFAULT 1,
+ state_case_id TEXT
+);
+
+CREATE TABLE claim_event ( -- append-only timeline
+ claim_event_id TEXT PRIMARY KEY,
+ claim_id TEXT NOT NULL REFERENCES claim_case(claim_id),
+ event_type TEXT NOT NULL,
+ payload TEXT NOT NULL,
+ idempotency_key TEXT NOT NULL,
+ created_at TIMESTAMP NOT NULL,
+ UNIQUE (claim_id, idempotency_key)
+);
+
+CREATE TABLE outbox_event ( -- transactional outbox → reliable state integration
+ outbox_event_id TEXT PRIMARY KEY,
+ aggregate_id TEXT NOT NULL,
+ event_type TEXT NOT NULL,
+ payload TEXT NOT NULL,
+ delivery_state TEXT NOT NULL DEFAULT 'pending' -- pending|sent|failed
+);
+
+-- ---------------------------------------------------------------------------
+-- Governance
+-- ---------------------------------------------------------------------------
+CREATE TABLE audit_event (
+ audit_event_id TEXT PRIMARY KEY,
+ actor TEXT,
+ action TEXT NOT NULL,
+ resource TEXT,
+ before_hash TEXT,
+ after_hash TEXT,
+ created_at TIMESTAMP NOT NULL
+);
+
+CREATE INDEX idx_owner_name_norm ON owner(owner_name_normalized);
+CREATE INDEX idx_owner_region ON owner(region);
+CREATE INDEX idx_property_jur ON property(jurisdiction_id);
diff --git a/docs/00-thesis-and-guardrails.md b/docs/00-thesis-and-guardrails.md
new file mode 100644
index 0000000..cb4272c
--- /dev/null
+++ b/docs/00-thesis-and-guardrails.md
@@ -0,0 +1,70 @@
+# Thesis & Guardrails
+
+## The thesis in one paragraph
+
+MissingMoney is a government-partnered public-service data platform, not a lead-gen
+site. It aggregates unclaimed-property records supplied by participating jurisdictions,
+lets consumers search free, and routes them to the state (or supports a state-approved
+claim). The defensible replication is a **B2G platform**: obtain authorized state feeds
+under data-use agreements, normalize into a common NAUPA-aligned schema, provide national
+masked search + state-specific claim routing, and earn revenue from **government**
+participation, implementation, white-label, and workflow modules — never from claimants.
+
+## The central challenge is institutional, not technical
+
+NAUPA reporting formats give a national *structural* standard, but they do not grant a
+right to obtain or republish any state's data. The platform needs either a master NAUPA
+relationship or individual state agreements. Public-records requests can supplement, but
+records supplied to third parties may be incomplete, delayed, redacted, or licensed on
+terms that forbid unrestricted republication. Engineering can build a national index in
+~12 months; it should NOT be represented as a complete national service until data
+rights, participation, refresh schedules, masking policies, and claim routes are in force.
+
+## The hard line: what the automated loop will NEVER do autonomously
+
+These are **human-gated** — they draft to `~/.claude/yolo-queue/pending-approval/`, they
+do not execute:
+
+- Acquire, download, or ingest any **real** state or NAUPA feed.
+- **Scrape** any state portal for bulk data (crawling is limited to availability
+ monitoring or a written-permission integration).
+- Sign, send, or negotiate any **LOI, data-use agreement, or contract**.
+- Retain counsel, buy insurance/bonds, or commit **spend**.
+- **Deploy** to any server, change **DNS**, or expose anything publicly.
+- **Send** anything to a state, a list, or a claimant.
+
+## What the loop CAN own (reversible, local, $0)
+
+- Scaffolding and evolving the software prototype against **synthetic** data.
+- The data model, ingestion adapters, masking rules, search, claim state-machine,
+ provenance/lineage, fraud-signal stubs.
+- Living planning docs: the state-acquisition matrix, architecture, roadmap, budget,
+ privacy/data-broker analysis, and the regulatory matrix.
+- Local tests that prove each increment works.
+
+## Revenue architecture (government-first)
+
+| Stream | Buyer | Basis | Risk |
+|---|---|---|---|
+| Annual platform participation | State / NAUPA | fixed / population / volume tier | low–moderate |
+| Feed onboarding & migration | State | one-time implementation | low |
+| White-label state search portal | State | annual software + support | low |
+| Claim-workflow module | State | annual or per-claim | moderate (needs explicit authority) |
+| Identity & document verification | State | pass-through + margin | moderate (privacy/biometric) |
+| Analytics & outreach | State | subscription / project | moderate |
+| Authorized partner API | Gov / regulated institution | usage tiers | moderate |
+
+**Excluded at launch:** selling claimant leads, ads against owner names, charging
+consumers for search, contingent recovery %, buying claims. These create finder-law
+exposure and conflict with the state agencies that are the real customer. State finder-fee
+caps vary (CA ~10%, FL ≤30%, IL ~10%, TX ≤10%, NY restricted) — the platform avoids the
+category entirely at launch.
+
+## Liability allocation
+
+| Responsibility | Owner |
+|---|---|
+| Accuracy of source records | Supplying jurisdiction (with contractual correction) |
+| Accuracy of normalization/display | Platform |
+| Identity proofing / evidence collection | Platform or approved vendor under state rules |
+| Entitlement determination & payment | **State**, unless expressly delegated |
diff --git a/docs/01-regulatory-state-acquisition-matrix.md b/docs/01-regulatory-state-acquisition-matrix.md
new file mode 100644
index 0000000..ca7d3dc
--- /dev/null
+++ b/docs/01-regulatory-state-acquisition-matrix.md
@@ -0,0 +1,90 @@
+# Regulatory & State Data-Acquisition Matrix
+
+**Production rule for every jurisdiction:** negotiate a **direct authorized feed** first;
+evaluate a **public-records bulk request** second; use the **public portal for claim
+handoff** rather than unattended extraction. The official NAUPA directory identifies a
+consumer search resource for all 50 states + DC but does **not** advertise open national
+bulk APIs — so "no public bulk API established by directory" is the default status.
+
+## Access method codes
+
+| Code | Meaning |
+|---|---|
+| `direct_feed` | Authorized full + incremental files over MFT / object storage / private API |
+| `records_request` | Bulk data under public-records law; reviewed for use, redistribution, retention, privacy |
+| `public_handoff` | Platform searches authorized data or shows a state-approved link; claim completed on the state portal |
+
+## Recommended production method (uniform)
+
+`direct_feed` → `records_request` (fallback) → `public_handoff` (claim completion).
+
+## 50 states + DC
+
+Every row's recommended method is identical (feed → records-request → handoff); the
+per-state differences that matter are captured in the machine-readable
+`jurisdictions.json` below (portal name, restrictive-disclosure flags, finder-fee cap).
+
+| Jurisdiction | Official portal (public source) | Notable legal note |
+|---|---|---|
+| Alabama | AL Treasury unclaimed property | — |
+| Alaska | AK Treasury program | — |
+| Arizona | AZ Dept of Revenue | — |
+| Arkansas | ClaimItAR (Auditor of State) | — |
+| California | CA State Controller | Field-level legal review before any records request; investigator agreements ~10% cap |
+| Colorado | Great Colorado Payback | — |
+| Connecticut | CT Big List | — |
+| Delaware | DE Office of Unclaimed Property | Aggressive holder-audit state; treat data terms carefully |
+| District of Columbia | DC Office of Unclaimed Property | Use DC records request as fallback |
+| Florida | FL Treasure Hunt | Recovery agreements ≤30%, professional requirements |
+| Georgia | GA Dept of Revenue | — |
+| Hawaii | HI Budget & Finance | — |
+| Idaho | Your Money Idaho | — |
+| Illinois | IL Treasurer I-Cash | Finders regulated; covered fees ~10% cap |
+| Indiana | IN Unclaimed Property (AG) | — |
+| Iowa | Great Iowa Treasure Hunt | — |
+| Kansas | KS Treasurer | — |
+| Kentucky | KY Treasury | — |
+| Louisiana | LA Treasury / LaCashClaim | — |
+| Maine | ME Treasurer | — |
+| Maryland | MD Comptroller | — |
+| Massachusetts | Find Mass Money | — |
+| Michigan | MI Treasury | — |
+| Minnesota | MN Commerce | — |
+| Mississippi | MS Treasury | — |
+| Missouri | MO Treasurer | — |
+| Montana | MT Dept of Revenue | — |
+| Nebraska | Nebraska Lost Cash | — |
+| Nevada | NV Unclaimed Property | — |
+| New Hampshire | NH unclaimed property | — |
+| New Jersey | NJ Treasury | — |
+| New Mexico | NM Taxation & Revenue | — |
+| New York | NY Office of the State Comptroller | Records request subject to NY disclosure rules; paid location arrangements restricted |
+| North Carolina | NCCash | — |
+| North Dakota | ND Unclaimed Property | — |
+| Ohio | OH Dept of Commerce | — |
+| Oklahoma | OK Treasurer | — |
+| Oregon | OR Unclaimed Property | — |
+| Pennsylvania | PA Treasury | — |
+| Rhode Island | Find RI Money | — |
+| South Carolina | SC Treasurer | — |
+| South Dakota | SD unclaimed property | — |
+| Tennessee | ClaimItTN | — |
+| Texas | ClaimItTexas / TX Comptroller | Finder fees ≤10% |
+| Utah | MyCash Utah | — |
+| Vermont | VT Treasurer | — |
+| Virginia | Virginia Money Search | — |
+| Washington | WA Dept of Revenue | — |
+| West Virginia | WV Treasurer | — |
+| Wisconsin | WI Dept of Revenue | — |
+| Wyoming | WY Treasurer | — |
+
+Puerto Rico and other NAUPA-listed jurisdictions are added after the 50-state + DC
+rollout, modeled as separate jurisdictions (own languages, IDs, addresses, claim rules).
+
+## Privacy / data-broker posture
+
+Design as if handling sensitive regulated data even when a government-record exemption
+could apply: data minimization, purpose limitation, per-state field controls, documented
+retention, consumer privacy-request handling, strict limits on enrichment, no sale of
+claimant search activity. Track CA (data-broker registration + DROP deletion, 2026
+schedule) and TX (data-broker registration + safeguards) applicability per launch state.
diff --git a/docs/02-architecture.md b/docs/02-architecture.md
new file mode 100644
index 0000000..9c1428d
--- /dev/null
+++ b/docs/02-architecture.md
@@ -0,0 +1,60 @@
+# Architecture
+
+Polyglot by design — public search, transactional claims, immutable source files, and
+analytics have different consistency, latency, and durability needs; forcing them into
+one engine is the wrong call.
+
+## Recommended stack
+
+| Role | Technology | Why |
+|---|---|---|
+| System of record | PostgreSQL (managed) | ACID, constraints, joins, JSON, trigram fuzzy for review tools |
+| National masked search | OpenSearch | distributed fuzzy search, faceting, near-real-time; **eventually consistent, not a claim store** |
+| Raw / replay / history | Object storage + open table format | cheap, immutable, versioned; mandatory forensic layer |
+| Analytics / model training | Warehouse (Snowflake/BigQuery/lakehouse) | large scans, workload isolation |
+| Cache / rate limits / sessions | Redis | low-latency atomic counters; not a store of record |
+| Fraud / entity investigations | Graph DB (later) | owners↔addresses↔holders↔claims |
+
+**Prototype substitution:** this repo runs the PostgreSQL + object-store + search roles
+against **SQLite + the local filesystem + an in-DB search fallback** so the whole thing
+runs at $0 with zero infrastructure. The service interfaces are unchanged; only the
+concrete adapters differ.
+
+## Service boundaries (anonymous search is isolated from claims)
+
+| API | Responsibility |
+|---|---|
+| Search API | query, filters, **masked** result cards, pagination tokens, rate limits |
+| Property-selection API | short-lived signed reference for one selected result |
+| Claim API | create case, eligibility, evidence metadata, status |
+| State-integration API | submit case, receive status, apply corrections & suppressions |
+| Administration API | per-jurisdiction field/policy config, support queues, reports |
+| Privacy API | access records, deletion, correction, restrictions |
+| Audit API | restricted evidence export for state & compliance reviewers |
+
+The authoritative property table is **never** directly queryable from the internet.
+
+## Two matching systems, deliberately separate
+
+| System | Objective | Error preference |
+|---|---|---|
+| Public search ranking | retrieve plausible records a user may recognize | favor **recall** |
+| Internal entity resolution | decide whether records refer to the same real owner | favor **precision**; keep a manual-review band |
+| Claimant↔owner comparison | support evidence review for one selected property | **never** auto-decide entitlement from a score |
+
+Fuzzy search must help a person *find* a record; it must never be the basis for
+automatically *approving* a claim.
+
+## Refresh cadence (per-contract, not one-size-fits-all)
+
+| Feed type | Cadence | Rule |
+|---|---|---|
+| Full snapshot | monthly/quarterly | reconcile totals; don't delete absent records until snapshot semantics confirmed |
+| Incremental adds | daily–weekly | idempotent upsert keyed by (jurisdiction, state record id) |
+| Corrections | daily | preserve prior version + effective timestamp |
+| Suppressions | near-real-time/daily | remove from public search promptly; retain restricted audit history |
+| Claim dispositions | daily–weekly | mark claimed/paid/denied/inactive per state definitions |
+
+Every batch carries: checksum, source URI, jurisdiction, parser version, record/error/
+accepted counts, effective date, transformation version. Reprocessing the same batch is
+idempotent unless the transformation version changes.
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..53383c5
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,13 @@
+# The core prototype + smoke test are STDLIB-ONLY (no installs needed).
+# These are for the PRODUCTION reference services only:
+
+# masked search API (services/search/search_api.py)
+fastapi>=0.111
+uvicorn>=0.30
+opensearch-py>=2.6
+pydantic>=2.7
+
+# production ingestion validation + faster fuzzy matching (swap-ins, optional)
+# pydantic>=2.7 # replaces the stdlib dataclass validation in ingest.py
+# rapidfuzz>=3.9 # replaces difflib in services/matching/entity_match.py
+# psycopg[binary]>=3.2 # PostgreSQL system-of-record repository
diff --git a/services/__init__.py b/services/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/claims/__init__.py b/services/claims/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/claims/claim_workflow.py b/services/claims/claim_workflow.py
new file mode 100644
index 0000000..d818abd
--- /dev/null
+++ b/services/claims/claim_workflow.py
@@ -0,0 +1,119 @@
+"""Claim workflow — explicit state machine + transactional outbox.
+
+Package preparation is deliberately separated from state adjudication. The platform never
+decides entitlement and never pays; the STATE does. Guarantees:
+ - explicit allowed transitions (illegal transitions raise)
+ - optimistic version increment
+ - idempotency keys on every event
+ - append-only event history
+ - transactional outbox so a transient integration failure can't double-submit to a state
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import Enum
+from typing import Protocol
+from uuid import UUID, uuid4
+
+
+class ClaimStatus(str, Enum):
+ DRAFT = "draft"
+ IDENTITY_PENDING = "identity_pending"
+ EVIDENCE_PENDING = "evidence_pending"
+ READY_FOR_SUBMISSION = "ready_for_submission"
+ SUBMITTING = "submitting"
+ SUBMITTED_TO_STATE = "submitted_to_state"
+ MORE_INFORMATION_REQUIRED = "more_information_required"
+ APPROVED = "approved"
+ DENIED = "denied"
+ PAID_BY_STATE = "paid_by_state"
+ CANCELLED = "cancelled"
+
+
+ALLOWED_TRANSITIONS: dict[ClaimStatus, set[ClaimStatus]] = {
+ ClaimStatus.DRAFT: {ClaimStatus.IDENTITY_PENDING, ClaimStatus.CANCELLED},
+ ClaimStatus.IDENTITY_PENDING: {ClaimStatus.EVIDENCE_PENDING, ClaimStatus.CANCELLED},
+ ClaimStatus.EVIDENCE_PENDING: {ClaimStatus.READY_FOR_SUBMISSION, ClaimStatus.CANCELLED},
+ ClaimStatus.READY_FOR_SUBMISSION: {ClaimStatus.SUBMITTING, ClaimStatus.EVIDENCE_PENDING},
+ ClaimStatus.SUBMITTING: {ClaimStatus.SUBMITTED_TO_STATE, ClaimStatus.READY_FOR_SUBMISSION},
+ ClaimStatus.SUBMITTED_TO_STATE: {
+ ClaimStatus.MORE_INFORMATION_REQUIRED, ClaimStatus.APPROVED, ClaimStatus.DENIED},
+ ClaimStatus.MORE_INFORMATION_REQUIRED: {ClaimStatus.EVIDENCE_PENDING, ClaimStatus.DENIED},
+ ClaimStatus.APPROVED: {ClaimStatus.PAID_BY_STATE},
+ ClaimStatus.DENIED: set(),
+ ClaimStatus.PAID_BY_STATE: set(),
+ ClaimStatus.CANCELLED: set(),
+}
+
+
+@dataclass
+class Claim:
+ claim_id: UUID
+ jurisdiction: str
+ public_property_reference: str
+ claimant_id: UUID
+ status: ClaimStatus
+ version: int
+ state_case_id: str | None = None
+
+
+class ClaimRepository(Protocol):
+ def get_for_update(self, claim_id: UUID) -> Claim: ...
+ def save(self, claim: Claim) -> None: ...
+ def append_event(self, claim_id: UUID, event_type: str, payload: dict,
+ idempotency_key: str) -> None: ...
+ def add_outbox_event(self, event_type: str, aggregate_id: UUID, payload: dict) -> None: ...
+
+
+class StateAdapter(Protocol):
+ def submit_claim(self, claim: Claim, idempotency_key: str) -> str: ...
+
+
+def transition_claim(repository: ClaimRepository, claim_id: UUID, target: ClaimStatus,
+ actor_id: str, idempotency_key: str) -> Claim:
+ claim = repository.get_for_update(claim_id)
+ if target not in ALLOWED_TRANSITIONS[claim.status]:
+ raise ValueError(f"Invalid transition from {claim.status} to {target}")
+ previous = claim.status
+ claim.status = target
+ claim.version += 1
+ repository.save(claim)
+ repository.append_event(
+ claim_id=claim.claim_id, event_type="claim_status_changed",
+ payload={"from": previous.value, "to": target.value,
+ "actor_id": actor_id, "version": claim.version},
+ idempotency_key=idempotency_key,
+ )
+ return claim
+
+
+def queue_state_submission(repository: ClaimRepository, claim_id: UUID, actor_id: str,
+ idempotency_key: str) -> Claim:
+ claim = transition_claim(repository, claim_id, ClaimStatus.SUBMITTING,
+ actor_id, idempotency_key)
+ # Written in the SAME transaction as the claim; a worker delivers it to the state.
+ repository.add_outbox_event(
+ event_type="state_claim_submission_requested",
+ aggregate_id=claim.claim_id,
+ payload={"claim_id": str(claim.claim_id), "jurisdiction": claim.jurisdiction,
+ "idempotency_key": idempotency_key, "correlation_id": str(uuid4())},
+ )
+ return claim
+
+
+def complete_state_submission(repository: ClaimRepository, adapter: StateAdapter,
+ claim_id: UUID, idempotency_key: str) -> Claim:
+ claim = repository.get_for_update(claim_id)
+ if claim.status != ClaimStatus.SUBMITTING:
+ raise ValueError("Claim is not awaiting submission")
+ external_case_id = adapter.submit_claim(claim=claim, idempotency_key=idempotency_key)
+ claim.state_case_id = external_case_id
+ claim.status = ClaimStatus.SUBMITTED_TO_STATE
+ claim.version += 1
+ repository.save(claim)
+ repository.append_event(
+ claim_id=claim.claim_id, event_type="claim_submitted_to_state",
+ payload={"state_case_id": external_case_id, "version": claim.version},
+ idempotency_key=idempotency_key,
+ )
+ return claim
diff --git a/services/common/__init__.py b/services/common/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/common/normalize.py b/services/common/normalize.py
new file mode 100644
index 0000000..a7ef97a
--- /dev/null
+++ b/services/common/normalize.py
@@ -0,0 +1,118 @@
+"""Shared normalization + masking primitives (stdlib only).
+
+Raw source values are never mutated by these functions — callers store the raw string
+alongside the normalized output. Production may swap in ICU/libpostal for parsing, but
+the CONTRACT (what 'normalized' means per field) is defined here so ingestion, search,
+and matching all agree.
+"""
+from __future__ import annotations
+
+import re
+import unicodedata
+from decimal import Decimal, InvalidOperation
+
+CORPORATE_SUFFIXES = {
+ "INC", "INCORPORATED", "LLC", "LLC.", "L L C", "LTD", "LIMITED",
+ "CORP", "CORPORATION", "CO", "COMPANY", "LP", "LLP", "PLLC",
+}
+
+
+def normalize_text(value: str | None) -> str:
+ """Uppercase, strip accents, collapse to single-spaced alphanumerics."""
+ if not value:
+ return ""
+ decomposed = unicodedata.normalize("NFKD", value)
+ ascii_like = "".join(ch for ch in decomposed if not unicodedata.combining(ch))
+ cleaned = re.sub(r"[^A-Z0-9]+", " ", ascii_like.upper())
+ return re.sub(r"\s+", " ", cleaned).strip()
+
+
+def normalize_business(value: str | None) -> str:
+ """Normalize a business name, dropping legal suffixes but keeping brand tokens."""
+ tokens = normalize_text(value).split()
+ meaningful = [t for t in tokens if t not in CORPORATE_SUFFIXES]
+ return " ".join(meaningful) or normalize_text(value)
+
+
+def parse_decimal(value: str | None) -> Decimal | None:
+ if not value or not value.strip():
+ return None
+ try:
+ return Decimal(value.replace("$", "").replace(",", "").strip())
+ except InvalidOperation as exc:
+ raise ValueError(f"Invalid amount: {value!r}") from exc
+
+
+# Soundex digit groups (the classic mapping). We build a phonetic key by coding EVERY
+# consonant (including the first letter) and dropping vowels/H/W/Y — so "Catherine" and
+# "Kathryn" collapse to the same code (C and K are both group 2), which first-letter
+# Soundex would miss. Production swaps this for Double Metaphone; the FEATURE contract
+# (phonetic agreement) is what matters here.
+_SOUNDEX_MAP = {
+ **{c: "1" for c in "BFPV"},
+ **{c: "2" for c in "CGJKQSXZ"},
+ **{c: "3" for c in "DT"},
+ "L": "4",
+ **{c: "5" for c in "MN"},
+ "R": "6",
+}
+
+
+def phonetic_key(value: str | None) -> str:
+ """Space-joined per-token phonetic code (vowel-stripped, adjacent-duplicate-collapsed)."""
+ out_tokens = []
+ for token in normalize_text(value).split():
+ digits: list[str] = []
+ for ch in token:
+ code = _SOUNDEX_MAP.get(ch)
+ if code and (not digits or digits[-1] != code):
+ digits.append(code)
+ elif not code:
+ # a vowel/H/W/Y breaks a run so a real repeated sound isn't over-collapsed
+ if digits and digits[-1] == "":
+ continue
+ digits.append("")
+ joined = "".join(d for d in digits if d)
+ if joined:
+ out_tokens.append(joined)
+ return " ".join(out_tokens)
+
+
+def normalize_postal(value: str | None) -> str | None:
+ if not value:
+ return None
+ digits = re.sub(r"\D", "", value)[:9]
+ return digits or None
+
+
+def mask_name(raw_name: str | None) -> str:
+ """Public-search masking: reveal first char of each token, mask the rest.
+
+ 'CATHERINE ONEILL' -> 'C******* O*****' — enough to recognize, not to enumerate.
+ """
+ norm = normalize_text(raw_name)
+ if not norm:
+ return ""
+ masked_tokens = []
+ for tok in norm.split():
+ masked_tokens.append(tok[0] + ("*" * (len(tok) - 1)) if len(tok) > 1 else tok)
+ return " ".join(masked_tokens)
+
+
+# Coarse public amount bands — never expose the exact figure through anonymous search.
+_BANDS = [
+ (Decimal("50"), "Under $50"),
+ (Decimal("100"), "$50–$100"),
+ (Decimal("500"), "$100–$500"),
+ (Decimal("1000"), "$500–$1,000"),
+ (Decimal("5000"), "$1,000–$5,000"),
+]
+
+
+def amount_band(amount: Decimal | None) -> str | None:
+ if amount is None:
+ return None
+ for ceiling, label in _BANDS:
+ if amount < ceiling:
+ return label
+ return "$5,000+"
diff --git a/services/common/sqlite_repo.py b/services/common/sqlite_repo.py
new file mode 100644
index 0000000..a39cc5b
--- /dev/null
+++ b/services/common/sqlite_repo.py
@@ -0,0 +1,182 @@
+"""Concrete SQLite-backed Repository + filesystem ObjectStore for the local prototype.
+
+Implements the same interface the production PostgreSQL repository and cloud object store
+will implement, so services/ingestion/ingest.py is written once and runs unchanged against
+either. Everything here is local and $0.
+
+Idempotency is enforced two ways:
+ 1. ingestion_batch UNIQUE(jurisdiction_id, checksum) — the same source FILE never
+ produces two batches.
+ 2. property UNIQUE(jurisdiction_id, source_property_id) — a re-delivered RECORD upserts
+ in place; it never creates a duplicate property row.
+"""
+from __future__ import annotations
+
+import os
+import sqlite3
+import uuid
+from datetime import datetime, timezone
+from pathlib import Path
+
+_SCHEMA_PATH = Path(__file__).resolve().parents[2] / "db" / "schema.sql"
+
+
+def _now() -> str:
+ return datetime.now(timezone.utc).isoformat()
+
+
+class FileObjectStore:
+ """Filesystem stand-in for cloud object storage (immutable raw archive)."""
+
+ def __init__(self, root: str | os.PathLike) -> None:
+ self.root = Path(root)
+ self.root.mkdir(parents=True, exist_ok=True)
+
+ def read_bytes(self, uri: str) -> bytes:
+ return (self.root / uri).read_bytes()
+
+ def write_bytes(self, uri: str, data: bytes) -> None:
+ target = self.root / uri
+ target.parent.mkdir(parents=True, exist_ok=True)
+ target.write_bytes(data)
+
+
+class SqliteRepository:
+ def __init__(self, db_path: str = ":memory:") -> None:
+ self.conn = sqlite3.connect(db_path)
+ self.conn.row_factory = sqlite3.Row
+ self.conn.execute("PRAGMA foreign_keys = ON")
+ self._init_schema()
+
+ def _init_schema(self) -> None:
+ # Load the PostgreSQL schema, downgrading the few types SQLite doesn't share.
+ ddl = _SCHEMA_PATH.read_text()
+ ddl = ddl.replace("NUMERIC", "REAL").replace("TIMESTAMP", "TEXT").replace("BOOLEAN", "INTEGER")
+ self.conn.executescript(ddl)
+ # Minimal jurisdiction seed so FK-free prototype inserts succeed.
+ self.conn.execute(
+ "INSERT OR IGNORE INTO jurisdiction(jurisdiction_id, name, status) VALUES (?,?,?)",
+ ("SAMPLE", "Synthetic Sample State", "prototype"),
+ )
+ self.conn.commit()
+
+ # --- ingestion Repository protocol -------------------------------------
+ def batch_exists(self, jurisdiction: str, checksum: str) -> bool:
+ row = self.conn.execute(
+ "SELECT 1 FROM ingestion_batch WHERE jurisdiction_id=? AND checksum=?",
+ (jurisdiction, checksum),
+ ).fetchone()
+ return row is not None
+
+ def begin_batch(self, jurisdiction: str, source_uri: str, checksum: str,
+ parser_version: str) -> str:
+ batch_id = str(uuid.uuid4())
+ self.conn.execute(
+ "INSERT OR IGNORE INTO jurisdiction(jurisdiction_id, name, status) VALUES (?,?,?)",
+ (jurisdiction, f"Jurisdiction {jurisdiction}", "prototype"),
+ )
+ self.conn.execute(
+ """INSERT INTO ingestion_batch
+ (batch_id, jurisdiction_id, source_uri, checksum, parser_version,
+ received_at, status)
+ VALUES (?,?,?,?,?,?, 'running')""",
+ (batch_id, jurisdiction, source_uri, checksum, parser_version, _now()),
+ )
+ self.conn.commit()
+ return batch_id
+
+ def upsert_property(self, batch_id: str, record) -> None:
+ """Idempotent upsert of one canonical property + its owner + search doc."""
+ jur = record.jurisdiction
+ spid = record.source_property_id
+ existing = self.conn.execute(
+ "SELECT property_id FROM property WHERE jurisdiction_id=? AND source_property_id=?",
+ (jur, spid),
+ ).fetchone()
+
+ amount = float(record.amount) if record.amount is not None else None
+ if existing:
+ property_id = existing["property_id"]
+ self.conn.execute(
+ "UPDATE property SET holder_name_raw=?, property_type=?, amount=? WHERE property_id=?",
+ (record.holder_name_raw, record.property_type, amount, property_id),
+ )
+ else:
+ property_id = str(uuid.uuid4())
+ self.conn.execute(
+ """INSERT INTO property
+ (property_id, jurisdiction_id, source_property_id, holder_name_raw,
+ property_type, amount, status)
+ VALUES (?,?,?,?,?,?, 'active')""",
+ (property_id, jur, spid, record.holder_name_raw, record.property_type, amount),
+ )
+
+ # Non-destructive version row (one per batch delivery).
+ self.conn.execute(
+ """INSERT INTO property_version
+ (property_version_id, property_id, batch_id, effective_from, raw_payload,
+ raw_record_hash)
+ VALUES (?,?,?,?,?,?)""",
+ (str(uuid.uuid4()), property_id, batch_id, _now(),
+ record.raw_payload, record.raw_record_hash),
+ )
+
+ # Owner: replace the current owner row for this property (prototype simplification).
+ self.conn.execute("DELETE FROM owner WHERE property_id=?", (property_id,))
+ self.conn.execute(
+ """INSERT INTO owner
+ (owner_id, property_id, owner_type, owner_name_raw, owner_name_normalized,
+ city_normalized, region, postal_code)
+ VALUES (?,?,?,?,?,?,?,?)""",
+ (str(uuid.uuid4()), property_id, record.owner_type, record.owner_name_raw,
+ record.owner_name_normalized, record.city_normalized, record.region,
+ record.postal_code),
+ )
+
+ # Search publication state (masked projection only).
+ self.conn.execute(
+ """INSERT INTO search_document_state
+ (property_id, index_version, is_public, is_suppressed, owner_name_masked, amount_band)
+ VALUES (?,?,?,?,?,?)
+ ON CONFLICT(property_id) DO UPDATE SET
+ index_version = index_version + 1,
+ owner_name_masked = excluded.owner_name_masked,
+ amount_band = excluded.amount_band""",
+ (property_id, 0, 1, 0, record.owner_name_masked, record.amount_band),
+ )
+ self.conn.commit()
+
+ def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str) -> None:
+ self.conn.execute(
+ "UPDATE ingestion_batch SET accepted_count=?, rejected_count=?, status=? WHERE batch_id=?",
+ (accepted, rejected, status, batch_id),
+ )
+ self.conn.commit()
+
+ # --- convenience reads for tests / review tools ------------------------
+ def count_properties(self, jurisdiction: str | None = None) -> int:
+ if jurisdiction:
+ return self.conn.execute(
+ "SELECT COUNT(*) FROM property WHERE jurisdiction_id=?", (jurisdiction,)
+ ).fetchone()[0]
+ return self.conn.execute("SELECT COUNT(*) FROM property").fetchone()[0]
+
+ def count_versions(self, jurisdiction: str | None = None) -> int:
+ return self.conn.execute("SELECT COUNT(*) FROM property_version").fetchone()[0]
+
+ def masked_search(self, name_query: str, limit: int = 20) -> list[dict]:
+ """In-DB fallback for the OpenSearch masked search (prototype only)."""
+ from services.common.normalize import normalize_text
+ norm = normalize_text(name_query)
+ rows = self.conn.execute(
+ """SELECT s.owner_name_masked, s.amount_band, p.jurisdiction_id, p.holder_name_raw,
+ p.property_type
+ FROM owner o
+ JOIN property p ON p.property_id = o.property_id
+ JOIN search_document_state s ON s.property_id = p.property_id
+ WHERE s.is_public=1 AND s.is_suppressed=0
+ AND o.owner_name_normalized LIKE ?
+ LIMIT ?""",
+ (f"%{norm.split()[0]}%" if norm else "%", limit),
+ ).fetchall()
+ return [dict(r) for r in rows]
diff --git a/services/ingestion/__init__.py b/services/ingestion/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/ingestion/ingest.py b/services/ingestion/ingest.py
new file mode 100644
index 0000000..77de406
--- /dev/null
+++ b/services/ingestion/ingest.py
@@ -0,0 +1,144 @@
+"""Authorized-feed ingestion with idempotent ETL + immutable raw archive.
+
+Accepts an AUTHORIZED file (state CSV here; NAUPA II fixed-width and NAUPA III XML
+adapters plug in the same way). It NEVER scrapes and NEVER fetches from a state portal —
+the source is always a file already handed to us under a data-use agreement.
+
+Idempotency: a re-run of the same file short-circuits at the batch checksum; a
+re-delivered record upserts in place (see SqliteRepository).
+"""
+from __future__ import annotations
+
+import csv
+import hashlib
+import io
+import json
+from dataclasses import dataclass
+from decimal import Decimal
+from typing import Iterable, Protocol
+
+from services.common.normalize import (
+ amount_band, mask_name, normalize_business, normalize_postal, normalize_text,
+ parse_decimal,
+)
+
+
+class ObjectStore(Protocol):
+ def read_bytes(self, uri: str) -> bytes: ...
+ def write_bytes(self, uri: str, data: bytes) -> None: ...
+
+
+class Repository(Protocol):
+ def batch_exists(self, jurisdiction: str, checksum: str) -> bool: ...
+ def begin_batch(self, jurisdiction: str, source_uri: str, checksum: str,
+ parser_version: str) -> str: ...
+ def upsert_property(self, batch_id: str, record: "CanonicalProperty") -> None: ...
+ def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str) -> None: ...
+
+
+@dataclass
+class CanonicalProperty:
+ jurisdiction: str
+ source_property_id: str
+ holder_name_raw: str
+ owner_type: str
+ owner_name_raw: str
+ owner_name_normalized: str
+ owner_name_masked: str
+ amount: Decimal | None
+ amount_band: str | None
+ raw_payload: str
+ raw_record_hash: str
+ city_normalized: str | None = None
+ region: str | None = None
+ postal_code: str | None = None
+ property_type: str | None = None
+
+
+@dataclass(frozen=True)
+class FeedDefinition:
+ jurisdiction: str
+ source_uri: str
+ format_name: str = "state_csv_v1"
+ parser_version: str = "2026.07.1"
+
+
+def _looks_like_business(name: str) -> bool:
+ from services.common.normalize import CORPORATE_SUFFIXES
+ return any(tok in CORPORATE_SUFFIXES for tok in normalize_text(name).split())
+
+
+def parse_csv_feed(data: bytes, jurisdiction: str) -> Iterable[CanonicalProperty]:
+ """Map a state's CSV columns into the canonical model.
+
+ Expected columns: property_id, holder_name, owner_name, address, city, state, zip,
+ property_type, amount. A real adapter maps each state's own field names + code tables.
+ """
+ reader = csv.DictReader(io.StringIO(data.decode("utf-8-sig")))
+ for row in reader:
+ serialized = json.dumps(row, sort_keys=True)
+ raw_hash = hashlib.sha256(serialized.encode()).hexdigest()
+ owner_name = (row.get("owner_name") or "").strip()
+ is_business = _looks_like_business(owner_name)
+ amount = parse_decimal(row.get("amount"))
+ owner_norm = (
+ normalize_business(owner_name) if is_business else normalize_text(owner_name)
+ )
+ yield CanonicalProperty(
+ jurisdiction=jurisdiction,
+ source_property_id=(row.get("property_id") or "").strip(),
+ holder_name_raw=(row.get("holder_name") or "").strip(),
+ owner_type="business" if is_business else "person",
+ owner_name_raw=owner_name,
+ owner_name_normalized=owner_norm,
+ owner_name_masked=mask_name(owner_name),
+ amount=amount,
+ amount_band=amount_band(amount),
+ raw_payload=serialized,
+ raw_record_hash=raw_hash,
+ city_normalized=normalize_text(row.get("city")) or None,
+ region=normalize_text(row.get("state")) or None,
+ postal_code=normalize_postal(row.get("zip")),
+ property_type=normalize_text(row.get("property_type")) or None,
+ )
+
+
+def ingest_authorized_feed(feed: FeedDefinition, object_store: ObjectStore,
+ repository: Repository) -> dict:
+ data = object_store.read_bytes(feed.source_uri)
+ checksum = hashlib.sha256(data).hexdigest()
+
+ # Idempotency gate 1: identical file already processed.
+ if repository.batch_exists(feed.jurisdiction, checksum):
+ return {"status": "duplicate", "accepted": 0, "rejected": 0}
+
+ # Immutable raw archive (forensic replay).
+ raw_archive_uri = f"raw/{feed.jurisdiction}/{checksum}.bin"
+ object_store.write_bytes(raw_archive_uri, data)
+
+ batch_id = repository.begin_batch(
+ jurisdiction=feed.jurisdiction, source_uri=feed.source_uri,
+ checksum=checksum, parser_version=feed.parser_version,
+ )
+
+ accepted = rejected = 0
+ try:
+ if feed.format_name != "state_csv_v1":
+ raise NotImplementedError(f"Unsupported format: {feed.format_name}")
+ for record in parse_csv_feed(data, feed.jurisdiction):
+ if not record.source_property_id or not record.owner_name_raw:
+ rejected += 1
+ continue
+ try:
+ repository.upsert_property(batch_id, record)
+ accepted += 1
+ except Exception:
+ rejected += 1
+ status = "completed_with_errors" if rejected else "completed"
+ repository.finish_batch(batch_id, accepted, rejected, status)
+ except Exception:
+ repository.finish_batch(batch_id, accepted, rejected, "failed")
+ raise
+
+ return {"status": status, "batch_id": batch_id,
+ "accepted": accepted, "rejected": rejected}
diff --git a/services/matching/__init__.py b/services/matching/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/matching/entity_match.py b/services/matching/entity_match.py
new file mode 100644
index 0000000..7a607a3
--- /dev/null
+++ b/services/matching/entity_match.py
@@ -0,0 +1,105 @@
+"""Entity resolution — probabilistic linkage, DISTINCT from public search ranking.
+
+Search ranking favors recall (help a person find a record). This favors precision (decide
+whether two records refer to the same real owner) and keeps a manual-review band. A score
+here NEVER auto-approves a claim.
+
+The weights below are an ILLUSTRATIVE logistic model. Production replaces them with a
+Fellegi-Sunter model whose agreement/disagreement weights are frequency-adjusted (rare-name
+agreement counts more than a common surname) and calibrated against state-adjudicated pairs,
+with per-jurisdiction error analysis and protected-class fairness testing.
+
+Stdlib-only: uses difflib.SequenceMatcher for string similarity in place of rapidfuzz.
+"""
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass, field
+from difflib import SequenceMatcher
+
+from services.common.normalize import (
+ normalize_business, normalize_postal, normalize_text, phonetic_key,
+)
+
+
+@dataclass(frozen=True)
+class MatchInput:
+ name: str
+ address: str | None = None
+ city: str | None = None
+ region: str | None = None
+ postal_code: str | None = None
+ is_business: bool = False
+
+
+@dataclass(frozen=True)
+class MatchResult:
+ probability: float
+ disposition: str # likely_duplicate_candidate | manual_review | distinct
+ features: dict = field(default_factory=dict)
+
+
+def _ratio(a: str, b: str) -> float:
+ if not a or not b:
+ return 0.0
+ return SequenceMatcher(None, a, b).ratio()
+
+
+def _token_set_ratio(a: str, b: str) -> float:
+ """Order-independent token overlap (Jaccard-ish), a cheap token_set_ratio stand-in."""
+ ta, tb = set(a.split()), set(b.split())
+ if not ta or not tb:
+ return 0.0
+ return len(ta & tb) / len(ta | tb)
+
+
+def similarity(left: str | None, right: str | None) -> float:
+ return _ratio(normalize_text(left), normalize_text(right))
+
+
+def entity_match(left: MatchInput, right: MatchInput) -> MatchResult:
+ # A person and a business are never the same entity.
+ if left.is_business != right.is_business:
+ return MatchResult(0.0, "distinct", {"type_conflict": 1.0})
+
+ ln = normalize_business(left.name) if left.is_business else normalize_text(left.name)
+ rn = normalize_business(right.name) if right.is_business else normalize_text(right.name)
+
+ features = {
+ "name_edit": _ratio(ln, rn),
+ "name_token": _token_set_ratio(ln, rn),
+ "phonetic": _ratio(phonetic_key(ln), phonetic_key(rn)),
+ "address": similarity(left.address, right.address),
+ "city": similarity(left.city, right.city),
+ "region_exact": float(
+ bool(left.region and right.region)
+ and normalize_text(left.region) == normalize_text(right.region)
+ ),
+ "postal_exact": float(
+ bool(left.postal_code and right.postal_code)
+ and (normalize_postal(left.postal_code) or "")[:5]
+ == (normalize_postal(right.postal_code) or "")[:5]
+ ),
+ }
+
+ # ILLUSTRATIVE weights — train + calibrate before production use.
+ log_odds = (
+ -8.0
+ + 3.6 * features["name_edit"]
+ + 2.8 * features["name_token"]
+ + 2.2 * features["phonetic"]
+ + 2.3 * features["address"]
+ + 1.1 * features["city"]
+ + 1.0 * features["region_exact"]
+ + 1.5 * features["postal_exact"]
+ )
+ probability = 1.0 / (1.0 + math.exp(-log_odds))
+
+ if probability >= 0.995:
+ disposition = "likely_duplicate_candidate"
+ elif probability >= 0.90:
+ disposition = "manual_review"
+ else:
+ disposition = "distinct"
+
+ return MatchResult(round(probability, 6), disposition, features)
diff --git a/services/search/__init__.py b/services/search/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/services/search/search_api.py b/services/search/search_api.py
new file mode 100644
index 0000000..02c2c8f
--- /dev/null
+++ b/services/search/search_api.py
@@ -0,0 +1,106 @@
+"""Masked public search API (FastAPI + OpenSearch reference).
+
+This is the PRODUCTION reference. It requires `fastapi` and `opensearch-py`
+(see requirements.txt) and a running OpenSearch. For the dependency-free prototype
+smoke test, the in-DB masked_search() on SqliteRepository is used instead.
+
+Design invariants:
+ - anonymous search is a SEPARATE service from claims; the authoritative property table
+ is never internet-queryable.
+ - only jurisdiction-approved, MASKED projections are returned (masked name, city,
+ holder, type, amount BAND, state reference) — never SSNs, full addresses, exact
+ amounts, or claim evidence.
+ - fuzzy queries are constrained (filters, prefix_length, max_expansions, result caps)
+ because broad fuzzy/wildcard queries are resource-expensive and enable enumeration.
+"""
+from __future__ import annotations
+
+import os
+from typing import Annotated, Any
+
+try:
+ from fastapi import Depends, FastAPI, Header, HTTPException, Query
+ from opensearchpy import OpenSearch
+ from pydantic import BaseModel
+ _DEPS = True
+except ImportError: # keep the module importable for docs/tests without the deps
+ _DEPS = False
+ BaseModel = object # type: ignore
+
+
+if _DEPS:
+ app = FastAPI(title="National Unclaimed Property Search API")
+
+ search_client = OpenSearch(
+ hosts=[os.environ.get("OPENSEARCH_URL", "https://localhost:9200")],
+ http_auth=(os.environ.get("OPENSEARCH_USERNAME", ""),
+ os.environ.get("OPENSEARCH_PASSWORD", "")),
+ use_ssl=True, verify_certs=True,
+ )
+
+ class SearchResult(BaseModel):
+ public_reference: str
+ jurisdiction: str
+ owner_name_masked: str
+ owner_city: str | None
+ holder_name: str
+ property_type: str | None
+ amount_band: str | None
+
+ class SearchResponse(BaseModel):
+ results: list[SearchResult]
+ total_relation: str
+ next_page_token: str | None = None
+
+ def require_rate_limit(
+ forwarded_for: Annotated[str | None, Header()] = None,
+ ) -> None:
+ """Replace with an atomic Redis limiter keyed by IP + device + session.
+ Raise 429 when anonymous-enumeration thresholds are exceeded."""
+ if forwarded_for and len(forwarded_for) > 500:
+ raise HTTPException(status_code=400, detail="Invalid forwarding header")
+
+ @app.get("/v1/search", response_model=SearchResponse)
+ def search_properties(
+ last_or_business: Annotated[str, Query(min_length=2, max_length=100)],
+ first_name: Annotated[str | None, Query(max_length=100)] = None,
+ jurisdiction: Annotated[str | None, Query(min_length=2, max_length=3)] = None,
+ city: Annotated[str | None, Query(max_length=100)] = None,
+ limit: Annotated[int, Query(ge=1, le=50)] = 20,
+ _: None = Depends(require_rate_limit),
+ ) -> "SearchResponse":
+ must: list[dict[str, Any]] = [{
+ "multi_match": {
+ "query": last_or_business,
+ "fields": ["owner_name_normalized^5", "owner_name_tokens^3",
+ "business_name_normalized^4"],
+ "type": "best_fields",
+ "fuzziness": "AUTO:4,7", "prefix_length": 1, "max_expansions": 25,
+ }
+ }]
+ filters: list[dict[str, Any]] = [
+ {"term": {"is_public": True}},
+ {"term": {"is_suppressed": False}},
+ ]
+ if first_name:
+ must.append({"match": {"first_name_normalized": {
+ "query": first_name, "fuzziness": "AUTO"}}})
+ if jurisdiction:
+ filters.append({"term": {"jurisdiction": jurisdiction.upper()}})
+ if city:
+ filters.append({"match": {"city_normalized": city}})
+
+ response = search_client.search(
+ index="unclaimed-property-public",
+ body={
+ "size": limit, "track_total_hits": False,
+ "_source": ["public_reference", "jurisdiction", "owner_name_masked",
+ "owner_city", "holder_name", "property_type", "amount_band"],
+ "query": {"bool": {"must": must, "filter": filters}},
+ },
+ )
+ hits = response["hits"]["hits"]
+ return SearchResponse(
+ results=[SearchResult(**h["_source"]) for h in hits],
+ total_relation=response["hits"]["total"]["relation"],
+ )
diff --git a/tests/__init__.py b/tests/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/test_ingest_and_match.py b/tests/test_ingest_and_match.py
new file mode 100644
index 0000000..5cab4cb
--- /dev/null
+++ b/tests/test_ingest_and_match.py
@@ -0,0 +1,112 @@
+"""End-to-end smoke test — proves the prototype runs at $0 on synthetic data (stdlib only).
+
+Run: python -m tests.test_ingest_and_match
+
+Asserts:
+ 1. Ingestion loads the synthetic feed and rejects the empty-owner row.
+ 2. Ingestion is IDEMPOTENT — re-running the same file is a no-op ('duplicate'), and
+ record count does not grow.
+ 3. Masking works — no raw owner name leaks into the search projection.
+ 4. Entity matching links spelling variants (person + business) while keeping distinct
+ people distinct.
+"""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# make repo root importable when run as a script
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.common.sqlite_repo import FileObjectStore, SqliteRepository
+from services.ingestion.ingest import FeedDefinition, ingest_authorized_feed
+from services.matching.entity_match import MatchInput, entity_match
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+SAMPLE = REPO_ROOT / "data" / "sample" / "sample_state_feed.csv"
+
+
+def _passed(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def main() -> int:
+ import tempfile
+
+ tmp = Path(tempfile.mkdtemp(prefix="upp-proto-"))
+ store = FileObjectStore(tmp / "objstore")
+ # stage the synthetic feed into the object store under a relative uri
+ (tmp / "objstore").mkdir(parents=True, exist_ok=True)
+ store.write_bytes("incoming/sample.csv", SAMPLE.read_bytes())
+ repo = SqliteRepository(str(tmp / "proto.db"))
+
+ feed = FeedDefinition(jurisdiction="SAMPLE", source_uri="incoming/sample.csv")
+
+ print("1) Ingestion + rejection")
+ r1 = ingest_authorized_feed(feed, store, repo)
+ assert r1["status"] in ("completed", "completed_with_errors"), r1
+ # 8 rows, 1 has empty owner_name -> rejected
+ assert r1["accepted"] == 7, f"expected 7 accepted, got {r1['accepted']}"
+ assert r1["rejected"] == 1, f"expected 1 rejected, got {r1['rejected']}"
+ _passed(f"accepted={r1['accepted']} rejected={r1['rejected']} (empty-owner row rejected)")
+
+ print("2) Idempotency")
+ count_after_first = repo.count_properties("SAMPLE")
+ r2 = ingest_authorized_feed(feed, store, repo)
+ assert r2["status"] == "duplicate", r2
+ assert repo.count_properties("SAMPLE") == count_after_first, "re-ingest changed count!"
+ _passed(f"re-ingest -> 'duplicate', property count stable at {count_after_first}")
+
+ print("3) Masking (no raw name leaks into search projection)")
+ hits = repo.masked_search("Catherine")
+ assert hits, "expected at least one masked hit for 'Catherine'"
+ leaked = [h for h in hits if "CATHERINE" in (h["owner_name_masked"] or "").upper()]
+ assert not leaked, f"raw name leaked into masked projection: {leaked}"
+ _passed(f"masked hit sample: {hits[0]['owner_name_masked']} | {hits[0]['amount_band']}")
+
+ print("4) Entity matching")
+ person = entity_match(
+ MatchInput("Catherine O'Neil", city="Springfield", region="SAMPLE",
+ postal_code="00001", address="100 Test St"),
+ MatchInput("Kathryn ONeill", city="Springfield", region="SAMPLE",
+ postal_code="00001", address="100 Test Street"),
+ )
+ assert person.disposition in ("manual_review", "likely_duplicate_candidate"), person
+ _passed(f"person variants -> {person.disposition} (p={person.probability})")
+
+ business = entity_match(
+ MatchInput("Acme Widgets Inc", city="Rivertown", region="SAMPLE",
+ postal_code="00002", address="300 Nowhere Blvd", is_business=True),
+ MatchInput("Acme Widgets Incorporated", city="Rivertown", region="SAMPLE",
+ postal_code="00002", address="300 Nowhere Blvd", is_business=True),
+ )
+ assert business.disposition in ("manual_review", "likely_duplicate_candidate"), business
+ _passed(f"business variants -> {business.disposition} (p={business.probability})")
+
+ distinct = entity_match(
+ MatchInput("Catherine O'Neil", city="Springfield", region="SAMPLE", postal_code="00001"),
+ MatchInput("Jonathan Doe", city="Springfield", region="SAMPLE", postal_code="00001"),
+ )
+ assert distinct.disposition == "distinct", distinct
+ _passed(f"different people -> {distinct.disposition} (p={distinct.probability})")
+
+ # person vs business must never link
+ cross = entity_match(
+ MatchInput("Acme Widgets Inc", is_business=True),
+ MatchInput("Acme Widgets Inc", is_business=False),
+ )
+ assert cross.disposition == "distinct", cross
+ _passed("person/business type conflict -> distinct (never linked)")
+
+ # Robust contract (survives weight re-tuning): a same-owner variant must ALWAYS
+ # out-score two clearly-different people at the same address.
+ assert person.probability > distinct.probability, (person, distinct)
+ assert business.probability > distinct.probability, (business, distinct)
+ _passed(f"ordering: variant p={person.probability} > distinct p={distinct.probability}")
+
+ print("\nALL SMOKE-TEST ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
(oldest)
·
back to Unclaimed Property Platform
·
auto-save: 2026-07-31T14:58:42 (8 files) — .gitignore servic f576a32 →