Stabilize preserved child record identities

docs/developer/hosted-ccld-backfill.md

@@ -93,6 +93,10 @@ returns exit code 1.
 - Repeating apply with unchanged inputs is idempotent. Checkpoint/resume skips
   completed Facility IDs, retries failed IDs without silently dropping them,
   and excludes newly appearing all-existing rows until a deliberate restart.
+- Preserved allegation, event, and repeated extraction-audit identities are
+  reconciled by exact generated identity and deterministic source occurrence
+  before any fallback. A newly extracted sibling therefore cannot reorder mixed
+  historical and current ID formats on the next replay.
 - Only canonical `facility_type`, `county`, and `status` allocations can be
   enriched. The command does not alter reviewer-created notes, status,
   decisions, or continuity state and does not activate ArcGIS, retrieve a live
@@ -109,6 +113,7 @@ precedence and conflicts, missing-value preservation, preserved-artifact and
 canonical-observation dry-run/apply/repeat behavior, explicit apply bounds,
 durable checkpoint/resume behavior, interrupted-run recovery, isolated failures,
 unchanged raw traceability, stable identities/import scope, preserved reviewer-
-created state and audit history, equivalent preserved-operation provenance,
-reviewer-detail visibility, and ordinary retrieval parity. Tests use local
-source-shaped fixtures and doubles only.
+created state and audit history, stable child identity when extraction adds a
+sibling, equivalent preserved-operation provenance, reviewer-detail visibility,
+and ordinary retrieval parity. Tests use local source-shaped fixtures and
+doubles only.

src/ccld_complaints/hosted_app/ccld_backfill.py

@@ -2,6 +2,7 @@
 
 import json
 import os
+import re
 import tempfile
 from collections.abc import Mapping, Sequence
 from dataclasses import dataclass
@@ -32,6 +33,7 @@
     "canonical-complaint-observations",
 ]
 _REPO_ROOT = Path(__file__).resolve().parents[3]
+_IDENTITY_ORDINAL_PATTERN = re.compile(r"(?:^|[-:])(\d+)(?=$|[-:])")
 
 
 @dataclass(frozen=True)
@@ -354,13 +356,11 @@ def _preserve_stable_identities(
         ("event", "events", "event_id"),
     ):
         items = cast(list[dict[str, Any]], normalized.get(list_name, []))
-        prior = sorted(
+        prior = tuple(
             (row for row in existing if row["entity_type"] == entity_type),
-            key=lambda row: str(row["stable_source_id"]),
         )
-        for index, item in enumerate(items):
-            if index < len(prior):
-                item[id_field] = prior[index]["stable_source_id"]
+        _preserve_sequence_stable_identities(items, prior, id_field)
+        for item in items:
             item["complaint_id"] = complaint_id
     prior_audits: dict[str, list[Mapping[str, Any]]] = {}
     for row in sorted(
@@ -371,22 +371,75 @@ def _preserve_stable_identities(
             cast(Mapping[str, Any], row["original_values"]).get("field_name")
         )
         prior_audits.setdefault(field_name, []).append(row)
-    audit_indexes: dict[str, int] = {}
-    for audit in cast(list[dict[str, Any]], normalized.get("extraction_audit", [])):
+    audits = cast(list[dict[str, Any]], normalized.get("extraction_audit", []))
+    audits_by_field: dict[str, list[dict[str, Any]]] = {}
+    for audit in audits:
         audit["document_id"] = document_id
         field_name = str(audit.get("field_name") or "")
-        field_index = audit_indexes.get(field_name, 0)
-        prior_for_field = prior_audits.get(field_name, [])
-        prior_audit = (
-            prior_for_field[field_index]
-            if field_index < len(prior_for_field)
-            else None
+        audits_by_field.setdefault(field_name, []).append(audit)
+    for field_name, field_audits in audits_by_field.items():
+        assigned_indexes = _preserve_sequence_stable_identities(
+            field_audits,
+            prior_audits.get(field_name, ()),
+            "audit_id",
         )
-        audit_indexes[field_name] = field_index + 1
-        if prior_audit is not None:
-            audit["audit_id"] = prior_audit["stable_source_id"]
-        elif str(audit.get("audit_id", "")).startswith(old_document_id):
-            audit["audit_id"] = document_id + str(audit["audit_id"])[len(old_document_id) :]
+        for index, audit in enumerate(field_audits):
+            if index in assigned_indexes:
+                continue
+            if str(audit.get("audit_id", "")).startswith(old_document_id):
+                audit["audit_id"] = (
+                    document_id
+                    + str(audit["audit_id"])[len(old_document_id) :]
+                )
+
+
+def _preserve_sequence_stable_identities(
+    items: Sequence[dict[str, Any]],
+    prior_rows: Sequence[Mapping[str, Any]],
+    id_field: str,
+) -> set[int]:
+    """Reuse stable child IDs without letting mixed ID formats reorder siblings."""
+
+    available = {
+        str(row["stable_source_id"]): row
+        for row in prior_rows
+    }
+    assignments: dict[int, str] = {}
+
+    for index, item in enumerate(items):
+        generated_id = str(item.get(id_field) or "")
+        if generated_id in available:
+            assignments[index] = generated_id
+            available.pop(generated_id)
+
+    for index, _item in enumerate(items, start=1):
+        item_index = index - 1
+        if item_index in assignments:
+            continue
+        ordinal_matches = [
+            stable_id
+            for stable_id in sorted(available)
+            if _stable_identity_ordinal(stable_id) == index
+        ]
+        if len(ordinal_matches) == 1:
+            assignments[item_index] = ordinal_matches[0]
+            available.pop(ordinal_matches[0])
+
+    for item_index, stable_id in zip(
+        (index for index in range(len(items)) if index not in assignments),
+        sorted(available),
+        strict=False,
+    ):
+        assignments[item_index] = stable_id
+
+    for index, stable_id in assignments.items():
+        items[index][id_field] = stable_id
+    return set(assignments)
+
+
+def _stable_identity_ordinal(stable_id: str) -> int | None:
+    matches = _IDENTITY_ORDINAL_PATTERN.findall(stable_id)
+    return int(matches[-1]) if matches else None
 
 
 def _related_records(

tests/unit/test_hosted_ccld_backfill.py

@@ -46,6 +46,7 @@
     hosted_seeded_import_metadata,
     hosted_source_derived_records,
     import_seeded_corpus_artifact,
+    load_seeded_corpus_artifact,
 )
 from ccld_complaints.utils.hash import sha256_bytes
 
@@ -57,6 +58,9 @@
     "https://www.ccld.dss.ca.gov/transparencyapi/api/FacilityReports"
     "?facNum=425802141&inx=1"
 )
+HISTORICAL_SEEDED_CORPUS_FIXTURE = Path(
+    "tests/fixtures/hosted_seeded_corpus/validated_seeded_corpus.json"
+)
 
 
 def test_apply_requires_explicit_bound_and_durable_checkpoint(tmp_path: Path) -> None:
@@ -678,6 +682,115 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
         assert _reviewer_snapshot(connection) == initial_reviewer_state
 
 
+def test_preserved_artifact_repeat_is_unchanged_when_extraction_adds_child(
+    tmp_path: Path,
+) -> None:
+    historical = load_seeded_corpus_artifact(HISTORICAL_SEEDED_CORPUS_FIXTURE)
+    record = copy.deepcopy(dict(historical.records[0]))
+    allegations = cast(list[dict[str, Any]], record["allegations"])
+    assert len(allegations) == 2
+    record["allegations"] = allegations[:1]
+    pre_correction = replace(historical, records=(record,))
+    complaint = cast(dict[str, Any], record["complaint"])
+    complaint_id = str(complaint["complaint_id"])
+    complaint_key = f"complaint:{complaint_id}"
+    source_document = cast(dict[str, Any], record["source_document"])
+    source_document_id = str(source_document["document_id"])
+
+    engine = create_engine("sqlite+pysqlite:///:memory:")
+    hosted_seeded_import_metadata.create_all(engine)
+    hosted_facility_reference_metadata.create_all(engine)
+    checkpoint = tmp_path / "new-child-checkpoint.json"
+    apply_request = CcldHostedBackfillRequest(
+        facility_numbers=("157806098",),
+        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,
+    )
+
+    with engine.connect() as connection:
+        import_seeded_corpus_artifact(connection, pre_correction)
+        _insert_reviewer_state_and_audit(
+            connection,
+            complaint_key,
+            complaint_stable_id=complaint_id,
+            source_document_id=source_document_id,
+        )
+        connection.commit()
+        initial_identities = _stable_identities(connection)
+        initial_import_scope = dict(_import_scope_snapshot(connection))
+        initial_reviewer_state = _reviewer_snapshot(connection)
+
+        first_dry_run = run_ccld_hosted_backfill(connection, dry_run_request)
+        first_apply = run_ccld_hosted_backfill(connection, apply_request)
+        connection.commit()
+        after_first_apply = _source_rows_snapshot(connection)
+        after_first_identities = _stable_identities(connection)
+        after_first_hashes = _source_hash_snapshot(connection)
+        after_first_import_scope = _import_scope_snapshot(connection)
+        after_first_reviewer_state = _reviewer_snapshot(connection)
+        after_first_conflicts = _entity_traceability(
+            connection,
+            "complaint",
+            complaint_id,
+        ).get("refresh_conflicts", [])
+
+        repeat_dry_run = run_ccld_hosted_backfill(connection, dry_run_request)
+        assert _source_rows_snapshot(connection) == after_first_apply
+
+        equivalent_operation_dry_run = run_ccld_hosted_backfill(
+            connection,
+            replace(
+                dry_run_request,
+                operation="canonical-complaint-observations",
+            ),
+        )
+        assert _source_rows_snapshot(connection) == after_first_apply
+
+        second_apply = run_ccld_hosted_backfill(
+            connection,
+            replace(apply_request, restart=True),
+        )
+        connection.commit()
+
+        final_conflicts = _entity_traceability(
+            connection,
+            "complaint",
+            complaint_id,
+        ).get("refresh_conflicts", [])
+
+        assert first_dry_run.updated == 1
+        assert first_dry_run.unchanged == 0
+        assert first_apply.updated == 1
+        assert first_apply.unchanged == 0
+        assert repeat_dry_run.updated == 0
+        assert repeat_dry_run.unchanged == 1
+        assert equivalent_operation_dry_run.updated == 0
+        assert equivalent_operation_dry_run.unchanged == 1
+        assert second_apply.updated == 0
+        assert second_apply.unchanged == 1
+        assert _source_rows_snapshot(connection) == after_first_apply
+        assert initial_identities < after_first_identities
+        assert _stable_identities(connection) == after_first_identities
+        assert _source_hash_snapshot(connection) == after_first_hashes
+        assert _import_scope_snapshot(connection) == after_first_import_scope
+        assert all(
+            dict(after_first_import_scope)[key] == import_batch_id
+            for key, import_batch_id in initial_import_scope.items()
+        )
+        assert after_first_reviewer_state == initial_reviewer_state
+        assert _reviewer_snapshot(connection) == initial_reviewer_state
+        assert final_conflicts == after_first_conflicts
+
+
 def test_repeated_supported_retrieval_refreshes_existing_rows_without_state_loss(
     tmp_path: Path,
 ) -> None:
@@ -889,7 +1002,13 @@ def _insert_reference(
     )
 
 
-def _insert_reviewer_state_and_audit(connection: Any, complaint_key: str) -> None:
+def _insert_reviewer_state_and_audit(
+    connection: Any,
+    complaint_key: str,
+    *,
+    complaint_stable_id: str = "ccld-complaint-31-CR-20240425094018",
+    source_document_id: str = "ccld-425802141-inx-1",
+) -> None:
     connection.execute(
         hosted_reviewer_created_state.insert().values(
             reviewer_state_id="reviewer-state:governed-refresh",
@@ -925,8 +1044,8 @@ def _insert_reviewer_state_and_audit(connection: Any, complaint_key: str) -> Non
             target_reviewer_state_id="reviewer-state:governed-refresh",
             source_record_key=complaint_key,
             source_entity_type="complaint",
-            source_stable_source_id="ccld-complaint-31-CR-20240425094018",
-            source_document_id="ccld-425802141-inx-1",
+            source_stable_source_id=complaint_stable_id,
+            source_document_id=source_document_id,
             context_metadata={"payload_kind": "reviewer_status_scaffold"},
         )
     )