← back to Unclaimed Property Platform
Cycle 8: concrete SqliteClaimRepository — claims subsystem runnable on real persistence
c3d81a06233d83b3e131a8ee23c3743619ba149b · 2026-08-01 20:56:28 -0700 · Steve Abrams
- services/claims/sqlite_claim_repo.py implements ClaimRepository against claim_case/
claim_event/outbox_event, sharing the SqliteRepository connection (one system-of-record).
- Closes the last 'runs only on a test fake' gap: the claim lifecycle now persists end-to-end
and the C3 idempotency fix is proven against the REAL UNIQUE(claim_id, idempotency_key)
constraint (duplicate -> sqlite3.IntegrityError), not an in-memory imitation.
- tests/test_cycle9_claims_persistence.py: full DRAFT->SUBMITTED_TO_STATE lifecycle, outbox,
DB-level idempotency, state-only guard, illegal-transition rejection.
Tests: 9/9 suites green. All local/synthetic/$0.
TK-10097
Files touched
A services/claims/sqlite_claim_repo.pyA tests/test_cycle9_claims_persistence.py
Diff
commit c3d81a06233d83b3e131a8ee23c3743619ba149b
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Aug 1 20:56:28 2026 -0700
Cycle 8: concrete SqliteClaimRepository — claims subsystem runnable on real persistence
- services/claims/sqlite_claim_repo.py implements ClaimRepository against claim_case/
claim_event/outbox_event, sharing the SqliteRepository connection (one system-of-record).
- Closes the last 'runs only on a test fake' gap: the claim lifecycle now persists end-to-end
and the C3 idempotency fix is proven against the REAL UNIQUE(claim_id, idempotency_key)
constraint (duplicate -> sqlite3.IntegrityError), not an in-memory imitation.
- tests/test_cycle9_claims_persistence.py: full DRAFT->SUBMITTED_TO_STATE lifecycle, outbox,
DB-level idempotency, state-only guard, illegal-transition rejection.
Tests: 9/9 suites green. All local/synthetic/$0.
TK-10097
---
services/claims/sqlite_claim_repo.py | 108 ++++++++++++++++++++++++++++++++
tests/test_cycle9_claims_persistence.py | 108 ++++++++++++++++++++++++++++++++
2 files changed, 216 insertions(+)
diff --git a/services/claims/sqlite_claim_repo.py b/services/claims/sqlite_claim_repo.py
new file mode 100644
index 0000000..768dd61
--- /dev/null
+++ b/services/claims/sqlite_claim_repo.py
@@ -0,0 +1,108 @@
+"""Concrete SQLite ClaimRepository — makes the claim subsystem runnable end-to-end.
+
+Implements the ClaimRepository protocol from claim_workflow against the schema's
+claim_case / claim_event / outbox_event tables, sharing the same connection as
+SqliteRepository (one system-of-record DB). Production is the PostgreSQL equivalent with
+real SELECT ... FOR UPDATE row locks; here the prototype is single-threaded so the
+optimistic version field + the UNIQUE(claim_id, idempotency_key) constraint carry integrity.
+
+Crucially the DB — not a mock — enforces idempotency: a duplicate (claim_id, idempotency_key)
+raises sqlite3.IntegrityError, so the C3 fix is proven against real constraints.
+"""
+from __future__ import annotations
+
+import json
+import sqlite3
+import uuid
+from datetime import datetime, timezone
+
+from services.claims.claim_workflow import Claim, ClaimStatus
+
+
+def _now() -> str:
+ return datetime.now(timezone.utc).isoformat()
+
+
+class SqliteClaimRepository:
+ def __init__(self, conn: sqlite3.Connection) -> None:
+ self.conn = conn
+ self.conn.row_factory = sqlite3.Row
+
+ def create_claim(self, jurisdiction: str, public_property_reference: str,
+ claimant_id: uuid.UUID) -> Claim:
+ claim = Claim(
+ claim_id=uuid.uuid4(), jurisdiction=jurisdiction,
+ public_property_reference=public_property_reference,
+ claimant_id=claimant_id, status=ClaimStatus.DRAFT, version=1,
+ )
+ self.conn.execute(
+ """INSERT INTO claim_case
+ (claim_id, jurisdiction_id, public_property_reference, claimant_id,
+ status, version, state_case_id)
+ VALUES (?,?,?,?,?,?,?)""",
+ (str(claim.claim_id), claim.jurisdiction, claim.public_property_reference,
+ str(claim.claimant_id), claim.status.value, claim.version, None),
+ )
+ self.conn.commit()
+ return claim
+
+ # --- ClaimRepository protocol -----------------------------------------
+ def get_for_update(self, claim_id: uuid.UUID) -> Claim:
+ row = self.conn.execute(
+ "SELECT * FROM claim_case WHERE claim_id=?", (str(claim_id),)
+ ).fetchone()
+ if row is None:
+ raise KeyError(f"claim {claim_id} not found")
+ return Claim(
+ claim_id=uuid.UUID(row["claim_id"]),
+ jurisdiction=row["jurisdiction_id"],
+ public_property_reference=row["public_property_reference"],
+ claimant_id=uuid.UUID(row["claimant_id"]),
+ status=ClaimStatus(row["status"]),
+ version=row["version"],
+ state_case_id=row["state_case_id"],
+ )
+
+ def save(self, claim: Claim) -> None:
+ self.conn.execute(
+ "UPDATE claim_case SET status=?, version=?, state_case_id=? WHERE claim_id=?",
+ (claim.status.value, claim.version, claim.state_case_id, str(claim.claim_id)),
+ )
+ self.conn.commit()
+
+ def append_event(self, claim_id: uuid.UUID, event_type: str, payload: dict,
+ idempotency_key: str) -> None:
+ # UNIQUE(claim_id, idempotency_key) is enforced by the DB — a collision raises
+ # sqlite3.IntegrityError (the real C3 guard, not a mock's imitation).
+ self.conn.execute(
+ """INSERT INTO claim_event
+ (claim_event_id, claim_id, event_type, payload, idempotency_key, created_at)
+ VALUES (?,?,?,?,?,?)""",
+ (str(uuid.uuid4()), str(claim_id), event_type, json.dumps(payload),
+ idempotency_key, _now()),
+ )
+ self.conn.commit()
+
+ def add_outbox_event(self, event_type: str, aggregate_id: uuid.UUID, payload: dict) -> None:
+ self.conn.execute(
+ """INSERT INTO outbox_event
+ (outbox_event_id, aggregate_id, event_type, payload, delivery_state)
+ VALUES (?,?,?,?, 'pending')""",
+ (str(uuid.uuid4()), str(aggregate_id), event_type, json.dumps(payload)),
+ )
+ self.conn.commit()
+
+ # --- reads for tests / workers ----------------------------------------
+ def events(self, claim_id: uuid.UUID) -> list[dict]:
+ rows = self.conn.execute(
+ "SELECT event_type, idempotency_key FROM claim_event WHERE claim_id=? ORDER BY created_at",
+ (str(claim_id),),
+ ).fetchall()
+ return [dict(r) for r in rows]
+
+ def pending_outbox(self, aggregate_id: uuid.UUID) -> list[dict]:
+ rows = self.conn.execute(
+ "SELECT event_type, delivery_state FROM outbox_event WHERE aggregate_id=?",
+ (str(aggregate_id),),
+ ).fetchall()
+ return [dict(r) for r in rows]
diff --git a/tests/test_cycle9_claims_persistence.py b/tests/test_cycle9_claims_persistence.py
new file mode 100644
index 0000000..58f3785
--- /dev/null
+++ b/tests/test_cycle9_claims_persistence.py
@@ -0,0 +1,108 @@
+"""Cycle 8 test — claim lifecycle against a REAL SQLite ClaimRepository.
+
+Run: python -m tests.test_cycle9_claims_persistence
+
+Proves the claims subsystem is runnable end-to-end on real persistence (not a mock):
+ 1. full lifecycle DRAFT -> ... -> SUBMITTED_TO_STATE persists across get_for_update reads;
+ 2. queue+complete write the outbox event and both status events WITHOUT colliding on the
+ real UNIQUE(claim_id, idempotency_key) constraint (C3 fix proven against the DB);
+ 3. the DB itself enforces idempotency (duplicate key -> IntegrityError);
+ 4. the state-only guard blocks claimant self-approval on a persisted claim;
+ 5. illegal transitions raise.
+"""
+from __future__ import annotations
+
+import sqlite3
+import sys
+import tempfile
+from pathlib import Path
+from uuid import uuid4
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+from services.claims.claim_workflow import (
+ ClaimStatus, complete_state_submission, queue_state_submission, transition_claim,
+)
+from services.claims.sqlite_claim_repo import SqliteClaimRepository
+from services.common.sqlite_repo import SqliteRepository
+
+
+class FakeStateAdapter:
+ def submit_claim(self, claim, idempotency_key) -> str:
+ return "STATE-CASE-1"
+
+
+def ok(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def main() -> int:
+ tmp = Path(tempfile.mkdtemp(prefix="upp-cycle9-"))
+ base = SqliteRepository(str(tmp / "p.db")) # creates full schema
+ repo = SqliteClaimRepository(base.conn) # shares the one system-of-record DB
+
+ print("1) create + walk the lifecycle (persisted)")
+ claim = repo.create_claim("SAMPLE", public_property_reference="ref-1", claimant_id=uuid4())
+ assert repo.get_for_update(claim.claim_id).status == ClaimStatus.DRAFT
+ steps = [
+ (ClaimStatus.IDENTITY_PENDING, "k1"),
+ (ClaimStatus.EVIDENCE_PENDING, "k2"),
+ (ClaimStatus.READY_FOR_SUBMISSION, "k3"),
+ ]
+ for target, key in steps:
+ transition_claim(repo, claim.claim_id, target, actor_id="worker", idempotency_key=key)
+ assert repo.get_for_update(claim.claim_id).status == ClaimStatus.READY_FOR_SUBMISSION
+ ok("DRAFT -> IDENTITY -> EVIDENCE -> READY_FOR_SUBMISSION persisted")
+
+ print("2) queue + complete submission (real UNIQUE constraint, no collision)")
+ queue_state_submission(repo, claim.claim_id, actor_id="worker", idempotency_key="sub")
+ assert repo.get_for_update(claim.claim_id).status == ClaimStatus.SUBMITTING
+ complete_state_submission(repo, FakeStateAdapter(), claim.claim_id, idempotency_key="sub")
+ final = repo.get_for_update(claim.claim_id)
+ assert final.status == ClaimStatus.SUBMITTED_TO_STATE, final.status
+ assert final.state_case_id == "STATE-CASE-1", final.state_case_id
+ outbox = repo.pending_outbox(claim.claim_id)
+ assert any(e["event_type"] == "state_claim_submission_requested" for e in outbox), outbox
+ ok(f"SUBMITTED_TO_STATE persisted; state_case_id set; outbox has request; "
+ f"{len(repo.events(claim.claim_id))} events, keys all unique")
+
+ print("3) DB itself enforces idempotency (duplicate key -> IntegrityError)")
+ repo.append_event(claim.claim_id, "probe", {"n": 1}, idempotency_key="dupe")
+ raised = False
+ try:
+ repo.append_event(claim.claim_id, "probe", {"n": 2}, idempotency_key="dupe")
+ except sqlite3.IntegrityError:
+ raised = True
+ assert raised, "duplicate (claim_id, idempotency_key) must raise IntegrityError"
+ ok("duplicate idempotency key rejected by the real UNIQUE constraint")
+
+ print("4) state-only guard on a persisted claim")
+ raised = False
+ try:
+ transition_claim(repo, claim.claim_id, ClaimStatus.APPROVED,
+ actor_id="claimant:self", idempotency_key="appr")
+ except PermissionError:
+ raised = True
+ assert raised, "claimant must not self-approve"
+ # a state actor can
+ transition_claim(repo, claim.claim_id, ClaimStatus.APPROVED,
+ actor_id="state:CA", idempotency_key="appr2")
+ assert repo.get_for_update(claim.claim_id).status == ClaimStatus.APPROVED
+ ok("claimant approval blocked; state:CA approval persisted")
+
+ print("5) illegal transition raises")
+ raised = False
+ try:
+ transition_claim(repo, claim.claim_id, ClaimStatus.DRAFT,
+ actor_id="worker", idempotency_key="bad")
+ except ValueError:
+ raised = True
+ assert raised, "APPROVED -> DRAFT is illegal"
+ ok("illegal transition (APPROVED -> DRAFT) rejected")
+
+ print("\nALL CYCLE-8 CLAIM-PERSISTENCE ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
← afbc9ad docs: ledger Cycle-7 record (suppression path); TK-10097
·
back to Unclaimed Property Platform
·
docs: ledger Cycle-8 record (claims persistence); TK-10097 e7a163f →