from __future__ import annotations import json import shutil import threading from dataclasses import replace from datetime import date, time, timedelta from pathlib import Path import pytest import checkin_cli.customer_admin as customer_admin_module from checkin_cli.adaptive_nutrition import ( digest, feature_config_digest, initialize_adaptive_customer, load_approved_adaptive_artifacts, ) from checkin_cli.customer_admin import ( CustomerAdminError, CustomerDraft, activate_customer, approve_adaptive_registration_inputs, approve_dual_coach_risk_policy, audit_gate_d_preflight, load_approved_adaptive_registration_inputs, load_runtime_customer_registry, main, prepare_adaptive_nutrition_runtime, reconcile_adaptive_nutrition_journals, register_customer, set_customer_ai_consent, set_customer_enabled, update_adaptive_registration_inputs, validate_committed_activation, ) from checkin_cli.customer_coaching import ( CONSENT_VERSION, AiProcessingConsent, TelegramAddress, ) from checkin_cli.customer_schedule import initialize_schedule_delivery_fence from checkin_cli.nutrition_onboarding_authority import ( validate_current_registry_authority, ) from checkin_cli.nutrition_onboarding_contract import OnboardingAuthority from checkin_cli.nutrition_onboarding_migration import ( commit_legacy_migration, legacy_migration_preflight, ) from checkin_cli.store import CanonicalEventTransaction def _record_fixture(tmp_path: Path) -> tuple[Path, Path, Path]: profile_root = tmp_path / "profile" registry_path = profile_root / "customers" / "registry.json" registry_path.parent.mkdir(parents=True) _empty_registry(registry_path) register_customer( registry_path, CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ), ) data_root = profile_root / "data" / "customers" / "client_001" data_root.mkdir(parents=True) return profile_root, registry_path, data_root def _empty_registry(path: Path) -> None: path.write_text( json.dumps({ "version": 1, "owner": {"user_id": "1", "chat_id": "-100", "topic_id": "10"}, "customers": [], }), encoding="utf-8", ) def _write_readiness_document(path: Path, document: dict[str, object]) -> dict[str, object]: payload = dict(document) payload["digest"] = digest(payload) path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) path.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") path.chmod(0o600) return payload def _write_activation_readiness(profile_root: Path, customer_key: str) -> None: input_reconciliation_digest = "9" * 64 kb = _write_readiness_document( profile_root / "data/global/nutrition-safety/restriction-kb-v1.json", { "schema_version": "1.0", "knowledge_version": "2026-08-01", "effective_at_kst": "2026-08-01T00:00:00+09:00", "reviewed_at_kst": "2026-08-01T00:00:00+09:00", "reviewed_by": "owner-receipt", "approved": True, "sources": [{"source_id": "source-1", "title": "approved source"}], "allergens": [{"rule_id": "allergen-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]}], "intolerances": [{"rule_id": "intolerance-1", "action": "require_human_review", "severity": "medium", "applicability": "all", "source_ids": ["source-1"]}], "religious_ethical_exclusions": [ {"rule_id": "ethical-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]} ], "medication_condition_rules": [ {"rule_id": "medical-1", "action": "require_human_review", "severity": "high", "applicability": "all", "source_ids": ["source-1"]} ], "hard_contraindications": [ {"rule_id": "contraindication-1", "action": "require_human_review", "severity": "critical", "applicability": "all", "source_ids": ["source-1"]} ], "cross_contact_rules": [{"rule_id": "contact-1", "action": "exclude", "severity": "high", "applicability": "all", "source_ids": ["source-1"]}], "substitution_rules": [{"rule_id": "substitution-1", "action": "inform", "severity": "low", "applicability": "all", "source_ids": ["source-1"]}], }, ) baseline = _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/baseline-v1.json", { "schema_version": "2.0", "customer_key": customer_key, "adult_age": 30, "equation_sex_basis": "female", "height_cm": 165.0, "weight_kg": 65.0, "activity_category": "moderate", "activity_rationale": "reviewed", "goal_type": "maintain", "target_weight_kg": 65.0, "target_date": "2026-10-01", "maintenance_intent": True, "dietary_preferences": [], "disliked_foods": [], "allergies": [], "intolerances": [], "religious_ethical_exclusions": [], "conditions": [], "medications": [], "pregnancy_breastfeeding": False, "cooking_access": "home", "budget_band": "standard", "meal_count": 3, "schedule_constraints": [], "customer_attested_at_kst": "2026-08-01T01:00:00+09:00", "owner_review_receipt": "owner-review", "input_reconciliation_digest": input_reconciliation_digest, }, ) reconciliation = _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/restriction-reconciliation-v1.json", { "schema_version": "1.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "resolution_status": "resolved", "unresolved_hard_contraindications": [], "reviewed_by": "owner-review", "reviewed_at_kst": "2026-08-01T01:30:00+09:00", }, ) calculation = _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/initial-plan-v1.json", { "schema_version": "1.0", "method_id": "mifflin_st_jeor_1990", "method_version": "1.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "normalized_inputs_digest": "a" * 64, "bmr_kcal": 1400, "tdee_kcal": 2100, "minimum_calories_kcal": 1200, "maximum_calories_kcal": 3000, "activity_assumption": "moderate", "goal_trajectory": "maintain", "goal_trajectory_safe": True, "safe_rate_guardrail": "reviewed", "weeks": [ { "week": week, "calories_kcal": 2100, "protein_g": 120, "carbohydrate_g": 250, "fat_g": 70, } for week in range(1, 13) ], "reviewed_by": "owner-review", "reviewed_at_kst": "2026-08-01T02:00:00+09:00", "review_decision": "approved", }, ) policy = _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/adjustment-policy-v1.json", { "schema_version": "1.0", "policy_version": "1.0", "effective_at_kst": "2026-08-01T00:00:00+09:00", "effective_from": "2026-08-01", "goal_mode": "maintain", "approved": True, "minimum_observation_days": 7, "minimum_current_samples": 4, "minimum_total_samples": 10, "minimum_adherent_days": 5, "weight_trend_method": "two_non_overlapping_seven_day_means", "adherence_inputs": ["calories", "meal_plan"], "actual_intake_inputs": ["calories"], "activity_change_inputs": ["training"], "adjustment_thresholds": { "minimum_current_samples": 4, "minimum_total_samples": 10, "minimum_adherent_days": 5, "goal_mode": "maintain", }, "maximum_step_kcal": 100, "calorie_step": 100, "calorie_floor": 1500, "calorie_ceiling": 4500, "cooldown_days": 7, "desired_weekly_change_min": "-0.25", "desired_weekly_change_max": "0.25", "safety_hold": False, "escalation_rules": ["contradictory_adherence", "safety_hold"], "safety_stop_rules": ["medical_review", "eating_disorder_risk"], "manual_override_receipt_required": True, "reviewed_by": "owner-review", }, ) receipt = _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/readiness-receipt-v1.json", { "schema_version": "2.0", "baseline_digest": baseline["digest"], "restriction_kb_digest": kb["digest"], "restriction_reconciliation_digest": reconciliation["digest"], "calculation_digest": calculation["digest"], "adjustment_policy_digest": policy["digest"], "input_reconciliation_digest": input_reconciliation_digest, "owner_review_receipt": "owner-review", "privacy_consent_version": CONSENT_VERSION, "privacy_consent_digest": "b" * 64, "feature_epoch_digest": "c" * 64, "delivery_enabled": False, "activation_enabled": False, "issued_at_kst": "2026-08-01T03:00:00+09:00", "expires_at_kst": "2099-08-01T03:00:00+09:00", }, ) readiness_digests = { "restriction_kb": kb["digest"], "baseline": baseline["digest"], "restriction_reconciliation": reconciliation["digest"], "calculation": calculation["digest"], "adjustment_policy": policy["digest"], "receipt": receipt["digest"], "input_reconciliation": input_reconciliation_digest, } _write_readiness_document( profile_root / f"data/customers/{customer_key}/nutrition-onboarding/readiness-current.json", { "schema_version": "nutrition_readiness_pointer_v1", "revision": 1, "bundle_digest": digest(readiness_digests), "readiness_receipt_digest": receipt["digest"], "input_reconciliation_digest": input_reconciliation_digest, }, ) def _activation_fixture( tmp_path: Path, *, grant_consent: bool = True, owner_matches_customer: bool = False, nutrition_readiness: bool = True, ) -> tuple[Path, Path, Path, Path]: profile_root = tmp_path / "profile" registry_path = profile_root / "customers" / "registry.json" registry_path.parent.mkdir(parents=True) _empty_registry(registry_path) register_customer( registry_path, CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ), ) if grant_consent: set_customer_ai_consent( registry_path, "client_001", AiProcessingConsent( granted=True, recorded_on=date(2026, 8, 1), notice_version=CONSENT_VERSION, ), ) if owner_matches_customer: payload = json.loads(registry_path.read_text(encoding="utf-8")) payload["owner"] = payload["customers"][0]["telegram"] registry_path.write_text(json.dumps(payload), encoding="utf-8") data_root = profile_root / "data" / "customers" / "client_001" data_root.mkdir(parents=True) if nutrition_readiness: _write_activation_readiness(profile_root, "client_001") checklist_path = tmp_path / "checklist.json" checklist_path.write_text( json.dumps( { "checklist": { "token_rotated": True, "missend_test_passed": True, "provider_terms_checked": { "checked": True, "version": CONSENT_VERSION, }, "withdrawal_deletion_doc": True, "retention_backup_doc": True, "manual_fallback_doc": True, } } ), encoding="utf-8", ) return profile_root, registry_path, data_root, checklist_path def _write_approved_adaptive_inputs(data_root: Path) -> None: root = data_root / "nutrition-plans" policy = { "starts_on": "2026-08-01", "goal_mode": "fat_loss", "weekly_rate_min": "-0.5", "weekly_rate_max": "-0.25", "calorie_step": 100, "calorie_floor": 1800, "calorie_ceiling": 2600, } constraints = { "meal_count": 1, "budget_tier": "standard", "cooking_access": "home", } catalog = [{ "food_id": "safe", "label": "safe", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] for name, key, value in ( ("policy.json", "policy", policy), ("meal-constraints.json", "meal_constraints", constraints), ("food-catalog.json", "catalog", catalog), ): (root / name).write_text( json.dumps({ "schema_version": "1.0", "version": "v1", "digest": digest(value), "approved": True, "approved_by": "1", "approved_at_kst": "2026-08-01T09:00:00+09:00", key: value, }), encoding="utf-8", ) (root / name).chmod(0o600) def _adaptive_fixture(tmp_path: Path) -> tuple[Path, Path, Path]: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_approved_adaptive_inputs(data_root) return profile_root, registry_path, data_root def test_risk_policy_writer_uses_current_owner_and_private_artifact(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1) ) document = approve_dual_coach_risk_policy( profile_root, "client_001", version="1", owner_actor=TelegramAddress(user_id="1", chat_id="-100", topic_id="10"), approved_at_kst="2026-08-01T09:00:00+09:00", ) path = data_root / "nutrition-plans" / "dual-coach-risk-policy.json" assert document["approved_by"] == {"user_id": "1", "chat_id": "-100", "topic_id": "10"} assert document["policy_digest"] == digest(document["policy"]) assert path.stat().st_mode & 0o777 == 0o600 with pytest.raises(CustomerAdminError, match="owner actor"): approve_dual_coach_risk_policy( profile_root, "client_001", version="2", owner_actor=TelegramAddress(user_id="wrong", chat_id="-100", topic_id="10"), ) assert document == json.loads(path.read_text(encoding="utf-8")) def _write_owner_approved_registration_artifacts(data_root: Path) -> None: root = data_root / "nutrition-plans" owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} catalog = [{ "food_id": "safe", "label": "safe", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] for filename, key, value in ( ("food-catalog.json", "catalog", catalog), ): document = { "schema_version": "1.0", "version": "v1", "digest": digest(value), "approved": True, "approved_by": owner, "approved_at_kst": "2026-08-01T09:00:00+09:00", key: value, } path = root / filename path.write_text(json.dumps(document), encoding="utf-8") path.chmod(0o600) def _registration_inputs(*, meal_count: int = 3, version: str = "v1") -> dict[str, object]: return { "version": version, "meal_count": meal_count, "budget_band": "standard", "cooking_access": "home", "preferences": ["한식"], "exclusions": ["고수"], "allergies": ["땅콩"], "training_schedule": [{ "date": "2026-08-03", "weekday": 0, "time": "18:00", "load_category": "high", }], } def _policy_rows(data_root: Path) -> list[dict[str, object]]: path = data_root / "nutrition-plans" / "policy-revisions.jsonl" return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line] def test_adaptive_prepare_without_extension_preserves_approved_default(tmp_path: Path) -> None: profile_root, _, data_root = _adaptive_fixture(tmp_path) policy_path = data_root / "nutrition-plans" / "policy.json" before = policy_path.read_bytes() prepare_adaptive_nutrition_runtime(profile_root, "client_001") assert policy_path.read_bytes() == before assert _policy_rows(data_root) == [] assert load_approved_adaptive_artifacts(data_root).policy.extended_through is None @pytest.mark.parametrize("offset", [28, 83], ids=["d-plus-29", "d-plus-84"]) def test_adaptive_prepare_persists_approved_extension_revision( tmp_path: Path, offset: int, ) -> None: profile_root, _, data_root = _adaptive_fixture(tmp_path) through = date(2026, 8, 1) + timedelta(days=offset) prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=through, ) rows = _policy_rows(data_root) assert [row["state"] for row in rows] == ["prepared", "committed"] committed = rows[-1] document = committed["policy_document"] assert isinstance(document, dict) assert document["approved"] is True assert document["extension_through"] == through.isoformat() assert document["supersedes_digest"] == rows[0]["supersedes_digest"] assert document["digest"] == committed["policy_revision_digest"] artifacts = load_approved_adaptive_artifacts(data_root) assert artifacts.policy.extended_through == through def test_adaptive_prepare_rejects_d_plus_85_without_writing(tmp_path: Path) -> None: profile_root, _, data_root = _adaptive_fixture(tmp_path) with pytest.raises(CustomerAdminError, match="D\\+29\\.\\.D\\+84"): prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=date(2026, 10, 25), ) assert _policy_rows(data_root) == [] def test_adaptive_policy_supersession_preserves_prior_revision_audit(tmp_path: Path) -> None: profile_root, _, data_root = _adaptive_fixture(tmp_path) first_day = date(2026, 8, 29) second_day = date(2026, 10, 23) prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=first_day, ) first_rows = _policy_rows(data_root) first_commit = first_rows[-1] prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=first_day, ) assert _policy_rows(data_root) == first_rows prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=second_day, ) rows = _policy_rows(data_root) assert [row["state"] for row in rows] == ["prepared", "committed", "prepared", "committed"] assert rows[2]["supersedes_digest"] == first_commit["policy_revision_digest"] assert rows[1]["policy_document"] == first_commit["policy_document"] audits = [ json.loads(line) for line in (data_root / "nutrition-plans" / "policy-audit.jsonl").read_text(encoding="utf-8").splitlines() if line ] assert [audit["policy_revision_digest"] for audit in audits] == [ first_commit["policy_revision_digest"], rows[-1]["policy_revision_digest"], ] assert load_approved_adaptive_artifacts(data_root).policy.extended_through == second_day def test_adaptive_policy_prepared_restart_recovers_and_tamper_fails_closed(tmp_path: Path) -> None: profile_root, _, data_root = _adaptive_fixture(tmp_path) through = date(2026, 8, 29) prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=through, ) revisions = data_root / "nutrition-plans" / "policy-revisions.jsonl" first_row = revisions.read_text(encoding="utf-8").splitlines()[0] revisions.write_text(first_row + "\n", encoding="utf-8") revisions.chmod(0o600) (data_root / "nutrition-plans" / "policy-audit.jsonl").write_text("", encoding="utf-8") with pytest.raises(ValueError, match="incomplete"): load_approved_adaptive_artifacts(data_root) prepare_adaptive_nutrition_runtime( profile_root, "client_001", extension_through=through, ) assert load_approved_adaptive_artifacts(data_root).policy.extended_through == through rows = _policy_rows(data_root) rows[-1]["policy_revision_digest"] = "0" * 64 revisions.write_text( "\n".join(json.dumps(row) for row in rows) + "\n", encoding="utf-8", ) revisions.chmod(0o600) with pytest.raises(ValueError, match="row digest"): load_approved_adaptive_artifacts(data_root) def test_register_customer_creates_disabled_twelve_week_draft_atomically(tmp_path: Path) -> None: path = tmp_path / "registry.json" _empty_registry(path) draft = CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ) registered = register_customer(path, draft) assert registered.enabled is False assert len(registered.plan.weeks) == 12 assert path.stat().st_mode & 0o777 == 0o600 with pytest.raises(ValueError, match="already exists"): register_customer(path, draft) def test_register_customer_allows_multiple_disabled_drafts(tmp_path: Path) -> None: path = tmp_path / "registry.json" _empty_registry(path) first = CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ) register_customer(path, first) second = replace( first, customer_key="client_002", display_name="김철수", user_id="4", topic_id="40", ) register_customer(path, second) payload = json.loads(path.read_text(encoding="utf-8")) assert [item["customer_key"] for item in payload["customers"]] == ["client_001", "client_002"] assert all(item["enabled"] is False for item in payload["customers"]) def test_direct_enable_is_rejected_after_plan_review(tmp_path: Path) -> None: path = tmp_path / "registry.json" _empty_registry(path) draft = CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), ) register_customer(path, draft) with pytest.raises(ValueError, match="direct enabling is not allowed"): set_customer_enabled(path, "client_001", enabled=True) def test_direct_enable_cli_is_removed() -> None: with pytest.raises(SystemExit) as exc_info: main(["--registry", "registry.json", "enable", "client_001"]) assert exc_info.value.code == 2 def test_retry_command_is_removed_from_cli() -> None: with pytest.raises(SystemExit) as exc_info: main(["--registry", "registry.json", "retry"]) assert exc_info.value.code == 2 def test_ai_processing_consent_is_explicit_and_revocable(tmp_path: Path, capsys) -> None: path = tmp_path / "registry.json" _empty_registry(path) register_customer(path, CustomerDraft( customer_key="client_001", display_name="홍길동", user_id="2", chat_id="-100", topic_id="20", starts_on=date(2026, 8, 1), daily_time=time(8, 0), weekly_weekday=0, monthly_day=1, calories_kcal=2300, protein_g=150, meals=("아침", "점심", "저녁"), primary_goal="식습관 안정", )) assert main([ "--registry", str(path), "consent", "client_001", "--recorded-on", "2026-08-01", "--notice-version", "privacy-v1", ]) == 0 assert "consent recorded" in capsys.readouterr().out payload = json.loads(path.read_text(encoding="utf-8")) assert payload["customers"][0]["ai_processing_consent"]["granted"] is True assert main(["--registry", str(path), "revoke-consent", "client_001"]) == 0 payload = json.loads(path.read_text(encoding="utf-8")) assert payload["customers"][0]["ai_processing_consent"]["granted"] is False def test_validate_command_reports_registry_and_plan_success(tmp_path: Path, capsys) -> None: path = tmp_path / "registry.json" _empty_registry(path) assert main(["--registry", str(path), "validate"]) == 0 assert "registry valid: 0 customers" in capsys.readouterr().out def test_activation_accepts_all_g1_through_g5_evidence(tmp_path: Path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) result = activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) assert result.enabled is True assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is True assert result.audit_path.exists() def test_activation_allows_one_operator_to_use_distinct_role_topics( tmp_path: Path, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) payload = json.loads(registry_path.read_text(encoding="utf-8")) shared_user_id = "shared-test-operator" payload["owner"]["user_id"] = shared_user_id payload["customers"][0]["telegram"]["user_id"] = shared_user_id registry_path.write_text(json.dumps(payload), encoding="utf-8") (data_root / "role-binding.json").write_text( json.dumps({"user_id": shared_user_id, "topic_id": "20"}), encoding="utf-8", ) result = activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) assert result.enabled is True def test_activation_allows_shared_current_role_component_on_disabled_customer( tmp_path: Path, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) payload = json.loads(registry_path.read_text(encoding="utf-8")) shared_user_id = "shared-test-operator" payload["owner"]["user_id"] = shared_user_id payload["customers"][0]["telegram"]["user_id"] = shared_user_id other = json.loads(json.dumps(payload["customers"][0])) other["customer_key"] = "disabled_history" other["display_name"] = "Disabled History" other["enabled"] = False other["telegram"]["chat_id"] = "other-chat" other["telegram"]["topic_id"] = "other-topic" payload["customers"].append(other) registry_path.write_text(json.dumps(payload), encoding="utf-8") (data_root / "role-binding.json").write_text( json.dumps({"user_id": shared_user_id, "topic_id": "20"}), encoding="utf-8", ) result = activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) assert result.enabled is True def test_withdraw_customer_is_one_recovered_authority_transition( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root, registry_path, _, _ = _activation_fixture(tmp_path) events: list[str] = [] writes: list[dict[str, object]] = [] real_recover = customer_admin_module._recover_activation_journal_locked real_write = customer_admin_module._write class ObservedLock: def __enter__(self) -> None: events.append("lock-enter") def __exit__(self, *args: object) -> None: events.append("lock-exit") def observed_recover(root: Path, path: Path) -> None: events.append("recover") real_recover(root, path) def observed_write(path: Path, document: object) -> None: events.append("write") writes.append(document.model_dump(mode="json")) real_write(path, document) monkeypatch.setattr( customer_admin_module, "profile_authority_lock", lambda _root: ObservedLock(), ) monkeypatch.setattr( customer_admin_module, "_recover_activation_journal_locked", observed_recover, ) monkeypatch.setattr(customer_admin_module, "_write", observed_write) withdrawn = customer_admin_module.withdraw_customer( profile_root, "client_001", kst_date=date(2026, 8, 4), ) assert events == ["lock-enter", "recover", "write", "lock-exit"] assert len(writes) == 1 written_customer = writes[0]["customers"][0] assert written_customer["enabled"] is False assert written_customer["ai_processing_consent"] == { "granted": False, "recorded_on": "2026-08-04", "notice_version": CONSENT_VERSION, } assert withdrawn == customer_admin_module._read(registry_path).customers[0] def test_withdraw_customer_recovers_activation_journal_and_is_idempotent( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal_path = profile_root / "data" / "customer-activation-journal.json" journal = json.loads(journal_path.read_text(encoding="utf-8")) journal["state"] = "prepared" journal.pop("committed_at") journal_path.write_text(json.dumps(journal), encoding="utf-8") write_count = 0 real_write = customer_admin_module._write def counted_write(path: Path, document: object) -> None: nonlocal write_count write_count += 1 real_write(path, document) monkeypatch.setattr(customer_admin_module, "_write", counted_write) first = customer_admin_module.withdraw_customer( profile_root, "client_001", kst_date=date(2026, 8, 4), ) journal_after_first = journal_path.read_bytes() second = customer_admin_module.withdraw_customer( profile_root, "client_001", kst_date=date(2026, 8, 4), ) assert first == second assert first.enabled is False assert first.ai_processing_consent == AiProcessingConsent( granted=False, recorded_on=date(2026, 8, 4), notice_version=CONSENT_VERSION, ) assert write_count == 1 assert json.loads(journal_after_first)["state"] == "committed" assert journal_path.read_bytes() == journal_after_first assert validate_committed_activation( profile_root, registry_path, "client_001", ) is True def test_direct_disable_remains_allowed_after_activation(tmp_path: Path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) disabled = set_customer_enabled(registry_path, "client_001", enabled=False) assert disabled.enabled is False assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is False @pytest.mark.parametrize( "terms", [ {"checked": True}, {"checked": True, "version": "privacy-v0"}, True, ], ) def test_activation_rejects_missing_or_wrong_provider_terms_version( tmp_path: Path, terms: object, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) payload = json.loads(checklist_path.read_text(encoding="utf-8")) payload["checklist"]["provider_terms_checked"] = terms checklist_path.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(CustomerAdminError, match="G3 provider terms check uses the wrong consent version"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) def test_activation_rejects_g1_data_root_escape(tmp_path: Path) -> None: profile_root, _, _, checklist_path = _activation_fixture(tmp_path) foreign_root = tmp_path / "foreign" foreign_root.mkdir() with pytest.raises(CustomerAdminError, match="G1 data root"): activate_customer( profile_root, foreign_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) def test_activation_rejects_g2_missing_customer_consent(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path, grant_consent=False) with pytest.raises(CustomerAdminError, match="G2/G3"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) def test_activation_rejects_g4_owner_customer_identity_collision(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture( tmp_path, owner_matches_customer=True, ) with pytest.raises(CustomerAdminError, match="G4"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) def test_activation_rejects_g5_false_checklist_item(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) payload = json.loads(checklist_path.read_text(encoding="utf-8")) payload["checklist"]["manual_fallback_doc"] = False checklist_path.write_text(json.dumps(payload), encoding="utf-8") with pytest.raises(CustomerAdminError, match="G5 checklist item"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) @pytest.mark.parametrize( "kst_date", [date(2026, 8, 1), date(2026, 8, 28)], ids=["plan-start", "day-28"], ) def test_activation_accepts_plan_start_and_day_28( tmp_path: Path, kst_date: date, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) result = activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=kst_date, ) assert result.enabled is True assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is True journal = json.loads( (profile_root / "data" / "customer-activation-journal.json").read_text(encoding="utf-8") ) assert journal["state"] == "committed" def test_activation_registry_restore_failure_retains_recovery_journal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) journal_path = profile_root / "data" / "customer-activation-journal.json" audit_path = profile_root / "data" / "customer-activation-audit.jsonl" real_write_journal = customer_admin_module._write_activation_journal real_restore = customer_admin_module._restore_bytes def fail_commit(path: Path, payload: dict[str, object]) -> None: if payload.get("state") == "committed": raise OSError("commit receipt injection") real_write_journal(path, payload) def fail_registry_restore(path: Path, previous: bytes | None) -> None: if path == registry_path: raise OSError("registry restore injection") real_restore(path, previous) monkeypatch.setattr(customer_admin_module, "_write_activation_journal", fail_commit) monkeypatch.setattr(customer_admin_module, "_restore_bytes", fail_registry_restore) with pytest.raises(CustomerAdminError, match="recovery required"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal = json.loads(journal_path.read_text(encoding="utf-8")) assert journal["state"] == "prepared" assert journal["recovery_required"] is True assert all(item["component"] != "journal" for item in journal["recovery_failures"]) assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is True assert not any( item.get("state") in {"committed", "rolled_back"} for item in [journal] ) monkeypatch.setattr(customer_admin_module, "_write_activation_journal", real_write_journal) monkeypatch.setattr(customer_admin_module, "_restore_bytes", real_restore) assert validate_committed_activation(profile_root) is True recovered = json.loads(journal_path.read_text(encoding="utf-8")) assert recovered["state"] == "abandoned" assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is False assert not audit_path.exists() or not audit_path.read_bytes() def test_activation_audit_restore_failure_retains_recovery_journal( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) journal_path = profile_root / "data" / "customer-activation-journal.json" audit_path = profile_root / "data" / "customer-activation-audit.jsonl" real_write_journal = customer_admin_module._write_activation_journal real_restore = customer_admin_module._restore_bytes def fail_commit(path: Path, payload: dict[str, object]) -> None: if payload.get("state") == "committed": raise OSError("commit receipt injection") real_write_journal(path, payload) def fail_audit_restore(path: Path, previous: bytes | None) -> None: if path == audit_path: raise OSError("audit restore injection") real_restore(path, previous) monkeypatch.setattr(customer_admin_module, "_write_activation_journal", fail_commit) monkeypatch.setattr(customer_admin_module, "_restore_bytes", fail_audit_restore) with pytest.raises(CustomerAdminError, match="recovery required"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal = json.loads(journal_path.read_text(encoding="utf-8")) assert journal["state"] == "prepared" assert journal["recovery_required"] is True assert any(item["component"] == "audit" for item in journal["recovery_failures"]) assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is False assert audit_path.exists() assert not any( item.get("state") in {"committed", "rolled_back"} for item in [journal] ) monkeypatch.setattr(customer_admin_module, "_write_activation_journal", real_write_journal) monkeypatch.setattr(customer_admin_module, "_restore_bytes", real_restore) assert validate_committed_activation(profile_root) is True recovered = json.loads(journal_path.read_text(encoding="utf-8")) assert recovered["state"] == "abandoned" assert not audit_path.exists() or not audit_path.read_bytes() def test_activation_restart_reconciles_prepared_commit_without_rollback( tmp_path: Path, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal_path = profile_root / "data" / "customer-activation-journal.json" journal = json.loads(journal_path.read_text(encoding="utf-8")) journal["state"] = "prepared" journal["recovery_required"] = True journal_path.write_text(json.dumps(journal), encoding="utf-8") assert validate_committed_activation(profile_root, customer_id="client_001") is True recovered = json.loads(journal_path.read_text(encoding="utf-8")) assert recovered["state"] == "committed" assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is True @pytest.mark.parametrize( "kst_date", [date(2026, 7, 31), date(2026, 8, 29)], ids=["day-before-start", "day-29"], ) def test_activation_rejects_outside_plan_window_without_receipt( tmp_path: Path, kst_date: date, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) registry_before = registry_path.read_bytes() journal_path = profile_root / "data" / "customer-activation-journal.json" audit_path = profile_root / "data" / "customer-activation-audit.jsonl" with pytest.raises(CustomerAdminError, match="outside the KST plan window"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=kst_date, ) assert registry_path.read_bytes() == registry_before assert json.loads(registry_path.read_text(encoding="utf-8"))["customers"][0]["enabled"] is False assert not journal_path.exists() assert not audit_path.exists() def test_typed_record_commands_append_canonical_events_and_safe_receipts( tmp_path: Path, capsys, ) -> None: profile_root, registry_path, data_root = _record_fixture(tmp_path) assert main([ "--registry", str(registry_path), "record-payment", "client_001", "--profile-root", str(profile_root), "--data-root", str(data_root), "--kind", "initial", "--paid-on", "2026-08-01", "--period-start-on", "2026-08-01", "--period-end-on", "2026-08-28", ]) == 0 payment_receipt = json.loads(capsys.readouterr().out) assert payment_receipt["status"] == "recorded" assert payment_receipt["receipt"].startswith("pilot_") assert "client_001" not in json.dumps(payment_receipt) private_note = "private satisfaction note" assert main([ "--registry", str(registry_path), "record-satisfaction", "client_001", "--profile-root", str(profile_root), "--score", "9", "--collected-on", "2026-08-07", "--note", private_note, ]) == 0 satisfaction_receipt = json.loads(capsys.readouterr().out) assert satisfaction_receipt["status"] == "recorded" assert private_note not in json.dumps(satisfaction_receipt) assert main([ "--registry", str(registry_path), "record-operator-time", "client_001", "--profile-root", str(profile_root), "--entry-id", "entry-00000000000001", "--attempt-id", "attempt-000000000001", "--minutes", "12", "--task", "draft_review", "--work-date", "2026-08-07", ]) == 0 operator_receipt = json.loads(capsys.readouterr().out) assert operator_receipt["status"] == "recorded" events = [ json.loads(line) for line in (data_root / "wizard" / "events.jsonl").read_text(encoding="utf-8").splitlines() if line ] assert [event["event_type"] for event in events] == [ "payment_record", "satisfaction_record", "operator_time_record", ] assert all(event["provenance"]["source_ref"] == f"pilot:client_001:{event['event_type']}" for event in events) assert events[0]["payment"]["amount_krw"] == 150000 assert events[1]["satisfaction"]["score_1to10"] == 9 assert events[2]["operator_time"]["minutes"] == 12 def test_typed_record_commands_fail_closed_for_scope_values_and_periods(tmp_path: Path) -> None: profile_root, registry_path, data_root = _record_fixture(tmp_path) events_path = data_root / "wizard" / "events.jsonl" with pytest.raises(CustomerAdminError, match="payment period"): main([ "--registry", str(registry_path), "record-payment", "client_001", "--profile-root", str(profile_root), "--kind", "initial", "--paid-on", "2026-08-01", "--period-start-on", "2026-08-02", "--period-end-on", "2026-08-29", ]) with pytest.raises(CustomerAdminError, match="amount"): main([ "--registry", str(registry_path), "record-payment", "client_001", "--profile-root", str(profile_root), "--kind", "initial", "--paid-on", "2026-08-01", "--period-start-on", "2026-08-01", "--period-end-on", "2026-08-28", "--amount-krw", "149999", ]) with pytest.raises(CustomerAdminError, match="satisfaction score"): main([ "--registry", str(registry_path), "record-satisfaction", "client_001", "--profile-root", str(profile_root), "--score", "11", "--collected-on", "2026-08-07", ]) with pytest.raises(CustomerAdminError, match="unknown customer"): main([ "--registry", str(registry_path), "record-satisfaction", "client_999", "--profile-root", str(profile_root), "--score", "8", "--collected-on", "2026-08-07", ]) with pytest.raises(CustomerAdminError, match="data root"): main([ "--registry", str(registry_path), "record-satisfaction", "client_001", "--profile-root", str(profile_root), "--data-root", str(tmp_path / "foreign"), "--score", "8", "--collected-on", "2026-08-07", ]) assert not events_path.exists() def test_operator_time_retries_are_noop_and_corrections_append_replacement( tmp_path: Path, capsys, ) -> None: profile_root, registry_path, data_root = _record_fixture(tmp_path) first_args = [ "--registry", str(registry_path), "record-operator-time", "client_001", "--profile-root", str(profile_root), "--entry-id", "entry-00000000000002", "--attempt-id", "attempt-000000000002", "--minutes", "60", "--task", "reporting", "--work-date", "2026-08-07", ] assert main(first_args) == 0 first_receipt = json.loads(capsys.readouterr().out) assert main(first_args[:6] + [ "--entry-id", "entry-00000000000002", "--attempt-id", "attempt-retry-0000002", "--minutes", "60", "--task", "reporting", "--work-date", "2026-08-07", ]) == 0 retry_receipt = json.loads(capsys.readouterr().out) assert retry_receipt["status"] == "duplicate" assert retry_receipt["receipt"] == first_receipt["receipt"] assert main([ "--registry", str(registry_path), "record-operator-time", "client_001", "--profile-root", str(profile_root), "--entry-id", "entry-00000000000003", "--attempt-id", "attempt-000000000003", "--minutes", "30", "--task", "reporting", "--work-date", "2026-08-07", "--supersedes-entry-id", "entry-00000000000002", ]) == 0 correction_receipt = json.loads(capsys.readouterr().out) assert correction_receipt["status"] == "recorded" events = [ json.loads(line) for line in (data_root / "wizard" / "events.jsonl").read_text(encoding="utf-8").splitlines() if line ] operator_events = [event for event in events if event["event_type"] == "operator_time_record"] assert len(operator_events) == 2 assert operator_events[1]["operator_time"]["supersedes_entry_id"] == "entry-00000000000002" def test_judge_kpi_uses_only_registered_customer_plan_and_events( tmp_path: Path, capsys, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) wizard_root = data_root / "wizard" wizard_root.mkdir(mode=0o700) canonical_path = wizard_root / "events.jsonl" canonical_path.write_text("", encoding="utf-8") canonical_path.chmod(0o600) transaction = CanonicalEventTransaction.for_customer_runtime( load_runtime_customer_registry(profile_root).customers[0] ) transaction.lock_path.touch(mode=0o600, exist_ok=True) transaction.lock_path.chmod(0o600) transaction.sequence_path.touch(mode=0o600, exist_ok=True) transaction.sequence_path.chmod(0o600) assert main([ "--registry", str(registry_path), "judge-kpi", "client_001", "--profile-root", str(profile_root), ]) == 0 report = json.loads(capsys.readouterr().out) assert report["customer_key"] == "client_001" assert report["window"] == {"starts_on": "2026-08-01", "ends_on": "2026-08-28"} assert report["passed"] is False assert set(report["failure_reasons"]) == { "checkin_rate", "satisfaction", "renewal", } assert "events_path" not in report assert "data_root" not in report def test_judge_kpi_has_no_caller_selected_event_path_or_window(tmp_path: Path) -> None: profile_root, registry_path, _ = _record_fixture(tmp_path) with pytest.raises(SystemExit): main([ "--registry", str(registry_path), "judge-kpi", "client_001", "--profile-root", str(profile_root), "--events-path", str(tmp_path / "foreign.jsonl"), ]) with pytest.raises(SystemExit): main([ "--registry", str(registry_path), "judge-kpi", "client_001", "--profile-root", str(profile_root), "--starts-on", "2020-01-01", ]) def test_adaptive_registration_inputs_reload_and_bind_all_artifacts(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} approved = approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) loaded = load_approved_adaptive_registration_inputs(profile_root, "client_001") adaptive_root = data_root / "nutrition-plans" assert (adaptive_root / "food-catalog.json").exists() assert not (adaptive_root / "catalog.json").exists() assert approved.digest == loaded.digest assert loaded.meal_count == 3 assert loaded.budget_band == "standard" assert set(loaded.artifact_digests or {}) == { "base_policy", "meal_constraints", "catalog", } rows = [ json.loads(line) for line in (data_root / "nutrition-plans" / "input-approvals.jsonl").read_text( encoding="utf-8" ).splitlines() if line ] assert [row["artifact_kind"] for row in rows] == [ "base_policy", "base_policy", "meal_constraints", "meal_constraints", "catalog", "catalog", ] assert all(row["state"] in {"prepared", "committed"} for row in rows) def test_adaptive_registration_update_is_append_only_and_supersedes_all_artifacts( tmp_path: Path, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} first = approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) second = update_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(meal_count=4, version="v2"), approved_by=owner, approved_at_kst="2026-08-02T10:00:00+09:00", ) assert second.digest != first.digest assert load_approved_adaptive_registration_inputs(profile_root, "client_001").meal_count == 4 rows = [ json.loads(line) for line in (data_root / "nutrition-plans" / "input-approvals.jsonl").read_text( encoding="utf-8" ).splitlines() if line ] committed = [row for row in rows if row["state"] == "committed"] assert len(committed) == 6 assert all( row["supersedes_digest"] == first.artifact_digests[row["artifact_kind"]] for row in committed[3:] ) def test_adaptive_registration_concurrent_same_predecessor_has_one_winner( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} barrier = threading.Barrier(2) results: list[object] = [None, None] def approve(index: int, meal_count: int) -> None: barrier.wait() try: results[index] = approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(meal_count=meal_count), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", supersedes_digest=customer_admin_module._REGISTRATION_ZERO_DIGEST, ) except BaseException as exc: results[index] = exc threads = [ threading.Thread(target=approve, args=(0, 3)), threading.Thread(target=approve, args=(1, 4)), ] for thread in threads: thread.start() for thread in threads: thread.join() successes = [result for result in results if isinstance(result, customer_admin_module.AdaptiveRegistrationInputs)] failures = [result for result in results if isinstance(result, CustomerAdminError)] assert len(successes) == 1 assert len(failures) == 1 revisions_path = data_root / "nutrition-plans" / "adaptive-registration-inputs.jsonl" revision_rows = [ json.loads(line) for line in revisions_path.read_text(encoding="utf-8").splitlines() if line ] committed, pending = customer_admin_module._validate_registration_revision_rows(revision_rows) assert not pending assert len(committed) == 1 assert committed[0]["supersedes_digest"] == customer_admin_module._REGISTRATION_ZERO_DIGEST config_path = data_root / "nutrition-plans" / "adaptive-registration-config.jsonl" config_rows = [ json.loads(line) for line in config_path.read_text(encoding="utf-8").splitlines() if line ] config_committed, config_pending = customer_admin_module._validate_registration_config_rows( config_rows ) assert not config_pending assert len(config_committed) == 1 input_digests = { customer_admin_module.adaptive_digest( customer_admin_module._registration_input_payload( _registration_inputs(meal_count=meal_count), "client_001", default_version="v1", ) ) for meal_count in (3, 4) } winner_digest = str(successes[0].digest) loser_digests = input_digests - {winner_digest} assert len(loser_digests) == 1 approval_path = data_root / "nutrition-plans" / "input-approvals.jsonl" approval_rows = [ json.loads(line) for line in approval_path.read_text(encoding="utf-8").splitlines() if line ] approval_committed, approval_pending = customer_admin_module._validate_registration_approval_rows( approval_rows ) assert not approval_pending assert len(approval_committed) == 3 loser_digest = next(iter(loser_digests)) all_rows = [ *revision_rows, *config_rows, *approval_rows, ] assert all( row.get("revision_digest", row.get("registration_digest")) != loser_digest for row in all_rows ) def test_adaptive_registration_lock_rejects_symlink(tmp_path: Path) -> None: adaptive_root = tmp_path / "nutrition-plans" adaptive_root.mkdir(mode=0o700) foreign_lock = tmp_path / "foreign.lock" foreign_lock.write_bytes(b"foreign") lock_path = adaptive_root / customer_admin_module._REGISTRATION_LOCK_FILE lock_path.symlink_to(foreign_lock) with pytest.raises(CustomerAdminError, match="registration lock"): with customer_admin_module._registration_lock(adaptive_root): pytest.fail("unsafe registration lock was acquired") assert foreign_lock.read_bytes() == b"foreign" def test_adaptive_registration_inputs_fail_closed_for_missing_wrong_owner_and_tamper( tmp_path: Path, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} with pytest.raises(CustomerAdminError, match="missing"): load_approved_adaptive_registration_inputs(profile_root, "client_001") _write_owner_approved_registration_artifacts(data_root) with pytest.raises(CustomerAdminError, match="approver"): approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by={"user_id": "9", "chat_id": "-9", "topic_id": "9"}, ) approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) policy_path = data_root / "nutrition-plans" / "policy.json" policy = json.loads(policy_path.read_text(encoding="utf-8")) policy["policy"]["calorie_step"] = 999 policy_path.write_text(json.dumps(policy), encoding="utf-8") policy_path.chmod(0o600) with pytest.raises(CustomerAdminError, match="stale|digest"): load_approved_adaptive_registration_inputs(profile_root, "client_001") def test_adaptive_registration_canonical_catalog_and_valid_legacy_fallback(tmp_path: Path) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) root = data_root / "nutrition-plans" canonical_path = root / "food-catalog.json" legacy_path = root / "catalog.json" assert canonical_path.exists() assert not legacy_path.exists() canonical_document = json.loads(canonical_path.read_text(encoding="utf-8")) canonical_path.unlink() legacy_path.write_text(json.dumps(canonical_document), encoding="utf-8") legacy_path.chmod(0o600) loaded = load_approved_adaptive_registration_inputs(profile_root, "client_001") assert loaded.artifact_digests["catalog"] == canonical_document["digest"] tampered = dict(canonical_document) tampered["catalog"] = [dict(canonical_document["catalog"][0], label="tampered")] legacy_path.write_text(json.dumps(tampered), encoding="utf-8") legacy_path.chmod(0o600) with pytest.raises(CustomerAdminError, match="digest|stale"): load_approved_adaptive_registration_inputs(profile_root, "client_001") canonical_path.write_text( json.dumps({**canonical_document, "approved": False}), encoding="utf-8", ) canonical_path.chmod(0o600) legacy_path.write_text(json.dumps(canonical_document), encoding="utf-8") legacy_path.chmod(0o600) with pytest.raises(CustomerAdminError, match="not approved"): load_approved_adaptive_registration_inputs(profile_root, "client_001") def test_adaptive_registration_inputs_reject_stale_activation_authority(tmp_path: Path) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) payload = json.loads(registry_path.read_text(encoding="utf-8")) payload["owner"]["topic_id"] = "11" registry_path.write_text(json.dumps(payload), encoding="utf-8") registry_path.chmod(0o600) with pytest.raises(CustomerAdminError, match="stale|receipt"): load_approved_adaptive_registration_inputs(profile_root, "client_001") def test_reapprove_registration_rebinds_only_current_activation_authority( tmp_path: Path, ) -> None: # Given: an approved registration whose activation receipt was rotated. profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} original = approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) registry = json.loads(registry_path.read_text(encoding="utf-8")) registry["customers"][0]["display_name"] = "새 테스트 라벨" registry_path.write_text(json.dumps(registry), encoding="utf-8") registry_path.chmod(0o600) set_customer_enabled(registry_path, "client_001", enabled=False) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) with pytest.raises(CustomerAdminError, match="stale"): load_approved_adaptive_registration_inputs(profile_root, "client_001") # When: the owner invokes the typed authority-only reapproval. rebound = customer_admin_module.reapprove_adaptive_registration_inputs( profile_root, "client_001", approved_by=owner, ) # Then: a child approval uses the new receipt without changing customer meaning. original_values = original.value_payload() rebound_values = rebound.value_payload() assert { key: value for key, value in rebound_values.items() if key != "version" } == { key: value for key, value in original_values.items() if key != "version" } assert rebound.version != original.version assert rebound.digest != original.digest assert rebound.supersedes_digest == original.digest assert rebound.activation_receipt_id != original.activation_receipt_id assert rebound.artifact_digests == original.artifact_digests assert load_approved_adaptive_registration_inputs( profile_root, "client_001", ) == rebound assert customer_admin_module.validate_adaptive_registration_reapproval( profile_root, "client_001", original.digest, ) def test_reapprove_registration_rejects_changed_artifact_without_appending( tmp_path: Path, ) -> None: # Given: a rotated receipt and a policy artifact changed after its approval. profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) set_customer_enabled(registry_path, "client_001", enabled=False) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) policy_path = data_root / "nutrition-plans" / "policy.json" policy = json.loads(policy_path.read_text(encoding="utf-8")) policy["policy"]["calorie_step"] = 999 policy_path.write_text(json.dumps(policy), encoding="utf-8") policy_path.chmod(0o600) journal_paths = tuple( data_root / "nutrition-plans" / name for name in ( "adaptive-registration-inputs.jsonl", "adaptive-registration-config.jsonl", "input-approvals.jsonl", ) ) before = tuple(path.read_bytes() for path in journal_paths) # When: authority-only reapproval validates the changed artifact. with pytest.raises(CustomerAdminError, match="artifact|digest"): customer_admin_module.reapprove_adaptive_registration_inputs( profile_root, "client_001", approved_by=owner, ) # Then: no registration or approval rows were appended. assert tuple(path.read_bytes() for path in journal_paths) == before def _gate_d_fixture(tmp_path: Path) -> tuple[Path, Path, Path, Path]: profile_root, registry_path, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) _write_owner_approved_registration_artifacts(data_root) owner = {"user_id": "1", "chat_id": "-100", "topic_id": "10"} approve_adaptive_registration_inputs( profile_root, "client_001", inputs=_registration_inputs(), approved_by=owner, approved_at_kst="2026-08-01T10:00:00+09:00", ) config = { "bot": {"bot_id": "main-profile-bot"}, "adaptive_nutrition": { "enabled": True, "delivery_enabled": False, "analytics_shadow": False, "operator_candidates": False, "activation": False, "delivery": False, "review_operator": { "user_id": "review-user", "chat_id": "-100", "topic_id": "59", "version": 1, }, "separate_bot": { "separate": True, "bot_id": "gate-d-test-bot", }, }, } config_path = profile_root / "config.json" config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) wizard_root = data_root / "wizard" runtime = load_runtime_customer_registry(profile_root).customers[0] transaction = CanonicalEventTransaction.for_customer_runtime(runtime) event = { "event_id": "event-preflight-0001", "event_type": "morning_checkin", "occurred_at_kst": "2026-08-03T09:00:00+09:00", "recorded_at_kst": "2026-08-03T09:00:00+09:00", "provenance": { "source_type": "manual", "source_ref": "pilot:client_001:morning", "content_sha256": "0" * 64, }, "status": "accepted", "dedupe_key": "event-preflight-dedupe", "check_in": { "body_weight_kg": 80, "calories_kcal": 2300, "sleep_hours": 8, "sleep_quality_1to5": 4, "readiness_1to5": 4, "pain_summary": "없음", "training_plan": "계획대로 진행", }, } canonical_event = transaction.append(event)["canonical_event"] reconcile_adaptive_nutrition_journals( profile_root, "client_001", canonical_events=[canonical_event], registry=load_runtime_customer_registry(profile_root), ) for path in ( profile_root, profile_root / "customers", profile_root / "data", data_root, wizard_root, ): path.chmod(0o700) registry_path.chmod(0o600) config_path.chmod(0o600) initialize_schedule_delivery_fence(profile_root) return profile_root, registry_path, data_root, config_path def _profile_snapshot(root: Path) -> tuple[tuple[object, ...], ...]: snapshot: list[tuple[object, ...]] = [] for path in sorted(root.rglob("*"), key=lambda candidate: candidate.relative_to(root).as_posix()): relative = path.relative_to(root).as_posix() if path.is_symlink(): snapshot.append(("symlink", relative, path.readlink().as_posix())) elif path.is_dir(): snapshot.append(("dir", relative, path.stat().st_mode & 0o777)) else: snapshot.append(("file", relative, path.stat().st_mode & 0o777, path.read_bytes())) return tuple(snapshot) def _assert_bounded_receipt_serialization( receipt: object, profile_root: Path, ) -> None: assert hasattr(receipt, "to_dict") payload = receipt.to_dict() # type: ignore[union-attr] assert set(payload) == { "schema_version", "ready", "checks", "counts", "digests", "epoch", "reason_codes", "checked_at_kst", } bounded = json.dumps(payload, ensure_ascii=False, sort_keys=True) for raw_value in ( "client_001", "홍길동", "1", "2", "3", "-100", "10", "20", "30", "59", "review-user", "main-profile-bot", "gate-d-test-bot", "event-preflight-0001", "pilot:client_001:morning", "telegram:-100:20", "Sensitive customer body", "gate-d-test-token", ): assert f'"{raw_value}"' not in bounded assert str(profile_root) not in bounded assert payload["checked_at_kst"].endswith("+09:00") def test_gate_d_preflight_is_filesystem_read_only( tmp_path: Path, ) -> None: profile_root, _, _, _ = _gate_d_fixture(tmp_path) before = _profile_snapshot(profile_root) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is True assert _profile_snapshot(profile_root) == before def test_preflight_simple_yaml_preserves_block_scalars_and_lists() -> None: # Given: the dependency-free subset used by the canonical Hermes profile. config = """\ system_prompt: | 코칭 원칙: - 승인된 내용만 사용한다. toolsets: - audit platforms: telegram: allowed_user_ids: - "1" - "2" extra: adaptive_nutrition: enabled: true delivery_enabled: false """ # When: Gate-D parses YAML without the optional PyYAML package. parsed = customer_admin_module._preflight_simple_yaml(config) # Then: block content cannot escape into configuration keys or corrupt lists. assert parsed["system_prompt"] == "코칭 원칙:\n- 승인된 내용만 사용한다.\n" assert parsed["toolsets"] == ["audit"] telegram = parsed["platforms"]["telegram"] assert telegram["allowed_user_ids"] == ["1", "2"] assert telegram["extra"]["adaptive_nutrition"] == { "enabled": True, "delivery_enabled": False, } def test_gate_d_preflight_rejects_stale_schedule_claim_read_only( tmp_path: Path, ) -> None: profile_root, _, _, _ = _gate_d_fixture(tmp_path) claim = ( profile_root / "data" / "customer-schedule-claims" / "client_001" / "2026-08-03" / "daily.claim" ) claim.parent.mkdir(parents=True) claim.parent.parent.chmod(0o700) claim.parent.chmod(0o700) claim.write_bytes(b"claimed-after-ready\n") claim.chmod(0o600) before = _profile_snapshot(profile_root) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "schedule_legacy_claim_stale" in receipt.reason_codes assert _profile_snapshot(profile_root) == before _assert_bounded_receipt_serialization(receipt, profile_root) def test_gate_d_preflight_accepts_complete_fixture_and_bounded_receipt( tmp_path: Path, ) -> None: profile_root, _, _, _ = _gate_d_fixture(tmp_path) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is True assert receipt.passed is True assert all(receipt.checks.values()) assert receipt.counts["approved_artifacts"] == 3 _assert_bounded_receipt_serialization(receipt, profile_root) @pytest.mark.parametrize( "category", ("customer", "owner_scheduled", "generic"), ) def test_gate_d_preflight_rejects_review_space_collision_categories( tmp_path: Path, category: str, ) -> None: profile_root, registry_path, _, config_path = _gate_d_fixture(tmp_path) if category != "generic": payload = json.loads(registry_path.read_text(encoding="utf-8")) if category == "customer": payload["customers"][0]["telegram"]["topic_id"] = "59" else: payload["owner"]["topic_id"] = "59" registry_path.write_text(json.dumps(payload), encoding="utf-8") registry_path.chmod(0o600) else: config = json.loads(config_path.read_text(encoding="utf-8")) config["telegram"] = { "extra": { "generic_reserved_routes": [ {"chat_id": "-100", "topic_id": "59"}, ], }, } config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "review_space_collision" in receipt.reason_codes assert f"review_space_collision_{category}" in receipt.reason_codes @pytest.mark.parametrize( "missing", ["user_id", "chat_id", "topic_id"], ids=["missing-review-user", "missing-review-chat", "missing-review-topic"], ) def test_gate_d_preflight_rejects_missing_full_review_operator_triple( tmp_path: Path, missing: str, ) -> None: profile_root, _, _, config_path = _gate_d_fixture(tmp_path) config = json.loads(config_path.read_text(encoding="utf-8")) review = config["adaptive_nutrition"]["review_operator"] assert isinstance(review, dict) review.pop(missing) config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "review_operator_missing" in receipt.reason_codes def test_gate_d_preflight_rejects_legacy_review_operator_aliases( tmp_path: Path, ) -> None: profile_root, _, _, config_path = _gate_d_fixture(tmp_path) config = json.loads(config_path.read_text(encoding="utf-8")) adaptive = config["adaptive_nutrition"] assert isinstance(adaptive, dict) adaptive.pop("review_operator") adaptive.update({ "operator_user_id": "review-user", "operator_chat_id": "-100", "operator_topic_id": 59, "review_operator_version": 1, }) config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "review_operator_missing" in receipt.reason_codes def test_gate_d_preflight_rejects_non_distinct_operator_customer_owner_identities( tmp_path: Path, ) -> None: profile_root, registry_path, _, _ = _gate_d_fixture(tmp_path) payload = json.loads(registry_path.read_text(encoding="utf-8")) payload["customers"][0]["telegram"] = payload["owner"] registry_path.write_text(json.dumps(payload), encoding="utf-8") registry_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "identity_roles_not_distinct" in receipt.reason_codes def test_gate_d_preflight_rejects_same_main_and_separate_test_bot( tmp_path: Path, ) -> None: profile_root, _, _, config_path = _gate_d_fixture(tmp_path) config = json.loads(config_path.read_text(encoding="utf-8")) config["adaptive_nutrition"]["separate_bot"]["bot_id"] = "main-profile-bot" config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "separate_bot_invalid" in receipt.reason_codes def test_gate_d_preflight_rejects_stale_kst_plan_window(tmp_path: Path) -> None: profile_root, _, _, _ = _gate_d_fixture(tmp_path) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 9, 1), ) assert receipt.ready is False assert "plan_window_stale" in receipt.reason_codes assert receipt.checks["current_kst_window"] is False def test_gate_d_preflight_rejects_public_profile_mode(tmp_path: Path) -> None: profile_root, _, _, _ = _gate_d_fixture(tmp_path) profile_root.chmod(0o755) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "private_modes_invalid" in receipt.reason_codes assert receipt.checks["private_modes"] is False def test_gate_d_preflight_rejects_customer_root_symlink(tmp_path: Path) -> None: profile_root, _, data_root, _ = _gate_d_fixture(tmp_path) foreign_root = tmp_path / "foreign-customer-root" data_root.rename(foreign_root) data_root.symlink_to(foreign_root, target_is_directory=True) with pytest.raises(CustomerAdminError, match="customer root is unsafe"): audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) @pytest.mark.parametrize( "case", [ "artifact-missing", "artifact-stale", "activation-missing", "activation-stale", "reconciliation-missing", "reconciliation-stale", ], ids=[ "missing-artifact", "stale-artifact", "missing-activation", "stale-activation", "missing-reconciliation", "stale-reconciliation", ], ) def test_gate_d_preflight_rejects_missing_or_stale_evidence( tmp_path: Path, case: str, ) -> None: profile_root, _, data_root, _ = _gate_d_fixture(tmp_path) adaptive_root = data_root / "nutrition-plans" expected_reason = { "artifact-missing": "registration_missing", "artifact-stale": "registration_missing", "activation-missing": "activation_receipt_missing", "activation-stale": "activation_receipt_missing", "reconciliation-missing": "canonical_reconciliation_missing", "reconciliation-stale": "canonical_reconciliation_stale", }[case] if case == "artifact-missing": (adaptive_root / "food-catalog.json").unlink() elif case == "artifact-stale": policy_path = adaptive_root / "policy.json" policy = json.loads(policy_path.read_text(encoding="utf-8")) policy["policy"]["calorie_step"] = 999 policy_path.write_text(json.dumps(policy), encoding="utf-8") policy_path.chmod(0o600) elif case == "activation-missing": (profile_root / "data" / "customer-activation-journal.json").unlink() elif case == "activation-stale": journal_path = profile_root / "data" / "customer-activation-journal.json" journal = json.loads(journal_path.read_text(encoding="utf-8")) journal["registry_sha256"] = "0" * 64 journal_path.write_text(json.dumps(journal), encoding="utf-8") journal_path.chmod(0o600) elif case == "reconciliation-missing": (adaptive_root / "source-days.jsonl").unlink() else: sequence_path = adaptive_root / "canonical-sequence.jsonl" sequence = json.loads(sequence_path.read_text(encoding="utf-8")) sequence["row_digest"] = "0" * 64 sequence_path.write_text(json.dumps(sequence), encoding="utf-8") sequence_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert expected_reason in receipt.reason_codes @pytest.mark.parametrize( "flag", ["analytics_shadow", "operator_candidates", "activation", "delivery"], ) def test_gate_d_preflight_rejects_enabled_profile_feature_flags( tmp_path: Path, flag: str, ) -> None: profile_root, _, _, config_path = _gate_d_fixture(tmp_path) config = json.loads(config_path.read_text(encoding="utf-8")) config["adaptive_nutrition"][flag] = True config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "feature_flags_enabled" in receipt.reason_codes def test_gate_d_preflight_rejects_enabled_delivery_config( tmp_path: Path, ) -> None: profile_root, _, _, config_path = _gate_d_fixture(tmp_path) config = json.loads(config_path.read_text(encoding="utf-8")) config["adaptive_nutrition"]["delivery_enabled"] = True config_path.write_text(json.dumps(config), encoding="utf-8") config_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "delivery_enabled" in receipt.reason_codes @pytest.mark.parametrize( "flag", ["analytics_shadow", "operator_candidates", "activation", "delivery"], ) def test_gate_d_preflight_rejects_enabled_feature_epoch_flag( tmp_path: Path, flag: str, ) -> None: profile_root, _, data_root, _ = _gate_d_fixture(tmp_path) feature_path = data_root / "nutrition-plans" / "feature-epoch.json" feature = json.loads(feature_path.read_text(encoding="utf-8")) flags = { name: feature[name] for name in ("analytics_shadow", "operator_candidates", "activation", "delivery") } flags[flag] = True feature["config_digest"] = feature_config_digest(feature["epoch"], flags) feature.update(flags) feature_path.write_text(json.dumps(feature), encoding="utf-8") feature_path.chmod(0o600) receipt = audit_gate_d_preflight( profile_root, "client_001", kst_now=date(2026, 8, 3), ) assert receipt.ready is False assert "feature_flags_enabled" in receipt.reason_codes def test_activation_writes_digest_pinned_nutrition_v2_receipt( tmp_path: Path, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal = json.loads( (profile_root / "data" / "customer-activation-journal.json").read_text( encoding="utf-8" ) ) receipt = journal["nutrition_activation_receipt"] assert journal["version"] == 2 assert receipt["schema_version"] == "nutrition_activation_v2" assert receipt["customer_key"] == "client_001" assert len(receipt["readiness_receipt_digest"]) == 64 assert len(receipt["readiness_bundle_digest"]) == 64 assert len(receipt["customer_projection_digest"]) == 64 assert len(receipt["digest"]) == 64 readiness_policy = json.loads( ( data_root / "nutrition-onboarding" / "adjustment-policy-v1.json" ).read_text(encoding="utf-8") ) promoted_policy = json.loads( (data_root / "nutrition-plans" / "policy.json").read_text(encoding="utf-8") ) assert promoted_policy["policy"] == { key: value for key, value in readiness_policy.items() if key != "digest" } assert promoted_policy["digest"] == readiness_policy["digest"] assert promoted_policy["approved_by"] == { "user_id": "1", "chat_id": "-100", "topic_id": "10", } def test_committed_v1_migration_remains_valid_without_readiness_artifacts( tmp_path: Path, ) -> None: profile_root, registry_path, data_root, checklist_path = _activation_fixture( tmp_path, ) activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), ) journal_path = profile_root / "data" / "customer-activation-journal.json" journal = json.loads(journal_path.read_text(encoding="utf-8")) journal["version"] = 1 journal.pop("nutrition_activation_receipt") journal_path.write_text(json.dumps(journal), encoding="utf-8") journal_path.chmod(0o600) registry = json.loads(registry_path.read_text(encoding="utf-8")) owner_digest = customer_admin_module._json_digest(registry["owner"]) preflight = legacy_migration_preflight( profile_root=profile_root, expected_enabled_customer="client_001", owner_digest=owner_digest, ) commit_legacy_migration( profile_root=profile_root, expected_enabled_customer="client_001", owner_digest=owner_digest, confirm_activation_receipt_digest=preflight["activation_receipt_digest"], ) shutil.rmtree(data_root / "nutrition-onboarding") assert validate_committed_activation( profile_root, registry_path, "client_001", ) manifest_path = ( profile_root / "data/migrations/nutrition-readiness-v1/legacy-activation-authority.json" ) outside_manifest = tmp_path / "outside-legacy-authority.json" outside_manifest.write_bytes(manifest_path.read_bytes()) outside_manifest.chmod(0o600) manifest_path.unlink() manifest_path.symlink_to(outside_manifest) with pytest.raises(CustomerAdminError, match="manifest"): validate_committed_activation( profile_root, registry_path, "client_001", ) def test_disabled_only_registry_needs_no_activation_journal(tmp_path: Path) -> None: profile_root, registry_path, _, _ = _activation_fixture( tmp_path, nutrition_readiness=False, ) assert validate_committed_activation(profile_root, registry_path) def test_current_onboarding_authority_rejects_revoked_registry_consent( tmp_path: Path, ) -> None: profile_root, registry_path, _, _ = _activation_fixture( tmp_path, nutrition_readiness=False, ) authority = OnboardingAuthority( customer_key="client_001", customer_user_id=2, customer_chat_id=-100, customer_topic_id=20, owner_user_id=1, owner_chat_id=-100, owner_topic_id=10, consent_notice_version="privacy-v1", consent_granted=True, customer_enabled=False, ) validate_current_registry_authority(profile_root, authority) set_customer_ai_consent( registry_path, "client_001", AiProcessingConsent(granted=False), ) with pytest.raises(ValueError, match="consent"): validate_current_registry_authority(profile_root, authority) @pytest.mark.parametrize("field", ("activation", "delivery")) def test_activation_rejects_pre_enabled_adaptive_runtime_flag( tmp_path: Path, field: str, ) -> None: profile_root, _, data_root, checklist_path = _activation_fixture(tmp_path) initialize_adaptive_customer(data_root) epoch_path = data_root / "nutrition-plans" / "feature-epoch.json" epoch = json.loads(epoch_path.read_text(encoding="utf-8")) epoch[field] = True epoch_path.write_text(json.dumps(epoch), encoding="utf-8") epoch_path.chmod(0o600) with pytest.raises(CustomerAdminError, match="adaptive runtime preparation"): activate_customer( profile_root, data_root, "client_001", checklist_path, kst_date=date(2026, 8, 1), )