Harden integrated US evidence review gates

Author: eliperez-dev

docs/research/integrated-sprint-review-2026-09-18.md

@@ -0,0 +1,34 @@
+# Integrated sprint review — 2026-09-18
+
+This private engineering review covers the integrated APHIS completeness
+accounting, FSIS stale-handoff guard, FSIS recall evidence parser, Brazil
+source documentation, and the shared country/source platform. No data was
+published or promoted.
+
+## Fixes applied
+
+- APHIS completeness now remains `incomplete` when duplicate export-page rows
+  are present, even if the inflated input count reaches the operator-supplied
+  displayed-row count. Duplicate observations remain quarantined.
+- FSIS recall establishment inference now requires an explicit `EST` marker
+  in free text. Digits in firm names or reasons cannot create a source-local
+  facility join.
+
+## Validation
+
+The focused package-qualified suite passed: 15 tests covering FSIS refresh,
+FSIS recall parsing, APHIS Wave 2 accounting, and the US refresh/rehearsal
+contracts. A direct `unittest discover -s pipeline/sources/us` invocation also
+ran 62 tests successfully but reported two loader errors for relative-import
+tests (`test_refresh` and `test_real_rehearsal`); running those modules with
+package-qualified names passed them. This is an invocation portability issue,
+not a product-test failure.
+
+## Residual risks
+
+The APHIS proof remains a bounded operator-saved subset, not a national
+completeness or currentness claim. FSIS recalls remain private, review-required
+evidence and source-local candidate edges; a recall is not itself a finding of
+wrongdoing. Registry coverage, source terms, privacy review, factual review,
+project approval, and publication remain separate gates. No public release or
+promotion was performed.

pipeline/sources/us/accountability/aphis_wave2.py

@@ -127,7 +127,7 @@ def _coverage_accounting(
         "failed_export_count": len(failed),
         "failure_states": dict(sorted(Counter(str(item.get("state", "unclassified")) for item in failed).items())),
         "not_observed_rows": max(expected - observed, 0) if expected is not None else None,
-        "accounting_state": "complete" if expected is not None and observed >= expected and not failed else "incomplete",
+        "accounting_state": "complete" if expected is not None and observed >= expected and not failed and not manifest["duplicate_page_rows"] else "incomplete",
         "not_observed_semantics": "not observed by this acquisition; not closure, non-use, or evidence of absence",
     }
 

pipeline/sources/us/accountability/test_aphis_wave2.py

@@ -66,6 +66,23 @@ def test_operator_expected_count_can_be_overridden_without_claiming_completion(s
             self.assertEqual(report["completeness"]["inspections"]["not_observed_rows"], 0)
             self.assertEqual(report["completeness"]["inspections"]["accounting_state"], "complete")
 
+    def test_duplicate_pages_do_not_claim_completeness(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            input_root = root / "inputs"
+            input_root.mkdir()
+            rows = {
+                "ExportData-registrations.csv": "Account Name,Customer Number,Certificate Number,Registration Type,Certificate Status,Status Date\nA,2,00-R-0002,Class R - Research Facility,Active,2026-01-01\n",
+                "ExportData-annual_reports.csv": "Customer Number,Certificate Number,Year,Dogs,Cats\n2,00-R-0002,2025,,1\n",
+                "ExportData-inspections.csv": "Customer Number,Certificate Number,Inspection Date,Direct NCIs,Non-Critical NCIs,Critical NCIs,Teachable Moments,Site Name,Legal Name,License-Registration Type,City,State,Zip\n2,00-R-0002,2026-02-01,,,,,S,A,Class R - Research Facility,T,TX,75001\n",
+            }
+            for name, content in rows.items():
+                (input_root / name).write_text(content, encoding="utf-8")
+                (input_root / (Path(name).stem + "-page2.csv")).write_text(content, encoding="utf-8")
+            report = run_wave2(input_root=input_root, run_dir=root / "run", expected_rows={"inspections": 1})
+            self.assertGreater(report["completeness"]["inspections"]["duplicate_page_rows"], 0)
+            self.assertEqual(report["completeness"]["inspections"]["accounting_state"], "incomplete")
+
 
 if __name__ == "__main__":
     unittest.main()

pipeline/sources/us/fsis/recall.py

@@ -45,7 +45,9 @@ def _establishment_number(row: dict[str, Any]) -> str | None:
     # Source text such as "EST. 1234" is retained as an explicit source clue,
     # but never treated as a join when multiple identifiers occur.
     import re
-    matches = sorted(set(re.findall(r"\b(?:EST\.?\s*)?(\d{1,6})\b", text, re.I)))
+    # Digits in names/reasons are not identity evidence. Require an explicit
+    # establishment marker before creating a source-local join.
+    matches = sorted(set(re.findall(r"\bEST\.?\s*(\d{1,6})\b", text, re.I)))
     return matches[0] if len(matches) == 1 else None
 
 

pipeline/sources/us/fsis/test_recall.py

@@ -34,6 +34,15 @@ def test_run_is_deterministic_and_row_free_manifest(self):
             self.assertEqual(json.loads((Path(left) / "aggregate-manifest.json").read_text())["quarantined_rows"], 1)
             self.assertFalse("records" in json.dumps(a))
 
+    def test_digits_in_free_text_do_not_create_establishment_join(self):
+        parsed = parse_bytes(json.dumps({"results": [{
+            "recall_number": "FSIS-2026-003",
+            "firm": "Foods 123 LLC",
+            "reason": "Product code 456",
+        }]}).encode("utf-8"))
+        self.assertEqual(len(parsed["accepted"]), 0)
+        self.assertIn("unresolved_establishment_identifier", parsed["quarantined"][0]["reasons"])
+
 
 if __name__ == "__main__":
     unittest.main()