import fcntl import hashlib import json import os import threading from datetime import date, timedelta from decimal import Decimal import pytest from checkin_cli import adaptive_nutrition as adaptive_module from checkin_cli.adaptive_nutrition import ( AdaptiveEventStore, AdaptiveOverlay, CustomerActionContinuity, CustomerPolicy, DailyObservation, Decision, DualCoachRiskEvidence, DualCoachRiskPolicyV1, Food, GLOBAL_MIN_CALORIES, GLOBAL_MIN_FAT_G, GLOBAL_MIN_PROTEIN_G, GLOBAL_MAX_CALORIES, GLOBAL_MAX_FAT_G, GLOBAL_MAX_PROTEIN_G, MacroTarget, MealConstraints, OverlayJournal, VersionedFoodCatalog, build_snapshot, CanonicalSequenceJournal, canonical_event_records, canonical_json, create_proposal_revision, derive_adherence_signal, digest, _evaluate_dual_coach_risk, feature_config_digest, feature_config_digest_preimage, held_dual_coach_risk_candidate, compile_meal_plan, compile_weekly_carb_cycle, evaluate_cooldown, initialize_adaptive_customer, load_approved_adaptive_artifacts, project_canonical_events, propose, render_customer_body, render_operator_card, render_explanation_fallback, render_operator_digest, solve_macros, validate_explanation, _policy_locked, ) def observations(day: date, *, rate_delta=Decimal("-0.2"), adherence=True): rows = [] for offset in range(14): d = day - timedelta(days=13-offset) weight = Decimal("80") if offset < 7 else Decimal("80") + rate_delta rows.append(DailyObservation(d, weight_kg=weight, adherence_ok=adherence)) return rows def policy(start: date, **overrides): values = dict(starts_on=start, goal_mode="fat_loss", weekly_rate_min=Decimal("-0.5"), weekly_rate_max=Decimal("-0.25"), calorie_step=100, calorie_floor=1800, calorie_ceiling=2600) values.update(overrides) return CustomerPolicy(**values) def _risk_policy() -> DualCoachRiskPolicyV1: value = { "weight_change_percent": {"normal": "<=2", "elevated": ">2-4", "high": ">4"}, "sleep_hours": {"normal": ">=7", "elevated": "5-<7", "high": "<5"}, "fatigue": ["low", "moderate", "high"], "pain": ["none", "present", "severe"], "exercise_feasibility": ["possible", "limited", "impossible"], "meal_deviation": ["none", "partial", "material"], "score_threshold": 4, "hard_overrides": ["pain_override", "exercise_impossible_override"], "missing_evidence_reason": "risk_evidence_unavailable", } return DualCoachRiskPolicyV1("1", value, digest(value), "a" * 64) def test_schedule_strategy_projection_is_pinned_and_idempotent(tmp_path): store = AdaptiveEventStore(tmp_path / "events.jsonl") digest_value = "a" * 64 baseline = store.project_schedule_baseline( customer_key="client_001", source_reference_id="schedule_001", source_reference_digest=digest_value, policy_version="dual-coach-risk-v1", policy_digest=digest_value, policy_document_digest=digest_value, epoch=7, parent_digest=digest_value, ) assert baseline["payload"]["strategy_state"] == "schedule_unconfirmed" assert baseline["payload"]["policy_document_digest"] == digest_value assert "categories" not in baseline["payload"] confirmed = store.project_confirmed_schedule_strategy( customer_key="client_001", source_reference_id="schedule_001", source_reference_digest=digest_value, confirmation_id="confirm_001", source_day_mapping_digest=digest_value, policy_version="dual-coach-risk-v1", policy_digest=digest_value, policy_document_digest=digest_value, epoch=7, parent_digest=digest_value, categories=("training", "rest", "training", "rest", "rest", "training", "rest"), last_change_note="첫 합의 일정", ) replay = store.project_confirmed_schedule_strategy( customer_key="client_001", source_reference_id="schedule_001", source_reference_digest=digest_value, confirmation_id="confirm_001", source_day_mapping_digest=digest_value, policy_version="dual-coach-risk-v1", policy_digest=digest_value, policy_document_digest=digest_value, epoch=7, parent_digest=digest_value, categories=("training", "rest", "training", "rest", "rest", "training", "rest"), last_change_note="첫 합의 일정", ) assert confirmed == replay assert confirmed["payload"]["policy_document_digest"] == digest_value assert confirmed["payload"]["categories"] == ["training", "rest", "training", "rest", "rest", "training", "rest"] def test_dual_coach_risk_is_deterministic_and_inclusive() -> None: result = _evaluate_dual_coach_risk( _risk_policy(), DualCoachRiskEvidence( Decimal("2"), Decimal("7"), "high", "none", "possible", "material" ), ) assert result.score == 4 assert result.reasons == ("risk_score_threshold",) assert result.held is True def test_dual_coach_risk_overrides_hold_independent_of_score() -> None: result = _evaluate_dual_coach_risk( _risk_policy(), DualCoachRiskEvidence( Decimal("0"), Decimal("8"), "low", "present", "impossible", "none" ), ) assert result.score == 3 assert result.reasons == ("pain_override", "exercise_impossible_override") assert result.held is True def test_unavailable_risk_candidate_uses_fixed_sentinels_and_dedupe() -> None: candidate = held_dual_coach_risk_candidate( customer_key="client_001", evaluation_kst_day=date(2026, 8, 1), terminal_checkin_id="checkin-1", terminal_checkin_digest="b" * 64, source_strategy_digest="c" * 64, epoch=7, ) assert candidate.reason_code == "risk_evidence_unavailable" assert candidate.policy_version == candidate.policy_digest == "unavailable" assert candidate.dedupe_key == held_dual_coach_risk_candidate( customer_key="client_001", evaluation_kst_day=date(2026, 8, 1), terminal_checkin_id="checkin-1", terminal_checkin_digest="b" * 64, source_strategy_digest="c" * 64, epoch=7, ).dedupe_key def test_missing_customer_goal_observes(): day = date(2026, 7, 14) snap = build_snapshot(observations(day), day, date(2026, 7, 1)) result = propose("client_001", snap, CustomerPolicy(date(2026, 7, 1), "fat_loss"), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60) assert result.decision is Decision.OBSERVE assert result.reasons == ("customer_goal_input_required",) def test_low_adherence_keeps_plan_and_investigates(): day = date(2026, 7, 14) snap = build_snapshot(observations(day, rate_delta=Decimal("0.5"), adherence=False), day, date(2026, 7, 1)) result = propose("client_001", snap, policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60) assert result.decision is Decision.MAINTAIN assert "low_adherence_investigate_barriers" in result.reasons def test_safety_always_holds(): day = date(2026, 7, 14) rows = observations(day) rows[-1] = DailyObservation(day, weight_kg=Decimal("80"), adherence_ok=True, safety_held=True) result = propose("client_001", build_snapshot(rows, day, date(2026, 7, 1)), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60) assert result.decision is Decision.HUMAN_REVIEW assert result.target is None @pytest.mark.parametrize("d_plus,allowed", [(0, False), (1, True), (28, True), (29, False), (84, False), (85, False)]) def test_pilot_boundaries(d_plus, allowed): start = date(2026, 7, 1) day = start + timedelta(days=d_plus - 1) snap = build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, start) result = propose("client_001", snap, policy(start), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60) assert (result.decision is Decision.CALORIE_ADJUSTMENT) is allowed def test_extension_is_customer_specific(): start = date(2026, 7, 1) day = start + timedelta(days=40) snap = build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, start) result = propose("client_001", snap, policy(start, extended_through=start + timedelta(days=83)), current_target=MacroTarget(2300, 300, 150, 55), protein_g=150, fat_g=60) assert result.decision is Decision.CALORIE_ADJUSTMENT def test_exact_macro_solver(): target = solve_macros(2300, 150, 60) assert target == MacroTarget(2300, 290, 150, 60) assert solve_macros(2301, 150, 60) is None def test_global_nutrition_floors_ignore_customer_policy(): start = date(2026, 7, 1) day = date(2026, 7, 14) snap = build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, start) result = propose( "client_001", snap, policy(start, calorie_floor=0), current_target=MacroTarget(1600, 110, 150, 60), protein_g=150, fat_g=60, ) assert result.target is not None assert result.target.calories >= GLOBAL_MIN_CALORIES assert result.target.protein_g >= GLOBAL_MIN_PROTEIN_G assert result.target.fat_g >= GLOBAL_MIN_FAT_G assert solve_macros(GLOBAL_MIN_CALORIES - 4, GLOBAL_MIN_PROTEIN_G, GLOBAL_MIN_FAT_G) is None def test_planned_schedule_requires_all_seven_days(): start = date(2026, 7, 1) day = date(2026, 7, 14) rows = observations(day, rate_delta=Decimal("0.5")) rows.append(DailyObservation(day + timedelta(days=1), exercise_load="high")) result = propose("client_001", build_snapshot(rows, day, start), policy(start), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60) assert result.carb_days == () assert "training_schedule_required" in result.reasons def test_same_day_later_row_cannot_erase_safety(): day = date(2026, 7, 14) rows = observations(day) rows.extend([ DailyObservation(day, safety_held=True), DailyObservation(day, weight_kg=Decimal("80"), adherence_ok=True), ]) snapshot = build_snapshot(rows, day, date(2026, 7, 1)) assert snapshot.safety_held is True def test_complete_planned_schedule_enables_carb_days(): start = date(2026, 7, 1) day = date(2026, 7, 14) rows = observations(day, rate_delta=Decimal("0.5")) loads = ("high", "medium", "low", "high", "medium", "low", "low") rows.extend( DailyObservation(day + timedelta(days=index), exercise_load=load) for index, load in enumerate(loads) ) result = propose( "client_001", build_snapshot(rows, day, start), policy(start), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert len(result.carb_days) == 7 assert "training_schedule_required" not in result.reasons def test_ambiguous_planned_day_requires_human_review_without_carb_assignment(): start = date(2026, 7, 1) day = date(2026, 7, 14) rows = observations(day, rate_delta=Decimal("0.5")) rows.extend([ DailyObservation(day + timedelta(days=1), exercise_load="high"), DailyObservation(day + timedelta(days=1), exercise_load="low"), ]) result = propose( "client_001", build_snapshot(rows, day, start), policy(start), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert result.decision is Decision.HUMAN_REVIEW assert result.target is None assert result.carb_days == () assert result.reasons == ("training_schedule_ambiguous",) def test_meal_constraints_and_allergens(): target = MacroTarget(600, 70, 40, 18) foods = [Food("safe", "닭고기 덮밥", 600, 70, 40, 18), Food("nuts", "견과", 600, 70, 40, 18, frozenset({"nut"}), True)] with pytest.raises(ValueError, match="VersionedFoodCatalog"): compile_meal_plan(target, MealConstraints(1, frozenset({"nut"}), budget_tier="standard", cooking_access="home"), foods) assert compile_meal_plan(target, MealConstraints(1), foods, shadow_test_only=True) is None catalog = VersionedFoodCatalog( foods=tuple(foods), version="v1", digest=digest(tuple(foods)), approved=True, ) plan = compile_meal_plan(target, MealConstraints(1, frozenset({"nut"}), budget_tier="standard", cooking_access="home"), catalog) assert plan is not None and plan.slots[0].food_ids == ("safe",) def test_customer_body_omits_zero_grams_for_unit_based_food(): day = date(2026, 7, 14) target = MacroTarget(2300, 290, 150, 60) catalog = VersionedFoodCatalog( foods=(Food("safe", "한 끼", 2300, 290, 150, 60),), version="v1", digest=digest(("safe",)), approved=True, ) proposal = propose( "client_001", build_snapshot( observations(day, rate_delta=Decimal("-0.3")), day, date(2026, 7, 1), ), policy(date(2026, 7, 1)), current_target=target, protein_g=150, fat_g=60, meal_constraints=MealConstraints( 1, budget_tier="standard", cooking_access="home", ), catalog=catalog, planned_sessions=tuple( (day + timedelta(days=offset), "medium") for offset in range(7) ), ) body = render_customer_body(proposal) assert "safe 1회" in body assert "(0g)" not in body def test_canonical_unicode_and_approval_replay(tmp_path): assert canonical_json({"x": "가"}) == canonical_json({"x": "가"}) day = date(2026, 7, 14) proposal = propose("client_001", build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, date(2026, 7, 1)), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60) store = AdaptiveEventStore(tmp_path / "events.jsonl") first = store.approve(proposal, operator_id="richard", expected_digest=proposal.digest) assert store.approve(proposal, operator_id="richard", expected_digest=proposal.digest) == first with pytest.raises(ValueError, match="stale"): store.approve(proposal, operator_id="richard", expected_digest="bad") assert (tmp_path / "events.jsonl").stat().st_mode & 0o777 == 0o600 assert "운영자 승인 전" in render_operator_card(proposal) def test_adaptive_customer_initialization_is_private_disabled_and_idempotent(tmp_path): first = initialize_adaptive_customer(tmp_path) second = initialize_adaptive_customer(tmp_path) assert first == second root = tmp_path / "nutrition-plans" assert root.stat().st_mode & 0o777 == 0o700 assert (root / "food-catalog.json").exists() assert not (root / "catalog.json").exists() assert all(path.stat().st_mode & 0o777 == 0o600 for path in root.iterdir()) assert '"enabled":false' in (root / "policy.json").read_text() assert '"delivery":false' in (root / "feature-epoch.json").read_text() def test_initializer_preserves_existing_legacy_catalog_for_migration(tmp_path): customer_root = tmp_path / "legacy-customer" runtime_root = customer_root / "nutrition-plans" runtime_root.mkdir(parents=True) legacy_path = runtime_root / "catalog.json" legacy_path.write_text("{}", encoding="utf-8") legacy_path.chmod(0o600) initialize_adaptive_customer(customer_root) assert not (runtime_root / "food-catalog.json").exists() def test_initializer_creates_fresh_disabled_customer_and_rejects_path_types(tmp_path): fresh = tmp_path / "new-customer" initialize_adaptive_customer(fresh) assert (fresh / "nutrition-plans" / "feature-epoch.json").exists() not_a_directory = tmp_path / "customer-file" not_a_directory.write_text("x", encoding="utf-8") with pytest.raises(ValueError, match="data root"): initialize_adaptive_customer(not_a_directory) symlink_target = tmp_path / "real-customer" symlink_target.mkdir() symlink = tmp_path / "customer-link" symlink.symlink_to(symlink_target, target_is_directory=True) with pytest.raises(ValueError, match="data root"): initialize_adaptive_customer(symlink) runtime_owner = tmp_path / "runtime-owner" runtime_owner.mkdir() runtime_target = tmp_path / "runtime-target" runtime_target.mkdir() (runtime_owner / "nutrition-plans").symlink_to(runtime_target, target_is_directory=True) with pytest.raises(ValueError, match="runtime symlink"): initialize_adaptive_customer(runtime_owner) def test_policy_lock_rejects_symlink_before_mutation(tmp_path): root = tmp_path / "nutrition-plans" root.mkdir() target = tmp_path / "outside.lock" target.write_bytes(b"untouched") target.chmod(0o600) (root / ".adaptive.lock").symlink_to(target) marker = root / "mutation-marker" with pytest.raises(OSError): with _policy_locked(root): marker.write_text("mutated", encoding="utf-8") assert not marker.exists() assert target.read_bytes() == b"untouched" def test_approved_private_artifacts_are_versioned_and_digest_pinned(tmp_path): initialize_adaptive_customer(tmp_path) root = tmp_path / "nutrition-plans" policy_value = { "starts_on": "2026-07-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_value = { "meal_count": 1, "budget_tier": "standard", "cooking_access": "home", } catalog_value = [{ "food_id": "safe", "label": "safe", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] for filename, key, value in ( ("policy.json", "policy", policy_value), ("meal-constraints.json", "meal_constraints", constraints_value), ("food-catalog.json", "catalog", catalog_value), ): document = { "version": "v1", "digest": digest(value), "approved": True, "approved_by": "richard", "approved_at_kst": "2026-07-01T09:00:00+09:00", key: value, } (root / filename).write_text(json.dumps(document), encoding="utf-8") artifacts = load_approved_adaptive_artifacts(tmp_path) assert artifacts.policy_digest == digest(policy_value) assert artifacts.meal_constraints.complete assert isinstance(artifacts.catalog, VersionedFoodCatalog) assert artifacts.catalog[0].food_id == "safe" policy_value["calorie_step"] = 200 (root / "policy.json").write_text( json.dumps({ "version": "v1", "digest": digest({ "starts_on": "2026-07-01", "goal_mode": "fat_loss", "weekly_rate_min": "-0.5", "weekly_rate_max": "-0.25", "calorie_step": 100, "calorie_floor": 1800, "calorie_ceiling": 2600, }), "approved": True, "approved_by": "richard", "approved_at_kst": "2026-07-01T09:00:00+09:00", "policy": policy_value, }), encoding="utf-8", ) with pytest.raises(ValueError, match="digest"): load_approved_adaptive_artifacts(tmp_path) def test_schema_two_catalog_parses_food_compatibility_metadata(tmp_path): initialize_adaptive_customer(tmp_path) root = tmp_path / "nutrition-plans" catalog_food = Food( "safe", "safe", 600, 70, 40, 18, allowed_budget_bands=frozenset({"standard"}), allowed_cooking_access=frozenset({"home"}), ) documents = ( ( "policy.json", "policy", { "starts_on": "2026-07-01", "goal_mode": "fat_loss", "weekly_rate_min": "-0.5", "weekly_rate_max": "-0.25", "calorie_step": 100, "calorie_floor": 1800, "calorie_ceiling": 2600, }, "1.0", ), ( "meal-constraints.json", "meal_constraints", { "meal_count": 1, "budget_tier": "standard", "cooking_access": "home", }, "1.0", ), ( "food-catalog.json", "catalog", [json.loads(canonical_json(catalog_food))], "2.0", ), ) for filename, key, value, schema_version in documents: (root / filename).write_text( json.dumps( { "schema_version": schema_version, "version": "v1", "digest": digest(value), "approved": True, "approved_by": "richard", "approved_at_kst": "2026-07-01T09:00:00+09:00", key: value, } ), encoding="utf-8", ) food = load_approved_adaptive_artifacts(tmp_path).catalog[0] assert food.allowed_budget_bands == frozenset({"standard"}) assert food.allowed_cooking_access == frozenset({"home"}) def test_approved_private_artifacts_accept_valid_legacy_catalog_fallback(tmp_path): initialize_adaptive_customer(tmp_path) root = tmp_path / "nutrition-plans" (root / "food-catalog.json").unlink() policy_value = { "starts_on": "2026-07-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_value = { "meal_count": 1, "budget_tier": "standard", "cooking_access": "home", } catalog_value = [{ "food_id": "legacy", "label": "legacy", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] for filename, key, value in ( ("policy.json", "policy", policy_value), ("meal-constraints.json", "meal_constraints", constraints_value), ("catalog.json", "catalog", catalog_value), ): path = root / filename path.write_text(json.dumps({ "version": "v1", "digest": digest(value), "approved": True, "approved_by": "richard", "approved_at_kst": "2026-07-01T09:00:00+09:00", key: value, }), encoding="utf-8") path.chmod(0o600) artifacts = load_approved_adaptive_artifacts(tmp_path) assert artifacts.catalog[0].food_id == "legacy" legacy_path = root / "catalog.json" legacy_document = json.loads(legacy_path.read_text(encoding="utf-8")) legacy_document["catalog"][0]["label"] = "tampered" legacy_path.write_text(json.dumps(legacy_document), encoding="utf-8") legacy_path.chmod(0o600) with pytest.raises(ValueError, match="digest"): load_approved_adaptive_artifacts(tmp_path) def test_approved_private_artifacts_reject_invalid_canonical_catalog_without_legacy_fallback( tmp_path, ): initialize_adaptive_customer(tmp_path) root = tmp_path / "nutrition-plans" policy_value = { "starts_on": "2026-07-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_value = { "meal_count": 1, "budget_tier": "standard", "cooking_access": "home", } canonical_value = [{ "food_id": "canonical", "label": "canonical", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] legacy_value = [{ "food_id": "legacy", "label": "legacy", "calories": 600, "carbs_g": 70, "protein_g": 40, "fat_g": 18, }] for filename, key, value in ( ("policy.json", "policy", policy_value), ("meal-constraints.json", "meal_constraints", constraints_value), ("food-catalog.json", "catalog", canonical_value), ("catalog.json", "catalog", legacy_value), ): path = root / filename path.write_text(json.dumps({ "version": "v1", "digest": digest(value), "approved": True, "approved_by": "richard", "approved_at_kst": "2026-07-01T09:00:00+09:00", key: value, }), encoding="utf-8") path.chmod(0o600) assert load_approved_adaptive_artifacts(tmp_path).catalog[0].food_id == "canonical" canonical_path = root / "food-catalog.json" canonical_document = json.loads(canonical_path.read_text(encoding="utf-8")) canonical_document["approved"] = False canonical_path.write_text(json.dumps(canonical_document), encoding="utf-8") canonical_path.chmod(0o600) with pytest.raises(ValueError, match="not approved"): load_approved_adaptive_artifacts(tmp_path) def test_canonical_projection_keeps_independent_flow_and_safety_union(): events = [ { "event_id": "morning", "event_type": "morning_checkin", "occurred_at_kst": "2026-07-14T08:00:00+09:00", "status": "accepted", "check_in": {"body_weight_kg": 80}, }, { "event_id": "nutrition", "event_type": "nutrition_checkin", "occurred_at_kst": "2026-07-14T21:00:00+09:00", "status": "accepted", "check_in": {"calories_kcal": 2300, "carbohydrate_g": 290, "protein_g": 150, "fat_g": 60}, }, { "event_id": "safety", "event_type": "safety_flag", "occurred_at_kst": "2026-07-14T21:01:00+09:00", "status": "unsafe", }, ] snapshot = project_canonical_events(events, date(2026, 7, 14), date(2026, 7, 1)) assert snapshot.current_samples == 1 assert snapshot.safety_held is True def test_canonical_projection_uses_terminal_superseding_event(): events = [ { "event_id": "root", "event_type": "morning_checkin", "occurred_at_kst": "2026-07-13T08:00:00+09:00", "status": "accepted", "check_in": {"body_weight_kg": 80}, }, { "event_id": "correction", "event_type": "correction", "occurred_at_kst": "2026-07-13T09:00:00+09:00", "status": "accepted", "supersedes": "root", "check_in": {"body_weight_kg": 79}, }, ] snapshot = project_canonical_events(events, date(2026, 7, 14), date(2026, 7, 1)) assert snapshot.current_mean_kg == Decimal("79") def test_superseded_safety_remains_a_hard_hold_and_forks_fail(): base = { "event_id": "safety", "event_type": "safety_flag", "occurred_at_kst": "2026-07-14T08:00:00+09:00", "status": "unsafe", } correction = { "event_id": "correction", "event_type": "correction", "occurred_at_kst": "2026-07-14T09:00:00+09:00", "status": "accepted", "supersedes": "safety", "check_in": {"body_weight_kg": 80}, } snapshot = project_canonical_events( [base, correction], date(2026, 7, 14), date(2026, 7, 1) ) assert snapshot.safety_held is True fork = {**correction, "event_id": "other"} with pytest.raises(ValueError, match="forked"): project_canonical_events( [base, correction, fork], date(2026, 7, 14), date(2026, 7, 1) ) def test_conflicting_duplicate_canonical_event_ids_are_rejected(): event = { "event_id": "same", "event_type": "morning_checkin", "occurred_at_kst": "2026-07-14T08:00:00+09:00", "status": "accepted", "check_in": {"body_weight_kg": 80}, } assert len(canonical_event_records([event, dict(event)])) == 1 with pytest.raises(ValueError, match="conflicting duplicate"): canonical_event_records([event, {**event, "check_in": {"body_weight_kg": 79}}]) def test_canonical_correction_cycle_is_rejected(): events = [ {"event_id": "a", "supersedes": "b"}, {"event_id": "b", "supersedes": "a"}, ] with pytest.raises(ValueError, match="cycle"): project_canonical_events(events, date(2026, 7, 14), date(2026, 7, 1)) def test_adaptive_event_rows_fail_closed_on_tamper(tmp_path): proposal_digest = "a" * 64 parent_digest = "b" * 64 payload = { "proposal_digest": proposal_digest, "parent_digest": parent_digest, "revision": 1, "operator_id": "", } store = AdaptiveEventStore(tmp_path / "events.jsonl") original = store.append( "plan_revision_created", payload, dedupe_key=f"revision:{proposal_digest}", ) path = tmp_path / "events.jsonl" tampered_rows = [] unknown = dict(original) unknown["event_type"] = "unknown_event" unknown["event_id"] = digest({ "event_type": unknown["event_type"], "payload": unknown["payload"], "dedupe_key": unknown["dedupe_key"], }) tampered_rows.append(unknown) invalid_payload = dict(original) invalid_payload["payload"] = { key: value for key, value in invalid_payload["payload"].items() if key != "parent_digest" } invalid_payload["event_id"] = digest({ "event_type": invalid_payload["event_type"], "payload": invalid_payload["payload"], "dedupe_key": invalid_payload["dedupe_key"], }) tampered_rows.append(invalid_payload) invalid_dedupe = dict(original) invalid_dedupe["dedupe_key"] = " " invalid_dedupe["event_id"] = digest({ "event_type": invalid_dedupe["event_type"], "payload": invalid_dedupe["payload"], "dedupe_key": invalid_dedupe["dedupe_key"], }) tampered_rows.append(invalid_dedupe) invalid_digest = dict(original) invalid_digest["event_id"] = "0" * 64 tampered_rows.append(invalid_digest) for row in tampered_rows: path.write_text(canonical_json(row) + "\n", encoding="utf-8") with pytest.raises(ValueError): store.read() with pytest.raises(ValueError): store.recover(path=path) def test_overlay_replay_rejects_unknown_missing_and_impossible_transitions(tmp_path): path = tmp_path / "overlays.jsonl" journal = OverlayJournal(path) with pytest.raises(ValueError, match="predecessor"): journal.append(AdaptiveOverlay( "child", "proposal", "2026-07-02T00:00:00+09:00", supersedes_revision_id="missing", )) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) with pytest.raises(ValueError, match="transition"): journal.replace( "base", AdaptiveOverlay("child", "proposal", "2026-07-01T00:00:00+09:00"), ) with pytest.raises(ValueError, match="target"): journal.rollback( "missing", as_of_kst="2026-07-03T00:00:00+09:00", reason="operator", ) row = { "schema_version": "1.0", "event_type": "adaptive_overlay_unknown", "append_sequence": 2, "payload": {}, } path.write_text( path.read_text(encoding="utf-8") + canonical_json({**row, "row_digest": digest(row)}) + "\n", encoding="utf-8", ) with pytest.raises(ValueError, match="unknown"): journal.resolve(effective_kst="2026-07-01T12:00:00+09:00") def test_overlay_recovery_is_owned_torn_tail_and_append_safe(tmp_path): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append( AdaptiveOverlay( "base", "proposal", "2026-07-01T00:00:00+09:00", effective_through="2026-07-02T00:00:00+09:00", ) ) complete = path.read_bytes() path.write_bytes(complete + b'{"torn":') result = journal.recover() assert result == {str(path): "truncated_torn_tail"} assert path.read_bytes() == complete with journal._locked() as token: assert journal._recover_locked(token) == {str(path): 1} with pytest.raises(TypeError): journal._recover_locked(object()) row = journal.append( AdaptiveOverlay( "child", "proposal", "2026-07-02T00:00:00+09:00", supersedes_revision_id="base", ) ) assert row["append_sequence"] == 2 def test_overlay_recovery_rejects_foreign_and_generic_adaptive_paths(tmp_path): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) before = path.read_bytes() foreign = tmp_path / "foreign.jsonl" foreign.write_bytes(b'{"untouched":true}\n') foreign_before = foreign.read_bytes() with pytest.raises(ValueError, match="foreign"): journal.recover(path=foreign) assert foreign.read_bytes() == foreign_before generic = AdaptiveEventStore(path, root=tmp_path) with pytest.raises(ValueError, match="overlay"): generic.recover(path=path) assert path.read_bytes() == before def test_overlay_recovery_reader_waits_for_authority_transition_writer(tmp_path, monkeypatch): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) complete = path.read_bytes() path.write_bytes(complete + b'{"torn":') reader_lock_attempt = threading.Event() reader_acquired = threading.Event() reader_values: list[bytes] = [] original_flock = adaptive_module.fcntl.flock def observed_flock(descriptor, operation): if operation == fcntl.LOCK_SH: reader_lock_attempt.set() original_flock(descriptor, operation) monkeypatch.setattr(adaptive_module.fcntl, "flock", observed_flock) def reader() -> None: with journal.read_locked(): reader_acquired.set() reader_values.append(path.read_bytes()) writer_context = journal._locked() token = writer_context.__enter__() reader_thread = threading.Thread(target=reader) reader_thread.start() try: assert reader_lock_attempt.wait(timeout=5) assert not reader_acquired.is_set() assert journal._recover_locked(token) == {str(path): "truncated_torn_tail"} finally: writer_context.__exit__(None, None, None) reader_thread.join(timeout=5) assert not reader_thread.is_alive() assert reader_values == [complete] def test_overlay_recovery_rejects_symlink_and_data_inode_replacement(tmp_path, monkeypatch): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) complete = path.read_bytes() outside = tmp_path / "outside.jsonl" outside.write_bytes(b"outside") path.unlink() path.symlink_to(outside) outside_before = outside.read_bytes() with pytest.raises(ValueError): journal.recover() assert outside.read_bytes() == outside_before path.unlink() path.write_bytes(complete + b'{"torn":') path.chmod(0o600) replacement = tmp_path / "replacement.jsonl" replacement.write_bytes(b"replacement") replacement.chmod(0o600) original_read = journal._read_overlay_rows_locked def replace_data_inode(token, *, allow_truncated_tail): result = original_read(token, allow_truncated_tail=allow_truncated_tail) os.replace(replacement, path) return result monkeypatch.setattr(journal, "_read_overlay_rows_locked", replace_data_inode) original_fd = os.open(path, os.O_RDONLY) try: with pytest.raises(ValueError, match="unsafe|replaced"): journal.recover() assert path.read_bytes() == b"replacement" assert os.read(original_fd, len(complete) + 8) == complete + b'{"torn":' finally: os.close(original_fd) def test_overlay_recovery_rejects_lock_inode_replacement_without_mutation(tmp_path, monkeypatch): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) before = path.read_bytes() lock = journal.lock_path replacement = tmp_path / ".authority-transition.lock.replacement" replacement.write_bytes(lock.read_bytes()) replacement.chmod(0o600) original_flock = adaptive_module.fcntl.flock replaced = False def replace_after_lock(descriptor, operation): nonlocal replaced original_flock(descriptor, operation) if operation == fcntl.LOCK_EX and not replaced: replaced = True os.replace(replacement, lock) monkeypatch.setattr(adaptive_module.fcntl, "flock", replace_after_lock) with pytest.raises(ValueError, match="replaced"): journal.recover() assert path.read_bytes() == before def test_overlay_recovery_rejects_authority_lock_symlink_without_mutation(tmp_path): path = tmp_path / "adaptive-overlays.jsonl" journal = OverlayJournal(path, root=tmp_path) journal.append(AdaptiveOverlay("base", "proposal", "2026-07-01T00:00:00+09:00")) before = path.read_bytes() outside = tmp_path / "outside.lock" outside.write_bytes(b"untouched") outside.chmod(0o600) journal.lock_path.unlink() journal.lock_path.symlink_to(outside) with pytest.raises(OSError): journal.recover() assert outside.read_bytes() == b"untouched" assert path.read_bytes() == before def test_partial_canonical_adherence_is_contradictory_and_requires_review(): day = date(2026, 7, 14) signal = derive_adherence_signal(DailyObservation( day, target_calories_kcal=2000, actual_protein_g=120, )) assert signal.status == "contradictory" assert signal.complete is False assert signal.adherent is None rows = observations(day, rate_delta=Decimal("0.5")) rows[-1] = DailyObservation( day, weight_kg=Decimal("80.5"), target_calories_kcal=2000, actual_protein_g=120, ) snapshot = build_snapshot(rows, day, date(2026, 7, 1)) result = propose( "client_001", snapshot, policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert result.decision is Decision.HUMAN_REVIEW assert result.reasons == ("adherence_evidence_contradictory",) def test_operator_note_is_bounded_operator_only_and_digest_pinned(): day = date(2026, 7, 14) proposal = propose( "client_001", build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, date(2026, 7, 1)), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) child = create_proposal_revision(proposal, operator_note="private operator note") assert "private operator note" in child.operator_body assert "private operator note" not in child.customer_body assert child.operator_body_digest == render_operator_digest(child) with pytest.raises(ValueError, match="too long"): create_proposal_revision(proposal, operator_note="x" * 2001) def test_config_epoch_public_prepared_committed_round_trip(tmp_path): path = tmp_path / "nutrition-plans" / "events.jsonl" store = AdaptiveEventStore(path) prepared = store.append_config_epoch( 7, "a" * 64, ("client_002", "client_001"), state="prepared", customer_states={ "client_001": "pending", "client_002": "pending", }, ) assert prepared["state"] == "prepared" reloaded = AdaptiveEventStore(path) committed = reloaded.append_config_epoch( 7, "a" * 64, ("client_001", "client_002"), state="committed", customer_states={ "client_001": "committed", "client_002": "committed", }, ) assert committed["state"] == "committed" rows = AdaptiveEventStore(path).journal_rows("config_epoch") assert [row["state"] for row in rows] == ["prepared", "committed"] assert rows[-1]["customer_keys"] == ["client_001", "client_002"] with pytest.raises(ValueError, match="requires append_config_epoch"): reloaded.append_journal( "config_epoch", { "epoch": 8, "config_digest": "b" * 64, "customer_keys": ["client_001"], }, intent_id="epoch:8", state="prepared", ) def test_config_epoch_typed_inputs_fail_before_persistence(tmp_path): path = tmp_path / "nutrition-plans" / "events.jsonl" store = AdaptiveEventStore(path) valid_keys = ("client_001",) valid_states = {"client_001": "pending"} invalid_calls = ( {"epoch": True}, {"epoch": -1}, {"config_digest": "z" * 64}, {"config_digest": "a" * 63}, {"customer_keys": ("client_001", "client_001")}, {"customer_keys": ("",)}, {"customer_keys": ("client_001", 2)}, {"customer_states": {"client_001": "committed"}}, {"customer_states": {"client_001": "pending", "extra": "pending"}}, ) for overrides in invalid_calls: kwargs = { "epoch": 1, "config_digest": "a" * 64, "customer_keys": valid_keys, "state": "prepared", "customer_states": valid_states, } kwargs.update(overrides) with pytest.raises(ValueError): store.append_config_epoch(**kwargs) assert not path.exists() def test_feature_config_digest_uses_one_canonical_preimage(tmp_path): flags = { "analytics_shadow": False, "operator_candidates": True, "activation": False, "delivery": False, } preimage = feature_config_digest_preimage(4, flags) assert preimage == {"epoch": 4, **flags} assert feature_config_digest(4, flags) == digest(preimage) initialize_adaptive_customer(tmp_path) feature = json.loads( (tmp_path / "nutrition-plans" / "feature-epoch.json").read_text(encoding="utf-8") ) current_flags = {key: feature[key] for key in flags} assert feature["config_digest"] == feature_config_digest(feature["epoch"], current_flags) def test_weekly_carb_cycle_reconciles_ties_and_actual_session_precedence(): start = date(2026, 7, 15) target = MacroTarget(2300, 290, 150, 60) planned = tuple( (start + timedelta(days=index), category) for index, category in enumerate(("high", "medium", "low", "high", "medium", "low", "low")) ) cycle = compile_weekly_carb_cycle(target, planned) assert cycle is not None assert cycle.reconciles_exactly assert cycle.weekly_calories == target.calories * 7 assert sum(item.target.calories for item in cycle.targets) == target.calories * 7 assert cycle.days == planned all_medium = {start + timedelta(days=index): "medium" for index in range(7)} actual = {start + timedelta(days=3): "high"} actual_cycle = compile_weekly_carb_cycle( target, planned_sessions=all_medium, actual_sessions=actual, ) assert actual_cycle is not None assert actual_cycle.days[3] == (start + timedelta(days=3), "high") assert actual_cycle.days.count((start + timedelta(days=3), "medium")) == 0 def test_actual_as_planned_resolves_against_approved_high_load(): start = date(2026, 7, 15) planned = tuple( (start + timedelta(days=index), category) for index, category in enumerate( ("high", "medium", "low", "high", "medium", "low", "low") ) ) actual = ( { "kst_day": start, "session_done": True, "intensity_vs_plan": "as_planned", }, ) cycle = compile_weekly_carb_cycle( MacroTarget(2300, 290, 150, 60), planned_sessions=planned, actual_sessions=actual, ) assert cycle is not None assert cycle.days[0] == (start, "high") def test_incomplete_actual_session_resolves_to_rest_day(): start = date(2026, 7, 15) planned = tuple( (start + timedelta(days=index), "medium") for index in range(7) ) actual = ( { "kst_day": start, "session_done": False, "intensity_vs_plan": "above", }, ) cycle = compile_weekly_carb_cycle( MacroTarget(2300, 290, 150, 60), planned_sessions=planned, actual_sessions=actual, ) assert cycle is not None assert cycle.days[0] == (start, "low") def test_completed_actual_session_against_planned_rest_fails_closed(): start = date(2026, 7, 15) planned = tuple( (start + timedelta(days=index), category) for index, category in enumerate( ("rest", "medium", "low", "high", "medium", "low", "low") ) ) actual = ( { "kst_day": start, "session_done": True, "intensity_vs_plan": "above", }, ) assert compile_weekly_carb_cycle( MacroTarget(2300, 290, 150, 60), planned_sessions=planned, actual_sessions=actual, ) is None def test_conflicting_actual_sessions_for_one_day_fail_closed(): start = date(2026, 7, 15) planned = tuple( (start + timedelta(days=index), "medium") for index in range(7) ) actual = ( { "kst_day": start, "session_done": True, "intensity_vs_plan": "below", }, { "kst_day": start, "session_done": True, "intensity_vs_plan": "above", }, ) assert compile_weekly_carb_cycle( MacroTarget(2300, 290, 150, 60), planned_sessions=planned, actual_sessions=actual, ) is None def test_weekly_nutrition_plan_requires_seven_consecutive_exact_days(): daily_plan_type = getattr(adaptive_module, "DailyNutritionPlan", None) weekly_plan_type = getattr(adaptive_module, "WeeklyNutritionPlan", None) assert daily_plan_type is not None assert weekly_plan_type is not None start = date(2026, 7, 15) target = MacroTarget(600, 70, 40, 18) catalog = VersionedFoodCatalog( foods=(Food("safe", "한 끼", 600, 70, 40, 18),), version="v1", digest=digest(("safe",)), approved=True, ) meal_plan = compile_meal_plan( target, MealConstraints(1, budget_tier="standard", cooking_access="home"), catalog, ) assert meal_plan is not None days = tuple( daily_plan_type( kst_day=start + timedelta(days=offset), category="medium", target=target, meal_plan=meal_plan, ) for offset in range(7) ) plan = weekly_plan_type( horizon_start=start, as_of_kst_day=start, frozen_through=None, base_target=target, days=days, ) assert tuple(item.kst_day for item in plan.days) == tuple( start + timedelta(days=offset) for offset in range(7) ) assert plan.weekly_calories == target.calories * 7 assert plan.weekly_carbs_g == target.carbs_g * 7 assert plan.weekly_protein_g == target.protein_g * 7 assert plan.weekly_fat_g == target.fat_g * 7 assert plan.reconciles_exactly def test_weekly_nutrition_plan_rejects_nonconsecutive_days(): daily_plan_type = getattr(adaptive_module, "DailyNutritionPlan", None) weekly_plan_type = getattr(adaptive_module, "WeeklyNutritionPlan", None) assert daily_plan_type is not None assert weekly_plan_type is not None start = date(2026, 7, 15) target = MacroTarget(600, 70, 40, 18) catalog = VersionedFoodCatalog( foods=(Food("safe", "한 끼", 600, 70, 40, 18),), version="v1", digest=digest(("safe",)), approved=True, ) meal_plan = compile_meal_plan( target, MealConstraints(1, budget_tier="standard", cooking_access="home"), catalog, ) assert meal_plan is not None days = tuple( daily_plan_type( kst_day=start + timedelta(days=offset + (1 if offset == 6 else 0)), category="medium", target=target, meal_plan=meal_plan, ) for offset in range(7) ) with pytest.raises(ValueError, match="consecutive"): weekly_plan_type( horizon_start=start, as_of_kst_day=start, frozen_through=None, base_target=target, days=days, ) def test_daily_nutrition_plan_rejects_inexact_meal_target(): daily_plan_type = getattr(adaptive_module, "DailyNutritionPlan", None) assert daily_plan_type is not None meal_target = MacroTarget(600, 70, 40, 18) catalog = VersionedFoodCatalog( foods=(Food("safe", "한 끼", 600, 70, 40, 18),), version="v1", digest=digest(("safe",)), approved=True, ) meal_plan = compile_meal_plan( meal_target, MealConstraints(1, budget_tier="standard", cooking_access="home"), catalog, ) assert meal_plan is not None with pytest.raises(ValueError, match="meal plan"): daily_plan_type( kst_day=date(2026, 7, 15), category="high", target=MacroTarget(604, 71, 40, 18), meal_plan=meal_plan, ) def test_compile_weekly_nutrition_plan_builds_seven_exact_meals(): compiler = getattr(adaptive_module, "compile_weekly_nutrition_plan", None) assert compiler is not None start = date(2026, 7, 15) base = MacroTarget(2300, 290, 150, 60) cycle = compile_weekly_carb_cycle( base, tuple( (start + timedelta(days=offset), category) for offset, category in enumerate( ("high", "medium", "low", "high", "medium", "low", "low") ) ), category_delta_calories=40, ) assert cycle is not None foods = tuple( Food( f"exact-{category}", f"{category} 식단", target.calories, target.carbs_g, target.protein_g, target.fat_g, allowed_budget_bands=frozenset({"standard"}), allowed_cooking_access=frozenset({"home"}), ) for category, target in ( ("high", cycle.high), ("medium", cycle.medium), ("low", cycle.low), ) if target is not None ) catalog = VersionedFoodCatalog( foods=foods, version="v2", digest=digest(foods), schema_version="2.0", approved=True, ) plan = compiler( cycle, MealConstraints( 1, budget_tier="standard", cooking_access="home", max_foods_per_meal=1, max_serving_units_per_meal=1, ), catalog, as_of_kst_day=start, ) assert plan is not None assert len(plan.days) == 7 assert all(item.meal_plan.exact for item in plan.days) assert all(item.meal_plan.target == item.target for item in plan.days) assert plan.reconciles_exactly def test_weekly_plan_excludes_budget_or_cooking_incompatible_food(): compiler = getattr(adaptive_module, "compile_weekly_nutrition_plan", None) assert compiler is not None start = date(2026, 7, 15) base = MacroTarget(2300, 290, 150, 60) cycle = compile_weekly_carb_cycle( base, tuple((start + timedelta(days=offset), "medium") for offset in range(7)), ) assert cycle is not None foods = ( Food( "premium-only", "프리미엄 외식", 2300, 290, 150, 60, allowed_budget_bands=frozenset({"premium"}), allowed_cooking_access=frozenset({"restaurant"}), ), ) catalog = VersionedFoodCatalog( foods=foods, version="v2", digest=digest(foods), schema_version="2.0", approved=True, ) assert compiler( cycle, MealConstraints( 1, budget_tier="standard", cooking_access="home", max_foods_per_meal=1, max_serving_units_per_meal=1, ), catalog, as_of_kst_day=start, ) is None def test_one_unsolved_daily_meal_rejects_entire_week(): compiler = getattr(adaptive_module, "compile_weekly_nutrition_plan", None) assert compiler is not None start = date(2026, 7, 15) base = MacroTarget(2300, 290, 150, 60) cycle = compile_weekly_carb_cycle( base, tuple( (start + timedelta(days=offset), category) for offset, category in enumerate( ("high", "medium", "low", "high", "medium", "low", "low") ) ), category_delta_calories=40, ) assert cycle is not None foods = tuple( Food( f"exact-{category}", f"{category} 식단", target.calories, target.carbs_g, target.protein_g, target.fat_g, allowed_budget_bands=frozenset({"standard"}), allowed_cooking_access=frozenset({"home"}), ) for category, target in ( ("medium", cycle.medium), ("low", cycle.low), ) if target is not None ) catalog = VersionedFoodCatalog( foods=foods, version="v2", digest=digest(foods), schema_version="2.0", approved=True, ) assert compiler( cycle, MealConstraints( 1, budget_tier="standard", cooking_access="home", max_foods_per_meal=1, max_serving_units_per_meal=1, ), catalog, as_of_kst_day=start, ) is None def _weekly_reconciliation_fixture(): start = date(2026, 7, 15) base = MacroTarget(2300, 290, 150, 60) planned = tuple( (start + timedelta(days=offset), category) for offset, category in enumerate( ("high", "medium", "low", "high", "medium", "low", "low") ) ) parent_cycle = compile_weekly_carb_cycle( base, planned, category_delta_calories=40, ) assert parent_cycle is not None foods = tuple( Food( f"exact-{carbs_g}", f"탄수 {carbs_g}g 식단", 4 * carbs_g + 4 * base.protein_g + 9 * base.fat_g, carbs_g, base.protein_g, base.fat_g, allowed_budget_bands=frozenset({"standard"}), allowed_cooking_access=frozenset({"home"}), ) for carbs_g in range(200, 381) ) catalog = VersionedFoodCatalog( foods=foods, version="v2", digest=digest(foods), schema_version="2.0", approved=True, ) constraints = MealConstraints( 1, budget_tier="standard", cooking_access="home", max_foods_per_meal=1, max_serving_units_per_meal=1, ) parent = adaptive_module.compile_weekly_nutrition_plan( parent_cycle, constraints, catalog, as_of_kst_day=start, ) assert parent is not None return start, base, planned, parent, constraints, catalog def test_actual_reconciliation_freezes_parent_prefix_and_horizon(): reconciler = getattr(adaptive_module, "reconcile_weekly_nutrition_plan", None) assert reconciler is not None start, base, planned, parent, constraints, catalog = ( _weekly_reconciliation_fixture() ) actual_day = start + timedelta(days=1) effective_cycle = compile_weekly_carb_cycle( base, planned_sessions=planned, actual_sessions=( { "kst_day": actual_day, "session_done": True, "intensity_vs_plan": "above", }, ), category_delta_calories=40, ) assert effective_cycle is not None child = reconciler( parent, effective_cycle, constraints, catalog, actual_day=actual_day, as_of_kst_day=actual_day, exercise_evidence_digest=digest(("actual-1",)), ) assert child is not None assert child.horizon_start == parent.horizon_start assert child.horizon_end == parent.horizon_end assert child.frozen_through == actual_day assert tuple( (item.target, item.meal_plan.digest) for item in child.days[:2] ) == tuple( (item.target, item.meal_plan.digest) for item in parent.days[:2] ) assert any( child.days[index].digest != parent.days[index].digest for index in range(2, 7) ) assert child.reconciles_exactly def test_late_actual_freezes_through_current_as_of_day(): reconciler = getattr(adaptive_module, "reconcile_weekly_nutrition_plan", None) assert reconciler is not None start, base, planned, parent, constraints, catalog = ( _weekly_reconciliation_fixture() ) actual_day = start + timedelta(days=1) as_of_day = start + timedelta(days=3) effective_cycle = compile_weekly_carb_cycle( base, planned_sessions=planned, actual_sessions=( { "kst_day": actual_day, "session_done": True, "intensity_vs_plan": "above", }, ), category_delta_calories=40, ) assert effective_cycle is not None child = reconciler( parent, effective_cycle, constraints, catalog, actual_day=actual_day, as_of_kst_day=as_of_day, exercise_evidence_digest=digest(("actual-late",)), ) assert child is not None assert child.frozen_through == as_of_day assert tuple( (item.target, item.meal_plan.digest) for item in child.days[:4] ) == tuple( (item.target, item.meal_plan.digest) for item in parent.days[:4] ) assert child.reconciles_exactly def test_last_day_actual_never_rewrites_nutrition(): reconciler = getattr(adaptive_module, "reconcile_weekly_nutrition_plan", None) assert reconciler is not None start, base, planned, parent, constraints, catalog = ( _weekly_reconciliation_fixture() ) actual_day = start + timedelta(days=6) effective_cycle = compile_weekly_carb_cycle( base, planned_sessions=planned, actual_sessions=( { "kst_day": actual_day, "session_done": True, "intensity_vs_plan": "above", }, ), category_delta_calories=40, ) assert effective_cycle is not None child = reconciler( parent, effective_cycle, constraints, catalog, actual_day=actual_day, as_of_kst_day=actual_day, exercise_evidence_digest=digest(("actual-last",)), ) assert child is not None assert tuple( (item.target, item.meal_plan.digest) for item in child.days ) == tuple( (item.target, item.meal_plan.digest) for item in parent.days ) assert child.reconciles_exactly def test_unsolved_future_meal_rejects_reconciliation(): reconciler = getattr(adaptive_module, "reconcile_weekly_nutrition_plan", None) assert reconciler is not None start, base, planned, parent, constraints, _ = ( _weekly_reconciliation_fixture() ) actual_day = start + timedelta(days=1) effective_cycle = compile_weekly_carb_cycle( base, planned_sessions=planned, actual_sessions=( { "kst_day": actual_day, "session_done": True, "intensity_vs_plan": "above", }, ), category_delta_calories=40, ) assert effective_cycle is not None impossible_foods = ( Food( "impossible", "맞지 않는 식단", 1200, 100, 100, 44, allowed_budget_bands=frozenset({"standard"}), allowed_cooking_access=frozenset({"home"}), ), ) impossible_catalog = VersionedFoodCatalog( foods=impossible_foods, version="v2", digest=digest(impossible_foods), schema_version="2.0", approved=True, ) assert reconciler( parent, effective_cycle, constraints, impossible_catalog, actual_day=actual_day, as_of_kst_day=actual_day, exercise_evidence_digest=digest(("actual-unsolved",)), ) is None def test_propose_emits_macro_redistribution_for_reconciled_cycle(): start = date(2026, 7, 1) day = date(2026, 7, 14) rows = observations(day) loads = ("high", "medium", "low", "high", "medium", "low", "low") rows.extend( DailyObservation(day + timedelta(days=index + 1), exercise_load=load) for index, load in enumerate(loads) ) result = propose( "client_001", build_snapshot(rows, day, start), policy(start), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert result.decision is Decision.MACRO_REDISTRIBUTION assert result.weekly_carb_cycle is not None assert result.weekly_carb_cycle.reconciles_exactly def test_proposal_pins_and_renders_seven_daily_meal_plans(): start, base, planned, _, constraints, catalog = ( _weekly_reconciliation_fixture() ) evaluation_day = start rows = observations(evaluation_day, rate_delta=Decimal("-0.3")) proposal = propose( "client_001", build_snapshot(rows, evaluation_day, start - timedelta(days=14)), policy(start - timedelta(days=14)), current_target=base, protein_g=base.protein_g, fat_g=base.fat_g, meal_constraints=constraints, catalog=catalog, planned_sessions=planned, ) weekly_plan = getattr(proposal, "weekly_nutrition_plan", None) assert weekly_plan is not None assert len(weekly_plan.days) == 7 assert proposal.weekly_nutrition_plan_digest == weekly_plan.digest body = render_customer_body(proposal) for daily in weekly_plan.days: assert daily.kst_day.isoformat() in body assert daily.category in body assert f"탄수 {daily.target.carbs_g}g" in body assert daily.meal_plan.slots[0].food_ids[0] in body preserved = create_proposal_revision(proposal) assert preserved.weekly_nutrition_plan == weekly_plan changed = create_proposal_revision( proposal, target=MacroTarget(2200, 265, 150, 60), ) assert changed.weekly_nutrition_plan is None def test_weekly_carb_cycle_infeasible_constraints_fail_closed(): start = date(2026, 7, 15) schedule = tuple( (start + timedelta(days=index), category) for index, category in enumerate(("high", "medium", "low", "high", "medium", "low", "low")) ) assert compile_weekly_carb_cycle( MacroTarget(2300, 290, 150, 60), schedule, max_category_delta_calories=0, require_distinct_categories=True, ) is None def test_global_upper_bounds_and_policy_infeasibility_fail_closed(): assert solve_macros(GLOBAL_MAX_CALORIES, GLOBAL_MIN_PROTEIN_G, GLOBAL_MIN_FAT_G) is not None assert solve_macros(GLOBAL_MAX_CALORIES + 1, GLOBAL_MIN_PROTEIN_G, GLOBAL_MIN_FAT_G) is None assert solve_macros(GLOBAL_MAX_CALORIES, GLOBAL_MAX_PROTEIN_G, GLOBAL_MIN_FAT_G) is not None assert solve_macros(GLOBAL_MIN_CALORIES, GLOBAL_MAX_PROTEIN_G + 1, GLOBAL_MIN_FAT_G) is None assert solve_macros(GLOBAL_MAX_CALORIES - 2, GLOBAL_MIN_PROTEIN_G, GLOBAL_MAX_FAT_G) is not None assert solve_macros(GLOBAL_MIN_CALORIES, GLOBAL_MIN_PROTEIN_G, GLOBAL_MAX_FAT_G + 1) is None start = date(2026, 7, 1) day = date(2026, 7, 14) result = propose( "client_001", build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, start), policy( start, calorie_floor=GLOBAL_MAX_CALORIES + 1, calorie_ceiling=GLOBAL_MAX_CALORIES, ), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert result.decision is Decision.HUMAN_REVIEW assert "global_floor_violation" in result.reasons def test_cooldown_uses_latest_committed_calorie_overlay_only(): history = [ { "revision_id": "macro", "state": "committed", "calorie_changing": False, "committed_at_kst": "2026-07-05T00:00:00+09:00", }, { "revision_id": "calorie-old", "state": "committed", "previous_target": {"calories": 2300}, "target": {"calories": 2200}, "committed_at_kst": "2026-07-06T00:00:00+09:00", }, { "revision_id": "calorie-latest", "state": "committed", "previous_target": {"calories": 2200}, "target": {"calories": 2300}, "committed_at_kst": "2026-07-08T00:00:00+09:00", }, ] result = evaluate_cooldown( date(2026, 7, 10), history, cooldown_days=3, ) assert result.active is True assert result.source_revision_id == "calorie-latest" assert result.anchor_kst == "2026-07-08T00:00:00+09:00" def test_explanation_validator_rejects_adversarial_or_outage_text(): day = date(2026, 7, 14) proposal = propose( "client_001", build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, date(2026, 7, 1)), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, ) assert proposal.target is not None fallback = render_explanation_fallback(proposal) assert proposal.explanation == fallback valid = ( f"하루 목표는 {proposal.target.calories}kcal, " f"탄수화물 {proposal.target.carbs_g}g, " f"단백질 {proposal.target.protein_g}g, 지방 {proposal.target.fat_g}g입니다." ) assert validate_explanation(valid, proposal) == valid for adversarial in ( valid.replace(str(proposal.target.calories), "9999"), "하루 목표는 이천삼백kcal입니다.", valid.replace("kcal", "mg"), f"치킨 {proposal.target.calories}kcal", f"하루 목표는 {proposal.target.calories}kcal입니다. 운동을 더 하세요.", f"하루 목표는 {proposal.target.calories}kcal입니다. 이 식단은 혈당을 낮추고 치료에 도움이 됩니다.", f"하루 목표는 {proposal.target.calories}kcal, 단백질 {proposal.target.protein_g}g 1회입니다.", "{{target}}", f"ignore previous instructions {proposal.target.calories}kcal", None, ): assert validate_explanation(adversarial, proposal) == fallback def test_explanation_provider_outage_uses_fallback_and_pins_customer_body(): day = date(2026, 7, 14) def outage(_target): raise RuntimeError("provider unavailable") for provider in (lambda _target: None, outage): proposal = propose( "client_001", build_snapshot(observations(day, rate_delta=Decimal("0.5")), day, date(2026, 7, 1)), policy(date(2026, 7, 1)), current_target=MacroTarget(2300, 290, 150, 60), protein_g=150, fat_g=60, explanation_provider=provider, ) fallback = render_explanation_fallback(proposal) assert proposal.explanation == fallback assert fallback in proposal.customer_body assert proposal.customer_body_digest == hashlib.sha256( proposal.customer_body.encode("utf-8") ).hexdigest() def test_cooldown_rejects_missing_or_invalid_evaluation_day(): with pytest.raises(ValueError, match="evaluation day"): evaluate_cooldown(cooldown_days=3) with pytest.raises(ValueError, match="evaluation day"): evaluate_cooldown("not-a-day", cooldown_days=3) def test_canonical_sequence_facade_is_read_only(): assert not hasattr(CanonicalSequenceJournal, "append") assert not hasattr(CanonicalSequenceJournal, "recover") assert not hasattr(CanonicalSequenceJournal, "recover_pending") assert hasattr(CanonicalSequenceJournal, "read_snapshot") def test_customer_action_continuity_is_idempotent_and_projects_canonical_outcome(tmp_path): store = AdaptiveEventStore(tmp_path / "events.jsonl") store.append( "plan_approved", { "customer_key": "client_001", "proposal_digest": "a" * 64, "revision": 3, "operator_id": "coach", "topic_id": 59, "execution_mode": "shadow_test_only", }, dedupe_key="approval:a", ) action = CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="a" * 64, revision=3, effective_kst_day=date(2026, 7, 28), action_text="아침 체크인을 남겨 주세요.", action_atom="morning_checkin", criterion_text="체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-30T09:00:00+09:00", ) first = store.append_customer_action_continuity(action) assert store.append_customer_action_continuity(action) == first second = CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="a" * 64, revision=3, effective_kst_day=date(2026, 7, 28), action_text="수분 섭취를 기록해 주세요.", action_atom="hydration_check", criterion_text="체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-30T09:00:00+09:00", ) store.append_customer_action_continuity(second) with pytest.raises(ValueError, match="conflicting"): store.append_customer_action_continuity( CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="a" * 64, revision=3, effective_kst_day=date(2026, 7, 28), action_text="다른 행동", action_atom="morning_checkin", criterion_text="체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-30T09:00:00+09:00", ) ) outcomes = store.project_customer_action_outcomes( customer_key="client_001", canonical_events=({ "event_id": "canonical-1", "observation_kst_day": "2026-07-29", "check_in": {"body_weight_kg": "80.0"}, },), as_of_kst_day=date(2026, 7, 29), ) assert len(outcomes) == 2 assert {outcome.outcome for outcome in outcomes} == {"met"} assert store.project_customer_action_outcomes( customer_key="other_customer", canonical_events=(), as_of_kst_day=date(2026, 7, 29), ) == () def test_customer_action_outcome_ignores_same_day_evidence_before_approval(tmp_path): store = AdaptiveEventStore(tmp_path / "anchored-actions.jsonl") store.append( "plan_approved", { "customer_key": "client_001", "proposal_digest": "b" * 64, "revision": 1, "operator_id": "coach", "topic_id": 59, "execution_mode": "shadow_test_only", }, dedupe_key="approval:b", ) action = CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="b" * 64, revision=1, effective_kst_day=date(2026, 7, 28), action_text="다음 체크인을 기록해 주세요.", action_atom="next_checkin", criterion_text="체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-29T08:00:00+09:00", approved_at_kst="2026-07-28T12:00:00+09:00", ) store.append_customer_action_continuity(action) before = { "event_id": "before-approval", "occurred_at_kst": "2026-07-28T08:00:00+09:00", "observation_kst_day": "2026-07-28", "check_in": {"body_weight_kg": "80.0"}, } after = { "event_id": "after-approval", "occurred_at_kst": "2026-07-28T13:00:00+09:00", "observation_kst_day": "2026-07-28", "check_in": {"body_weight_kg": "80.0"}, } before_only = store.project_customer_action_outcomes( customer_key="client_001", canonical_events=(before,), as_of_kst_day=date(2026, 7, 29), ) with_after = store.project_customer_action_outcomes( customer_key="client_001", canonical_events=(before, after), as_of_kst_day=date(2026, 7, 29), ) assert before_only[0].outcome == "insufficient" assert with_after[0].outcome == "met" with pytest.raises(ValueError, match="customer boundary"): store.project_customer_action_outcomes( customer_key="client_001", canonical_events=({**after, "customer_key": "other_customer"},), as_of_kst_day=date(2026, 7, 29), ) def test_prepared_customer_action_is_invisible_until_plan_approval(tmp_path): store = AdaptiveEventStore(tmp_path / "prepared-actions.jsonl") action = CustomerActionContinuity( customer_key="client_001", approved_proposal_digest="c" * 64, revision=1, effective_kst_day=date(2026, 7, 28), action_text="다음 체크인을 기록해 주세요.", action_atom="next_checkin", criterion_text="체크인 기록", criterion_atom="checkin_recorded", next_check_kst="2026-07-29T08:00:00+09:00", ) store.append_customer_action_continuity(action) evidence = ({ "event_id": "later-checkin", "occurred_at_kst": "2026-07-28T13:00:00+09:00", "observation_kst_day": "2026-07-28", "check_in": {"body_weight_kg": "80.0"}, },) assert store.project_customer_action_outcomes( customer_key="client_001", canonical_events=evidence, as_of_kst_day=date(2026, 7, 29), ) == () store.append( "plan_approved", { "customer_key": "client_001", "proposal_digest": "c" * 64, "revision": 1, "operator_id": "coach", "topic_id": 59, "execution_mode": "shadow_test_only", }, dedupe_key="approval:c", ) assert store.project_customer_action_outcomes( customer_key="client_001", canonical_events=evidence, as_of_kst_day=date(2026, 7, 29), )[0].outcome == "met"