Add read-only preserved artifact diagnostics

docs/developer/hosted-ccld-backfill.md

@@ -66,11 +66,37 @@ The Python entry point exposes the equivalent `--facility-number`,
 `--max-facilities`, `--checkpoint-file`, `--restart`, `--dry-run`, and
 `--apply` arguments.
 
-Output is limited to candidate, excluded, examined, eligible, intended-update,
-updated, unchanged, skipped, conflicted, warning, and failed counts. It does not print raw paths, source
-URLs, report narrative, database values, or credentials. A configuration error
-returns exit code 2; an isolated facility failure or safe runtime failure
-returns exit code 1.
+For a bounded residual-idempotence investigation, `-DiagnoseDifferences`
+requires `-Operation preserved-artifacts` and an explicit Facility ID or
+Facility-ID file. It rejects `-AllExisting`, apply, checkpoint, restart, and
+maximum-facility options. This non-default mode performs only preserved-artifact
+reads and preparation, compares the exact candidate values used by the import
+write decision, and emits one deterministic JSON document. Each difference
+contains the public Facility ID, entity type, stable source-record key, and
+differing field paths. String and structured values are represented only by
+type, length or count, and SHA-256; complaint and allegation narratives are not
+printed. Reviewer-created state and audit tables are outside the diagnostic read
+path.
+
+The equivalent direct Python forms are:
+
+```powershell
+python scripts/backfill_hosted_ccld_data.py --facility-number <facility-id> --operation preserved-artifacts --diagnose-differences
+python scripts/backfill_hosted_ccld_data.py --facility-number-file <facility-id-file> --operation preserved-artifacts --diagnose-differences
+```
+
+Diagnostic mode is a third mutually exclusive mode: the wrapper passes
+`--diagnose-differences` without `--dry-run` or `--apply`. Ordinary wrapper
+invocation still passes `--dry-run`, and explicit `-Apply` still passes
+`--apply`.
+
+Backfill mode output is limited to candidate, excluded, examined, eligible,
+intended-update, updated, unchanged, skipped, conflicted, warning, and failed
+counts. Diagnostic mode instead emits one deterministic JSON document containing
+redacted differences. Neither mode prints raw paths, source URLs, report
+narrative, database values, or credentials. A configuration error returns exit
+code 2; an isolated facility failure or safe runtime failure returns exit code
+1.
 
 ## Prerequisites and safeguards
 
@@ -115,5 +141,5 @@ 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, equivalent preserved-operation provenance, reviewer-detail visibility,
-and ordinary retrieval parity. Tests use local source-shaped fixtures and
-doubles only.
+SELECT-only redacted difference diagnosis, and ordinary retrieval parity. Tests
+use local source-shaped fixtures and doubles only.

scripts/backfill-hosted-ccld-data.ps1

@@ -20,6 +20,7 @@ param(
     [switch]$Restart,
     [switch]$Apply,
     [switch]$DryRun,
+    [switch]$DiagnoseDifferences,
     [switch]$QnapContainer,
     [string]$ComposeFile = "docker-compose.qnap.yml",
     [string]$EnvFile = ".env"
@@ -33,6 +34,15 @@ if ($selectionCount -ne 1) {
 if ($Apply -and $DryRun) {
     throw "Use either -Apply or -DryRun, not both. Omit both for dry-run."
 }
+if ($DiagnoseDifferences -and ($Apply -or $AllExisting)) {
+    throw "-DiagnoseDifferences requires explicit Facility IDs and cannot apply."
+}
+if ($DiagnoseDifferences -and $Operation -ne "preserved-artifacts") {
+    throw "-DiagnoseDifferences supports only -Operation preserved-artifacts."
+}
+if ($DiagnoseDifferences -and ($MaxFacilities -gt 0 -or $CheckpointFile -or $Restart)) {
+    throw "-DiagnoseDifferences does not accept apply or checkpoint options."
+}
 if ($MaxFacilities -lt 0 -or $MaxFacilities -gt 1000) {
     throw "-MaxFacilities must be between 1 and 1000 when provided."
 }
@@ -65,7 +75,10 @@ if (-not [string]::IsNullOrWhiteSpace($CheckpointFile)) {
 if ($Restart) {
     $arguments += "--restart"
 }
-if ($Apply) {
+if ($DiagnoseDifferences) {
+    $arguments += "--diagnose-differences"
+}
+elseif ($Apply) {
     $arguments += "--apply"
 }
 else {

scripts/backfill_hosted_ccld_data.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import argparse
+import json
 import sys
+from dataclasses import asdict
 from pathlib import Path
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -11,6 +13,7 @@
 
 from ccld_complaints.hosted_app.ccld_backfill import (  # noqa: E402
     CcldHostedBackfillRequest,
+    diagnose_ccld_preserved_artifact_differences,
     run_ccld_hosted_backfill,
 )
 from ccld_complaints.hosted_app.ccld_source_refresh import (  # noqa: E402
@@ -54,13 +57,35 @@ def build_parser() -> argparse.ArgumentParser:
     mode = parser.add_mutually_exclusive_group()
     mode.add_argument("--dry-run", action="store_true")
     mode.add_argument("--apply", action="store_true")
+    mode.add_argument("--diagnose-differences", action="store_true")
     return parser
 
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
     try:
         facility_numbers = _facility_numbers(args)
+        if args.diagnose_differences:
+            _validate_diagnostic_arguments(args, facility_numbers)
+            with open_configured_facility_reference_connection() as connection:
+                differences = diagnose_ccld_preserved_artifact_differences(
+                    connection,
+                    facility_numbers,
+                )
+                connection.rollback()
+            print(
+                json.dumps(
+                    {
+                        "schema_version": "recordstracker.ccld-backfill-differences.v1",
+                        "facility_count": len(facility_numbers),
+                        "difference_count": len(differences),
+                        "differences": [asdict(row) for row in differences],
+                    },
+                    sort_keys=True,
+                    separators=(",", ":"),
+                )
+            )
+            return 0
         request = CcldHostedBackfillRequest(
             facility_numbers=facility_numbers,
             all_existing=bool(args.all_existing),
@@ -126,6 +151,18 @@ def _facility_numbers(args: argparse.Namespace) -> tuple[str, ...]:
     return normalized
 
 
+def _validate_diagnostic_arguments(
+    args: argparse.Namespace,
+    facility_numbers: tuple[str, ...],
+) -> None:
+    if args.all_existing or not facility_numbers:
+        raise ValueError("Difference diagnosis requires explicit Facility IDs or a file.")
+    if args.operation != "preserved-artifacts":
+        raise ValueError("Difference diagnosis supports only preserved-artifacts.")
+    if args.max_facilities is not None or args.checkpoint_file or args.restart:
+        raise ValueError("Difference diagnosis does not accept apply or checkpoint options.")
+
+
 def _safe_error(error: Exception) -> str:
     message = str(error)
     if DATABASE_URL_ENV in message:
@@ -143,6 +180,8 @@ def _safe_error(error: Exception) -> str:
         "checkpoint",
         "Select either",
         "Unsupported CCLD hosted backfill operation",
+        "Difference diagnosis",
+        "Diagnostic",
     )
     if any(fragment in message for fragment in allowed_fragments):
         return message

src/ccld_complaints/hosted_app/ccld_backfill.py

@@ -21,6 +21,10 @@
 )
 from ccld_complaints.hosted_app.seeded_import import (
     SeededCorpusArtifact,
+    SeededSourceDerivedRecord,
+    _deduplicate_source_record_projections,
+    _prepared_source_record_values,
+    flatten_seeded_corpus_records,
     hosted_source_derived_records,
     import_seeded_corpus_artifact,
 )
@@ -64,6 +68,16 @@ class CcldHostedBackfillResult:
     intended_updates: int = 0
 
 
+@dataclass(frozen=True)
+class CcldHostedBackfillDifference:
+    facility_number: str
+    entity_type: str
+    source_record_key: str
+    differing_fields: tuple[str, ...]
+    persisted: Mapping[str, object]
+    prepared: Mapping[str, object]
+
+
 @dataclass
 class _BackfillCheckpoint:
     selected_facility_numbers: tuple[str, ...]
@@ -198,6 +212,183 @@ def run_ccld_hosted_backfill(
     )
 
 
+def diagnose_ccld_preserved_artifact_differences(
+    connection: Connection,
+    facility_numbers: Sequence[str],
+) -> tuple[CcldHostedBackfillDifference, ...]:
+    """Compare replay values with persisted rows using SELECT-only preparation."""
+
+    requested = tuple(dict.fromkeys(str(value).strip() for value in facility_numbers))
+    if not requested or any(not value.isdigit() for value in requested):
+        raise ValueError("Diagnostic Facility IDs must be an explicit digit-only selection.")
+    request = CcldHostedBackfillRequest(
+        facility_numbers=requested,
+        operation="preserved-artifacts",
+        apply_changes=False,
+    )
+    _validate_request(connection, request)
+    selected, _candidate_count, selection_conflicts, missing_count = _selected_facilities(
+        connection,
+        request,
+    )
+    if selection_conflicts or missing_count or len(selected) != len(requested):
+        raise ValueError(
+            "Diagnostic selection must resolve every explicit Facility ID exactly once."
+        )
+
+    differences: list[CcldHostedBackfillDifference] = []
+    for facility_number, facility_row in selected:
+        source_documents = _source_documents_for_facility(
+            connection,
+            str(facility_row["facility_id"]),
+        )
+        records: list[Mapping[str, Any]] = []
+        for source_document_row in source_documents:
+            normalized = _reprocess_preserved_document(
+                connection,
+                facility_number,
+                facility_row,
+                source_document_row,
+            )
+            if normalized is None:
+                raise ValueError(
+                    "Diagnostic selection contains an unsupported preserved source document."
+                )
+            records.append(normalized)
+        if not records:
+            raise ValueError("Diagnostic selection has no preserved CCLD source documents.")
+        prepared = prepare_ccld_hosted_source_records(
+            connection,
+            records,
+            include_facility_reference=False,
+        )
+        artifact = _backfill_artifact(
+            facility_number,
+            _deduplicate_facility_projections(prepared.records),
+            operation="preserved-artifacts",
+            now=datetime(2000, 1, 1, tzinfo=UTC),
+        )
+        flattened = _deduplicate_source_record_projections(
+            flatten_seeded_corpus_records(artifact)
+        )
+        for record in flattened:
+            difference = _source_record_difference(
+                connection,
+                facility_number,
+                record,
+            )
+            if difference is not None:
+                differences.append(difference)
+    return tuple(
+        sorted(
+            differences,
+            key=lambda row: (
+                row.facility_number,
+                row.entity_type,
+                row.source_record_key,
+            ),
+        )
+    )
+
+
+def _source_record_difference(
+    connection: Connection,
+    facility_number: str,
+    record: SeededSourceDerivedRecord,
+) -> CcldHostedBackfillDifference | None:
+    existing = cast(
+        Mapping[str, Any] | None,
+        connection.execute(
+            select(hosted_source_derived_records).where(
+                hosted_source_derived_records.c.source_record_key
+                == record.source_record_key
+            )
+        ).mappings().first(),
+    )
+    values, _conflicts = _prepared_source_record_values(
+        connection,
+        record,
+        existing,
+        preserve_existing_import_batch=True,
+    )
+    if existing is None:
+        return CcldHostedBackfillDifference(
+            facility_number=facility_number,
+            entity_type=record.entity_type,
+            source_record_key=record.source_record_key,
+            differing_fields=("persisted_record",),
+            persisted={"persisted_record": {"type": "missing"}},
+            prepared={"persisted_record": _diagnostic_value_summary(values)},
+        )
+    fields, persisted, prepared = _differing_source_record_fields(existing, values)
+    if not fields:
+        return None
+    return CcldHostedBackfillDifference(
+        facility_number=facility_number,
+        entity_type=record.entity_type,
+        source_record_key=record.source_record_key,
+        differing_fields=fields,
+        persisted=persisted,
+        prepared=prepared,
+    )
+
+
+def _differing_source_record_fields(
+    existing: Mapping[str, Any],
+    values: Mapping[str, Any],
+) -> tuple[tuple[str, ...], dict[str, object], dict[str, object]]:
+    fields: list[str] = []
+    persisted: dict[str, object] = {}
+    prepared: dict[str, object] = {}
+    for key in sorted(values):
+        before = existing.get(key)
+        after = values[key]
+        if before == after:
+            continue
+        if key in {"original_values", "source_traceability"} and isinstance(
+            before, Mapping
+        ) and isinstance(after, Mapping):
+            nested_keys = sorted(set(before) | set(after))
+            for nested_key in nested_keys:
+                nested_before = before.get(nested_key)
+                nested_after = after.get(nested_key)
+                if nested_before == nested_after:
+                    continue
+                path = f"{key}.{nested_key}"
+                fields.append(path)
+                persisted[path] = _diagnostic_value_summary(nested_before)
+                prepared[path] = _diagnostic_value_summary(nested_after)
+            continue
+        fields.append(key)
+        persisted[key] = _diagnostic_value_summary(before)
+        prepared[key] = _diagnostic_value_summary(after)
+    return tuple(fields), persisted, prepared
+
+
+def _diagnostic_value_summary(value: object) -> Mapping[str, object]:
+    if value is None:
+        return {"type": "null"}
+    if isinstance(value, bool):
+        return {"type": "boolean", "value": value}
+    if isinstance(value, int | float):
+        return {"type": "number", "value": value}
+    if isinstance(value, str):
+        encoded = value.encode("utf-8")
+        return {
+            "type": "string",
+            "length": len(value),
+            "sha256": sha256_bytes(encoded),
+        }
+    canonical = json.dumps(value, sort_keys=True, separators=(",", ":"), default=str)
+    summary: dict[str, object] = {
+        "type": "mapping" if isinstance(value, Mapping) else "sequence",
+        "sha256": sha256_bytes(canonical.encode("utf-8")),
+    }
+    if isinstance(value, Mapping | Sequence) and not isinstance(value, str | bytes):
+        summary["count"] = len(value)
+    return summary
+
+
 def _process_facility(
     connection: Connection,
     facility_number: str,