"""Durable finite model selection and privacy tests.""" from __future__ import annotations import json import stat import sys from collections.abc import Callable, Mapping from dataclasses import dataclass, replace from datetime import datetime from pathlib import Path from typing import TypeAlias import pytest PROFILE_PACKAGE = Path(__file__).resolve().parents[2] / "dualcoach" / "profile" if str(PROFILE_PACKAGE) not in sys.path: sys.path.insert(0, str(PROFILE_PACKAGE)) from checkin_cli.weekly_operations_grounding import FALLBACK_TOKEN, GroundedWeeklyReviewSource, deterministic_weekly_fallback from checkin_cli.weekly_operations_owner_binding import BoundWeeklySummaryForOwnerDraft from tests.gateway._weekly_operations_authority_support import ( test_registry_identity_binding_digest as registry_identity_binding_digest, test_registry_identity_document as registry_identity_document, ) from gateway.platforms.nutrition_weekly_operations_authority import WeeklyOperationsRuntimeContext, parse_weekly_operations_authority from gateway.platforms.nutrition_weekly_operations_config import parse_weekly_operations_config from gateway.platforms.nutrition_weekly_owner_contract import WeeklyLifecycleResult from gateway.platforms.nutrition_weekly_owner_draft import WeeklyOwnerDraftRequest, WeeklyOwnerDraftService, weekly_generation_base from gateway.platforms.nutrition_weekly_owner_ledger import WeeklyGenerationLedger, WeeklyGenerationOutcome, WeeklyGenerationState from gateway.platforms.nutrition_weekly_owner_model import JsonValue, weekly_model_request, weekly_model_request_digest from gateway.platforms.nutrition_weekly_owner_storage import WeeklyOwnerStorageAuthority, bind_weekly_owner_storage from checkin_cli.weekly_operations_knowledge import load_shipped_weekly_public_knowledge from tests.gateway._weekly_owner_summary_support import bound_summary_at ModelOutput: TypeAlias = Mapping[str, JsonValue] ModelFactory: TypeAlias = Callable[[Mapping[str, JsonValue]], ModelOutput] _MALFORMED: ModelOutput = {"malformed": True} _KOREAN_CONTRADICTION: ModelOutput = {"explanation": "상태는 incomplete이며 자동 발송합니다."} _ENGLISH_CONTRADICTION: ModelOutput = {"explanation": "status changed and automatic delivery"} @dataclass(frozen=True, slots=True) class ModelResponse: value: ModelOutput def resolve(self, request: Mapping[str, JsonValue]) -> ModelOutput: del request return self.value @dataclass(frozen=True, slots=True) class ModelGeneratedResponse: factory: ModelFactory def resolve(self, request: Mapping[str, JsonValue]) -> ModelOutput: return self.factory(request) @dataclass(frozen=True, slots=True) class ModelRaised: error: BaseException def resolve(self, request: Mapping[str, JsonValue]) -> ModelOutput: del request raise self.error ModelOutcome: TypeAlias = ModelResponse | ModelGeneratedResponse | ModelRaised class FakeLifecycle: def __init__(self, root: Path) -> None: self.root: Path = root self.storage: WeeklyOwnerStorageAuthority = bind_weekly_owner_storage(root) self.bound_summary: BoundWeeklySummaryForOwnerDraft | None = None self.sources: list[GroundedWeeklyReviewSource] = [] self.results: dict[str, WeeklyLifecycleResult] = {} def weekly_owner_key(self) -> tuple[str, str, str]: return "coach", "owner-dm", "owner" def weekly_owner_storage_authority(self): return self.storage def persist_grounded_weekly_review( self, customer_key: str, owner_key: tuple[str, str, str], source: GroundedWeeklyReviewSource, ) -> WeeklyLifecycleResult: assert customer_key == "client_001" and owner_key == self.weekly_owner_key() and source.verify() self.sources.append(source) result = WeeklyLifecycleResult(True, "weekly-1", source.render_customer_body()) self.results["weekly-1"] = result return result def grounded_weekly_review_result( self, draft_id: str, owner_key: tuple[str, str, str], ) -> WeeklyLifecycleResult: assert owner_key == self.weekly_owner_key() return self.results.get(draft_id, WeeklyLifecycleResult(False, error="missing")) def owner_draft_request(lifecycle: FakeLifecycle): config = parse_weekly_operations_config({"nutrition_coaching": { "operator_review": {"user_id": "operator", "chat_id": "review", "topic_id": 59}, "weekly_operations": {"enabled": True, "reminder_time": "20:00:00", "missed_cutoff_time": "23:00:00", "weekly_weekday": 0, "feature_epoch": "weekly-operations-v1", "registry_identity_binding_digest": registry_identity_binding_digest()}, }}) runtime = WeeklyOperationsRuntimeContext( "a" * 64, "client_001", "coach", "owner-dm", 7, "b" * 64, True, "weekly-operations-v1", datetime.fromisoformat("2026-08-24T08:00:00+09:00"), ) receipt = parse_weekly_operations_authority({ "schema": "nutricoach-weekly-operations-authority-v2", "candidate_digest": runtime.candidate_digest, "config_digest": config.digest, "enabled_customer_keys": [runtime.customer_key], "owner": {"user_id": "coach", "chat_id": "owner-dm", "version": 7}, "consent_digest": runtime.consent_digest, "registry_identity": registry_identity_document(), "issued_at": "2026-08-24T00:00:00+09:00", "expires_at": "2026-08-25T00:00:00+09:00", "feature_epoch": runtime.feature_epoch, }) if lifecycle.bound_summary is None: lifecycle.bound_summary = bound_summary_at(lifecycle.root / "summary-authority") bound = lifecycle.bound_summary return WeeklyOwnerDraftRequest( "client_001", lifecycle.weekly_owner_key(), config, receipt, runtime, bound, lifecycle.storage.binding_digest, lambda: bound, ) def _valid(request: Mapping[str, JsonValue]) -> ModelOutput: return {"grounding_digest": request["grounding_digest"], "principle_ids": ["recorded_trend"], "emphasis": "summary_first"} class SequenceModel: def __init__(self, outcomes: tuple[ModelOutcome, ...]) -> None: self.outcomes: list[ModelOutcome] = list(outcomes) self.calls: int = 0 self.requests: list[Mapping[str, JsonValue]] = [] def generate(self, request: Mapping[str, JsonValue]) -> ModelOutput: self.calls += 1 self.requests.append(request) outcome = self.outcomes.pop(0) return outcome.resolve(request) def test_valid_replay_has_one_total_provider_call_and_private_ledger(tmp_path: Path) -> None: lifecycle = FakeLifecycle(tmp_path) model = SequenceModel((ModelGeneratedResponse(_valid),)) service = WeeklyOwnerDraftService(lifecycle, model) request = owner_draft_request(lifecycle) first = service.create(request) replay = service.create(request) ledger = tmp_path / "weekly-draft-generations.jsonl" assert (first.accepted, replay.provider_calls, model.calls) == (True, 0, 1) assert stat.S_IMODE(ledger.stat().st_mode) == 0o600 assert len(ledger.read_text().splitlines()) == 3 def test_invalid_first_uses_one_correction_and_never_a_third_call(tmp_path: Path) -> None: lifecycle = FakeLifecycle(tmp_path) invalid: ModelOutput = {"grounding_digest": "0" * 64, "principle_ids": ["unknown"], "emphasis": "other", "extra": True} model = SequenceModel((ModelResponse(invalid), ModelGeneratedResponse(_valid))) result = WeeklyOwnerDraftService(lifecycle, model).create(owner_draft_request(lifecycle)) replay = WeeklyOwnerDraftService(lifecycle, model).create(owner_draft_request(lifecycle)) assert (result.provider_calls, replay.provider_calls, model.calls) == (2, 0, 2) assert model.requests[0]["correction"] is False assert model.requests[1]["correction"] is True assert model.requests[0]["locked_facts"] == model.requests[1]["locked_facts"] @pytest.mark.parametrize( "outcome", (TimeoutError(), OSError(), _MALFORMED, _KOREAN_CONTRADICTION, _ENGLISH_CONTRADICTION), ) def test_outage_and_two_invalid_results_use_deterministic_fallback( tmp_path: Path, outcome: ModelOutput | BaseException, ) -> None: lifecycle = FakeLifecycle(tmp_path) outcomes = ( (ModelRaised(outcome),) if isinstance(outcome, BaseException) else (ModelResponse(outcome), ModelResponse(outcome)) ) model = SequenceModel(outcomes) result = WeeklyOwnerDraftService(lifecycle, model).create(owner_draft_request(lifecycle)) assert result.fallback_used is True assert result.provider_calls <= 2 assert result.draft is not None and result.draft.text is not None assert FALLBACK_TOKEN in result.draft.text assert result.draft.text == deterministic_weekly_fallback(owner_draft_request(lifecycle).facts).render_customer_body() def test_provider_request_ledger_source_and_draft_exclude_all_sentinels(tmp_path: Path) -> None: lifecycle = FakeLifecycle(tmp_path) model = SequenceModel((ModelGeneratedResponse(_valid),)) sentinels = ( "RAW-MEAL", "RAW-SYMPTOM", "RAW-FREE-TEXT", "RAW-HEALTH", "CUSTOMER-NAME", "CUSTOMER-KEY", "TELEGRAM-USER", "TELEGRAM-CHAT", "WEIGHT-RAW", "CALORIE-RAW", "DIGESTION-RAW", ) result = WeeklyOwnerDraftService(lifecycle, model).create(owner_draft_request(lifecycle)) assert result.draft is not None and result.draft.text is not None surfaces = json.dumps({ "provider": model.requests, "ledger": (tmp_path / "weekly-draft-generations.jsonl").read_text(), "source": repr(lifecycle.sources), "draft": result.draft.text, }, ensure_ascii=False) assert all(sentinel not in surfaces for sentinel in sentinels) output_fields = model.requests[0]["output_fields"] assert isinstance(output_fields, list) assert all(isinstance(item, str) for item in output_fields) assert set(output_fields) == {"grounding_digest", "principle_ids", "emphasis"} assert "explanation" not in surfaces def test_interrupted_sending_is_terminal_fallback_without_provider_retry(tmp_path: Path) -> None: lifecycle = FakeLifecycle(tmp_path) request = owner_draft_request(lifecycle) base = weekly_generation_base(request) model_request = weekly_model_request( request.facts, load_shipped_weekly_public_knowledge(), correction=False, ) ledger = WeeklyGenerationLedger(lifecycle.storage) with ledger.locked() as transaction: _ = transaction.append( base, state=WeeklyGenerationState.SENDING, outcome=WeeklyGenerationOutcome.PROVIDER_AUTHORIZED, ordinal=1, request_digest=weekly_model_request_digest(model_request), ) model = SequenceModel(()) result = WeeklyOwnerDraftService(lifecycle, model).create(request) assert (result.accepted, result.provider_calls, result.fallback_used, model.calls) == (True, 0, True, 0) def test_tampered_source_and_ledger_corruption_fail_closed(tmp_path: Path) -> None: lifecycle = FakeLifecycle(tmp_path) service = WeeklyOwnerDraftService(lifecycle, SequenceModel((ModelGeneratedResponse(_valid),))) result = service.create(owner_draft_request(lifecycle)) assert result.accepted and lifecycle.sources assert replace(lifecycle.sources[0], source_digest="f" * 64).verify() is False ledger = tmp_path / "weekly-draft-generations.jsonl" _ = ledger.write_text(ledger.read_text().replace("draft_bound", "draft_b0und"), encoding="utf-8") with pytest.raises(RuntimeError): _ = service.create(owner_draft_request(lifecycle))