from __future__ import annotations import json import hashlib from datetime import date, datetime, timedelta from pathlib import Path from zoneinfo import ZoneInfo import pytest import checkin_cli import checkin_cli.customer_schedule as customer_schedule from checkin_cli.weekly_operations_schedule_host_models_r4 import ( ApprovedReminderScheduleEvidence, CustomerScheduleError, ) from checkin_cli.weekly_operations_schedule_host_review_r4 import ( non_response_review_candidate, ) from tests.test_customer_coaching_domain import _registry_payload KST = ZoneInfo("Asia/Seoul") def _registry(tmp_path: Path, *, starts_on: date | None = None): payload = _registry_payload() customers = payload["customers"] assert isinstance(customers, list) assert isinstance(customers[0], dict) and isinstance(customers[1], dict) customers[0]["schedule"] = {"daily_time": "08:00", "weekly_weekday": 0, "monthly_day": 1} customers[1]["schedule"] = {"daily_time": "09:00", "weekly_weekday": 4, "monthly_day": 15} if starts_on is not None: for customer in customers: plan = customer["plan"] assert isinstance(plan, dict) plan["starts_on"] = starts_on.isoformat() path = tmp_path / "registry.json" path.write_text(json.dumps(payload), encoding="utf-8") return checkin_cli.load_customer_registry(path, tmp_path) def test_late_tick_returns_only_due_customer_daily_task(tmp_path: Path) -> None: registry = _registry(tmp_path) now = datetime(2026, 7, 21, 8, 17, tzinfo=KST) tasks = checkin_cli.build_due_customer_tasks(registry, now) assert [(task.customer_key, task.kind) for task in tasks] == [("client_001", "daily")] def test_sunday_tick_returns_customer_daily_task(tmp_path: Path) -> None: registry = _registry(tmp_path) now = datetime(2026, 7, 26, 8, 17, tzinfo=KST) tasks = checkin_cli.build_due_customer_tasks(registry, now) assert [(task.customer_key, task.kind) for task in tasks] == [ ("client_001", "daily"), ] def test_weekly_task_runs_without_monthly_inside_pilot_window(tmp_path: Path) -> None: registry = _registry(tmp_path) now = datetime(2026, 7, 20, 8, 5, tzinfo=KST) tasks = checkin_cli.build_due_customer_tasks(registry, now) first_tasks = [task for task in tasks if task.customer_key == "client_001"] assert [task.kind for task in first_tasks] == ["daily", "weekly"] def test_missing_morning_checkin_reaches_only_the_reminder_branch_once(tmp_path: Path) -> None: registry = _registry(tmp_path) today = date(2026, 7, 20) evidence = ApprovedReminderScheduleEvidence("6" * 64, "7" * 64) tasks = checkin_cli.build_due_customer_tasks( registry, datetime(2026, 7, 20, 8, 17, tzinfo=KST), missing_morning_checkins={"client_001": today}, reminder_evidence=evidence, ) assert [(task.customer_key, task.kind) for task in tasks] == [ ("client_001", "daily"), ("client_001", "weekly"), ("client_001", "reminder"), ] assert len({(task.customer_key, task.kst_day, task.kind) for task in tasks}) == len(tasks) def test_reminder_is_not_due_when_answered_or_customer_is_ineligible(tmp_path: Path) -> None: registry = _registry(tmp_path) evidence = ApprovedReminderScheduleEvidence("6" * 64, "7" * 64) answered = checkin_cli.build_due_customer_tasks( registry, datetime(2026, 7, 21, 8, 17, tzinfo=KST), missing_morning_checkins={}, reminder_evidence=evidence, ) disabled = checkin_cli.build_due_customer_tasks( registry, datetime(2026, 8, 30, 8, 17, tzinfo=KST), missing_morning_checkins={"client_001": date(2026, 8, 30)}, reminder_evidence=evidence, ) assert all(task.kind != "reminder" for task in answered) assert disabled == () def test_missing_checkin_reminder_requires_pinned_approval_evidence(tmp_path: Path) -> None: with pytest.raises(CustomerScheduleError, match="approved policy and config evidence"): checkin_cli.build_due_customer_tasks( _registry(tmp_path), datetime(2026, 7, 21, 8, 17, tzinfo=KST), missing_morning_checkins={"client_001": date(2026, 7, 21)}, ) def test_legacy_claim_producer_is_not_exported() -> None: assert not hasattr(checkin_cli, "claim_customer_task") def _legacy_claim_path(profile_root: Path, task: checkin_cli.CustomerScheduleTask) -> Path: return ( profile_root / "data" / "customer-schedule-claims" / task.customer_key / task.kst_day.isoformat() / f"{task.kind}.claim" ) def test_legacy_claim_migrates_only_during_cutover_and_preserves_bytes(tmp_path: Path) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) claim = _legacy_claim_path(tmp_path, task) claim.parent.mkdir(parents=True) claim.write_bytes(b"claimed\n") claim.chmod(0o600) before = claim.read_bytes() fence = checkin_cli.initialize_schedule_delivery_fence(tmp_path) assert fence.state == "ready" assert claim.read_bytes() == before rows = _schedule_rows(tmp_path) assert len(rows) == 1 assert rows[0]["state"] == "unknown" assert rows[0]["reason"] == "legacy_claim_unknown" def test_ready_fence_stale_legacy_claim_blocks_and_cutover_preserves_bytes( tmp_path: Path, ) -> None: checkin_cli.initialize_schedule_delivery_fence(tmp_path) task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) claim = _legacy_claim_path(tmp_path, task) claim.parent.mkdir(parents=True) claim.write_bytes(b"claimed-after-ready\n") claim.chmod(0o600) before = claim.read_bytes() ledger = tmp_path / "data" / "scheduled-deliveries.jsonl" ledger_before = ledger.read_bytes() if ledger.exists() else b"" with pytest.raises(CustomerScheduleError, match="tombstone has no ledger pair"): checkin_cli.schedule_delivery_ledger(tmp_path) assert claim.read_bytes() == before assert (ledger.read_bytes() if ledger.exists() else b"") == ledger_before fence = json.loads( (tmp_path / "data" / "scheduled-deliveries-fence.json").read_text( encoding="utf-8" ) ) assert fence["state"] == "recovery_required" assert checkin_cli.prepare_schedule_delivery_cutover(tmp_path).state == "preparing" assert checkin_cli.finalize_schedule_delivery_cutover(tmp_path).state == "ready" assert claim.read_bytes() == before rows = _schedule_rows(tmp_path) assert rows[-1]["state"] == "unknown" def test_day_28_is_last_scheduled_day_and_day_29_is_dormant(tmp_path: Path) -> None: starts_on = date(2026, 7, 1) registry = _registry(tmp_path, starts_on=starts_on) day_28 = starts_on + timedelta(days=27) day_29 = day_28 + timedelta(days=1) day_28_tasks = checkin_cli.build_due_customer_tasks( registry, datetime(day_28.year, day_28.month, day_28.day, 9, 5, tzinfo=KST), ) day_29_tasks = checkin_cli.build_due_customer_tasks( registry, datetime(day_29.year, day_29.month, day_29.day, 9, 5, tzinfo=KST), ) assert [(task.customer_key, task.kind) for task in day_28_tasks] == [ ("client_001", "daily"), ] assert all(task.kst_day == day_28 for task in day_28_tasks) assert day_29_tasks == () def test_monthly_task_stays_dormant_on_configured_monthly_day(tmp_path: Path) -> None: registry = _registry(tmp_path, starts_on=date(2026, 7, 1)) tasks = checkin_cli.build_due_customer_tasks( registry, datetime(2026, 7, 15, 9, 5, tzinfo=KST), ) assert [(task.customer_key, task.kind) for task in tasks] == [ ("client_001", "daily"), ] assert all(task.kind != "monthly" for task in tasks) def _schedule_rows(profile_root: Path) -> list[dict[str, object]]: ledger = profile_root / "data" / "scheduled-deliveries.jsonl" if not ledger.exists(): return [] return [ json.loads(line) for line in ledger.read_text(encoding="utf-8").splitlines() if line ] def _schedule_delivery_args() -> dict[str, object]: return { "body": "private customer delivery body", "destination": { "user_id": "raw-user-id", "chat_id": "raw-chat-id", "topic_id": "raw-topic-id", }, "template_digest": "1" * 64, "registry_digest": "2" * 64, "config_digest": "3" * 64, "reservation_id": "reservation-00000001", } def _reminder_args(approval: str = "approval-0001") -> dict[str, object]: return { "destination": {"chat_id": "reminder-chat"}, "registry_digest": "4" * 64, "config_digest": "5" * 64, "operator_approval": approval, } def test_static_reminder_is_exactly_once_and_unknown_is_never_retried(tmp_path: Path) -> None: reserve = customer_schedule.reserve_missing_checkin_reminder first = reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args()) duplicate = reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args()) sending = checkin_cli.mark_customer_task_sending(tmp_path, first) unknown = checkin_cli.mark_customer_task_unknown( tmp_path, sending, reason="provider_timeout" ) assert duplicate == first assert unknown.state == "unknown" with pytest.raises(CustomerScheduleError, match="already terminal"): reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args("approval-0002")) def test_terminal_response_abandonment_is_no_send_and_never_replaceable( tmp_path: Path, ) -> None: reserve = customer_schedule.reserve_missing_checkin_reminder first = reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args()) abandoned = customer_schedule.abandon_missing_checkin_reminder_for_terminal_morning_response( tmp_path, checkin_cli.mark_customer_task_sending(tmp_path, first) ) assert abandoned.state == "abandoned" assert abandoned.provider_receipt is None assert abandoned.message_id is None assert abandoned.reason == "terminal_morning_response_before_provider" with pytest.raises(CustomerScheduleError, match="already terminal"): reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args("approval-0002")) def test_known_reminder_failure_requires_new_operator_approval(tmp_path: Path) -> None: reserve = customer_schedule.reserve_missing_checkin_reminder first = reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args()) sending = checkin_cli.mark_customer_task_sending(tmp_path, first) lease_key = customer_schedule._lease_key(tmp_path, first.reservation_id) lease_handle = customer_schedule._SCHEDULE_PROVIDER_LEASES[lease_key] failed = customer_schedule.mark_customer_task_known_failure( tmp_path, sending, "provider_rejected" ) assert lease_key not in customer_schedule._SCHEDULE_PROVIDER_LEASES assert lease_handle.closed replacement = reserve( tmp_path, "client_001", date(2026, 8, 3), **_reminder_args("approval-0002") ) assert failed.state == "known_failure" assert replacement.state == "prepared" assert replacement.reservation_id != first.reservation_id claim = _legacy_claim_path( tmp_path, checkin_cli.CustomerScheduleTask("client_001", "reminder", date(2026, 8, 3)) ) assert claim.read_bytes() == ( b"scheduled-delivery-tombstone-v1\n" + replacement.reservation_id.encode("ascii") + b"\n" ) sending_replacement = checkin_cli.mark_customer_task_sending(tmp_path, replacement) delivered_replacement = checkin_cli.mark_customer_task_delivered( tmp_path, sending_replacement, provider_receipt="replacement-receipt-0001" ) audited_replacement = checkin_cli.mark_customer_task_sent_audited( tmp_path, delivered_replacement ) assert audited_replacement.state == "sent_audited" assert [row["state"] for row in _schedule_rows(tmp_path)] == [ "prepared", "sending", "known_failure", "prepared", "sending", "delivered", "sent_audited", ] assert reserve( tmp_path, "client_001", date(2026, 8, 3), **_reminder_args("approval-0002") ) == audited_replacement def test_replacement_claim_crash_stays_fenced_and_never_reopens_the_old_attempt( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: reserve = customer_schedule.reserve_missing_checkin_reminder first = reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args()) failed = customer_schedule.mark_customer_task_known_failure( tmp_path, checkin_cli.mark_customer_task_sending(tmp_path, first), "provider_rejected" ) monkeypatch.setattr( customer_schedule, "_append_row", lambda *_args, **_kwargs: (_ for _ in ()).throw(CustomerScheduleError("disk failure")), ) with pytest.raises(CustomerScheduleError, match="disk failure"): reserve(tmp_path, "client_001", date(2026, 8, 3), **_reminder_args("approval-0002")) claim = _legacy_claim_path( tmp_path, checkin_cli.CustomerScheduleTask("client_001", "reminder", date(2026, 8, 3)) ) assert first.reservation_id.encode("ascii") not in claim.read_bytes() assert failed.reservation_id != customer_schedule._tombstone_reservation(claim.read_bytes()) fence = json.loads( (tmp_path / "data" / "scheduled-deliveries-fence.json").read_text(encoding="utf-8") ) assert fence["state"] == "recovery_required" with pytest.raises(CustomerScheduleError, match="not ready"): checkin_cli.schedule_delivery_ledger(tmp_path) def test_review_candidate_waits_for_audited_reminder_and_no_response(tmp_path: Path) -> None: reminder = customer_schedule.reserve_missing_checkin_reminder( tmp_path, "client_001", date(2026, 8, 3), **_reminder_args() ) sending = checkin_cli.mark_customer_task_sending(tmp_path, reminder) delivered = checkin_cli.mark_customer_task_delivered( tmp_path, sending, provider_receipt="provider-receipt-0001" ) deadline = datetime(2026, 8, 4, 9, tzinfo=KST) assert non_response_review_candidate( delivered, response_window_ends_at=deadline, now=deadline + timedelta(hours=1) ) is None audited = checkin_cli.mark_customer_task_sent_audited(tmp_path, delivered) assert non_response_review_candidate( audited, response_window_ends_at=deadline, now=deadline + timedelta(hours=1), checkin_received_at=deadline + timedelta(minutes=1), ) is None first = non_response_review_candidate( audited, response_window_ends_at=deadline, now=deadline + timedelta(hours=1) ) duplicate = non_response_review_candidate( audited, response_window_ends_at=deadline, now=deadline + timedelta(hours=2) ) assert first is not None assert duplicate == first assert first.correlation_id == duplicate.correlation_id def test_scheduled_delivery_lifecycle_is_prepared_sending_delivered_sent_audited( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) arguments = _schedule_delivery_args() prepared = checkin_cli.reserve_customer_task_delivery( tmp_path, task, **arguments, ) sending = checkin_cli.mark_customer_task_sending(tmp_path, prepared) delivered = checkin_cli.mark_customer_task_delivered( tmp_path, sending, provider_receipt="provider-receipt-0001", message_id="provider-message-0001", ) audited = checkin_cli.mark_customer_task_sent_audited( tmp_path, delivered, audit_receipt="audit-receipt-0001", ) assert [row["state"] for row in _schedule_rows(tmp_path)] == [ "prepared", "sending", "delivered", "sent_audited", ] assert audited.state == "sent_audited" assert audited.provider_receipt == "provider-receipt-0001" assert audited.message_id == "provider-message-0001" assert audited.body_digest == hashlib.sha256( json.dumps( str(arguments["body"]), ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") ).hexdigest() bounded = json.dumps(audited.to_dict(), ensure_ascii=False) assert "private customer delivery body" not in bounded assert "raw-user-id" not in bounded assert "raw-chat-id" not in bounded assert "raw-topic-id" not in bounded def test_scheduled_delivery_sending_transition_grants_one_provider_authority( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) prepared = checkin_cli.reserve_customer_task_delivery( tmp_path, task, **_schedule_delivery_args(), ) winner = checkin_cli.mark_customer_task_sending(tmp_path, prepared) contender = checkin_cli.mark_customer_task_sending(tmp_path, prepared) assert winner.state == "sending" assert winner.provider_authority is True assert contender.state == "sending" assert contender.provider_authority is False assert [row["state"] for row in _schedule_rows(tmp_path)] == [ "prepared", "sending", ] checkin_cli.mark_customer_task_unknown( tmp_path, winner, reason="test_cleanup", ) def test_scheduled_delivery_timeout_unknown_is_terminal_and_never_retried( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) arguments = _schedule_delivery_args() prepared = checkin_cli.reserve_customer_task_delivery(tmp_path, task, **arguments) sending = checkin_cli.mark_customer_task_sending(tmp_path, prepared) unknown = checkin_cli.mark_customer_task_unknown( tmp_path, sending, reason="provider_timeout", ) replay = checkin_cli.mark_customer_task_sending(tmp_path, unknown) duplicate_reservation = checkin_cli.reserve_customer_task_delivery( tmp_path, task, **arguments, ) assert unknown.state == "unknown" assert unknown.reason == "provider_timeout" assert replay == unknown assert duplicate_reservation == unknown assert [row["state"] for row in _schedule_rows(tmp_path)] == [ "prepared", "sending", "unknown", ] assert bool(unknown) is False def test_scheduled_delivery_receipt_reconciliation_never_invokes_provider( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) arguments = _schedule_delivery_args() prepared = checkin_cli.reserve_customer_task_delivery(tmp_path, task, **arguments) sending = checkin_cli.mark_customer_task_sending(tmp_path, prepared) delivered = checkin_cli.mark_customer_task_delivered( tmp_path, sending, provider_receipt="provider-receipt-0002", message_id="provider-message-0002", ) provider_calls = 0 reconciled = checkin_cli.reconcile_customer_task_delivery( tmp_path, delivered, provider_receipt="provider-receipt-0002", message_id="provider-message-0002", audit_receipt="audit-receipt-0002", ) assert provider_calls == 0 assert reconciled.state == "sent_audited" assert reconciled.provider_receipt == "provider-receipt-0002" assert [row["state"] for row in _schedule_rows(tmp_path)] == [ "prepared", "sending", "delivered", "sent_audited", ] def test_scheduled_delivery_reservation_pairs_tombstone_before_ledger( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "weekly", date(2026, 8, 3)) arguments = _schedule_delivery_args() receipt = checkin_cli.reserve_customer_task_delivery(tmp_path, task, **arguments) claim = ( tmp_path / "data" / "customer-schedule-claims" / task.customer_key / task.kst_day.isoformat() / f"{task.kind}.claim" ) assert claim.exists() claim_bytes = claim.read_bytes() rows = _schedule_rows(tmp_path) assert len(rows) == 1 assert rows[0]["state"] == "prepared" assert rows[0]["reservation_id"] == receipt.reservation_id assert rows[0]["legacy_claim_digest"] == hashlib.sha256(claim_bytes).hexdigest() assert claim_bytes.startswith(b"scheduled-delivery-tombstone-v1\n") def test_scheduled_delivery_orphan_tombstone_recovers_terminal_unknown_without_deleting_bytes( tmp_path: Path, ) -> None: claim = ( tmp_path / "data" / "customer-schedule-claims" / "client_001" / "2026-08-03" / "daily.claim" ) claim.parent.mkdir(parents=True) claim.write_bytes(b"scheduled-delivery-tombstone-v1\norphan-reservation\n") claim.chmod(0o600) before = claim.read_bytes() fence = checkin_cli.initialize_schedule_delivery_fence(tmp_path) assert fence.state == "ready" assert claim.read_bytes() == before rows = _schedule_rows(tmp_path) assert len(rows) == 1 assert rows[0]["state"] == "unknown" assert rows[0]["reservation_id"] == "orphan-reservation" assert rows[0]["reason"] == "tombstone_ledger_missing_recovered" assert checkin_cli.initialize_schedule_delivery_fence(tmp_path).state == "ready" def test_tombstone_durable_before_ledger_failure_has_idempotent_recovery( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: import checkin_cli.customer_schedule as schedule task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) original_append = schedule._append_row def fail_ledger_append(*_args: object, **_kwargs: object) -> dict[str, object]: raise CustomerScheduleError("simulated ledger fsync failure") monkeypatch.setattr(schedule, "_append_row", fail_ledger_append) with pytest.raises(CustomerScheduleError, match="simulated ledger fsync failure"): checkin_cli.reserve_customer_task_delivery( tmp_path, task, **_schedule_delivery_args(), ) claim = _legacy_claim_path(tmp_path, task) claim_bytes = claim.read_bytes() assert claim_bytes.startswith(b"scheduled-delivery-tombstone-v1\n") assert _schedule_rows(tmp_path) == [] monkeypatch.setattr(schedule, "_append_row", original_append) assert checkin_cli.prepare_schedule_delivery_cutover(tmp_path).state == "preparing" assert checkin_cli.finalize_schedule_delivery_cutover(tmp_path).state == "ready" assert claim.read_bytes() == claim_bytes rows = _schedule_rows(tmp_path) assert rows[-1]["state"] == "unknown" assert rows[-1]["reason"] == "tombstone_ledger_missing_recovered" assert checkin_cli.initialize_schedule_delivery_fence(tmp_path).state == "ready" def test_scheduled_delivery_ledger_without_tombstone_refuses_startup( tmp_path: Path, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) checkin_cli.reserve_customer_task_delivery( tmp_path, task, **_schedule_delivery_args(), ) claim = ( tmp_path / "data" / "customer-schedule-claims" / task.customer_key / task.kst_day.isoformat() / f"{task.kind}.claim" ) claim.unlink() with pytest.raises(CustomerScheduleError, match="no tombstone"): checkin_cli.schedule_delivery_ledger(tmp_path) fence = json.loads( (tmp_path / "data" / "scheduled-deliveries-fence.json").read_text( encoding="utf-8" ) ) assert fence["state"] == "recovery_required" def test_scheduled_delivery_preparing_fence_blocks_writes_until_ready( tmp_path: Path, ) -> None: preparing = checkin_cli.prepare_schedule_delivery_cutover(tmp_path) assert preparing.state == "preparing" with pytest.raises(CustomerScheduleError, match="not ready"): checkin_cli.reserve_customer_task_delivery( tmp_path, checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)), **_schedule_delivery_args(), ) ready = checkin_cli.finalize_schedule_delivery_cutover(tmp_path) assert ready.state == "ready" receipt = checkin_cli.reserve_customer_task_delivery( tmp_path, checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)), **_schedule_delivery_args(), ) assert receipt.state == "prepared" @pytest.mark.parametrize( "corruption", ["invalid-json", "mixed-version"], ids=["corrupt-ledger", "mixed-schema-version"], ) def test_scheduled_delivery_corrupt_or_mixed_version_ledger_fails_closed( tmp_path: Path, corruption: str, ) -> None: task = checkin_cli.CustomerScheduleTask("client_001", "daily", date(2026, 8, 3)) checkin_cli.reserve_customer_task_delivery( tmp_path, task, **_schedule_delivery_args(), ) ledger = tmp_path / "data" / "scheduled-deliveries.jsonl" if corruption == "invalid-json": ledger.write_text("{not-json}\n", encoding="utf-8") else: with ledger.open("a", encoding="utf-8") as handle: handle.write(json.dumps({"schema_version": 2}) + "\n") with pytest.raises(CustomerScheduleError): checkin_cli.schedule_delivery_ledger(tmp_path) def test_scheduled_delivery_daily_and_weekly_duplicates_are_idempotent( tmp_path: Path, ) -> None: day = date(2026, 8, 3) daily = checkin_cli.CustomerScheduleTask("client_001", "daily", day) weekly = checkin_cli.CustomerScheduleTask("client_001", "weekly", day) daily_args = _schedule_delivery_args() daily_again = checkin_cli.reserve_customer_task_delivery( tmp_path, daily, **daily_args, ) daily_replay = checkin_cli.reserve_customer_task_delivery( tmp_path, daily, **daily_args, ) weekly_replay = checkin_cli.reserve_customer_task_delivery( tmp_path, weekly, **{ **daily_args, "reservation_id": "reservation-00000002", }, ) assert daily_replay == daily_again assert weekly_replay.schedule_key.endswith(":weekly") assert len(checkin_cli.schedule_delivery_ledger(tmp_path)) == 2 assert [row["schedule_key"] for row in _schedule_rows(tmp_path)] == [ "client_001:2026-08-03:daily", "client_001:2026-08-03:weekly", ] with pytest.raises(CustomerScheduleError, match="pins conflict"): checkin_cli.reserve_customer_task_delivery( tmp_path, daily, **{**daily_args, "body": "different body"}, )