from __future__ import annotations import copy import importlib.util import sys from pathlib import Path from typing import Any import pytest HERE = Path(__file__).parent SPEC = importlib.util.spec_from_file_location( "continuous_v3", HERE / "continuous_lifecycle.py" ) assert SPEC and SPEC.loader m = importlib.util.module_from_spec(SPEC) sys.modules[SPEC.name] = m SPEC.loader.exec_module(m) def answers(*, activity: str = "moderate") -> dict[str, object]: return { "date_of_birth": "1996-08-01", "equation_sex_basis": "male", "height_cm": "180", "weight_kg": "80", "activity_category": activity, "activity_rationale": "주 3회 근력운동", "goal_type": "maintain", "target_weight_kg": None, "target_date": None, "allergies": {"status": "none", "items": []}, "intolerances": {"status": "none", "items": []}, "religious_ethical_exclusions": {"status": "none", "items": []}, "disliked_foods": {"status": "none", "items": []}, "dietary_preferences": {"status": "none", "items": []}, "conditions": {"status": "none", "items": []}, "medications": {"status": "none", "items": []}, "pregnancy_breastfeeding": False, "eating_disorder_risk": False, "cooking_access": "기본 조리 가능", "budget_band": "보통", "meal_count": 3, "schedule_constraints": "없음", } def reconciliation( value: dict[str, object], *, state: str = "resolved", current_index: int = 0 ) -> dict[str, object]: clarifications: list[dict[str, object]] = [] if state == "clarifying" or current_index: clarifications = [ { "field": "activity_category", "kind": "ambiguity", "question_ko": "평소 활동량은 어느 정도인가요?", } ] record: dict[str, object] = { "schema_version": "nutrition_input_reconciliation_v1", "state": state, "answers_digest": m.canonical_digest(value), "advisory": { "summary_ko": "고객 입력 요약", "facts_ko": [], "ambiguities_ko": [], "contradictions_ko": [], "safety_observations_ko": [], "clarifications": clarifications, }, "clarifications": clarifications, "current_index": current_index, } record["digest"] = m.canonical_digest(record) return record def base() -> dict[str, Any]: value = answers() payload = { "state": "customer_attestation", "answers_digest": m.canonical_digest(value), "body_digest": "a" * 64, } publication = { "session_id": m.SESSION, "generation": 24, "state": "COMMITTED", "message_id": 207, "payload": payload, } snapshot: dict[str, Any] = { "bootstrap": { "state": "AWAITING_ACTIVATION", "generation": 5, "session_id": m.SESSION, "sid_hash": m.SID, }, "customer": {"enabled": False, "route": [m.CUSTOMER, m.CUSTOMER, "0"]}, "workflow": { "state": "customer_attestation", "cursor": 22, "answers": value, "reconciliation": reconciliation(value, current_index=1), "consumed_updates": [159], }, "publication": publication, "outbox": [ { **publication, "role": "customer", "route": [m.CUSTOMER, "0"], } ], "owner_callbacks": [], "wizard": None, "generations": [], "cards": [], "drafts": {}, "deliveries": {}, "activation": [], } text = m.candidate_attestation_text(snapshot["workflow"], value, 1) assert text is not None body_digest = m.hashlib.sha256(text.encode()).hexdigest() snapshot["publication"]["payload"]["body_digest"] = body_digest snapshot["outbox"][0]["payload"]["body_digest"] = body_digest return snapshot def publish( s: dict[str, Any], generation: int, message_id: int, payload: dict[str, object] ) -> None: publication = { "session_id": m.SESSION, "generation": generation, "state": "COMMITTED", "message_id": message_id, "payload": payload, } s["publication"] = publication s["outbox"].append({**publication, "role": "customer", "route": [m.CUSTOMER, "0"]}) def test_candidate_bindings() -> None: assert ( m.SUCCESSOR == "e2f1699e8586ed3f835eaa2a3eaeaecaf5e52a4ad82aa931caade0ed377077ce" ) assert m.CORE == "b025a71695b620555f701f1c248545bd343428c430db37b2d8bd16148e0b0b7a" assert m.WHEEL == "bd5f44592e95b152cc466446fcbb607df586dc13482d0d3413fed80a476d2a70" def test_q22_stale_session_publication_waits() -> None: s = base() s["publication"].update( generation=21, message_id=203, payload={"state": "collecting"} ) s["outbox"] = [{**s["publication"], "role": "customer", "route": [m.CUSTOMER, "0"]}] assert m.classify(s).status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" def test_committed_clarification_is_field_bound() -> None: s = base() value = s["workflow"]["answers"] s["workflow"]["reconciliation"] = reconciliation(value, state="clarifying") payload: dict[str, object] = { "state": "customer_attestation", "answers_digest": str(m.canonical_digest(value)), "body_digest": "b" * 64, "clarification_field": "activity_category", } s["outbox"] = [] publish(s, 23, 205, payload) prompt = m.classify(s) assert prompt.status == "READY_CUSTOMER_CLARIFICATION" assert prompt.message_id == "205" assert prompt.action == "submit_clarification:activity_category" def test_resolved_authority_before_publication_waits() -> None: s = base() s["publication"]["payload"]["answers_digest"] = "0" * 64 s["outbox"][0]["payload"]["answers_digest"] = "0" * 64 assert m.classify(s).status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" def test_resolved_matching_publication_authorizes_attestation() -> None: prompt = m.classify(base()) assert prompt.status == "READY_CUSTOMER_ATTEST" assert prompt.message_id == "207" assert prompt.action == m.callback("attest", 24) assert "candidate-render-bound Korean summary passed" in prompt.detail @pytest.mark.parametrize( "fault", ["field", "digest", "generation", "message", "uncommitted"] ) def test_mismatch_fails_closed(fault: str) -> None: s = base() if fault == "field": s["publication"]["payload"]["clarification_field"] = "goal_type" s["outbox"][0]["payload"]["clarification_field"] = "goal_type" elif fault == "digest": s["publication"]["payload"]["answers_digest"] = "f" * 64 s["outbox"][0]["payload"]["answers_digest"] = "f" * 64 elif fault == "generation": s["publication"]["generation"] = 23 elif fault == "message": s["publication"]["message_id"] = 206 else: s["publication"]["state"] = "PREPARED" assert not m.classify(s).status.startswith("READY_CUSTOMER_ATTEST") def test_duplicate_and_non_monotonic_publications_fail_closed() -> None: s = base() s["outbox"].append(copy.deepcopy(s["outbox"][0])) with pytest.raises(m.LifecycleError, match="duplicate publication"): m.classify(s) a = base() b = base() b["publication"]["generation"] = 23 with pytest.raises(m.LifecycleError, match="regressed"): m.validate_transition(a, b) def test_candidate_surface_quality_rejects_internal_stale_and_unbound_copy() -> None: s = base() s["publication"]["payload"]["body_digest"] = "0" * 64 s["outbox"][0]["payload"]["body_digest"] = "0" * 64 prompt = m.classify(s) assert prompt.status == "STOP_CUSTOMER_SUMMARY_QUALITY" assert "render-bound" in prompt.detail def test_clarified_free_text_activity_is_not_attestable() -> None: s = base() value = answers(activity="활동 수준: 보통. 주 3회 근력운동") s["workflow"]["answers"] = value s["workflow"]["reconciliation"] = reconciliation(value, current_index=1) digest = m.canonical_digest(value) s["publication"]["payload"]["answers_digest"] = digest s["outbox"][0]["payload"]["answers_digest"] = digest prompt = m.classify(s) assert prompt.status == "STOP_CUSTOMER_SUMMARY_QUALITY" assert prompt.action == "do-not-attest" assert "activity" in prompt.detail def test_normal_customer_revision_reopens_only_last_field_with_same_value() -> None: s = base() value = answers(activity="활동 수준: 보통. 주 3회 근력운동") s["workflow"]["answers"] = value s["workflow"]["reconciliation"] = reconciliation(value, current_index=1) digest = m.canonical_digest(value) s["publication"]["payload"]["answers_digest"] = digest s["outbox"][0]["payload"]["answers_digest"] = digest handoff = m.candidate_revision_handoff(s) assert handoff is not None assert handoff["action"] == m.callback("revise", 24) assert handoff["next_field"] == "schedule_constraints" assert handoff["next_value"] == "없음" assert m.FIELDS[-1] == "schedule_constraints" def test_append_ledger_is_private_and_hash_chained(tmp_path: Path) -> None: target = tmp_path / "events.jsonl" m.append_event(target, {"status": "A"}) m.append_event(target, {"status": "B"}) rows = [m.json.loads(line) for line in target.read_text().splitlines()] assert target.stat().st_mode & 0o777 == 0o600 assert rows[1]["previous_event_sha256"] == rows[0]["event_sha256"]