from __future__ import annotations import copy import importlib.util import json import subprocess 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_final", 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(*, legacy: bool = True, include_schedule: bool = True) -> dict[str, object]: value: dict[str, object] = { "date_of_birth": "1996-08-01", "equation_sex_basis": "male", "height_cm": "180", "weight_kg": "80", "activity_category": m.LEGACY_ACTIVITY if legacy else "moderate", "activity_rationale": "주 3회 근력운동" if legacy else m.DETAILED_RATIONALE, "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, } if include_schedule: value["schedule_constraints"] = "없음" return value def reconciliation( value: dict[str, object], *, current_index: int, state: str = "resolved", canonical_summary: bool = False, ) -> 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": ( m.authoritative_summary(value) if canonical_summary else "수정 전 활동 요약입니다." ), "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": "3230cb1a3c13f8e164c2171148c6e3183cbce0d36289cc647c10634b840d8060", } publication = { "session_id": m.SESSION, "generation": 24, "state": "COMMITTED", "message_id": 207, "payload": payload, } return { "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": [], } def publish( snapshot: dict[str, Any], generation: int, message_id: int, payload: dict[str, object], *, role: str = "customer", ) -> None: publication = { "session_id": m.SESSION, "generation": generation, "state": "COMMITTED", "message_id": message_id, "payload": payload, } snapshot["publication"] = publication route = [m.CUSTOMER, "0"] if role == "customer" else [m.OWNER, "0"] snapshot["outbox"].append({**publication, "role": role, "route": route}) def candidate_payload( snapshot: dict[str, Any], *, current_index: int ) -> dict[str, object]: value = snapshot["workflow"]["answers"] text = m.candidate_attestation_text(snapshot["workflow"], value, current_index) assert text is not None return { "state": "customer_attestation", "answers_digest": m.canonical_digest(value), "body_digest": m.hashlib.sha256(text.encode()).hexdigest(), } def startup(snapshot: dict[str, Any]) -> None: publish(snapshot, 25, 209, candidate_payload(snapshot, current_index=1)) def normalized_rewind(snapshot: dict[str, Any]) -> None: snapshot["workflow"] = { "state": "collecting", "cursor": 21, "answers": answers(legacy=False, include_schedule=False), "consumed_updates": [159, 300], } publish( snapshot, 26, 211, { "state": "collecting", "body_digest": "c" * 64, "force_reply_mode": "broadcast", }, ) def canonical_attestation(snapshot: dict[str, Any]) -> None: value = answers(legacy=False) snapshot["workflow"] = { "state": "customer_attestation", "cursor": 22, "answers": value, "reconciliation": reconciliation( value, current_index=0, canonical_summary=True ), "consumed_updates": [159, 300, 301], } publish(snapshot, 27, 213, candidate_payload(snapshot, current_index=0)) def test_final_candidate_bindings() -> None: assert ( m.SUCCESSOR == "4e9962be49b72951c3b9ed7e1a4fe36d0ca03e757872ae03364c1ab7a790f32b" ) assert m.CORE == "88caabf3dee923e720463d917d35b6dd32ecc31102ed3ab4df4c6a203df4a760" assert m.WHEEL == "33688875a0d1ce20955bd8272257ee34d84cb138f6a23a43e5d16682368ae5e5" def test_product_runtime_recovers_exact_legacy_activity_and_startup_generation() -> ( None ): code = ( """ import json from types import SimpleNamespace from gateway.platforms.telegram_nutrition_onboarding_copy import parse_legacy_activity_answer from gateway.platforms.telegram_nutrition_onboarding_runtime_publication_transport import publication_generation legacy = %r parsed = parse_legacy_activity_answer(legacy) status = SimpleNamespace(answer_count=22, state=SimpleNamespace(value='customer_attestation')) current = SimpleNamespace(generation=24, payload={'body_digest': 'old'}) generation = publication_generation(status, current=current, payload={'body_digest': 'candidate-changed'}) print(json.dumps({'parsed': parsed, 'generation': generation}, ensure_ascii=False)) """ % m.LEGACY_ACTIVITY ) result = subprocess.run( [sys.executable, "-c", code], cwd="/home/cube/projects/richard/hermes-agent", check=True, capture_output=True, text=True, ) assert json.loads(result.stdout) == { "parsed": ["moderate", m.DETAILED_RATIONALE], "generation": 25, } def test_current_inactive_authority_emits_wait_deploy() -> None: prompt = m.classify(base()) assert prompt.status == "WAIT_DEPLOY" assert prompt.message_id is None assert prompt.action == "deploy-final-candidate" def test_startup_changed_body_generation_25_emits_latest_revision() -> None: snapshot = base() startup(snapshot) prompt = m.classify(snapshot) assert prompt.status == "READY_CUSTOMER_REVISION" assert prompt.message_id == "209" assert prompt.action == m.callback("revise", 25) assert prompt.action != m.callback("revise", 24) def test_stale_generation_24_message_207_callback_is_rejected() -> None: snapshot = base() startup(snapshot) snapshot["publication"] = copy.deepcopy(snapshot["outbox"][0]) prompt = m.classify(snapshot) assert prompt.status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" assert prompt.message_id is None def test_revision_handoff_helper_never_authorizes_predeploy_card() -> None: snapshot = base() assert m.candidate_revision_handoff(snapshot) is None startup(snapshot) handoff = m.candidate_revision_handoff(snapshot) assert handoff is not None assert handoff["status"] == "READY_CUSTOMER_REVISION" assert handoff["message_id"] == "209" assert handoff["action"] == m.callback("revise", 25) def test_authorized_rewind_normalizes_activity_and_preserves_exact_rationale() -> None: snapshot = base() startup(snapshot) normalized_rewind(snapshot) workflow = snapshot["workflow"] assert workflow["answers"]["activity_category"] == "moderate" assert workflow["answers"]["activity_rationale"] == m.DETAILED_RATIONALE assert "schedule_constraints" not in workflow["answers"] assert "reconciliation" not in workflow prompt = m.classify(snapshot) assert prompt.status == "READY_CUSTOMER_RESUBMIT" assert prompt.message_id == "211" assert prompt.action == "submit_answer:schedule_constraints" assert prompt.command == "없음" @pytest.mark.parametrize( "fault", ["activity", "rationale", "reconciliation", "generation"] ) def test_rewind_rejects_noncanonical_or_stale_authority(fault: str) -> None: snapshot = base() startup(snapshot) normalized_rewind(snapshot) if fault == "activity": snapshot["workflow"]["answers"]["activity_category"] = m.LEGACY_ACTIVITY elif fault == "rationale": snapshot["workflow"]["answers"]["activity_rationale"] = "주 3회 근력운동" elif fault == "reconciliation": snapshot["workflow"]["reconciliation"] = {} else: snapshot["publication"]["generation"] = 25 assert m.classify(snapshot).status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" return with pytest.raises(m.LifecycleError): m.classify(snapshot) def test_new_reconciliation_summary_is_canonical_baseline_valid_and_attestable() -> ( None ): snapshot = base() startup(snapshot) normalized_rewind(snapshot) canonical_attestation(snapshot) value = snapshot["workflow"]["answers"] assert m.baseline_valid_answers(value) text = m.candidate_attestation_text(snapshot["workflow"], value, 0) assert text is not None for forbidden in ( "{", "items", "status", "maintain", "moderate", "미입력", "위험 신호", "수정 전", "실제 총활동량은 불명확", ): assert forbidden not in text for required in ( "유지", "보통", m.DETAILED_RATIONALE, "별도 검토 입력이 없습니다.", ): assert required in text prompt = m.classify(snapshot) assert prompt.status == "READY_CUSTOMER_ATTEST" assert prompt.message_id == "213" assert prompt.action == m.callback("attest", 27) def test_final_summary_rejects_body_digest_and_baseline_drift() -> None: snapshot = base() startup(snapshot) normalized_rewind(snapshot) canonical_attestation(snapshot) snapshot["publication"]["payload"]["body_digest"] = "0" * 64 snapshot["outbox"][-1]["payload"]["body_digest"] = "0" * 64 assert m.classify(snapshot).status == "WAIT_CANDIDATE_PUBLICATION" snapshot = base() startup(snapshot) normalized_rewind(snapshot) canonical_attestation(snapshot) snapshot["workflow"]["answers"]["activity_rationale"] = "주 3회 근력운동" digest = m.canonical_digest(snapshot["workflow"]["answers"]) snapshot["workflow"]["reconciliation"]["answers_digest"] = digest unsigned = { k: v for k, v in snapshot["workflow"]["reconciliation"].items() if k != "digest" } snapshot["workflow"]["reconciliation"]["digest"] = m.canonical_digest(unsigned) snapshot["publication"]["payload"]["answers_digest"] = digest snapshot["outbox"][-1]["payload"]["answers_digest"] = digest assert m.classify(snapshot).status != "READY_CUSTOMER_ATTEST" def test_matching_clarification_remains_field_bound() -> None: snapshot = base() value = answers(legacy=False) snapshot["workflow"] = { "state": "customer_attestation", "cursor": 22, "answers": value, "reconciliation": reconciliation(value, current_index=0, state="clarifying"), } payload: dict[str, object] = { "state": "customer_attestation", "answers_digest": str(m.canonical_digest(value)), "body_digest": "d" * 64, "clarification_field": "activity_category", } publish(snapshot, 27, 213, payload) prompt = m.classify(snapshot) assert prompt.status == "READY_CUSTOMER_CLARIFICATION" assert prompt.action == "submit_clarification:activity_category" def test_duplicate_publication_generation_fails_closed() -> None: snapshot = base() snapshot["outbox"].append(copy.deepcopy(snapshot["outbox"][0])) with pytest.raises(m.LifecycleError, match="duplicate publication"): m.classify(snapshot) def test_owner_review_uses_only_matching_owner_publication() -> None: snapshot = base() snapshot["workflow"]["state"] = "owner_review" publish( snapshot, 29, 217, {"state": "owner_review", "body_digest": "e" * 64}, role="owner", ) prompt = m.classify(snapshot) assert prompt.status == "READY_OWNER_ONBOARDING_APPROVE" assert prompt.message_id == "217" assert prompt.action == m.callback("owner_ok", 29) 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"]