# -*- coding: utf-8 -*- """task-2644 regression: callback_next_action_runner × 8 fixture (보강-2/3/4/5).""" from __future__ import annotations import sys from pathlib import Path import pytest _HERE = Path(__file__).resolve().parent if str(_HERE) not in sys.path: sys.path.insert(0, str(_HERE)) from _fx_loader import load_fixture # noqa: E402 from utils import callback_adjudicator, callback_next_action_runner # noqa: E402 from utils.callback_next_action_runner import ( # noqa: E402 BRANCH_OF, NextAction, NextActionResult, validate_branch_invariant, ) def _decide_for(fx): env = fx["envelope"] if env is None: return None, None adj = callback_adjudicator.adjudicate(env) return adj, callback_next_action_runner.decide(adj, envelope=env) def test_next_action_decided_matches_expected(fx): expected = fx["expected"] adj, decision = _decide_for(fx) if decision is None or "next_action_decided" not in (expected or {}): pytest.skip("no envelope/expected for next_action assertion") # 1C0F6F52 fixture 는 runner 가 HOLD_FOR_CHAIR 결정 if fx["name"] == "fallback_safety_net_log_recovery_without_control_plane_adjudication": assert decision["next_action_decided"] == "HOLD_FOR_CHAIR" return assert decision["next_action_decided"] == expected["next_action_decided"], ( f"{fx['name']}: next_action mismatch — got={decision['next_action_decided']} " f"expected={expected['next_action_decided']}" ) def test_branch_matches_3_mutex_invariant(fx): """보강-2: 3 분기 mutually exclusive.""" adj, decision = _decide_for(fx) if decision is None: pytest.skip("no envelope") action = decision["next_action_decided"] branch = decision["next_action_branch"] assert branch == BRANCH_OF[action], ( f"{fx['name']}: branch invariant violated — action={action} " f"branch={branch} expected={BRANCH_OF[action]}" ) def test_merge_ready_hardcoded_policy_lock(): """보강-3: MERGE_READY → REQUEST_CHAIR_MERGE_APPROVAL hardcoded.""" fx = load_fixture("merge_ready") _, decision = _decide_for(fx) assert decision["next_action_decided"] == NextAction.REQUEST_CHAIR_MERGE_APPROVAL.value assert decision["next_action_branch"] == "chair-required" assert decision["merge_policy_lock_applied"] is True def test_telegram_only_in_chair_required_branch(all_fixtures): """보강-2: Telegram 은 chair-required 에서만 발사 (auto/terminal_noop 금지).""" for fx in all_fixtures: adj, decision = _decide_for(fx) if decision is None: continue result_enum = (fx["expected"] or {}).get("next_action_result") if result_enum != NextActionResult.TELEGRAM_SENT.value: continue assert decision["next_action_branch"] == "chair-required", ( f"{fx['name']}: TELEGRAM_SENT 는 chair-required branch 에서만 허용" ) def test_state_freshness_missing_forces_hold_for_chair(): """보강-4: state freshness MISSING/STALE/MISMATCH → HOLD_FOR_CHAIR fail-closed.""" adj = { "terminal_state": "TERMINAL_PASS", "policy_class": "LEDGER_ONLY_NOOP", "state_freshness_status": "FRESH", } # envelope 에는 snapshot 박았는데 anu_state 가 없음 → MISSING decision = callback_next_action_runner.decide( adj, envelope={"dispatch_state_snapshot_id": "snap-X"}, anu_state=None, ) assert decision["next_action_decided"] == NextAction.HOLD_FOR_CHAIR.value assert decision["state_freshness_status"] == "MISSING" def test_state_freshness_stale_forces_hold_for_chair(): adj = {"terminal_state": "TERMINAL_PASS", "policy_class": "LEDGER_ONLY_NOOP"} decision = callback_next_action_runner.decide( adj, envelope={"dispatch_state_snapshot_id": "snap-old"}, anu_state={"snapshot_id": "snap-new", "state_version": "v2"}, ) assert decision["next_action_decided"] == NextAction.HOLD_FOR_CHAIR.value assert decision["state_freshness_status"] == "STALE" def test_validate_state_freshness_paths(): assert ( callback_next_action_runner.validate_state_freshness({}, {"snapshot_id": "x"}) == "MISMATCH" ) assert ( callback_next_action_runner.validate_state_freshness( {"dispatch_state_snapshot_id": "x"}, {"snapshot_id": "x"} ) == "FRESH" ) assert ( callback_next_action_runner.validate_state_freshness( {"dispatch_state_snapshot_id": "x"}, {"snapshot_id": "y"} ) == "STALE" ) assert ( callback_next_action_runner.validate_state_freshness({"snapshot_id": "x"}, None) == "MISSING" ) def test_record_result_4_fields_present(): """보강-5: 4 필수 필드 — decided/attempted/result/evidence_path.""" adj = {"terminal_state": "TERMINAL_PASS", "policy_class": "LEDGER_ONLY_NOOP"} decision = callback_next_action_runner.decide(adj, envelope={}) recorded = callback_next_action_runner.record_result( decision, attempted=True, result=NextActionResult.LEDGER_ONLY.value, evidence_path="memory/system/.callback_ledger.jsonl#test", ) assert "next_action_decided" in recorded assert recorded["next_action_attempted"] is True assert recorded["next_action_result"] == "LEDGER_ONLY" assert recorded["next_action_evidence_path"].startswith("memory/") assert validate_branch_invariant(recorded) is None def test_failed_result_without_recovery_action_kept_none(): adj = {"terminal_state": "TERMINAL_FAIL", "policy_class": "LEDGER_ONLY_NOOP"} decision = callback_next_action_runner.decide(adj, envelope={}) recorded = callback_next_action_runner.record_result( decision, attempted=True, result=NextActionResult.FAILED.value, evidence_path="memory/system/.callback_ledger.jsonl#failed", ) # Stop hook 추가 차단 조건: result=FAILED + recovery=None → Stop hook 에서 차단 assert recorded["recovery_action"] is None def test_forbidden_auto_flag_blocks_to_hold_for_chair(): """자동 금지 11 中 하나라도 hit 면 HOLD_FOR_CHAIR.""" adj = {"terminal_state": "TERMINAL_PASS", "policy_class": "LEDGER_ONLY_NOOP"} decision = callback_next_action_runner.decide( adj, envelope={"merge_execution": True, "dispatch_state_snapshot_id": "x"}, anu_state={"snapshot_id": "x"}, ) assert decision["next_action_decided"] == NextAction.HOLD_FOR_CHAIR.value assert "FORBIDDEN_AUTO_FLAG_merge_execution" in decision["reason"]