Allow bounded preserved artifact backfill apply

docs/developer/hosted-ccld-backfill.md

@@ -47,13 +47,19 @@ The PowerShell wrapper defaults to dry-run:
 
 Selectors are mutually exclusive: `-FacilityNumber`,
 `-FacilityNumberFile <path>`, or `-AllExisting`. Operations are `all`,
-`facility-reference`, and `preserved-artifacts`. Use `-BatchSize 1..1000` for
-transaction grouping. Apply additionally requires both an explicit
-`-MaxFacilities 1..1000` per-run bound and `-CheckpointFile <path>`;
-`-Restart` explicitly starts a new checkpoint selection. Writes require
-explicit `-Apply`, and apply accepts only `-Operation facility-reference`.
-The `all` and `preserved-artifacts` operations remain dry-run diagnostics and
-cannot write complaint-derived records.
+`facility-reference`, `preserved-artifacts`, and
+`canonical-complaint-observations`. Use `-BatchSize 1..1000` for transaction
+grouping. Apply additionally requires both an explicit `-MaxFacilities 1..1000`
+per-run bound and `-CheckpointFile <path>`; `-Restart` explicitly starts a new
+checkpoint selection. Writes require explicit `-Apply`, and apply accepts only
+`-Operation facility-reference`, `-Operation preserved-artifacts`, or
+`-Operation canonical-complaint-observations`. The `all` operation remains a
+dry-run diagnostic and cannot write.
+
+`preserved-artifacts` and `canonical-complaint-observations` replay the same
+preserved-document preparation path and therefore use the same preserved-
+artifact provenance identity. Switching between those equivalent operations
+does not create a provenance-only source-derived update.
 
 The Python entry point exposes the equivalent `--facility-number`,
 `--facility-number-file`, `--all-existing`, `--operation`, `--batch-size`,
@@ -73,10 +79,11 @@ returns exit code 1.
   that requests facility-reference enrichment.
 - Preserved-artifact processing requires the stored raw path to be available to
   the runtime and its bytes to match the stored SHA-256.
-- Dry-run uses rollback and is the default. Apply validates reference identity
-  before writes, processes at most the explicit per-run bound, preserves
-  stable source-derived identities and the original import batch, and continues
-  after an isolated facility failure.
+- Dry-run uses rollback and is the default. When facility-reference enrichment
+  is requested, apply validates reference identity before writes. Every apply
+  processes at most the explicit per-run bound, preserves stable source-derived
+  identities and the original import batch, and continues after an isolated
+  facility failure.
 - Version 2 checkpoints freeze the selected public Facility IDs and the exact
   operation/selector, write through an atomic durable replacement, retain safe
   per-Facility-ID failure-attempt counts, and reject mismatched resume requests.
@@ -98,9 +105,10 @@ container configuration, environment file, or mounts.
 ## Validation coverage
 
 Fixture and unit coverage proves exact field/date extraction, approved-source
-precedence and conflicts, missing-value preservation, dry-run/apply/repeat,
-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, reviewer-detail visibility, and ordinary retrieval parity.
-Tests use local source-shaped fixtures and doubles only.
+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.

scripts/backfill-hosted-ccld-data.ps1

@@ -42,8 +42,8 @@ if ($Apply -and $MaxFacilities -eq 0) {
 if ($Apply -and [string]::IsNullOrWhiteSpace($CheckpointFile)) {
     throw "-Apply requires -CheckpointFile for durable recovery."
 }
-if ($Apply -and $Operation -notin @("facility-reference", "canonical-complaint-observations")) {
-    throw "-Apply supports only -Operation facility-reference or canonical-complaint-observations."
+if ($Apply -and $Operation -notin @("facility-reference", "preserved-artifacts", "canonical-complaint-observations")) {
+    throw "-Apply supports only -Operation facility-reference, preserved-artifacts, or canonical-complaint-observations."
 }
 
 $arguments = @("scripts/backfill_hosted_ccld_data.py", "--operation", $Operation, "--batch-size", "$BatchSize")

src/ccld_complaints/hosted_app/ccld_backfill.py

@@ -504,11 +504,13 @@ def _backfill_artifact(
 ) -> SeededCorpusArtifact:
     fingerprint_source = json.dumps(records, sort_keys=True, separators=(",", ":"), default=str)
     fingerprint = sha256_bytes(fingerprint_source.encode("utf-8"))
+    identity_operation = _source_artifact_identity_operation(operation)
     return SeededCorpusArtifact(
         import_batch_id=f"ccld-backfill-{fingerprint[:32]}",
         imported_at=now.replace(microsecond=0).isoformat(),
         source_artifact_identity=(
-            f"preserved-ccld-backfill:{operation}:{facility_number}:{fingerprint[:16]}"
+            f"preserved-ccld-backfill:{identity_operation}:{facility_number}:"
+            f"{fingerprint[:16]}"
         ),
         source_pipeline_version="governed-ccld-hosted-refresh-1.0.0",
         validation_status="validated",
@@ -575,11 +577,13 @@ def _validate_request(connection: Connection, request: CcldHostedBackfillRequest
         raise ValueError("apply requires an explicit max_facilities bound.")
     if request.apply_changes and request.operation not in {
         "facility-reference",
+        "preserved-artifacts",
         "canonical-complaint-observations",
     }:
         raise ValueError(
-            "apply supports only the approved facility-reference or historical "
-            "complaint-observation canonical allocation."
+            "apply supports only the approved facility-reference canonical allocation, "
+            "preserved-artifact normalization, or historical complaint-observation "
+            "canonical allocation."
         )
 
 
@@ -690,6 +694,16 @@ def _uses_preserved_artifacts(operation: BackfillOperation) -> bool:
     }
 
 
+def _source_artifact_identity_operation(
+    operation: BackfillOperation,
+) -> BackfillOperation:
+    """Use one provenance identity for equivalent preserved-document replays."""
+
+    if operation == "canonical-complaint-observations":
+        return "preserved-artifacts"
+    return operation
+
+
 def _required_text(value: object, field_name: str) -> str:
     normalized = _optional_text(value)
     if normalized is None:

tests/unit/test_hosted_ccld_backfill.py

@@ -68,6 +68,7 @@ def test_apply_requires_explicit_bound_and_durable_checkpoint(tmp_path: Path) ->
                 connection,
                 CcldHostedBackfillRequest(
                     facility_numbers=("425802141",),
+                    operation="preserved-artifacts",
                     apply_changes=True,
                     max_facilities=1,
                 ),
@@ -77,6 +78,7 @@ def test_apply_requires_explicit_bound_and_durable_checkpoint(tmp_path: Path) ->
                 connection,
                 CcldHostedBackfillRequest(
                     facility_numbers=("425802141",),
+                    operation="preserved-artifacts",
                     apply_changes=True,
                     checkpoint_file=tmp_path / "checkpoint.json",
                 ),
@@ -86,7 +88,6 @@ def test_apply_requires_explicit_bound_and_durable_checkpoint(tmp_path: Path) ->
                 connection,
                 CcldHostedBackfillRequest(
                     facility_numbers=("425802141",),
-                    operation="preserved-artifacts",
                     apply_changes=True,
                     checkpoint_file=tmp_path / "checkpoint.json",
                     max_facilities=1,
@@ -555,7 +556,7 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
     checkpoint = tmp_path / "duplicate-complaint-projections.json"
     apply_request = CcldHostedBackfillRequest(
         facility_numbers=("425802141",),
-        operation="canonical-complaint-observations",
+        operation="preserved-artifacts",
         batch_size=1,
         apply_changes=True,
         checkpoint_file=checkpoint,
@@ -610,6 +611,9 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
             "complaint",
             str(first_complaint["complaint_id"]),
         )
+        applied_artifact_identity = applied_traceability[
+            "source_artifact_identity"
+        ]
         applied_conflicts = tuple(applied_traceability["refresh_conflicts"])
 
         repeat_dry_run = run_ccld_hosted_backfill(
@@ -619,6 +623,16 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
         )
         assert _source_rows_snapshot(connection) == after_apply_source_rows
 
+        equivalent_operation_dry_run = run_ccld_hosted_backfill(
+            connection,
+            replace(
+                dry_run_request,
+                operation="canonical-complaint-observations",
+            ),
+            now=datetime(2026, 7, 14, tzinfo=UTC),
+        )
+        assert _source_rows_snapshot(connection) == after_apply_source_rows
+
         second_apply = run_ccld_hosted_backfill(
             connection,
             replace(apply_request, restart=True),
@@ -637,14 +651,25 @@ def test_preserved_artifact_repeat_is_unchanged_with_duplicate_complaint_project
         assert first_apply.unchanged == 0
         assert applied_complaint["first_investigation_activity_date"] == "2025-11-07"
         assert applied_complaint["finding"] == "Substantiated"
+        assert isinstance(applied_artifact_identity, str)
+        assert applied_artifact_identity.startswith(
+            "preserved-ccld-backfill:preserved-artifacts:425802141:"
+        )
         assert applied_conflicts
         assert repeat_dry_run.updated == 0
         assert repeat_dry_run.unchanged == 1
         assert repeat_dry_run.conflicted == 0
+        assert equivalent_operation_dry_run.updated == 0
+        assert equivalent_operation_dry_run.unchanged == 1
+        assert equivalent_operation_dry_run.conflicted == 0
         assert second_apply.updated == 0
         assert second_apply.unchanged == 1
         assert second_apply.conflicted == 0
         assert _source_rows_snapshot(connection) == after_apply_source_rows
+        assert (
+            final_traceability["source_artifact_identity"]
+            == applied_artifact_identity
+        )
         assert tuple(final_traceability["refresh_conflicts"]) == applied_conflicts
         assert _source_counts(connection) == initial_counts
         assert _stable_identities(connection) == initial_identities
@@ -972,11 +997,10 @@ def _source_hash_snapshot(connection: Any) -> tuple[tuple[str, str], ...]:
     return tuple(
         connection.execute(
             select(
-                hosted_source_derived_records.c.stable_source_id,
+                hosted_source_derived_records.c.source_record_key,
                 hosted_source_derived_records.c.raw_sha256,
             )
-            .where(hosted_source_derived_records.c.entity_type == "source_document")
-            .order_by(hosted_source_derived_records.c.stable_source_id)
+            .order_by(hosted_source_derived_records.c.source_record_key)
         ).tuples()
     )
 

tests/unit/test_hosted_ccld_backfill_cli.py

@@ -115,7 +115,7 @@ def test_cli_apply_commits_and_failure_exit_is_nonzero(
     assert "failed=1" in capsys.readouterr().out
 
 
-def test_cli_exposes_bounded_historical_complaint_observation_operation(
+def test_cli_exposes_bounded_preserved_artifact_apply_operations(
     monkeypatch: Any,
     capsys: Any,
 ) -> None:
@@ -146,25 +146,35 @@ def fake_run(_connection: Any, request: Any) -> CcldHostedBackfillResult:
 
     monkeypatch.setattr(cli, "run_ccld_hosted_backfill", fake_run)
 
-    assert cli.main(
-        [
-            "--facility-number",
-            "900000001",
-            "--operation",
-            "canonical-complaint-observations",
-            "--apply",
-            "--checkpoint-file",
-            "checkpoint.json",
-            "--max-facilities",
-            "1",
-        ]
-    ) == 0
-    request = captured_request[0]
-    assert request.operation == "canonical-complaint-observations"
-    assert request.apply_changes is True
-    assert request.max_facilities == 1
-    assert request.checkpoint_file == Path("checkpoint.json")
-    assert connection.commits == 1
+    for operation in (
+        "preserved-artifacts",
+        "canonical-complaint-observations",
+    ):
+        assert cli.main(
+            [
+                "--facility-number",
+                "900000001",
+                "--operation",
+                operation,
+                "--apply",
+                "--checkpoint-file",
+                f"{operation}.json",
+                "--max-facilities",
+                "1",
+            ]
+        ) == 0
+
+    assert [request.operation for request in captured_request] == [
+        "preserved-artifacts",
+        "canonical-complaint-observations",
+    ]
+    assert all(request.apply_changes is True for request in captured_request)
+    assert all(request.max_facilities == 1 for request in captured_request)
+    assert [request.checkpoint_file for request in captured_request] == [
+        Path("preserved-artifacts.json"),
+        Path("canonical-complaint-observations.json"),
+    ]
+    assert connection.commits == 2
     assert connection.rollbacks == 0
     assert "intended_updates=1" in capsys.readouterr().out
 
@@ -193,6 +203,11 @@ def test_powershell_wrapper_exposes_bounded_restartable_interface() -> None:
         assert token in wrapper
     assert "Omit both for dry-run" in wrapper
     assert "canonical-complaint-observations" in wrapper
+    assert (
+        '@("facility-reference", "preserved-artifacts", '
+        '"canonical-complaint-observations")'
+        in wrapper
+    )
 
 
 def _load_cli_module() -> ModuleType: