"""Persistent button-wizard domain behavior.""" from __future__ import annotations import json import os from datetime import date from multiprocessing import get_context from multiprocessing.queues import Queue from pathlib import Path import pytest from checkin_cli.models import SafetyReason, validate_event from checkin_cli.customer_grounding import referral_guidance, safe_hold_reasons from checkin_cli.store import current_terminal_morning_response from checkin_cli.wizard import ( WizardContext, WizardBranch, WizardFlow, WizardService, WizardStatus, deterministic_branch, project_customer_event_state, ) OWNER = "owner-1" TOPIC = "topic-1" DAY = "2026-07-20" OWNER_CUSTOMER = "customer-1" AC21_SAFETY_CASES = ( pytest.param( { "fixture_id": "SF-S1-C", "flow": WizardFlow.MORNING, "field": "pain", "prefix": ( ("value", "70.2"), ("value", "7.5"), ("select", "4"), ("select", "4"), ), "action": "value", "raw": "흉통이 있어요", "rule_id": "S1", "class_name": "urgent", "source_flow": "customer_checkin", "matched_field": "pain_summary", "signal": "urgent_symptom", "level": "stop_and_escalate", "referral_marker": "의료기관", }, id="SF-S1-C", ), pytest.param( { "fixture_id": "SF-S2-C", "flow": WizardFlow.MORNING, "field": "pain", "prefix": ( ("value", "70.2"), ("value", "7.5"), ("select", "4"), ("select", "4"), ), "action": "value", "raw": "무릎 통증이 있어요", "rule_id": "S2", "class_name": "pain", "source_flow": "customer_checkin", "matched_field": "pain_summary", "signal": "pain", "level": "monitor", "referral_marker": "의료진", }, id="SF-S2-C", ), pytest.param( { "fixture_id": "SF-S3-C", "flow": WizardFlow.MORNING, "field": "pain", "prefix": ( ("value", "70.2"), ("value", "7.5"), ("select", "4"), ("select", "4"), ), "action": "value", "raw": "병원 진단을 받았어요", "rule_id": "S3", "class_name": "disease", "source_flow": "customer_checkin", "matched_field": "pain_summary", "signal": "disease", "level": "monitor", "referral_marker": "의료진", }, id="SF-S3-C", ), pytest.param( { "fixture_id": "SF-S4-C", "flow": WizardFlow.NUTRITION, "field": "calories", "prefix": (("value", "70.2"),), "action": "value", "raw": "900", "rule_id": "S4", "class_name": "eating_risk", "source_flow": "customer_checkin", "matched_field": "calories_kcal", "signal": "eating_risk", "level": "monitor", "referral_marker": "의료진", }, id="SF-S4-C", ), pytest.param( { "fixture_id": "SF-S5-C", "flow": WizardFlow.MORNING, "field": "pain", "prefix": ( ("value", "70.2"), ("value", "7.5"), ("select", "4"), ("select", "4"), ), "action": "value", "raw": "스테로이드를 복용 중이에요", "rule_id": "S5", "class_name": "drugs", "source_flow": "customer_checkin", "matched_field": "pain_summary", "signal": "drugs", "level": "monitor", "referral_marker": "의료진", }, id="SF-S5-C", ), pytest.param( { "fixture_id": "SF-S6-C", "flow": WizardFlow.MORNING, "field": "training_plan", "prefix": ( ("value", "70.2"), ("value", "7.5"), ("select", "4"), ("select", "4"), ("select", "none"), ("value", "2300"), ), "action": "value", "raw": "lose 5 kg in 2 weeks", "rule_id": "S6", "class_name": "extreme_manipulation", "source_flow": "customer_checkin", "matched_field": "stated_goal", "signal": "extreme_manipulation", "level": "monitor", "referral_marker": "의료진", }, id="SF-S6-C", ), ) def _start_morning_in_process(home: str, queue: Queue[str]) -> None: """Start the exact same session from an independent local process.""" result = WizardService.for_standalone(Path(home)).start_morning(WizardContext(OWNER, TOPIC), DAY) queue.put(result.session_id) def _morning(service: WizardService) -> str: """Start one owner/topic-bound morning session.""" return service.start_morning(WizardContext(OWNER, TOPIC), DAY).session_id def _answer( service: WizardService, session_id: str, version: int, action: str, value: str | None = None, ) -> int: """Submit one expected wizard transition and return the new version.""" result = service.answer(WizardContext(OWNER, TOPIC), session_id, version, action, value) assert result.status is WizardStatus.ADVANCED return result.version def _complete_morning(service: WizardService, session_id: str) -> int: """Advance the daily flow to its confirmation summary.""" version = 0 version = _answer(service, session_id, version, "value", "70.2") version = _answer(service, session_id, version, "value", "7.5") version = _answer(service, session_id, version, "select", "4") version = _answer(service, session_id, version, "select", "4") version = _answer(service, session_id, version, "select", "none") version = _answer(service, session_id, version, "value", "2300") version = _answer(service, session_id, version, "value", "feasible") return _answer(service, session_id, version, "value", "none") def test_deterministic_branch_fixtures_use_canonical_customer_state() -> None: normal_answers = { "bodyweight": "70.2", "sleep_duration": "7.5", "sleep_quality": "4", "condition": "4", "pain": "none", "calories": "2300", "completion": "done", "workout_quality": "4", } normal = deterministic_branch(normal_answers, prior_week_same_weekday_weight=70.0) assert normal.branch is WizardBranch.NORMAL assert normal.follow_up_ids == () assert normal.detailed is False performance = deterministic_branch({**normal_answers, "completion": "missed"}) assert performance.branch is WizardBranch.ANOMALY assert performance.follow_up_ids == ("Q-PERF-REASON", "Q-PERF-NEXT") assert performance.detailed is True state = { "kst_day": DAY, "events": ( { "event_id": "prior-event", "event_type": "morning_checkin", "occurred_at_kst": "2026-07-13T08:00:00+09:00", "status": "accepted", "check_in": {"body_weight_kg": 70.2}, }, ), } projected = project_customer_event_state(state) assert projected["prior_week_same_weekday_weight"] == 70.2 changed = deterministic_branch( {**normal_answers, "bodyweight": "71.4"}, canonical_event_state=state, ) assert changed.branch is WizardBranch.CHANGE assert changed.follow_up_ids == () assert changed.reasons == ("weight_change",) def test_missed_workout_reaches_bounded_performance_followups(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_workout(WizardContext(OWNER, TOPIC), DAY) result = service.answer( WizardContext(OWNER, TOPIC), started.session_id, started.version, "select", "missed", ) assert result.status is WizardStatus.ADVANCED assert result.step == "Q-PERF-REASON" assert result.follow_up_ids == ("Q-PERF-REASON", "Q-PERF-NEXT") def test_wizard_direct_constructor_requires_explicit_standalone_factory(tmp_path: Path) -> None: with pytest.raises(TypeError, match="explicit persistence factory"): WizardService(tmp_path) service = WizardService.for_standalone(tmp_path) assert service._events._canonical_transaction is None def test_morning_final_event_only_exists_after_explicit_save(tmp_path: Path) -> None: # Given: a started, owner-bound morning draft. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) # When: its six compact fields are completed but confirmation is not saved. version = _complete_morning(service, session_id) # Then: no trend-eligible event exists before save. assert not (tmp_path / "events.jsonl").exists() saved = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "save") assert saved.status is WizardStatus.SAVED event = json.loads((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert event["event_type"] == "morning_checkin" assert event["check_in"]["body_weight_kg"] == 70.2 def test_latest_finalized_through_uses_prior_day_when_today_is_unfinished( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) finalized = service.start_morning(WizardContext(OWNER, TOPIC), DAY) version = _complete_morning(service, finalized.session_id) saved = service.answer( WizardContext(OWNER, TOPIC), finalized.session_id, version, "save", ) assert saved.status is WizardStatus.SAVED _ = service.start_morning( WizardContext(OWNER, TOPIC), date(2026, 7, 21).isoformat(), ) latest = service._storage.find_latest_finalized_through( OWNER, TOPIC, date(2026, 7, 21).isoformat(), ) assert latest is not None assert latest.session_id == finalized.session_id def test_latest_finalized_through_breaks_equal_mtime_ties_deterministically( tmp_path: Path, ) -> None: service = WizardService.for_standalone(tmp_path) finalized = service.start_morning(WizardContext(OWNER, TOPIC), DAY) version = _complete_morning(service, finalized.session_id) saved = service.answer( WizardContext(OWNER, TOPIC), finalized.session_id, version, "save", ) assert saved.status is WizardStatus.SAVED original = service._storage.load(finalized.session_id) assert original is not None tied = original.model_copy( update={"session_id": "f" * 32, "finalized_event_id": "event-tied"} ) original_path = tmp_path / "drafts" / f"{finalized.session_id}.json" tied_path = tmp_path / "drafts" / f"{tied.session_id}.json" _ = tied_path.write_text(tied.model_dump_json(), encoding="utf-8") tied_path.chmod(0o600) same_mtime = original_path.stat().st_mtime_ns os.utime(original_path, ns=(same_mtime, same_mtime)) os.utime(tied_path, ns=(same_mtime, same_mtime)) latest = service._storage.find_latest_finalized_through(OWNER, TOPIC, DAY) assert latest is not None assert latest.session_id == tied.session_id def test_session_rejects_foreign_topic_stale_and_duplicate_actions(tmp_path: Path) -> None: # Given: an untouched morning session. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) # When: a foreign topic and owner, then a valid action, then its replay act on it. foreign = service.answer(WizardContext(OWNER, "topic-2"), session_id, 0, "value", "70.2") foreign_owner = service.answer(WizardContext("owner-2", TOPIC), session_id, 0, "value", "70.2") advanced = service.answer(WizardContext(OWNER, TOPIC), session_id, 0, "value", "70.2") replay = service.answer(WizardContext(OWNER, TOPIC), session_id, 0, "value", "70.2") # Then: only the current owner/topic/version advances. assert foreign.status is WizardStatus.REJECTED assert foreign_owner.status is WizardStatus.REJECTED assert advanced.status is WizardStatus.ADVANCED assert replay.status is WizardStatus.REJECTED @pytest.mark.parametrize("value", ["0", "700", "7O"]) def test_invalid_bodyweight_cannot_advance(tmp_path: Path, value: str) -> None: # Given: a new morning draft awaiting bodyweight. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) # When: an invalid bodyweight is supplied. result = service.answer(WizardContext(OWNER, TOPIC), session_id, 0, "value", value) # Then: the draft remains on the same version and has no event. assert result.status is WizardStatus.INVALID assert result.version == 0 assert not (tmp_path / "events.jsonl").exists() @pytest.mark.parametrize("value", ["-1", "30001", "24OO"]) def test_invalid_calories_cannot_advance(tmp_path: Path, value: str) -> None: # Given: a morning draft awaiting calorie input. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) version = 0 version = _answer(service, session_id, version, "value", "70.2") version = _answer(service, session_id, version, "value", "7.5") version = _answer(service, session_id, version, "select", "4") version = _answer(service, session_id, version, "select", "3") version = _answer(service, session_id, version, "select", "none") # When: a malformed calorie value is submitted. result = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "value", value) # Then: the session remains at calories and no event is present. assert result.status is WizardStatus.INVALID assert result.version == version assert result.step == "calories" assert not (tmp_path / "events.jsonl").exists() def test_restart_resumes_same_day_draft_at_exact_next_step(tmp_path: Path) -> None: # Given: a process that has recorded only bodyweight. first = WizardService.for_standalone(tmp_path) session_id = _morning(first) _answer(first, session_id, 0, "value", "70.2") # When: a new service instance starts that same KST day. resumed = WizardService.for_standalone(tmp_path).start_morning(WizardContext(OWNER, TOPIC), DAY) # Then: it returns the same opaque session at the persisted next step. assert resumed.session_id == session_id assert resumed.step == "sleep_duration" assert resumed.version == 1 def test_completed_morning_day_reopens_review_without_second_event(tmp_path: Path) -> None: # Given: a fully saved morning record for the owner/topic/day. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) version = _complete_morning(service, session_id) saved = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "save") # When: a restarted service starts the same morning again. resumed = WizardService.for_standalone(tmp_path).start_morning(WizardContext(OWNER, TOPIC), DAY) # Then: it returns the completed session for review, never a second canonical event. assert resumed.status is WizardStatus.SAVED assert resumed.session_id == saved.session_id assert resumed.message == "completed_review" assert len((tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines()) == 1 def test_concurrent_same_day_starts_share_one_morning_draft(tmp_path: Path) -> None: # Given: multiple isolated processes starting the exact owner/topic/day together. context = get_context("fork") queue = context.Queue() workers = [context.Process(target=_start_morning_in_process, args=(str(tmp_path), queue)) for _ in range(8)] # When: all starts race under the private wizard lock. for worker in workers: worker.start() for worker in workers: worker.join() # Then: each process observes the sole persisted session instead of creating one. session_ids = [queue.get() for _ in workers] assert all(worker.exitcode == 0 for worker in workers) assert len(set(session_ids)) == 1 assert len(list((tmp_path / "drafts").glob("*.json"))) == 1 def test_explicit_same_day_correction_supersedes_completed_morning(tmp_path: Path) -> None: # Given: one completed canonical morning session. service = WizardService.for_standalone(tmp_path) original_id = _morning(service) version = _complete_morning(service, original_id) saved = service.answer(WizardContext(OWNER, TOPIC), original_id, version, "save") assert saved.status is WizardStatus.SAVED # When: an explicit correction session is completed and saved. correction = service.start_morning_correction(WizardContext(OWNER, TOPIC), DAY) correction_version = _complete_morning(service, correction.session_id) correction_saved = service.answer(WizardContext(OWNER, TOPIC), correction.session_id, correction_version, "save") # Then: the second event is a correction that links to the original event. assert correction_saved.status is WizardStatus.SAVED events = [json.loads(line) for line in (tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines()] assert len(events) == 2 assert events[1]["event_type"] == "correction" assert events[1]["supersedes"] == events[0]["event_id"] assert current_terminal_morning_response( service._events._read_events(), date.fromisoformat(events[0]["occurred_at_kst"][:10]) ).event_id == events[1]["event_id"] def test_correction_start_creates_one_open_draft_per_predecessor(tmp_path: Path) -> None: # Given: one completed morning event eligible for correction. service = WizardService.for_standalone(tmp_path) original = _morning(service) original_version = _complete_morning(service, original) service.answer(WizardContext(OWNER, TOPIC), original, original_version, "save") predecessor = service._storage.load(original) assert predecessor is not None and predecessor.finalized_event_id is not None # When: its correction is started and then resumed. started = service.start_morning_correction(WizardContext(OWNER, TOPIC), DAY) resumed = service.start_morning_correction(WizardContext(OWNER, TOPIC), DAY) # Then: both calls identify the sole open draft for that predecessor. open_corrections = [ draft for path in (tmp_path / "drafts").glob("*.json") if (draft := service._storage.load(path.stem)) is not None and draft.supersedes == predecessor.finalized_event_id and draft.finalized_event_id is None ] assert resumed.session_id == started.session_id assert len(open_corrections) == 1 assert open_corrections[0].session_id == started.session_id def test_repeated_same_day_corrections_chain_from_latest_event(tmp_path: Path) -> None: # Given: an original morning and its first explicit correction. service = WizardService.for_standalone(tmp_path) original = _morning(service) original_version = _complete_morning(service, original) service.answer(WizardContext(OWNER, TOPIC), original, original_version, "save") first = service.start_morning_correction(WizardContext(OWNER, TOPIC), DAY) first_version = _complete_morning(service, first.session_id) service.answer(WizardContext(OWNER, TOPIC), first.session_id, first_version, "save") # When: a second correction is explicitly created and saved. second = service.start_morning_correction(WizardContext(OWNER, TOPIC), DAY) second_version = _complete_morning(service, second.session_id) service.answer(WizardContext(OWNER, TOPIC), second.session_id, second_version, "save") # Then: every replacement forms one append-only chain and one current value remains. events = [json.loads(line) for line in (tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines()] assert len(events) == 3 assert events[1]["supersedes"] == events[0]["event_id"] assert events[2]["supersedes"] == events[1]["event_id"] current = json.loads((tmp_path / "views" / "current.json").read_text(encoding="utf-8")) assert len(current["eligible_checkins"]) == 1 @pytest.mark.parametrize("invalid_day", ["2026-02-30", "2026-13-01", "2026/07/20"]) def test_calendar_invalid_kst_day_is_rejected_at_start_boundary(tmp_path: Path, invalid_day: str) -> None: # Given: a service with no existing draft. service = WizardService.for_standalone(tmp_path) # When: an invalid calendar day is supplied. result = service.start_morning(WizardContext(OWNER, TOPIC), invalid_day) # Then: a typed safe outcome is returned and no draft is persisted. assert result.status is WizardStatus.INVALID assert not (tmp_path / "drafts").exists() @pytest.mark.parametrize( ("step_value", "field"), [("x" * 4001, "training_plan"), ("x" * 4001, "optional_note"), ("x" * 2001, "pain")], ) def test_oversized_sensitive_text_is_rejected_before_model_finalization( tmp_path: Path, step_value: str, field: str, ) -> None: # Given: a morning draft advanced to the requested free-text boundary. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) version = 0 version = _answer(service, session_id, version, "value", "70.2") version = _answer(service, session_id, version, "value", "7.5") version = _answer(service, session_id, version, "select", "4") version = _answer(service, session_id, version, "select", "3") if field == "pain": pass else: version = _answer(service, session_id, version, "select", "none") version = _answer(service, session_id, version, "value", "2350") if field == "optional_note": version = _answer(service, session_id, version, "select", "rest") # When: an oversized value arrives for the current private text field. result = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "value", step_value) # Then: it is an ordinary invalid response, never an uncaught model error. assert result.status is WizardStatus.INVALID assert result.step == field assert not (tmp_path / "events.jsonl").exists() def test_oversized_workout_training_text_is_rejected_before_finalization(tmp_path: Path) -> None: # Given: a workout draft awaiting its concise actual-training note. service = WizardService.for_standalone(tmp_path) started = service.start_workout(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, started.session_id, 0, "select", "complete") # When: a value larger than the event contract arrives. result = service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "value", "x" * 4001) # Then: it fails safely at the input boundary without an event. assert result.status is WizardStatus.INVALID assert result.step == "training_summary" assert not (tmp_path / "events.jsonl").exists() def test_urgent_pain_requires_acknowledgement_and_never_enters_trends(tmp_path: Path) -> None: # Given: a morning draft at the pain prompt. service = WizardService.for_standalone(tmp_path) session_id = _morning(service) version = 0 version = _answer(service, session_id, version, "value", "70.2") version = _answer(service, session_id, version, "value", "7.5") version = _answer(service, session_id, version, "select", "4") version = _answer(service, session_id, version, "select", "3") # When: an urgent symptom arrives, then the user acknowledges the stop. urgent = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "value", "흉통이 있어요") acknowledged = service.answer(WizardContext(OWNER, TOPIC), session_id, urgent.version, "acknowledge") # Then: only a non-trend safety audit is appended after acknowledgement. assert urgent.status is WizardStatus.SAFETY_STOP assert acknowledged.status is WizardStatus.SAVED event = json.loads((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert event["event_type"] == "safety_audit" current = json.loads((tmp_path / "views" / "current.json").read_text(encoding="utf-8")) assert current["eligible_checkins"] == [] def test_urgent_nutrition_free_text_is_held_before_finalization(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_nutrition(WizardContext(OWNER, TOPIC), DAY) result = started for action, value in ( ("value", "70.2"), ("value", "2300"), ("value", "280 150 65"), ("value", "계획대로 3식"), ("value", "2.5"), ("value", "7.5"), ("select", "4"), ("select", "normal"), ("select", "4"), ("value", "식욕 3, 스트레스 2"), ("value", "하체 70분"), ): result = service.answer(WizardContext(OWNER, TOPIC), result.session_id, result.version, action, value) urgent = service.answer( WizardContext(OWNER, TOPIC), result.session_id, result.version, "value", "흉통이 있고 호흡 곤란이 있습니다", ) acknowledged = service.answer( WizardContext(OWNER, TOPIC), result.session_id, urgent.version, "acknowledge", ) assert urgent.status is WizardStatus.SAFETY_STOP assert acknowledged.status is WizardStatus.SAVED event = json.loads((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert event["status"] == "unsafe" assert event["safety"]["coaching_held"] is True def test_urgent_nutrition_text_stops_before_numeric_parsing(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_nutrition(WizardContext(OWNER, TOPIC), DAY) urgent = service.answer( WizardContext(OWNER, TOPIC), started.session_id, started.version, "value", "체중은 모르겠고 흉통과 호흡 곤란이 있습니다", ) assert started.step == "bodyweight" assert urgent.status is WizardStatus.SAFETY_STOP assert urgent.step == "safety_ack" assert urgent.message == "stop_and_escalate" def test_urgent_nutrition_text_stops_from_summary(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) result = service.start_nutrition(WizardContext(OWNER, TOPIC), DAY) for action, value in ( ("value", "70.2"), ("value", "2300"), ("value", "280 150 65"), ("value", "계획대로 3식"), ("value", "2.5"), ("value", "7.5"), ("select", "4"), ("select", "normal"), ("select", "4"), ("value", "식욕 3, 스트레스 2"), ("value", "하체 70분"), ("select", "skip"), ): result = service.answer( WizardContext(OWNER, TOPIC), result.session_id, result.version, action, value, ) urgent = service.answer( WizardContext(OWNER, TOPIC), result.session_id, result.version, "value", "흉통과 호흡 곤란이 있습니다", ) assert result.step == "summary" assert urgent.status is WizardStatus.SAFETY_STOP assert urgent.step == "safety_ack" def test_workout_record_is_final_without_bodyweight_or_calories(tmp_path: Path) -> None: # Given: a separate post-workout wizard. service = WizardService.for_standalone(tmp_path) started = service.start_workout(WizardContext(OWNER, TOPIC), DAY) # When: it is completed and explicitly saved. version = _answer(service, started.session_id, 0, "select", "complete") version = _answer(service, started.session_id, version, "value", "랫풀다운 4세트") version = _answer(service, started.session_id, version, "select", "4") version = _answer(service, started.session_id, version, "select", "none") saved = service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "save") # Then: a final workout event has no weight/calorie and cannot alter trends. assert saved.status is WizardStatus.SAVED event = json.loads((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert event["event_type"] == "workout_record" assert "body_weight_kg" not in event["check_in"] assert "calories_kcal" not in event["check_in"] current = json.loads((tmp_path / "views" / "current.json").read_text(encoding="utf-8")) assert current["eligible_checkins"] == [] def test_final_wizard_events_conform_to_versioned_contract(tmp_path: Path) -> None: # Given: one saved morning check-in and one saved workout record. service = WizardService.for_standalone(tmp_path) morning_id = _morning(service) morning_version = _complete_morning(service, morning_id) service.answer(WizardContext(OWNER, TOPIC), morning_id, morning_version, "save") workout = service.start_workout(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, workout.session_id, 0, "select", "complete") version = _answer(service, workout.session_id, version, "value", "등 운동") version = _answer(service, workout.session_id, version, "select", "3") version = _answer(service, workout.session_id, version, "select", "none") service.answer(WizardContext(OWNER, TOPIC), workout.session_id, version, "save") # When/Then: persisted final events pass the package-owned contract. events = [json.loads(line) for line in (tmp_path / "events.jsonl").read_text(encoding="utf-8").splitlines()] assert all(validate_event(event) for event in events) def _customer_context() -> WizardContext: return WizardContext(OWNER, TOPIC, OWNER_CUSTOMER) def test_sleep_followups_are_answered_before_morning_summary(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_morning(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, started.session_id, 0, "value", "70.2") version = _answer(service, started.session_id, version, "value", "5") result = service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "select", "2") assert result.step == "Q-SLEEP-CAUSE" assert result.follow_up_ids == ("Q-SLEEP-CAUSE", "Q-SLEEP-ADJUST") version = _answer(service, started.session_id, result.version, "select", "stress") version = _answer(service, started.session_id, version, "select", "yes") version = _answer(service, started.session_id, version, "select", "4") version = _answer(service, started.session_id, version, "select", "none") version = _answer(service, started.session_id, version, "value", "2300") version = _answer(service, started.session_id, version, "select", "rest") version = _answer(service, started.session_id, version, "select", "skip") assert service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "save").status is WizardStatus.SAVED def test_condition_followups_are_answered_before_morning_summary(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_morning(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, started.session_id, 0, "value", "70.2") version = _answer(service, started.session_id, version, "value", "7.5") version = _answer(service, started.session_id, version, "select", "4") result = service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "select", "2") assert result.step == "Q-COND-SYMPTOM" assert result.follow_up_ids == ("Q-COND-SYMPTOM", "Q-COND-INTENSITY") version = _answer(service, started.session_id, result.version, "select", "fatigue") version = _answer(service, started.session_id, version, "select", "reduce") version = _answer(service, started.session_id, version, "select", "none") version = _answer(service, started.session_id, version, "value", "2300") version = _answer(service, started.session_id, version, "select", "rest") version = _answer(service, started.session_id, version, "select", "skip") assert service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "save").status is WizardStatus.SAVED def test_performance_followups_are_answered_before_workout_summary(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_workout(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, started.session_id, 0, "select", "complete") version = _answer(service, started.session_id, version, "value", "랫풀다운") result = service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "select", "2") assert result.step == "Q-PERF-REASON" assert result.follow_up_ids == ("Q-PERF-REASON", "Q-PERF-NEXT") version = _answer(service, started.session_id, result.version, "select", "condition") version = _answer(service, started.session_id, version, "select", "no") version = _answer(service, started.session_id, version, "select", "none") assert service.answer(WizardContext(OWNER, TOPIC), started.session_id, version, "save").status is WizardStatus.SAVED def test_safety_reason_retrieval_and_referral_guidance_are_bounded(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) started = service.start_morning(WizardContext(OWNER, TOPIC), DAY) version = _answer(service, started.session_id, 0, "value", "70.2") version = _answer(service, started.session_id, version, "value", "7.5") version = _answer(service, started.session_id, version, "select", "4") version = _answer(service, started.session_id, version, "select", "4") stopped = service.answer( WizardContext(OWNER, TOPIC), started.session_id, version, "value", "흉통이 있어요", ) saved = service.answer(WizardContext(OWNER, TOPIC), started.session_id, stopped.version, "acknowledge") event = service.finalized_event(saved.session_id) reasons = safe_hold_reasons(event) assert saved.status is WizardStatus.SAVED assert reasons and reasons[0].startswith("S1 urgent/") assert "흉통" in reasons[0] assert "의료기관" in referral_guidance(event) def _start_ac21_flow(service: WizardService, case: dict[str, object]) -> tuple[WizardContext, object]: flow = case["flow"] context = WizardContext(OWNER, TOPIC) started = ( service.start_nutrition(context, DAY) if flow is WizardFlow.NUTRITION else service.start_morning(context, DAY) ) result = started for action, value in case["prefix"]: result = service.answer(context, result.session_id, result.version, action, value) assert result.status is WizardStatus.ADVANCED, (case["fixture_id"], result) assert result.step == case["field"] stopped = service.answer( context, result.session_id, result.version, case["action"], case["raw"], ) return context, stopped @pytest.mark.parametrize("case", AC21_SAFETY_CASES) def test_ac21_all_customer_safety_fixtures_hold_and_send_nothing( tmp_path: Path, case: dict[str, object], ) -> None: """Exercise each declared fixture through typed storage and owner-held evidence.""" service = WizardService.for_standalone(tmp_path) context, stopped = _start_ac21_flow(service, case) assert stopped.status is WizardStatus.SAFETY_STOP assert stopped.step == "safety_ack" assert stopped.message == "stop_and_escalate" assert len(stopped.safety_reasons) == 1 reason = stopped.safety_reasons[0] assert isinstance(reason, SafetyReason) assert reason.rule_id.value == case["rule_id"] assert reason.class_name == case["class_name"] assert reason.source_flow.value == case["source_flow"] assert reason.matched_field.value == case["matched_field"] assert reason.excerpt == case["raw"] assert len(reason.excerpt) <= 160 assert stopped.safety_signals == (case["signal"],) acknowledged = service.answer( context, stopped.session_id, stopped.version, "acknowledge", ) assert acknowledged.status is WizardStatus.SAVED event = service.finalized_event(acknowledged.session_id) assert event is not None assert event.event_type.value == "safety_audit" assert event.status.value == "unsafe" assert event.safety is not None safety = event.safety assert safety.coaching_held is True assert safety.level.value == case["level"] assert len(safety.reasons) == 1 stored_reason = safety.reasons[0] assert isinstance(stored_reason, SafetyReason) assert stored_reason == reason owner_reasons = safe_hold_reasons(event) assert owner_reasons == ( f"{case['rule_id']} {case['class_name']}/{case['matched_field']}: {case['raw']}", ) referral = referral_guidance(event) assert owner_reasons[0] in referral assert case["referral_marker"] in referral events_path = tmp_path / "events.jsonl" events = [json.loads(line) for line in events_path.read_text(encoding="utf-8").splitlines()] assert [item["event_type"] for item in events] == ["safety_audit"] assert not any(item["event_type"] in {"draft_created", "draft_sent"} for item in events) serialized_events = events_path.read_text(encoding="utf-8") assert all( token not in serialized_events for token in ("draft_created", "draft_sent", "request_token", "send_token") ) def test_morning_captures_six_source_facts(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) session_id = _morning(service) version = _complete_morning(service, session_id) saved = service.answer(WizardContext(OWNER, TOPIC), session_id, version, "save") assert saved.status is WizardStatus.SAVED event = json.loads((tmp_path / "events.jsonl").read_text(encoding="utf-8")) assert event["check_in"] == { "body_weight_kg": 70.2, "calories_kcal": 2300, "sleep_hours": 7.5, "sleep_quality_1to5": 4, "readiness_1to5": 4, "pain_summary": "none", "training_plan": "feasible", "notes": "none", } def test_owner_schedule_reference_is_dual_confirmed_source_fact(tmp_path: Path) -> None: service = WizardService.for_standalone(tmp_path) context = _customer_context() started = service.start_schedule_reference(context, DAY) version = 0 for action, value in ( ("value", DAY), ("value", "18:30"), ("select", "yes"), ("select", "yes"), ("value", "고객과 최근 변경 사항을 확인함"), ): result = service.answer(context, started.session_id, version, action, value) assert result.status is WizardStatus.ADVANCED version = result.version saved = service.answer(context, started.session_id, version, "save") assert saved.status is WizardStatus.INVALID assert saved.message == "schedule_reference_requires_registered_runtime" assert not (tmp_path / "events.jsonl").exists() def test_nutrition_progress_defer_resume_and_optional_skip(tmp_path: Path) -> None: context = WizardContext(OWNER, TOPIC) service = WizardService.for_standalone(tmp_path) started = service.start_nutrition(context, DAY) assert (started.position, started.total) == (1, 12) assert started.can_previous is False assert started.can_skip is False required_skip = service.answer( context, started.session_id, started.version, "skip", ) assert required_skip.status is WizardStatus.INVALID assert required_skip.version == started.version bodyweight = service.answer( context, started.session_id, started.version, "value", "70.2", ) deferred = service.answer( context, started.session_id, bodyweight.version, "defer", ) assert deferred.status is WizardStatus.DEFERRED assert deferred.step == "calories" resumed = WizardService.for_standalone(tmp_path).start_nutrition(context, DAY) assert resumed.session_id == started.session_id assert resumed.version == deferred.version assert resumed.step == deferred.step assert resumed.position == 2 assert resumed.can_previous is True result = resumed for action, value in ( ("value", "2300"), ("value", "280 150 65"), ("value", "계획대로 3식"), ("value", "2.5"), ("value", "7.5"), ("select", "4"), ("select", "normal"), ("select", "4"), ("value", "식욕 3, 스트레스 2"), ("value", "하체 70분"), ): result = service.answer( context, started.session_id, result.version, action, value, ) assert result.step == "optional_note" assert (result.position, result.total) == (12, 12) assert result.can_skip is True summary = service.answer( context, started.session_id, result.version, "skip", ) assert summary.status is WizardStatus.ADVANCED assert summary.step == "summary" assert (summary.position, summary.total) == (12, 12) def test_previous_invalidates_dynamic_followups_and_stale_version( tmp_path: Path, ) -> None: context = WizardContext(OWNER, TOPIC) service = WizardService.for_standalone(tmp_path) result = service.start_morning(context, DAY) for action, value in ( ("value", "70.2"), ("value", "5"), ("select", "2"), ): result = service.answer( context, result.session_id, result.version, action, value, ) assert result.step == "Q-SLEEP-CAUSE" assert result.total == 10 previous = service.answer( context, result.session_id, result.version, "previous", ) assert previous.status is WizardStatus.ADVANCED assert previous.step == "sleep_quality" assert previous.version == result.version + 1 stale = service.answer( context, result.session_id, result.version, "select", "4", ) assert stale.status is WizardStatus.REJECTED rewound = service.answer( context, result.session_id, previous.version, "previous", ) assert rewound.step == "sleep_duration" changed = service.answer( context, result.session_id, rewound.version, "value", "7.5", ) assert changed.step == "sleep_quality" assert changed.total == 8 assert changed.follow_up_ids == () persisted = service._storage.load(result.session_id) assert persisted is not None assert not any(key.startswith("Q-SLEEP-") for key in persisted.answers)