from __future__ import annotations import ast import hashlib import inspect import json import textwrap from dataclasses import replace from datetime import date, datetime, timedelta from pathlib import Path from types import SimpleNamespace import pytest from tests.gateway.test_nutrition_coaching import ( _coach_v2_artifacts, _coach_v2_review, _draft_coordinator, _generation_record, _generation_transition, _generation_pins, ) from gateway.platforms.nutrition_coaching import ( DeliveryReconciliationEvidence, DraftAction, DraftGenerationClaim, DraftGenerationError, DraftGenerationRecord, DraftGenerationState, DraftGenerationTransitionError, IncomingAddress, ) from gateway.platforms.nutrition_coaching_judgment import judgment_revision_binding from gateway.platforms.nutrition_service_state import CustomerServiceStateStore from gateway.platforms.telegram import TelegramAdapter def _v2_draft(tmp_path: Path): coordinator, owner, _events = _draft_coordinator(tmp_path) selection = coordinator.resolve_draft("draft-001", owner) assert selection is not None binding = judgment_revision_binding( "client_001", "session-001", selection.snapshot.model_dump() ) created = coordinator.create_draft( "draft-001", owner, "검증된 초안입니다.", expected_revision_binding_digest=binding, coach_review=_coach_v2_review(binding), coach_artifacts=_coach_v2_artifacts(binding, "검증된 초안입니다."), ) assert created.accepted return coordinator, owner, created def test_v2_approval_requires_authoritative_generation_history_and_all_cas_pins( tmp_path: Path, ) -> None: coordinator, owner, created = _v2_draft(tmp_path) generation = coordinator.draft_generation("draft-001") assert generation is not None assert generation.checkin_event_id == "finalized-event-001" assert coordinator.approve_draft("draft-001", owner).error == "draft_generation_stale" approved = coordinator.approve_draft( "draft-001", owner, expected_generation=generation.generation, expected_record_digest=generation.record_digest, expected_checkin_revision=generation.checkin_revision, expected_draft_revision=generation.draft_revision, ) assert approved.accepted def _tree_snapshot(root: Path) -> dict[str, bytes]: return { str(path.relative_to(root)): path.read_bytes() for path in sorted(root.rglob("*")) if path.is_file() } def _record_digest(payload: dict[str, object]) -> str: unsigned = {key: value for key, value in payload.items() if key != "record_digest"} encoded = json.dumps( unsigned, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _rechain(history: list[dict[str, object]]) -> None: predecessor = None for record in history: record["predecessor_digest"] = predecessor record["record_digest"] = _record_digest(record) predecessor = record["record_digest"] def test_deleted_generation_history_blocks_hold_without_mutation( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) coordinator._generation_ledger_path().unlink() before = _tree_snapshot(tmp_path) rejected = coordinator.hold_draft("draft-001", owner) assert rejected.accepted is False assert rejected.error == "draft_generation_missing" assert _tree_snapshot(tmp_path) == before def test_deleted_parent_generation_blocks_edit_before_any_mutation( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) coordinator._generation_ledger_path().unlink() before = _tree_snapshot(tmp_path) rejected = coordinator.edit_draft( "draft-001", owner, "권위 이력 없이 수정하면 안 됩니다." ) assert rejected.accepted is False assert rejected.error == "draft_generation_missing" assert _tree_snapshot(tmp_path) == before def test_deleted_established_journal_blocks_create_replay_without_mutation( tmp_path: Path, ) -> None: coordinator, owner, created = _v2_draft(tmp_path) coordinator._generation_ledger_path().unlink() before = _tree_snapshot(tmp_path) replay = coordinator.create_draft( "draft-001", owner, created.text or "", expected_revision_binding_digest=created.coach_review.revision_binding_digest, coach_review=created.coach_review, coach_artifacts=_coach_v2_artifacts( created.coach_review.revision_binding_digest, created.text or "", ), ) assert replay.accepted is False assert _tree_snapshot(tmp_path) == before def test_child_generation_write_failure_leaves_exact_pre_state( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) before = _tree_snapshot(tmp_path) real_write = coordinator._write_draft_generations def fail_child_generation(payload): if set(payload) != {"draft-001"}: raise OSError("injected child generation failure") real_write(payload) coordinator._write_draft_generations = fail_child_generation rejected = coordinator.edit_draft( "draft-001", owner, "자식 권위 기록 실패를 주입합니다." ) assert rejected.accepted is False assert _tree_snapshot(tmp_path) == before def test_global_lineage_validation_rejects_orphan_child_digest( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) child = coordinator.edit_draft("draft-001", owner, "계보 검증용 수정본입니다.") assert child.accepted path = coordinator._generation_ledger_path() payload = json.loads(path.read_text(encoding="utf-8")) del payload["draft-001"] path.write_text(json.dumps(payload), encoding="utf-8") assert coordinator.draft_generation_history(child.draft_id) == () def test_worker_completion_requires_claim_and_terminal_failure_stays_closed( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) receipt = coordinator.draft_generation("draft-001").generation_provider_receipt assert receipt is not None assert coordinator.complete_draft_generation("draft-001", receipt) is False coordinator2, owner2, _events = _draft_coordinator(tmp_path / "terminal") claim = coordinator2.claim_draft_generation("draft-001", owner2, "worker-a") assert isinstance(claim, DraftGenerationClaim) assert coordinator2.release_draft_generation( claim, DraftGenerationError("invalid_response", "schema rejected", False), ) assert coordinator2.claim_draft_generation( "draft-001", owner2, "worker-b" ) is None def test_deleted_generation_history_cannot_fall_back_to_legacy_v2_ledgers( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) coordinator._generation_ledger_path().unlink() rejected = coordinator.approve_draft("draft-001", owner) assert rejected.accepted is False assert rejected.error == "draft_generation_missing" def test_edited_child_has_predecessor_linked_generation_and_parent_cannot_deliver( tmp_path: Path, ) -> None: coordinator, owner, created = _v2_draft(tmp_path) parent = coordinator.draft_generation("draft-001") assert parent is not None child = coordinator.edit_draft("draft-001", owner, "운영자가 수정한 초안입니다.") assert child.accepted child_generation = coordinator.draft_generation(child.draft_id) assert child_generation is not None assert child_generation.lineage_predecessor_digest == parent.record_digest assert coordinator.prepare_delivery("draft-001", owner).accepted is False def test_generation_persisted_shape_rejects_extra_keys_and_wrong_receipt_state( tmp_path: Path, ) -> None: coordinator, _owner, _created = _v2_draft(tmp_path) path = coordinator._generation_ledger_path() payload = json.loads(path.read_text(encoding="utf-8")) payload["draft-001"][-1]["extra"] = True path.write_text(json.dumps(payload), encoding="utf-8") assert coordinator.draft_generation_history("draft-001") == () def test_record_digest_and_draft_revision_pins_reject_independently( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) generation = coordinator.draft_generation("draft-001") assert generation is not None for overrides in ( {"expected_record_digest": "0" * 64}, {"expected_draft_revision": "1" * 64}, ): pins = _generation_pins(generation) pins.update(overrides) rejected = coordinator.approve_draft("draft-001", owner, **pins) assert rejected.accepted is False assert rejected.error == "draft_generation_stale" def test_newer_canonical_checkin_event_rejects_older_draft( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) generation = coordinator.draft_generation("draft-001") assert generation is not None coordinator._by_key["client_001"].bridge.finalized_event = ( lambda _session_id: SimpleNamespace(event_id="finalized-event-002") ) rejected = coordinator.approve_draft( "draft-001", owner, **_generation_pins(generation) ) assert rejected.accepted is False assert rejected.error == "draft_generation_stale" def test_delivery_intent_failure_refreshes_current_card_and_recovers_without_transport( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) generation = coordinator.draft_generation("draft-001") assert generation is not None assert coordinator.approve_draft( "draft-001", owner, **_generation_pins(generation) ).accepted approved = coordinator.draft_generation("draft-001") assert approved is not None real_write = coordinator._write_json_private failed_once = False def fail_intent(path: Path, payload: object) -> None: nonlocal failed_once if path == coordinator._delivery_ledger_path() and not failed_once: failed_once = True raise OSError("injected delivery intent failure") real_write(path, payload) coordinator._write_json_private = fail_intent failed = coordinator.prepare_delivery( "draft-001", owner, **_generation_pins(approved) ) assert failed.accepted is False assert failed.error == "draft_delivery_write_failed:OSError" coordinator._write_json_private = real_write refreshed = coordinator.draft("draft-001", owner) recovered = coordinator.prepare_delivery( "draft-001", owner, **_generation_pins(refreshed) ) assert recovered.accepted is True assert recovered.transport_required is True def test_approval_projection_recovers_after_injected_cross_ledger_failure( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) generation = coordinator.draft_generation("draft-001") assert generation is not None real_write = coordinator._write_json_private failed_once = False def fail_draft_projection(path: Path, payload: object) -> None: nonlocal failed_once if path == coordinator._drafts_path and not failed_once: failed_once = True raise OSError("injected draft projection failure") real_write(path, payload) coordinator._write_json_private = fail_draft_projection failed = coordinator.approve_draft( "draft-001", owner, **_generation_pins(generation) ) assert failed.accepted is False assert coordinator.draft_generation("draft-001").state is DraftGenerationState.APPROVED coordinator._write_json_private = real_write recovery_card = coordinator.draft("draft-001", owner) recovered = coordinator.approve_draft( "draft-001", owner, **_generation_pins(recovery_card) ) assert recovered.accepted is True assert recovered.status == "approved" def test_generation_rejects_timestamp_regression_terminal_retry_and_bad_receipt_matrix() -> None: generating = _generation_record(DraftGenerationState.GENERATING) with pytest.raises(DraftGenerationTransitionError, match="timestamp"): generating.transition( DraftGenerationState.DRAFT_CREATED, expected_generation=generating.generation, expected_record_digest=generating.record_digest, expected_checkin_event_id=generating.checkin_event_id, expected_checkin_revision=generating.checkin_revision, expected_draft_revision=None, actor="worker", authority_digest="c" * 64, occurred_at=generating.updated_at - timedelta(microseconds=1), draft_revision="d" * 64, generation_provider_receipt="e" * 64, ) terminal = generating.transition( DraftGenerationState.GENERATION_FAILED, expected_generation=generating.generation, expected_record_digest=generating.record_digest, expected_checkin_event_id=generating.checkin_event_id, expected_checkin_revision=generating.checkin_revision, expected_draft_revision=None, actor="worker", authority_digest="c" * 64, occurred_at=generating.updated_at, error=DraftGenerationError("invalid_response", "schema rejected", False), ) with pytest.raises(DraftGenerationTransitionError, match="not retryable"): _generation_transition(terminal, DraftGenerationState.GENERATION_PENDING) created = _generation_record(DraftGenerationState.DRAFT_CREATED) payload = created.as_dict() payload["delivery_provider_receipt"] = "telegram-early" payload["record_digest"] = "0" * 64 with pytest.raises(DraftGenerationTransitionError, match="premature"): DraftGenerationRecord.from_dict(payload) def test_generation_attempt_policy_is_exactly_two() -> None: failed = _generation_record(DraftGenerationState.GENERATION_FAILED) retry_pending = _generation_transition( failed, DraftGenerationState.GENERATION_PENDING ) assert retry_pending.attempt == 2 retry_generating = _generation_transition( retry_pending, DraftGenerationState.GENERATING ) exhausted = _generation_transition( retry_generating, DraftGenerationState.GENERATION_FAILED ) assert exhausted.attempt == exhausted.max_attempts == 2 with pytest.raises(DraftGenerationTransitionError, match="exhausted"): _generation_transition(exhausted, DraftGenerationState.GENERATION_PENDING) def test_edited_child_recovers_after_restart_from_generation_only_projection( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) real_write = coordinator._write_json_private failed_once = False def fail_child_request(path: Path, payload: object) -> None: nonlocal failed_once if path == coordinator._requests_path and not failed_once: failed_once = True raise OSError("injected child request projection failure") real_write(path, payload) coordinator._write_json_private = fail_child_request failed = coordinator.edit_draft( "draft-001", owner, "재시작 후 복구할 운영자 수정본입니다." ) assert failed.accepted is False generation_payload = json.loads( coordinator._generation_ledger_path().read_text(encoding="utf-8") ) child_ids = set(generation_payload) - {"draft-001"} assert len(child_ids) == 1 child_id = child_ids.pop() restarted, restarted_owner, _events = _draft_coordinator(tmp_path) recovered = restarted.edit_draft( "draft-001", restarted_owner, "재시작 후 복구할 운영자 수정본입니다." ) assert recovered.accepted is True assert recovered.draft_id == child_id parent = restarted.draft_generation("draft-001") child = restarted.draft_generation(child_id) assert parent is not None and child is not None assert child.lineage_predecessor_digest == parent.record_digest drafts = json.loads(restarted._drafts_path.read_text(encoding="utf-8")) assert drafts["draft-001"]["superseded_by_draft_id"] == child_id assert drafts[child_id]["parent_draft_id"] == "draft-001" def test_create_recovers_after_restart_from_generation_only_projection( tmp_path: Path, ) -> None: coordinator, owner, _events = _draft_coordinator(tmp_path) selection = coordinator.resolve_draft("draft-001", owner) assert selection is not None binding = judgment_revision_binding( "client_001", "session-001", selection.snapshot.model_dump() ) review = _coach_v2_review(binding) artifacts = _coach_v2_artifacts(binding, "재시작 복구 초안입니다.") real_write = coordinator._write_json_private failed_once = False def fail_draft_projection(path: Path, payload: object) -> None: nonlocal failed_once if path == coordinator._drafts_path and not failed_once: failed_once = True raise OSError("injected create projection failure") real_write(path, payload) coordinator._write_json_private = fail_draft_projection failed = coordinator.create_draft( "draft-001", owner, "재시작 복구 초안입니다.", expected_revision_binding_digest=binding, coach_review=review, coach_artifacts=artifacts, ) assert failed.accepted is False assert coordinator.draft_generation("draft-001") is not None restarted, restarted_owner, _events = _draft_coordinator(tmp_path) recovered = restarted.create_draft( "draft-001", restarted_owner, "재시작 복구 초안입니다.", expected_revision_binding_digest=binding, coach_review=review, coach_artifacts=artifacts, ) assert recovered.accepted is True assert recovered.status == "created" assert restarted.draft_generation("draft-001").state is DraftGenerationState.DRAFT_CREATED def test_worker_receipt_and_failure_outcomes_survive_restart(tmp_path: Path) -> None: retry_root = tmp_path / "retryable" retryable, retry_owner, _events = _draft_coordinator(retry_root) first = retryable.claim_draft_generation("draft-001", retry_owner, "worker-a") assert isinstance(first, DraftGenerationClaim) assert retryable.release_draft_generation( first, DraftGenerationError("provider_timeout", "timed out", True) ) retry_restart, retry_owner, _events = _draft_coordinator(retry_root) second = retry_restart.claim_draft_generation( "draft-001", retry_owner, "worker-b" ) assert isinstance(second, DraftGenerationClaim) assert retry_restart.complete_draft_generation(first, "0" * 64) is False second_restart, _owner, _events = _draft_coordinator(retry_root) assert second_restart.release_draft_generation( second, DraftGenerationError("invalid_response", "schema rejected", False) ) terminal_restart, terminal_owner, _events = _draft_coordinator(retry_root) assert terminal_restart.claim_draft_generation( "draft-001", terminal_owner, "worker-c" ) is None failed = terminal_restart.draft_generation("draft-001") assert failed is not None and failed.error is not None assert failed.error.code == "invalid_response" assert failed.error.retryable is False def test_reloaded_history_rejects_persisted_timestamp_regression( tmp_path: Path, ) -> None: coordinator, owner, _events = _draft_coordinator(tmp_path) assert coordinator.claim_draft_generation("draft-001", owner, "worker-a") path = coordinator._generation_ledger_path() payload = json.loads(path.read_text(encoding="utf-8")) history = payload["draft-001"] successor_time = datetime.fromisoformat(history[1]["updated_at"]) history[0]["updated_at"] = (successor_time + timedelta(seconds=1)).isoformat( timespec="microseconds" ) _rechain(history) path.write_text(json.dumps(payload), encoding="utf-8") restarted, _owner, _events = _draft_coordinator(tmp_path) assert restarted.draft_generation_history("draft-001") == () def test_global_lineage_rejects_existing_but_non_tip_parent_digest( tmp_path: Path, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) child = coordinator.edit_draft("draft-001", owner, "잘못된 계보 검증용 수정본입니다.") assert child.accepted path = coordinator._generation_ledger_path() payload = json.loads(path.read_text(encoding="utf-8")) parent_history = payload["draft-001"] child_history = payload[child.draft_id] child_history[0]["lineage_predecessor_digest"] = parent_history[0]["record_digest"] for record in child_history[1:]: record["lineage_predecessor_digest"] = parent_history[0]["record_digest"] _rechain(child_history) path.write_text(json.dumps(payload), encoding="utf-8") restarted, _owner, _events = _draft_coordinator(tmp_path) assert restarted.draft_generation_history(child.draft_id) == () @pytest.mark.parametrize("delete_ledger", [False, True]) def test_deleted_authority_marker_fails_closed_after_hard_restart_byte_exact( tmp_path: Path, delete_ledger: bool, ) -> None: coordinator, owner, created = _v2_draft(tmp_path) # Force legacy candidate state through its one durable migration. assert coordinator.draft_generation("draft-001") is not None coordinator._generation_authority_ledger_path().unlink() if delete_ledger: coordinator._generation_ledger_path().unlink() before = _tree_snapshot(tmp_path) restarted, restarted_owner, _events = _draft_coordinator(tmp_path) replay = restarted.create_draft( "draft-001", restarted_owner, created.text or "", expected_revision_binding_digest=( created.coach_review.revision_binding_digest ), coach_review=created.coach_review, coach_artifacts=_coach_v2_artifacts( created.coach_review.revision_binding_digest, created.text or "", ), ) assert replay.accepted is False assert _tree_snapshot(tmp_path) == before @pytest.mark.parametrize("damage", ("token", "empty", "digest")) def test_established_v2_authority_damage_fails_closed_without_reconstruction( tmp_path: Path, damage: str, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) journal = coordinator._generation_ledger_path() marker = coordinator._generation_authority_ledger_path() payload = json.loads(journal.read_text(encoding="utf-8")) if damage == "token": del payload["draft-001"] journal.write_text(json.dumps(payload), encoding="utf-8") elif damage == "empty": journal.write_text("{}", encoding="utf-8") else: authority = json.loads(marker.read_text(encoding="utf-8")) authority["source_digest"] = "0" * 64 marker.write_text(json.dumps(authority), encoding="utf-8") before = _tree_snapshot(tmp_path) assert coordinator.draft_generation_history("draft-001") == () assert coordinator.approve_draft("draft-001", owner).accepted is False assert _tree_snapshot(tmp_path) == before def test_generation_marker_digest_tracks_every_journal_write(tmp_path: Path) -> None: coordinator, owner, _created = _v2_draft(tmp_path) child = coordinator.edit_draft("draft-001", owner, "새 권위 자식입니다.") assert child.accepted journal_bytes = coordinator._generation_ledger_path().read_bytes() marker = json.loads( coordinator._generation_authority_ledger_path().read_text(encoding="utf-8") ) import hashlib assert marker["source_digest"] == hashlib.sha256(journal_bytes).hexdigest() def test_hold_is_authoritative_when_draft_projection_rolls_back(tmp_path: Path) -> None: coordinator, owner, _created = _v2_draft(tmp_path) held = coordinator.hold_draft("draft-001", owner) assert held.accepted drafts = json.loads(coordinator._drafts_path.read_text(encoding="utf-8")) drafts["draft-001"]["status"] = "created" drafts["draft-001"]["coach_review_state"] = "validated_model" coordinator._drafts_path.write_text(json.dumps(drafts), encoding="utf-8") current = coordinator.draft_generation("draft-001") assert current is not None and current.state is DraftGenerationState.HELD rejected = coordinator.approve_draft( "draft-001", owner, **_generation_pins(current) ) assert rejected.accepted is False def test_worker_claim_receipt_is_attempt_bound_and_release_conflicts_fail( tmp_path: Path, ) -> None: coordinator, owner, _events = _draft_coordinator(tmp_path) first = coordinator.claim_draft_generation("draft-001", owner, "worker-a") assert isinstance(first, DraftGenerationClaim) retryable = DraftGenerationError( "provider_unavailable", "provider unavailable", True ) assert coordinator.release_draft_generation(first, retryable) assert not coordinator.release_draft_generation(first, retryable) assert not coordinator.release_draft_generation( first, DraftGenerationError("invalid_response", "schema rejected", False) ) second = coordinator.claim_draft_generation("draft-001", owner, "worker-a") assert isinstance(second, DraftGenerationClaim) assert second.attempt == 2 assert not coordinator.complete_draft_generation( first, generation_provider_receipt="0" * 64, ) def test_generation_and_delivery_contract_versions_are_independent() -> None: record = _generation_record(DraftGenerationState.SENT_AUDITED) assert record.generation_provider_contract_version == "chat-completions-v1" assert record.delivery_transport_contract_version == "telegram-send-v1" assert record.generation_provider_contract_version != record.delivery_transport_contract_version def test_child_lineage_requires_declared_parent_tip(tmp_path: Path) -> None: coordinator, owner, _created = _v2_draft(tmp_path) child = coordinator.edit_draft("draft-001", owner, "계보 자식입니다.") assert child.accepted drafts = json.loads(coordinator._drafts_path.read_text(encoding="utf-8")) drafts[child.draft_id]["parent_draft_id"] = "wrong-parent" coordinator._drafts_path.write_text(json.dumps(drafts), encoding="utf-8") assert coordinator.draft_generation_history(child.draft_id) == () def test_authoritative_child_intent_blocks_parent_approval(tmp_path: Path) -> None: coordinator, owner, _created = _v2_draft(tmp_path) parent = coordinator.draft_generation("draft-001") assert parent is not None real_write = coordinator._write_json_private def fail_request(path: Path, payload: object) -> None: if path == coordinator._requests_path: raise OSError("injected request projection failure") real_write(path, payload) coordinator._write_json_private = fail_request failed = coordinator.edit_draft("draft-001", owner, "미완료 자식입니다.") assert failed.accepted is False coordinator._write_json_private = real_write rejected = coordinator.approve_draft( "draft-001", owner, **_generation_pins(parent) ) assert rejected.accepted is False @pytest.mark.asyncio async def test_real_callback_recovers_receipt_projection_after_restart_exactly_once( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) service_state_store = CustomerServiceStateStore( tmp_path / "data" / "owner-actions" / "customer-service-state.json" ) service_state_store.ensure() setattr(coordinator, "_service_state_store", service_state_store) setattr(coordinator, "_service_state_error", None) destination = IncomingAddress("client", "customer-chat", "customer-topic") resolved = coordinator._by_key["client_001"] setattr(resolved.customer.spec, "telegram", destination) setattr(coordinator, "_kst_date_provider", lambda: date(2026, 7, 19)) generation = coordinator.draft_generation("draft-001") assert generation is not None approved = coordinator.approve_draft( "draft-001", owner, **_generation_pins(generation) ) assert approved.accepted nonce = TelegramAdapter._nutrition_generation_card_nonce(approved) assert nonce is not None class Query: def __init__(self) -> None: self.answers: list[str | None] = [] self.edits: list[dict[str, object]] = [] async def answer(self, text: str | None = None) -> None: self.answers.append(text) async def edit_message_text(self, **kwargs: object) -> None: self.edits.append(dict(kwargs)) adapter = object.__new__(TelegramAdapter) adapter._adaptive_nutrition_config = SimpleNamespace(delivery_enabled=True) adapter._get_nutrition_coaching = lambda: coordinator adapter._nutrition_address = lambda *_args: owner adapter._nutrition_operator_actor = lambda *_args: owner transport_calls: list[str] = [] async def fake_transport(**_kwargs: object) -> object: claimed = next(iter(coordinator._read_deliveries().values())) assert claimed["status"] == "unknown_provider_outcome" assert claimed["provider_chat_id"] == destination.chat_id assert claimed["provider_topic_id"] == destination.topic_id transport_calls.append("sent") return SimpleNamespace(message_id="fake-message-101") adapter._send_nutrition_topic = fake_transport real_write = coordinator._write_json_private attempted_outbox_states: list[str] = [] failed_once = False def fail_transport_receipt_write(path: Path, payload: object) -> None: nonlocal failed_once if path == coordinator._delivery_ledger_path() and isinstance(payload, dict): states = { status for value in payload.values() if isinstance(value, dict) for status in (value.get("status"),) if isinstance(status, str) } assert len(states) == 1 state = next(iter(states)) attempted_outbox_states.append(state) if state == "sent" and not failed_once: failed_once = True raise OSError("injected transport receipt write failure") real_write(path, payload) monkeypatch.setattr( coordinator, "_write_json_private", fail_transport_receipt_write, ) first_query = Query() await adapter._handle_nutrition_draft_callback( first_query, f"n3:draft-001:s:{nonce}:65", SimpleNamespace(message_id=101), ) assert failed_once is True assert attempted_outbox_states == ["pending", "unknown_provider_outcome", "sent"] assert transport_calls == ["sent"] unknown_row = next(iter(coordinator._read_deliveries().values())) assert unknown_row["status"] == "unknown_provider_outcome" assert "message_id" not in unknown_row assert unknown_row["approved_revision"] == approved.generation_draft_revision assert unknown_row["checkin_revision"] == approved.generation_checkin_revision assert unknown_row["canonical_payload_digest"] == approved.generation_draft_revision assert first_query.edits reconcile_markup = first_query.edits[-1]["reply_markup"] reconcile_data = reconcile_markup.inline_keyboard[0][0].callback_data assert ":c:" in reconcile_data approval_events = resolved.bridge._events approval_event_count = len(approval_events) restarted, restarted_owner, _restart_events = _draft_coordinator(tmp_path) setattr(restarted, "_service_state_store", service_state_store) setattr(restarted, "_service_state_error", None) restarted_resolved = restarted._by_key["client_001"] setattr(restarted_resolved, "bridge", resolved.bridge) setattr(restarted_resolved.customer.spec, "telegram", destination) setattr(restarted, "_kst_date_provider", lambda: date(2026, 7, 19)) adapter._get_nutrition_coaching = lambda: restarted adapter._nutrition_address = lambda *_args: restarted_owner adapter._nutrition_operator_actor = lambda *_args: restarted_owner restarted_generation = restarted.draft_generation("draft-001") assert restarted_generation is not None assert restarted_generation.state is DraftGenerationState.DELIVERY_PENDING unknown = restarted.prepare_delivery( "draft-001", restarted_owner, **_generation_pins(restarted_generation), ) assert unknown.accepted and unknown.status == "unknown_provider_outcome" assert unknown.transport_required is False second_query = Query() await adapter._handle_nutrition_draft_callback( second_query, reconcile_data, SimpleNamespace(message_id=101), ) assert second_query.answers[-1] == "전달 기록을 아직 복구하지 못했습니다." assert transport_calls == ["sent"] still_unknown = next(iter(restarted._read_deliveries().values())) assert still_unknown == unknown_row idempotency_key = still_unknown.get("idempotency_key") provider_chat_id = still_unknown.get("provider_chat_id") provider_topic_id = still_unknown.get("provider_topic_id") payload_digest = still_unknown.get("canonical_payload_digest") assert isinstance(idempotency_key, str) assert isinstance(provider_chat_id, str) assert isinstance(provider_topic_id, str) assert isinstance(payload_digest, str) evidence = DeliveryReconciliationEvidence( outcome="sent", idempotency_key=idempotency_key, provider_chat_id=provider_chat_id, provider_topic_id=provider_topic_id, canonical_payload_digest=payload_digest, provider_message_id="fake-message-101", ) rejected_evidence = restarted.reconcile_delivery( "draft-001", restarted_owner, replace(evidence, provider_chat_id="forged-chat"), **_generation_pins(unknown), ) assert rejected_evidence.accepted is False assert rejected_evidence.error == "delivery_reconciliation_evidence_invalid" assert next(iter(restarted._read_deliveries().values())) == unknown_row assert transport_calls == ["sent"] reconciled = restarted.reconcile_delivery( "draft-001", restarted_owner, evidence, **_generation_pins(unknown), ) assert reconciled.accepted and reconciled.status == "sent_audited" assert reconciled.message_id == "fake-message-101" final = restarted.draft_generation("draft-001") assert final is not None and final.state is DraftGenerationState.SENT_AUDITED assert final.delivery_provider_receipt == "fake-message-101" audited_row = next(iter(restarted._read_deliveries().values())) assert audited_row == { **unknown_row, "status": "sent_audited", "message_id": "fake-message-101", } replay = restarted.reconcile_delivery( "draft-001", restarted_owner, evidence, **_generation_pins(reconciled), ) assert replay.accepted and replay.status == "sent_audited" assert transport_calls == ["sent"] assert len(approval_events) == approval_event_count + 1 def test_generation_root_rejects_forged_rechained_journal( tmp_path: Path, ) -> None: coordinator, _owner, _created = _v2_draft(tmp_path) journal_path = coordinator._generation_ledger_path() marker_path = coordinator._generation_authority_ledger_path() root_path = coordinator._generation_authority_root_path() root_before = root_path.read_bytes() payload = json.loads(journal_path.read_text(encoding="utf-8")) history = payload["draft-001"] history[-1]["actor"] = "forged-actor" history[-1]["authority_digest"] = "f" * 64 _rechain(history) journal_bytes = coordinator._private_json_bytes(payload) journal_path.write_bytes(journal_bytes) authority = json.loads(marker_path.read_text(encoding="utf-8")) authority["source_digest"] = hashlib.sha256(journal_bytes).hexdigest() marker_path.write_bytes( coordinator._private_json_bytes(authority) ) assert root_path.read_bytes() == root_before before = _tree_snapshot(tmp_path) restarted, _owner, _events = _draft_coordinator(tmp_path) assert restarted.draft_generation_history("draft-001") == () assert restarted.draft_generation("draft-001") is None assert _tree_snapshot(tmp_path) == before def test_deleted_complete_generation_authority_cannot_reconstruct_from_projection( tmp_path: Path, ) -> None: coordinator, _owner, created = _v2_draft(tmp_path) coordinator._generation_authority_root_path().unlink() coordinator._generation_authority_ledger_path().unlink() coordinator._generation_ledger_path().unlink() before = _tree_snapshot(tmp_path) restarted, restarted_owner, _events = _draft_coordinator(tmp_path) replay = restarted.create_draft( "draft-001", restarted_owner, created.text or "", expected_revision_binding_digest=( created.coach_review.revision_binding_digest ), coach_review=created.coach_review, coach_artifacts=_coach_v2_artifacts( created.coach_review.revision_binding_digest, created.text or "", ), ) assert replay.accepted is False assert restarted.draft_generation("draft-001") is None assert _tree_snapshot(tmp_path) == before @pytest.mark.asyncio async def test_delivery_restart_after_transport_before_receipt_does_not_resend( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: coordinator, owner, _created = _v2_draft(tmp_path) service_state_store = CustomerServiceStateStore( tmp_path / "data" / "owner-actions" / "customer-service-state.json" ) service_state_store.ensure() setattr(coordinator, "_service_state_store", service_state_store) setattr(coordinator, "_service_state_error", None) destination = IncomingAddress("client", "customer-chat", "customer-topic") resolved = coordinator._by_key["client_001"] setattr(resolved.customer.spec, "telegram", destination) setattr(coordinator, "_kst_date_provider", lambda: date(2026, 7, 19)) generation = coordinator.draft_generation("draft-001") assert generation is not None approved = coordinator.approve_draft( "draft-001", owner, **_generation_pins(generation) ) assert approved.accepted nonce = TelegramAdapter._nutrition_generation_card_nonce(approved) assert nonce is not None class Query: def __init__(self) -> None: self.answers: list[str | None] = [] self.edits: list[dict[str, object]] = [] async def answer(self, text: str | None = None) -> None: self.answers.append(text) async def edit_message_text(self, **kwargs: object) -> None: self.edits.append(dict(kwargs)) adapter = object.__new__(TelegramAdapter) adapter._adaptive_nutrition_config = SimpleNamespace(delivery_enabled=True) adapter._get_nutrition_coaching = lambda: coordinator adapter._nutrition_address = lambda *_args: owner adapter._nutrition_operator_actor = lambda *_args: owner transport_calls: list[str] = [] async def fake_transport(**_kwargs: object) -> object: claimed = next(iter(coordinator._read_deliveries().values())) assert claimed["status"] == "unknown_provider_outcome" assert claimed["provider_chat_id"] == destination.chat_id assert claimed["provider_topic_id"] == destination.topic_id transport_calls.append("sent") return SimpleNamespace(message_id="fake-message-hard-crash") adapter._send_nutrition_topic = fake_transport def crash_before_receipt( *_args: object, **_kwargs: object, ) -> DraftAction: raise OSError("injected post-send pre-receipt crash") monkeypatch.setattr( coordinator, "record_delivery_transport_receipt", crash_before_receipt, ) callback_data = f"n3:draft-001:s:{nonce}:65" first_query = Query() with pytest.raises(OSError, match="injected post-send pre-receipt crash"): await adapter._handle_nutrition_draft_callback( first_query, callback_data, SimpleNamespace(message_id=101), ) assert transport_calls == ["sent"] unknown_row = next(iter(coordinator._read_deliveries().values())) assert unknown_row["status"] == "unknown_provider_outcome" assert "message_id" not in unknown_row assert unknown_row["approved_revision"] == approved.generation_draft_revision assert unknown_row["checkin_revision"] == approved.generation_checkin_revision assert unknown_row["canonical_payload_digest"] == approved.generation_draft_revision approval_events = resolved.bridge._events approval_event_count = len(approval_events) restarted, restarted_owner, _restart_events = _draft_coordinator(tmp_path) setattr(restarted, "_service_state_store", service_state_store) setattr(restarted, "_service_state_error", None) restarted_resolved = restarted._by_key["client_001"] setattr(restarted_resolved, "bridge", resolved.bridge) setattr(restarted_resolved.customer.spec, "telegram", destination) setattr(restarted, "_kst_date_provider", lambda: date(2026, 7, 19)) adapter._get_nutrition_coaching = lambda: restarted adapter._nutrition_address = lambda *_args: restarted_owner adapter._nutrition_operator_actor = lambda *_args: restarted_owner restarted_generation = restarted.draft_generation("draft-001") assert restarted_generation is not None assert restarted_generation.state is DraftGenerationState.DELIVERY_PENDING unknown = restarted.prepare_delivery( "draft-001", restarted_owner, **_generation_pins(restarted_generation), ) assert unknown.accepted and unknown.status == "unknown_provider_outcome" assert unknown.transport_required is False replay_query = Query() await adapter._handle_nutrition_draft_callback( replay_query, callback_data, SimpleNamespace(message_id=101), ) assert replay_query.edits assert transport_calls == ["sent"] still_unknown = next(iter(restarted._read_deliveries().values())) assert still_unknown == unknown_row idempotency_key = still_unknown.get("idempotency_key") provider_chat_id = still_unknown.get("provider_chat_id") provider_topic_id = still_unknown.get("provider_topic_id") payload_digest = still_unknown.get("canonical_payload_digest") assert isinstance(idempotency_key, str) assert isinstance(provider_chat_id, str) assert isinstance(provider_topic_id, str) assert isinstance(payload_digest, str) evidence = DeliveryReconciliationEvidence( outcome="sent", idempotency_key=idempotency_key, provider_chat_id=provider_chat_id, provider_topic_id=provider_topic_id, canonical_payload_digest=payload_digest, provider_message_id="fake-message-hard-crash", ) rejected_evidence = restarted.reconcile_delivery( "draft-001", restarted_owner, replace(evidence, canonical_payload_digest="0" * 64), **_generation_pins(unknown), ) assert rejected_evidence.accepted is False assert rejected_evidence.error == "delivery_reconciliation_evidence_invalid" assert next(iter(restarted._read_deliveries().values())) == unknown_row assert transport_calls == ["sent"] reconciled = restarted.reconcile_delivery( "draft-001", restarted_owner, evidence, **_generation_pins(unknown), ) assert reconciled.accepted and reconciled.status == "sent_audited" assert reconciled.message_id == "fake-message-hard-crash" final = restarted.draft_generation("draft-001") assert final is not None and final.state is DraftGenerationState.SENT_AUDITED assert final.delivery_provider_receipt == "fake-message-hard-crash" audited_row = next(iter(restarted._read_deliveries().values())) assert audited_row == { **unknown_row, "status": "sent_audited", "message_id": "fake-message-hard-crash", } replay = restarted.reconcile_delivery( "draft-001", restarted_owner, evidence, **_generation_pins(reconciled), ) assert replay.accepted and replay.status == "sent_audited" assert transport_calls == ["sent"] assert len(approval_events) == approval_event_count + 1 def test_adaptive_callback_has_no_inline_orchestration_calls() -> None: tree = ast.parse( textwrap.dedent( inspect.getsource(TelegramAdapter._handle_adaptive_review_callback) ) ) called_attributes = { node.func.attr for node in ast.walk(tree) if isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute) } awaited_delivery_results = [ node for node in ast.walk(tree) if isinstance(node, ast.Await) and isinstance(node.value, ast.Subscript) and isinstance(node.value.slice, ast.Constant) and node.value.slice.value == "delivery" ] assert "_adaptive_coach_card_text" not in called_attributes assert awaited_delivery_results == []