← back to Unclaimed Property Platform
Cycle 10: feed-completeness reconciliation (catches silent truncation)
edfce96a103575c3824e59c0249979e814e79b51 · 2026-08-01 21:07:53 -0700 · Steve Abrams
- FeedDefinition.expected_count (state control total; NAUPA trailer in prod); _reconcile()
compares parsed (accepted+rejected) vs expected -> ok|short|over|unknown + delta.
- 'short' is the silent-truncation signal (a transfer that cut off looks 'successful' otherwise).
Persisted on ingestion_batch (expected_count + reconciliation cols); batch_reconciliation() reads it.
- tests/test_cycle11_reconciliation.py: ok/short/over/unknown + persistence.
Tests: 11/11 suites green. All local/synthetic/$0.
TK-10097
Files touched
M db/schema.sqlM services/common/sqlite_repo.pyM services/ingestion/ingest.pyA tests/test_cycle11_reconciliation.py
Diff
commit edfce96a103575c3824e59c0249979e814e79b51
Author: Steve Abrams <steve@designerwallcoverings.com>
Date: Sat Aug 1 21:07:53 2026 -0700
Cycle 10: feed-completeness reconciliation (catches silent truncation)
- FeedDefinition.expected_count (state control total; NAUPA trailer in prod); _reconcile()
compares parsed (accepted+rejected) vs expected -> ok|short|over|unknown + delta.
- 'short' is the silent-truncation signal (a transfer that cut off looks 'successful' otherwise).
Persisted on ingestion_batch (expected_count + reconciliation cols); batch_reconciliation() reads it.
- tests/test_cycle11_reconciliation.py: ok/short/over/unknown + persistence.
Tests: 11/11 suites green. All local/synthetic/$0.
TK-10097
---
db/schema.sql | 2 +
services/common/sqlite_repo.py | 17 +++++++--
services/ingestion/ingest.py | 23 +++++++++--
tests/test_cycle11_reconciliation.py | 74 ++++++++++++++++++++++++++++++++++++
4 files changed, 110 insertions(+), 6 deletions(-)
diff --git a/db/schema.sql b/db/schema.sql
index b5efb41..a06c31e 100644
--- a/db/schema.sql
+++ b/db/schema.sql
@@ -44,6 +44,8 @@ CREATE TABLE ingestion_batch (
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
+ expected_count INTEGER, -- state-declared control total (NAUPA trailer in prod)
+ reconciliation TEXT, -- ok|short|over|unknown (accepted+rejected vs expected)
UNIQUE (jurisdiction_id, checksum) -- idempotency: same file never processed twice
);
diff --git a/services/common/sqlite_repo.py b/services/common/sqlite_repo.py
index 36adba3..4530c15 100644
--- a/services/common/sqlite_repo.py
+++ b/services/common/sqlite_repo.py
@@ -168,13 +168,24 @@ class SqliteRepository:
)
self.conn.commit()
- def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str) -> None:
+ def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str,
+ expected: int | None = None, reconciliation: str | None = None) -> None:
self.conn.execute(
- "UPDATE ingestion_batch SET accepted_count=?, rejected_count=?, status=? WHERE batch_id=?",
- (accepted, rejected, status, batch_id),
+ """UPDATE ingestion_batch
+ SET accepted_count=?, rejected_count=?, status=?, expected_count=?, reconciliation=?
+ WHERE batch_id=?""",
+ (accepted, rejected, status, expected, reconciliation, batch_id),
)
self.conn.commit()
+ def batch_reconciliation(self, batch_id: str) -> dict | None:
+ r = self.conn.execute(
+ """SELECT accepted_count, rejected_count, expected_count, reconciliation, status
+ FROM ingestion_batch WHERE batch_id=?""",
+ (batch_id,),
+ ).fetchone()
+ return dict(r) if r else None
+
# --- convenience reads for tests / review tools ------------------------
def count_properties(self, jurisdiction: str | None = None) -> int:
if jurisdiction:
diff --git a/services/ingestion/ingest.py b/services/ingestion/ingest.py
index 70103f1..39f7be3 100644
--- a/services/ingestion/ingest.py
+++ b/services/ingestion/ingest.py
@@ -33,7 +33,8 @@ class Repository(Protocol):
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: ...
+ def finish_batch(self, batch_id: str, accepted: int, rejected: int, status: str,
+ expected: int | None = None, reconciliation: str | None = None) -> None: ...
@dataclass
@@ -67,6 +68,19 @@ class FeedDefinition:
source_uri: str
format_name: str = "state_csv_v1"
parser_version: str = "2026.07.1"
+ expected_count: int | None = None # state-declared control total (NAUPA trailer in prod)
+
+
+def _reconcile(expected: int | None, parsed: int) -> tuple[str, int]:
+ """Compare records PARSED (accepted+rejected) against the state's control total.
+ Returns (status, delta). 'short' = silent-truncation signal (parsed < expected)."""
+ if expected is None:
+ return "unknown", 0
+ if parsed < expected:
+ return "short", expected - parsed
+ if parsed > expected:
+ return "over", parsed - expected
+ return "ok", 0
# --- No-scrape / synthetic-only red line, ENFORCED IN CODE (Security finding 2.1) ---------
@@ -188,10 +202,13 @@ def ingest_authorized_feed(feed: FeedDefinition, object_store: ObjectStore,
except Exception:
rejected += 1
status = "completed_with_errors" if rejected else "completed"
- repository.finish_batch(batch_id, accepted, rejected, status)
+ recon, delta = _reconcile(feed.expected_count, accepted + rejected)
+ repository.finish_batch(batch_id, accepted, rejected, status,
+ expected=feed.expected_count, reconciliation=recon)
except Exception:
repository.finish_batch(batch_id, accepted, rejected, "failed")
raise
return {"status": status, "batch_id": batch_id,
- "accepted": accepted, "rejected": rejected}
+ "accepted": accepted, "rejected": rejected,
+ "reconciliation": recon, "reconciliation_delta": delta}
diff --git a/tests/test_cycle11_reconciliation.py b/tests/test_cycle11_reconciliation.py
new file mode 100644
index 0000000..3002658
--- /dev/null
+++ b/tests/test_cycle11_reconciliation.py
@@ -0,0 +1,74 @@
+"""Cycle 10 test — feed-completeness reconciliation (catches silent truncation).
+
+Run: python -m tests.test_cycle11_reconciliation
+
+The sample feed has 8 rows (7 accepted + 1 rejected empty-owner) => 8 records PARSED.
+Reconciliation compares parsed vs the state-declared control total:
+ 1. expected == parsed -> 'ok'
+ 2. expected > parsed -> 'short' (SILENT-TRUNCATION signal + missing delta)
+ 3. expected < parsed -> 'over'
+ 4. no expected declared -> 'unknown' (not flagged)
+and the result is persisted on the batch (queryable).
+"""
+from __future__ import annotations
+
+import sys
+import tempfile
+from pathlib import Path
+
+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
+
+SAMPLE = Path(__file__).resolve().parents[1] / "data" / "sample" / "sample_state_feed.csv"
+
+
+def ok(msg: str) -> None:
+ print(f" ✓ {msg}")
+
+
+def _ingest(tmp: Path, sub: str, expected):
+ store = FileObjectStore(tmp / sub)
+ (tmp / sub).mkdir(parents=True, exist_ok=True)
+ store.write_bytes("incoming/s.csv", SAMPLE.read_bytes())
+ repo = SqliteRepository(str(tmp / sub / "p.db"))
+ feed = FeedDefinition("SAMPLE", "incoming/s.csv", expected_count=expected)
+ return repo, ingest_authorized_feed(feed, store, repo)
+
+
+def main() -> int:
+ tmp = Path(tempfile.mkdtemp(prefix="upp-cycle11-"))
+
+ print("0) baseline parse counts")
+ repo, r = _ingest(tmp, "ok", expected=8)
+ assert r["accepted"] == 7 and r["rejected"] == 1, r
+ ok("8 parsed (7 accepted + 1 rejected)")
+
+ print("1) expected == parsed -> ok")
+ assert r["reconciliation"] == "ok" and r["reconciliation_delta"] == 0, r
+ stored = repo.batch_reconciliation(r["batch_id"])
+ assert stored["reconciliation"] == "ok" and stored["expected_count"] == 8, stored
+ ok("reconciliation 'ok' (persisted on batch)")
+
+ print("2) expected > parsed -> short (silent-truncation signal)")
+ _, r2 = _ingest(tmp, "short", expected=10)
+ assert r2["reconciliation"] == "short" and r2["reconciliation_delta"] == 2, r2
+ ok(f"expected 10, parsed 8 -> 'short', missing {r2['reconciliation_delta']} (would alert)")
+
+ print("3) expected < parsed -> over")
+ _, r3 = _ingest(tmp, "over", expected=5)
+ assert r3["reconciliation"] == "over" and r3["reconciliation_delta"] == 3, r3
+ ok("expected 5, parsed 8 -> 'over' (delta 3)")
+
+ print("4) no control total -> unknown (not flagged)")
+ _, r4 = _ingest(tmp, "unknown", expected=None)
+ assert r4["reconciliation"] == "unknown", r4
+ ok("no expected_count -> 'unknown'")
+
+ print("\nALL CYCLE-10 RECONCILIATION ASSERTIONS PASSED ✅")
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
← 2a43af4 docs: ledger Cycle-9 record (outbox worker); TK-10097
·
back to Unclaimed Property Platform
·
docs: ledger Cycle-10 record (reconciliation); TK-10097 d4ba8b0 →