from __future__ import annotations import hashlib import json from pathlib import Path from typing import cast import pytest from checkin_cli.customer_cleanup import ( CleanupBlockedError, CleanupIntegrityError, SharedLedgerProjection, archive_customer_cleanup, post_cleanup_authority_inventory, resume_customer_cleanup, ) TARGET = "client_001" FOREIGN = "client_002" LEDGER = Path("data/owner-actions/draft-generations.json") _RECORD_KEYS = { "schema_version", "generation", "state", "customer_key", "checkin_event_id", "checkin_revision", "draft_revision", "model_contract_version", "provider_contract_version", "delivery_provider_contract_version", "actor", "authority_digest", "predecessor_digest", "lineage_predecessor_digest", "lineage_parent_token", "created_at", "updated_at", "attempt", "max_attempts", "error", "generation_provider_receipt", "delivery_provider_receipt", "idempotency_key", "record_digest", } def _private_write(path: Path, value: object) -> None: path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) path.parent.chmod(0o700) path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True), encoding="utf-8") path.chmod(0o600) def _profile(tmp_path: Path) -> Path: root = tmp_path / "profile" registry = { "version": 1, "owner": {"user_id": "1", "chat_id": "-100", "topic_id": "10"}, "customers": [ { "customer_key": key, "display_name": key, "telegram": {"user_id": user, "chat_id": "-100", "topic_id": topic}, "enabled": False, "ai_processing_consent": { "granted": False, "recorded_on": "2026-08-18", "notice_version": "v1", }, } for key, user, topic in ((TARGET, "2", "20"), (FOREIGN, "3", "30")) ], } _private_write(root / "customers/registry.json", registry) customer = root / f"data/customers/{TARGET}/onboarding/state.json" _private_write(customer, {"answer": 1}) return root def _generation(customer: str, state: str, marker: str) -> dict[str, object]: # This is the exact draft-generations.json record shape written by the Golden Path. row: dict[str, object] = { "schema_version": "nutrition-draft-generation-v1", "generation": 8, "state": state, "customer_key": customer, "checkin_event_id": f"checkin-{marker}", "checkin_revision": "a" * 64, "draft_revision": "b" * 64, "model_contract_version": "nutrition-coach-response-v2", "provider_contract_version": "chat-completions-v1", "delivery_provider_contract_version": "telegram-send-v1", "actor": "owner:1", "authority_digest": "c" * 64, "predecessor_digest": "d" * 64, "lineage_predecessor_digest": None, "lineage_parent_token": None, "created_at": "2026-08-18T00:00:00+00:00", "updated_at": "2026-08-18T00:01:00+00:00", "attempt": 1, "max_attempts": 2, "error": None, "generation_provider_receipt": "e" * 64, "delivery_provider_receipt": f"telegram-{marker}", "idempotency_key": "f" * 64, "record_digest": "0" * 64, } assert set(row) == _RECORD_KEYS unsigned = dict(row) unsigned.pop("record_digest") row["record_digest"] = hashlib.sha256( json.dumps(unsigned, sort_keys=True, separators=(",", ":")).encode() ).hexdigest() return row def _write_ledger(root: Path, target_state: str, foreign_state: str = "sent_audited") -> bytes: payload = { "target-draft": [_generation(TARGET, target_state, "TARGET-ONLY")], "foreign-draft": [_generation(FOREIGN, foreign_state, "FOREIGN-SECRET")], } path = root / LEDGER _private_write(path, payload) return path.read_bytes() def _cleanup(root: Path): return archive_customer_cleanup( root, TARGET, shared_ledger_projections=(SharedLedgerProjection.DRAFT_GENERATIONS,), ) def test_two_customer_projection_archives_only_target_and_preserves_source(tmp_path: Path) -> None: root = _profile(tmp_path) source_before = _write_ledger(root, "sent_audited") receipt = _cleanup(root) assert (root / LEDGER).read_bytes() == source_before manifest = json.loads(receipt.manifest_path.read_text()) binding = manifest["shared_ledger_projections"][0] projection_path = receipt.archive_root / binding["projection_archive_path"] projection_bytes = projection_path.read_bytes() projection = json.loads(projection_bytes) assert projection["customer_key"] == TARGET assert list(projection["histories"]) == ["target-draft"] assert projection["histories"]["target-draft"][0]["customer_key"] == TARGET assert b"FOREIGN-SECRET" not in projection_bytes assert FOREIGN.encode() not in projection_bytes assert binding["source_relative_path"] == LEDGER.as_posix() assert binding["selector_version"] == "draft-generations-customer-v1" assert binding["selected_record_count"] == 1 assert len(binding["ordered_record_digests"]) == 1 assert binding["projection_sha256"] == hashlib.sha256(projection_bytes).hexdigest() assert {"source_sha256", "source_inode", "source_mode"} <= set(binding) inventory = post_cleanup_authority_inventory(root, TARGET) assert inventory.terminal assert inventory.categories["owner_action"] == 1 @pytest.mark.parametrize("state", ["generation_pending", "future_unrecognized_state"]) def test_target_pending_or_unknown_shared_authority_blocks(tmp_path: Path, state: str) -> None: root = _profile(tmp_path) _write_ledger(root, state) with pytest.raises(CleanupBlockedError, match="pending or unknown"): _cleanup(root) inventory = post_cleanup_authority_inventory(root, TARGET) assert not inventory.terminal assert (inventory.pending_count if state == "generation_pending" else inventory.unknown_count) == 1 def test_foreign_pending_shared_rows_do_not_block_target(tmp_path: Path) -> None: root = _profile(tmp_path) source_before = _write_ledger(root, "sent_audited", "generation_pending") receipt = _cleanup(root) assert receipt.phase == "committed" assert (root / LEDGER).read_bytes() == source_before assert post_cleanup_authority_inventory(root, TARGET).terminal def test_shared_projection_tamper_is_rejected_on_resume(tmp_path: Path) -> None: root = _profile(tmp_path) _write_ledger(root, "sent_audited") def crash(phase: str) -> None: if phase == "copied_verified": raise RuntimeError("stop") with pytest.raises(RuntimeError, match="stop"): archive_customer_cleanup( root, TARGET, shared_ledger_projections=(SharedLedgerProjection.DRAFT_GENERATIONS,), fault_injector=crash, ) projection = next((root / "data/customer-cleanup/archives").glob("*/projections/**/*.json")) projection.write_text('{"foreign":"client_002"}\n', encoding="utf-8") with pytest.raises(CleanupIntegrityError, match="archive verification"): resume_customer_cleanup(root, TARGET) def test_shared_source_mutation_after_prepare_blocks_resume(tmp_path: Path) -> None: root = _profile(tmp_path) _write_ledger(root, "sent_audited") def crash(phase: str) -> None: if phase == "prepared": raise RuntimeError("stop") with pytest.raises(RuntimeError, match="stop"): archive_customer_cleanup( root, TARGET, shared_ledger_projections=(SharedLedgerProjection.DRAFT_GENERATIONS,), fault_injector=crash, ) _write_ledger(root, "sent_audited", "generation_pending") with pytest.raises(CleanupIntegrityError, match="shared ledger changed"): resume_customer_cleanup(root, TARGET) def test_unsupported_shared_path_and_whole_file_candidate_remain_strict(tmp_path: Path) -> None: root = _profile(tmp_path) _write_ledger(root, "sent_audited") unsupported = root / "data/owner-actions/arbitrary.json" _private_write(unsupported, {"customer_key": TARGET, "secret": "no"}) unsupported_projection = cast( SharedLedgerProjection, "data/owner-actions/arbitrary.json", ) with pytest.raises(CleanupIntegrityError, match="unsupported shared ledger"): archive_customer_cleanup( root, TARGET, shared_ledger_projections=(unsupported_projection,), ) with pytest.raises(CleanupIntegrityError, match="candidate path is not customer-scoped"): archive_customer_cleanup(root, TARGET, candidate_paths=(root / LEDGER,)) def test_shared_ledger_missing_or_ambiguous_customer_binding_is_rejected(tmp_path: Path) -> None: root = _profile(tmp_path) row = _generation(TARGET, "sent_audited", "TARGET") row.pop("customer_key") _private_write(root / LEDGER, {"target-draft": [row]}) with pytest.raises(CleanupIntegrityError, match="customer binding"): _cleanup(root)