diff --git a/docs/developer/hosted-ccld-backfill.md b/docs/developer/hosted-ccld-backfill.md index 73293ef..0f08d7d 100644 --- a/docs/developer/hosted-ccld-backfill.md +++ b/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. diff --git a/src/ccld_complaints/hosted_app/ccld_backfill.py b/src/ccld_complaints/hosted_app/ccld_backfill.py index 04d0332..eb88c19 100644 --- a/src/ccld_complaints/hosted_app/ccld_backfill.py +++ b/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( diff --git a/tests/unit/test_hosted_ccld_backfill.py b/tests/unit/test_hosted_ccld_backfill.py index d9e4c06..c7f199b 100644 --- a/tests/unit/test_hosted_ccld_backfill.py +++ b/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"}, ) )