from __future__ import annotations import copy import importlib.util import json import shlex import subprocess import sys from datetime import date from pathlib import Path from typing import Any import pytest HERE = Path(__file__).parent SPEC = importlib.util.spec_from_file_location( "continuous_v6_current", 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, "owner_id": m.OWNER, }, "customer": {"enabled": False, "route": [m.CUSTOMER, m.CUSTOMER, "0"]}, "owner": {"user_id": m.OWNER, "chat_id": m.OWNER, "topic_id": "0"}, "baseline": None, "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 == "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb" ) assert m.CORE == "3374be765c7e53fc431b0c6e48cacbacb09618f77f13ef87016cb0ad2575d060" assert m.WHEEL == "9d22e89a0b1a14d4eb7f1c2005fc01860bec5f3b4eda6e01bb1390542a90b875" 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_bad_body_and_accepts_domain_valid_rationale() -> 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_customer_attestation_to_owner_publication_gap_waits_without_exit() -> None: snapshot = base() owner_review_state(snapshot, commit_publication=False) assert ( snapshot["workflow"]["baseline_digest"] == "4d3d925fe27f953252ed0f108ed7c65900d05546cbbe46ee42cb7004a069be45" ) prompt = m.classify(snapshot) assert prompt.status == "WAIT_MATCHING_ROLE_PUBLICATION" assert prompt.role == "system" assert prompt.message_id is None publish( snapshot, 29, 215, { "state": "owner_review", "body_digest": "4903900d9c98c474006e512d14207439262ce405b696f4c6765dd65e1927cbd9", }, role="owner", ) prompt = m.classify(snapshot) assert prompt.status == "READY_OWNER_REVIEW" assert prompt.actor == m.OWNER assert prompt.route == (m.OWNER, "0") assert prompt.message_id == "215" assert prompt.action == m.callback("owner_ok", 29) assert prompt.action != m.callback("attest", 28) def test_owner_review_rejects_baseline_role_route_body_and_stale_message() -> None: snapshot = base() owner_review_state(snapshot, commit_publication=True) assert m.classify(snapshot).status == "READY_OWNER_REVIEW" bad_baseline = copy.deepcopy(snapshot) bad_baseline["workflow"]["baseline_digest"] = "0" * 64 with pytest.raises(m.LifecycleError, match="baseline"): m.classify(bad_baseline) bad_route = copy.deepcopy(snapshot) bad_route["outbox"][-1]["route"] = [m.CUSTOMER, "0"] with pytest.raises(m.LifecycleError, match="route"): m.classify(bad_route) bad_body = copy.deepcopy(snapshot) bad_body["publication"]["payload"]["body_digest"] = "0" * 64 bad_body["outbox"][-1]["payload"]["body_digest"] = "0" * 64 assert m.classify(bad_body).status == "WAIT_MATCHING_ROLE_PUBLICATION" stale_message = copy.deepcopy(snapshot) stale_message["publication"]["message_id"] = 212 stale_message["outbox"][-1]["message_id"] = 212 assert m.classify(stale_message).status == "WAIT_MATCHING_ROLE_PUBLICATION" def test_owner_finalization_ready_activation_checkin_delivery_cleanup_continues() -> ( None ): snapshot = base() owner_review_state(snapshot, commit_publication=True) snapshot["workflow"]["state"] = "finalizing" assert m.classify(snapshot).status == "WAIT_MATCHING_ROLE_PUBLICATION" publish( snapshot, 30, 217, {"state": "finalizing", "body_digest": "f" * 64}, role="owner", ) assert m.classify(snapshot).status == "WAIT_AUTOMATIC_READINESS" snapshot["workflow"]["state"] = "ready" assert m.classify(snapshot).status == "WAIT_MATCHING_ROLE_PUBLICATION" publish( snapshot, 31, 219, {"state": "ready", "body_digest": "a" * 64}, role="owner", ) cutover = m.classify(snapshot) assert cutover.status == "READY_OPERATOR_ACTIVATION_CUTOVER" assert cutover.command == m.exact_cutover() assert str(m.CHECKLIST) in cutover.command assert "<" not in cutover.command and ">" not in cutover.command snapshot["bootstrap"].update(state="ACTIVE", generation=6) snapshot["customer"]["enabled"] = True start = m.classify(snapshot) assert start.status == "READY_CUSTOMER_START_CHECKIN" assert start.action == "send:오늘 체크인" assert start.detail == ( "after arming, send one fresh exact production alias 오늘 체크인, then press inline " "오늘 체크인 시작; the pre-arm 2026-08-17 12:10 message produced no state/card " "and is not replayed or accepted" ) ready = m.ready_doc(snapshot, start) assert ready["event_subscription"] == "recursive-inotify-before-snapshot-and-actions" assert ready["fresh_alias_requirement"] == { "required_after_arming": "오늘 체크인", "expected_inline": "오늘 체크인 시작", "pre_arm_message": "2026-08-17 12:10", "pre_arm_result": "no state/card; not replayed or accepted", } snapshot["wizard"] = { "session_id": "wiz-current", "version": 13, "step": "summary", "message_id": "221", "route": [m.CUSTOMER, "0"], "finalized_event_id": "evt-current", } assert m.classify(snapshot).status == "WAIT_AUTOMATIC_GENERATION" snapshot["generations"] = [ { "token": "tok-current", "customer_key": m.KEY, "session_id": "wiz-current", "state": "generated", } ] snapshot["cards"] = [ { "token": "tok-current", "state": "published", "message_id": "223", "destination": { "user_id": m.OWNER, "chat_id": m.OWNER, "topic_id": "0", }, } ] snapshot["drafts"] = { "draft-current": { "customer_key": m.KEY, "session_id": "wiz-current", "status": "created", "generation_token": "tok-current", } } assert m.classify(snapshot).status == "READY_OWNER_DRAFT_REVIEW" snapshot["drafts"]["draft-current"]["status"] = "approved" assert m.classify(snapshot).status == "READY_OWNER_EXPLICIT_SEND" snapshot["deliveries"] = { "delivery-current": { "customer_key": m.KEY, "session_id": "wiz-current", "draft_id": "draft-current", "status": "sent_audited", "message_id": "225", } } assert m.classify(snapshot).status == "READY_OPERATOR_DISABLE" snapshot["customer"]["enabled"] = False assert m.classify(snapshot).status == "READY_CLEANUP_HANDOFF" def current_generation_28(snapshot: dict[str, Any]) -> dict[str, object]: rationale = ( "현재도 주 3회, 회당 60분 정도의 중간 강도 근력운동을 꾸준히 하고 있으며,\n" " 앞으로도 같은 수준으로 유지할 계획입니다. 평소 하루 6,000~8,000보 정도 걷습\n" " 니다." ) value = answers(legacy=False) value["activity_rationale"] = rationale snapshot["workflow"] = { "state": "customer_attestation", "cursor": 22, "answers": value, "reconciliation": reconciliation(value, current_index=1), "consumed_updates": [159, 300, 301, 302], } payload = candidate_payload(snapshot, current_index=1) publish(snapshot, 28, 213, payload) return value def owner_review_state(snapshot: dict[str, Any], *, commit_publication: bool) -> None: source_answers = current_generation_28(snapshot) package = str(m.DOMAIN_PACKAGE_ROOT) if package not in sys.path: sys.path.insert(0, package) models = importlib.import_module("checkin_cli.nutrition_onboarding_models") raw = dict(source_answers) raw.update( schema_version="1.0", customer_key=m.KEY, session_status="completed", review_decision="pending", publication_status="unpublished", ) baseline_payload = models.build_onboarding_baseline( raw, as_of=date(2026, 8, 16) ).model_dump(mode="json") baseline_digest = m.canonical_digest(baseline_payload) reconciliation_record = snapshot["workflow"]["reconciliation"] snapshot["baseline"] = { "schema_version": "1.0", "digest": baseline_digest, "baseline": baseline_payload, "source_answers": source_answers, "input_reconciliation_digest": reconciliation_record["digest"], } snapshot["workflow"].update( state="owner_review", answers={}, baseline_digest=baseline_digest, consumed_updates=[159, 629525114], ) if commit_publication: publish( snapshot, 29, 215, { "state": "owner_review", "body_digest": "4903900d9c98c474006e512d14207439262ce405b696f4c6765dd65e1927cbd9", }, role="owner", ) def test_current_transport_whitespace_rationale_is_domain_valid_and_attestable() -> ( None ): snapshot = base() value = current_generation_28(snapshot) assert ( m.canonical_digest(value) == "280dabc7a1705face34a68fb61f62f079e20f6fc3844afe82f9752db0f5cf669" ) assert ( snapshot["publication"]["payload"]["body_digest"] == "f760b5ac4c2109067a2798ec3480898afe6f77722efa18635684a45e51406b65" ) valid, normalized = m.shipped_baseline_validation(value) assert valid is True assert normalized == " ".join(str(value["activity_rationale"]).split()) prompt = m.classify(snapshot) assert prompt.status == "READY_CUSTOMER_ATTEST" assert prompt.message_id == "213" assert prompt.action == m.callback("attest", 28) @pytest.mark.parametrize( ("category", "rationale"), (("moderate", ""), ("moderate", "x" * 501), ("arbitrary", "valid rationale")), ) def test_shipped_baseline_gate_rejects_invalid_activity( category: str, rationale: str ) -> None: value = answers(legacy=False) value["activity_category"] = category value["activity_rationale"] = rationale valid, normalized = m.shipped_baseline_validation(value) assert valid is False assert normalized is None def test_current_authority_rejects_stale_digest_message_and_bad_body() -> None: snapshot = base() current_generation_28(snapshot) stale_digest = copy.deepcopy(snapshot) stale_digest["publication"]["payload"]["answers_digest"] = "0" * 64 stale_digest["outbox"][-1]["payload"]["answers_digest"] = "0" * 64 assert m.classify(stale_digest).status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" stale_message = copy.deepcopy(snapshot) stale_message["publication"]["message_id"] = 200 stale_message["outbox"][-1]["message_id"] = 200 assert m.classify(stale_message).status == "WAIT_MATCHING_CUSTOMER_PUBLICATION" bad_body = copy.deepcopy(snapshot) bad_body["publication"]["payload"]["body_digest"] = "f" * 64 bad_body["outbox"][-1]["payload"]["body_digest"] = "f" * 64 assert m.classify(bad_body).status == "WAIT_CANDIDATE_PUBLICATION" 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"] def test_current_cutover_bindings_are_exact_and_placeholder_free() -> None: command = m.exact_cutover() arguments = shlex.split(command) expected_data_root = ( m.DOMAIN_PACKAGE_ROOT.parent.parent / "data/customers" / m.KEY ) assert m.SUCCESSOR == "4a6c7ee54cf9526a30de8bb576c1d71b411938beba33a04914738f6e1b6ed1cb" assert m.CORE == "3374be765c7e53fc431b0c6e48cacbacb09618f77f13ef87016cb0ad2575d060" assert m.PROFILE_WHEEL == "f75856d6d986b64d3d2f083aec2f865c7aea19f5950b84ff06e519f2f6505af6" assert m.CHECKLIST_SHA256 == "b093aeb888ce7325cddd7d56d52a7e4a7ec8a180ac80eef21fea9453a8ef0e8e" assert m.VALIDATION_RECEIPT_SHA256 == "acb21c8c02260497c197427cfc93417f2be65c8746768ba788bb33400e3e6e78" assert command.count("--checklist-evidence") == 1 assert str(m.CHECKLIST) in command assert arguments[arguments.index("--data-root") + 1] == str(expected_data_root) assert "" not in command def test_real_registered_root_resolver_rejects_profile_data_and_accepts_customer_root() -> None: package = str(m.DOMAIN_PACKAGE_ROOT) if package not in sys.path: sys.path.insert(0, package) admin = importlib.import_module("checkin_cli.customer_admin") profile = m.DOMAIN_PACKAGE_ROOT.parent.parent registry = profile / "customers/registry.json" wrong = profile / "data" canonical = profile / "data/customers" / m.KEY with pytest.raises( admin.CustomerAdminError, match="data root does not match the registered customer root", ): admin._registered_customer(registry, profile, m.KEY, wrong) runtime = admin._registered_customer(registry, profile, m.KEY, canonical) assert runtime.data_root.resolve() == canonical.resolve() document = admin._read_profile_registry(registry, profile) with pytest.raises(admin.CustomerAdminError, match="G1 data root"): admin._validate_activation_data_root( profile, wrong, runtime.data_root, document, runtime.spec ) # Post-activation customer artifacts are intentionally outside the pre-activation # G1 content audit; canonical root identity is resolved above without mutation. def test_installed_telegram_adapter_accepts_only_production_customer_start_aliases() -> None: telegram = importlib.import_module("gateway.platforms.telegram") classifier = telegram.TelegramAdapter._is_nutrition_customer_start_command assert classifier("체크인 시작") is True assert classifier("오늘 체크인") is True assert classifier(" 오늘 체크인 ") is True assert classifier("checkin") is False assert classifier("오늘 체크인 시작") is False def test_one_delivery_is_terminal_and_duplicate_delivery_fails_closed() -> None: snapshot = base() owner_review_state(snapshot, commit_publication=True) snapshot["workflow"]["state"] = "ready" publish(snapshot, 30, 216, {"state": "ready", "body_digest": "a" * 64}, role="owner") snapshot["bootstrap"].update(state="ACTIVE", generation=6) snapshot["customer"]["enabled"] = True delivery = {"customer_key": m.KEY, "status": "sent_audited", "message_id": "225"} snapshot["deliveries"] = {"one": delivery} assert m.classify(snapshot).status == "READY_OPERATOR_DISABLE" snapshot["deliveries"]["duplicate"] = {**delivery, "message_id": "226"} with pytest.raises(m.LifecycleError, match="duplicate sent_audited delivery"): m.classify(snapshot) def timeout_args(*extra: str) -> Any: return m.parse_args( [ "observe", "--profile", "/authority/profile", "--ready", "/fresh/ready.json", "--events", "/fresh/events.jsonl", *extra, ] ) def test_timeout_defaults_remain_1800_stage_and_21600_total() -> None: parsed = timeout_args() assert parsed.stage_timeout == 1800 assert parsed.total_timeout == 21600 m.validate_timeout_bounds(parsed) def test_six_hour_stage_timeout_is_accepted_with_existing_total_cap() -> None: parsed = timeout_args("--stage-timeout", "21600") assert parsed.stage_timeout == m.MAX_STAGE_TIMEOUT assert parsed.total_timeout == m.MAX_TOTAL_TIMEOUT m.validate_timeout_bounds(parsed) @pytest.mark.parametrize( "arguments", ( ("--stage-timeout", "21600.001"), ("--total-timeout", "21600.001"), ), ) def test_timeout_values_above_explicit_caps_are_rejected( arguments: tuple[str, str], ) -> None: with pytest.raises(m.LifecycleError, match="timeout bound invalid"): m.validate_timeout_bounds(timeout_args(*arguments))