Stabilize preserved extraction audit identities

docs/developer/hosted-ccld-backfill.md

@@ -125,6 +125,14 @@ code 2; an isolated facility failure or safe runtime failure returns exit code
   unmatched allegations, events, and repeated extraction-audit rows. A newly
   extracted allegation receives a collision-free identity and therefore cannot
   shift the stable identities or values of unchanged siblings.
+- Preserved extraction-audit reconciliation prefers identities already scoped
+  to the current source document. When a field has no current-document history,
+  it may reuse a legacy non-document-scoped audit identity from the same
+  facility; this preserves historical multi-document projections that the
+  importer stores under one final stable key. Audit semantic values match before
+  exact identity, ordinal, and deterministic fallback. A genuinely new audit
+  retains a collision-free document-scoped identity, and changed or removed
+  governed audit evidence still prevents equivalence.
 - Before a write decision, preserved-artifact provenance is stabilized only when
   the complete final source-record key set and every importer-visible governed
   value other than artifact identity exactly match persisted state. Superseded
@@ -149,7 +157,8 @@ durable checkpoint/resume behavior, interrupted-run recovery, isolated failures,
 unchanged raw traceability, stable identities/import scope, preserved reviewer-
 created state and audit history, stable child identity when extraction adds a
 sibling, semantic allegation identity when a sibling is inserted, final-
-projection artifact identity with duplicated document projections, equivalent
+projection artifact identity with duplicated document projections, historical
+shared extraction-audit identity replay across multiple documents, equivalent
 preserved-operation provenance, reviewer-detail visibility, SELECT-only redacted
 difference diagnosis, and ordinary retrieval parity. Tests use local source-
 shaped fixtures and doubles only.

src/ccld_complaints/hosted_app/ccld_backfill.py

@@ -570,9 +570,10 @@ def _preserve_stable_identities(
         for item in items:
             item["complaint_id"] = complaint_id
     prior_audits: dict[str, list[Mapping[str, Any]]] = {}
-    for row in sorted(
-        (row for row in existing if row["entity_type"] == "extraction_audit"),
-        key=lambda row: str(row["stable_source_id"]),
+    for row in _prior_extraction_audits(
+        connection,
+        facility_id,
+        document_id,
     ):
         field_name = str(
             cast(Mapping[str, Any], row["original_values"]).get("field_name")
@@ -589,6 +590,16 @@ def _preserve_stable_identities(
             field_audits,
             prior_audits.get(field_name, ()),
             "audit_id",
+            semantic_fields=(
+                "field_name",
+                "extraction_method",
+                "extractor_version",
+                "extracted_value",
+                "confidence",
+                "source_text",
+                "source_section",
+                "warning",
+            ),
         )
         for index, audit in enumerate(field_audits):
             if index in assigned_indexes:
@@ -735,6 +746,51 @@ def _related_records(
     )
 
 
+def _prior_extraction_audits(
+    connection: Connection,
+    facility_id: str,
+    source_document_id: str,
+) -> tuple[Mapping[str, Any], ...]:
+    rows = tuple(
+        dict(row)
+        for row in connection.execute(
+            select(hosted_source_derived_records).where(
+                hosted_source_derived_records.c.entity_type == "extraction_audit",
+                hosted_source_derived_records.c.facility_id == facility_id,
+            )
+        ).mappings()
+    )
+    current_document_fields = {
+        str(cast(Mapping[str, Any], row["original_values"]).get("field_name") or "")
+        for row in rows
+        if row["source_document_id"] == source_document_id
+    }
+    return tuple(
+        sorted(
+            (
+                row
+                for row in rows
+                if row["source_document_id"] == source_document_id
+                or (
+                    not _audit_identity_is_document_scoped(row)
+                    and str(cast(Mapping[str, Any], row["original_values"]).get("field_name") or "")
+                    not in current_document_fields
+                )
+            ),
+            key=lambda row: str(row["stable_source_id"]),
+        )
+    )
+
+
+def _audit_identity_is_document_scoped(row: Mapping[str, Any]) -> bool:
+    values = cast(Mapping[str, Any], row["original_values"])
+    document_id = _optional_text(values.get("document_id"))
+    if document_id is None:
+        return False
+    stable_id = str(row["stable_source_id"])
+    return stable_id.startswith(f"{document_id}-") or stable_id.startswith(f"{document_id}:")
+
+
 def _source_documents_for_facility(
     connection: Connection,
     facility_id: str,

tests/unit/test_hosted_ccld_backfill.py

@@ -684,6 +684,193 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
         assert _reviewer_snapshot(connection) == initial_reviewer_state
 
 
+def test_preserved_artifact_reuses_shared_historical_audit_identities(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    first_source_url = SOURCE_URL.replace("425802141", "900000001")
+    first_record = _normalized_record(
+        raw_fixture=STRUCTURED_RAW_FIXTURE,
+        source_url=first_source_url,
+    )
+    second_record = _normalized_record(
+        raw_fixture=STRUCTURED_RAW_FIXTURE,
+        source_url=first_source_url.replace("inx=1", "inx=33"),
+    )
+    first_document = cast(dict[str, Any], first_record["source_document"])
+    second_document = cast(dict[str, Any], second_record["source_document"])
+    assert first_document["document_id"] != second_document["document_id"]
+
+    for record in (first_record, second_record):
+        occurrences: dict[str, int] = {}
+        for audit in cast(list[dict[str, Any]], record["extraction_audit"]):
+            field_name = str(audit["field_name"])
+            occurrences[field_name] = occurrences.get(field_name, 0) + 1
+            if record is first_record and field_name == "facility_number":
+                continue
+            audit["audit_id"] = f"historical-audit:{field_name}:{occurrences[field_name]}"
+    assert (
+        sum(
+            audit["field_name"] == "deficiency_text"
+            for audit in cast(list[dict[str, Any]], first_record["extraction_audit"])
+        )
+        == 2
+    )
+
+    historical = replace(
+        _artifact("shared-historical-audits", first_record),
+        records=(first_record, second_record),
+    )
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    hosted_seeded_import_metadata.create_all(engine)
+    hosted_facility_reference_metadata.create_all(engine)
+    checkpoint = tmp_path / "shared-historical-audits.json"
+    apply_request = CcldHostedBackfillRequest(
+        facility_numbers=("900000001",),
+        operation="preserved-artifacts",
+        batch_size=1,
+        apply_changes=True,
+        checkpoint_file=checkpoint,
+        max_facilities=1,
+    )
+    dry_run_request = replace(
+        apply_request,
+        apply_changes=False,
+        checkpoint_file=None,
+        max_facilities=None,
+    )
+    change_mode: str | None = None
+    original_reprocess = ccld_backfill._reprocess_preserved_document
+
+    def reprocess_preserved_document(
+        connection: Any,
+        facility_number: str,
+        facility_row: Mapping[str, Any],
+        source_document_row: Mapping[str, Any],
+    ) -> Mapping[str, Any] | None:
+        normalized = original_reprocess(
+            connection,
+            facility_number,
+            facility_row,
+            source_document_row,
+        )
+        target_document_id = (
+            first_document["document_id"]
+            if change_mode == "new"
+            else second_document["document_id"]
+        )
+        if normalized is None or source_document_row["stable_source_id"] != target_document_id:
+            return normalized
+        audits = cast(list[dict[str, Any]], normalized["extraction_audit"])
+        if change_mode == "changed":
+            next(audit for audit in audits if audit["field_name"] == "facility_number")[
+                "extracted_value"
+            ] = "governed-change"
+        elif change_mode == "new":
+            audits.append(
+                {
+                    "audit_id": f"{first_document['document_id']}-new-governed-audit",
+                    "document_id": first_document["document_id"],
+                    "field_name": "new_governed_audit",
+                    "extraction_method": "ccld_facility_report_html_labels",
+                    "extractor_version": "0.1.0",
+                    "extracted_value": "new governed evidence",
+                    "confidence": 1.0,
+                    "source_text": "new governed evidence",
+                    "source_section": "fixture section",
+                    "warning": None,
+                }
+            )
+        elif change_mode == "removed":
+            audits[:] = [audit for audit in audits if audit["field_name"] != "agency_name"]
+        return normalized
+
+    monkeypatch.setattr(
+        ccld_backfill,
+        "_reprocess_preserved_document",
+        reprocess_preserved_document,
+    )
+
+    with engine.connect() as connection:
+        import_seeded_corpus_artifact(connection, historical)
+        complaint = cast(dict[str, Any], second_record["complaint"])
+        complaint_id = str(complaint["complaint_id"])
+        _insert_reviewer_state_and_audit(
+            connection,
+            f"complaint:{complaint_id}",
+            complaint_stable_id=complaint_id,
+            source_document_id=str(second_document["document_id"]),
+        )
+        connection.commit()
+        initial_rows = _source_rows_snapshot(connection)
+        initial_audit_ids = {
+            stable_id
+            for entity_type, stable_id in _stable_identities(connection)
+            if entity_type == "extraction_audit"
+        }
+        initial_hashes = _source_hash_snapshot(connection)
+        initial_import_scope = _import_scope_snapshot(connection)
+        initial_reviewer_state = _reviewer_snapshot(connection)
+
+        differences = diagnose_ccld_preserved_artifact_differences(
+            connection,
+            ("900000001",),
+        )
+        first_dry_run = run_ccld_hosted_backfill(connection, dry_run_request)
+        first_apply = run_ccld_hosted_backfill(connection, apply_request)
+        connection.commit()
+        repeat = run_ccld_hosted_backfill(connection, dry_run_request)
+        equivalent_operation = run_ccld_hosted_backfill(
+            connection,
+            replace(dry_run_request, operation="canonical-complaint-observations"),
+        )
+        second_apply = run_ccld_hosted_backfill(
+            connection,
+            replace(apply_request, restart=True),
+        )
+        connection.commit()
+
+        assert differences == ()
+        assert first_dry_run.updated == 0
+        assert first_dry_run.unchanged == 1
+        assert first_apply.updated == 0
+        assert first_apply.unchanged == 1
+        assert repeat.updated == 0
+        assert repeat.unchanged == 1
+        assert equivalent_operation.updated == 0
+        assert equivalent_operation.unchanged == 1
+        assert second_apply.updated == 0
+        assert second_apply.unchanged == 1
+        assert {
+            stable_id
+            for entity_type, stable_id in _stable_identities(connection)
+            if entity_type == "extraction_audit"
+        } == initial_audit_ids
+
+        change_mode = "changed"
+        changed = run_ccld_hosted_backfill(connection, dry_run_request)
+        change_mode = "new"
+        new_differences = diagnose_ccld_preserved_artifact_differences(
+            connection,
+            ("900000001",),
+        )
+        new_audit_key = f"extraction_audit:{first_document['document_id']}-new-governed-audit"
+        change_mode = "removed"
+        removed = run_ccld_hosted_backfill(connection, dry_run_request)
+
+        assert changed.updated == 1
+        assert any(
+            difference.source_record_key == new_audit_key
+            and difference.differing_fields == ("persisted_record",)
+            for difference in new_differences
+        )
+        assert removed.updated == 1
+        assert _source_rows_snapshot(connection) == initial_rows
+        assert _source_hash_snapshot(connection) == initial_hashes
+        assert _import_scope_snapshot(connection) == initial_import_scope
+        assert _reviewer_snapshot(connection) == initial_reviewer_state
+
+
 def test_preserved_artifact_identity_uses_final_persisted_projections() -> None:
     first_record = _initial_missing_record()
     second_record = _initial_missing_record(