# pyright: reportMissingImports=false """Regression suite for ANU v3.1 ANU-Codex Micro Refinement Loop Core (task-2662). Verbatim coverage of the 10 regressions listed in `memory/tasks/task-2662.md`: R1 safe document refinement → GO_READY R2 PASS_WITH_RECOMMENDATIONS → GO_READY R3 Critical7 finding → HOLD_FOR_CHAIR R4 permission expansion → HOLD_FOR_CHAIR R5 GitHub write required → HOLD_FOR_CHAIR R6 forbidden_write_target touch → HOLD_FOR_CHAIR R7 scope expansion → HOLD_FOR_CHAIR R8 allowed_write_paths violation → HOLD_FOR_CHAIR R9 repeated safe rounds allowed (hard cap 0 verification) R10 GO_READY packet contains chair_command """ from __future__ import annotations import sys from pathlib import Path import pytest REPO_ROOT = Path(__file__).resolve().parents[2] if str(REPO_ROOT) in sys.path: sys.path.remove(str(REPO_ROOT)) sys.path.insert(0, str(REPO_ROOT)) from utils.anu_codex_micro_refinement_loop import ( # noqa: E402 RoundOutcome, enforce_allowed_write_paths, run_micro_refinement, ) TASK_ID = "task-2662" BASE_ALLOWED_PATHS = [ "utils/anu_codex_micro_refinement_loop.py", "utils/codex_cc_decision_loop.py", "schemas/anu_v3_1_*.json", "tests/anu_codex_micro_refinement_loop/**", "memory/tasks/task-2662.md", "memory/reports/task-2662.md", "memory/events/task-2662.*", ] BASE_FORBIDDEN = [ "/home/jay/.claude/settings.json", "/home/jay/.claude/settings.local.json", "/home/jay/.claude/hooks/**", "/usr/local/bin/cokacdir", ".github/**", "hooks/**", "dispatch.py", "scripts/finish-task.sh", "utils/anu_callback_registrar.py", "utils/canonical_workspace_resolver.py", "utils/worktree_resolver.py", "utils/callback_collector_helper_integration.py", "utils/**axis_1**", "utils/**axis_2**", "utils/axis_3_canary_scale_aware_guard/**", "**/.env*", "**/credentials*", ] def _target(**overrides): base = { "schema": "anu_v3.micro_refinement_target.v1", "task_id": TASK_ID, "target_kind": "task_md", "allowed_write_paths": list(BASE_ALLOWED_PATHS), "forbidden_write_targets": list(BASE_FORBIDDEN), "codex_findings_path": "memory/events/task-2662.codex-findings.json", "goal_condition": "CODEX_PASS_OR_PASS_WITH_RECOMMENDATIONS", "max_rounds": None, "round_limit_policy": "NO_HARD_CAP_FOR_MICRO_REFINEMENT", "safety_gates": { "critical_7": "hold", "security_high_or_critical": "hold", "permission_expansion": "hold", "github_write_required": "hold", "dev_bot_reactivation_required": "hold", "real_write_required": "hold", "forbidden_write_target": "hold", "scope_expansion": "hold", }, } base.update(overrides) return base def _round(verdict="PASS", critical_7=False, proposed=None, safety=None, actions=()): return RoundOutcome( codex_payload={ "codex_final_verdict": verdict, "critical_7": critical_7, "recommendations": [], }, proposed_changes=list(proposed or ()), safety_probe=dict(safety or {}), actions=list(actions), ) def test_R1_safe_document_refinement_go_ready(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["memory/tasks/task-2662.md"], actions=("refine task md wording",), ) ], ) assert result["final_verdict"] == "GO_READY" assert result["codex_final_verdict"] == "PASS" assert "go_ready_packet" in result assert result["go_ready_packet"]["codex_verdict"] == "PASS" assert result["go_ready_packet"]["chair_command"] def test_R2_pass_with_recommendations_go_ready(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS_WITH_RECOMMENDATIONS", proposed=["memory/reports/task-2662.md"], actions=("incorporate codex recommendations",), ) ], ) assert result["final_verdict"] == "GO_READY" assert result["codex_final_verdict"] == "PASS_WITH_RECOMMENDATIONS" assert result["go_ready_packet"]["codex_verdict"] == "PASS_WITH_RECOMMENDATIONS" assert result["go_ready_packet"]["chair_command"] def test_R3_critical_7_finding_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="FAIL", critical_7=True, proposed=[], actions=("codex flagged critical-7 issue",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert result["critical_7"] is True assert result["hold_for_chair_packet"]["hold_reason"] == "CRITICAL_7" assert "critical_7" in result["hold_for_chair_packet"]["triggered_gates"] def test_R4_permission_expansion_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", safety={"permission_expansion": True}, actions=("codex suggested broader perms",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert result["permission_expansion"] is True assert result["hold_for_chair_packet"]["hold_reason"] == "PERMISSION_EXPANSION" assert "permission_expansion" in result["hold_for_chair_packet"]["triggered_gates"] def test_R5_github_write_required_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", safety={"github_write_required": True}, actions=("would call gh api repos/.../merge",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert result["github_write_required"] is True assert result["hold_for_chair_packet"]["hold_reason"] == "GITHUB_WRITE_REQUIRED" assert "github_write_required" in result["hold_for_chair_packet"]["triggered_gates"] def test_R6_forbidden_write_target_touch_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["dispatch.py"], actions=("attempted to edit dispatch.py",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert result["forbidden_write_target_touched"] is True assert "forbidden_write_target" in result["hold_for_chair_packet"]["triggered_gates"] assert "dispatch.py" in result["hold_for_chair_packet"]["offending_paths"] def test_R7_scope_expansion_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS_WITH_RECOMMENDATIONS", safety={"scope_expansion": True}, actions=("codex demands extra modules beyond scope",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert result["scope_expansion_detected"] is True assert result["hold_for_chair_packet"]["hold_reason"] == "SCOPE_EXPANSION" def test_R8_allowed_write_paths_violation_hold_for_chair(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["src/unauthorized/file.py"], actions=("attempted out-of-bounds write",), ) ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" assert "allowed_write_path_violation" in result["hold_for_chair_packet"]["triggered_gates"] assert "src/unauthorized/file.py" in result["hold_for_chair_packet"]["offending_paths"] def test_R9_repeated_safe_rounds_allowed_no_hard_cap(): """ANCHOR-3: safe micro refinement has NO hard cap; many rounds OK if safety_gates stay clean.""" safe_rounds = [ _round( verdict="FAIL", proposed=["memory/reports/task-2662.md"], actions=(f"safe refinement round {i}",), ) for i in range(1, 8) ] safe_rounds.append( _round( verdict="PASS", proposed=["memory/reports/task-2662.md"], actions=("final pass",), ) ) result = run_micro_refinement(target=_target(), round_outcomes=safe_rounds) assert result["final_verdict"] == "GO_READY" assert result["rounds"] == 8 assert len(result["round_history"]) == 8 for entry in result["round_history"]: assert entry["safety_gates_triggered"] == [] def test_R10_go_ready_packet_contains_chair_command(): result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["schemas/anu_v3_1_micro_refinement_target.json"], actions=("schema refinement",), ) ], ) assert result["final_verdict"] == "GO_READY" packet = result["go_ready_packet"] assert "chair_command" in packet assert packet["chair_command"] assert "APPROVE FINAL_GO" in packet["chair_command"] assert packet["task_id"] in packet["chair_command"] assert packet["required_chair_decision"] == "FINAL_GO_ONLY" assert packet["critical_7"] is False assert packet["permission_expansion"] is False assert packet["forbidden_action"] is False def test_helper_enforce_allowed_write_paths_basic(): check = enforce_allowed_write_paths( ["memory/tasks/task-2662.md", "dispatch.py", "src/foreign.py"], BASE_ALLOWED_PATHS, BASE_FORBIDDEN, ) assert check["forbidden_hit"] == ["dispatch.py"] assert check["out_of_bounds"] == ["src/foreign.py"] def test_target_validation_rejects_max_rounds_nonzero(): from utils.anu_codex_micro_refinement_loop import ( MicroRefinementTargetError, validate_target, ) bad = _target(max_rounds=5) with pytest.raises(MicroRefinementTargetError): validate_target(bad) def test_R11_dual_gate_individual_decision_items(): """재발 방지: forbidden_write_target + allowed_write_path_violation 동시 trigger 시 각 gate별 individual decision_item 생성 · 정확한 paths 매핑 강제. """ result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", # dual violation: forbidden 1 + out_of_bounds 1 proposed=["dispatch.py", "src/foreign.py"], actions=["dual-violation-attempt"], ), ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" decision_items = result["hold_for_chair_packet"]["decision_items"] gates_in_items = [item["gate"] for item in decision_items] # 두 gate 모두 individual decision_item 으로 박제 assert "forbidden_write_target" in gates_in_items assert "allowed_write_path_violation" in gates_in_items # gate 별 정확 path 매핑 by_gate = {item["gate"]: item for item in decision_items} assert by_gate["forbidden_write_target"]["offending_paths"] == ["dispatch.py"] assert by_gate["allowed_write_path_violation"]["offending_paths"] == ["src/foreign.py"] def test_R12_aggregate_changed_dedup(): """재발 방지: 여러 라운드에 걸쳐 동일 file 수정 시 aggregate_changed dedup 강제. 동일 path 중복 없이 유일하게 유지. """ same_file = "utils/anu_codex_micro_refinement_loop.py" result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS_WITH_RECOMMENDATIONS", proposed=[same_file], actions=["round-1-noop"], ), _round( verdict="PASS_WITH_RECOMMENDATIONS", proposed=[same_file], actions=["round-2-noop"], ), ], ) # GO_READY 도달 + changed_files dedup 확인 assert result["final_verdict"] == "GO_READY" changed = result["changed_files"] # 동일 file 한 번만 등장 (★ 중복 없음) assert changed.count(same_file) == 1 assert same_file in changed def test_R13_path_match_case_sensitive(): """재발 방지: _path_matches_any 가 대소문자 엄격 구분(fnmatchcase) 으로 동작. 대소문자만 다른 path 는 forbidden_write_target 패턴에 매치되면 안 됨 (★ Linux 대소문자 구분 파일 시스템에서 path bypass 방지). """ forbidden = ["scripts/finish-task.sh"] allowed = ["scripts/finish-task.sh"] # 대소문자 정확 일치 → forbidden_hit 발생 exact = enforce_allowed_write_paths( ["scripts/finish-task.sh"], allowed, forbidden ) assert "scripts/finish-task.sh" in exact["forbidden_hit"] # 대소문자만 다른 path → forbidden 매치 안 됨 (★ fnmatchcase 엄격 구분) mixed = enforce_allowed_write_paths( ["scripts/Finish-Task.SH"], allowed, forbidden ) assert "scripts/Finish-Task.SH" not in mixed["forbidden_hit"] def test_R15_safety_signal_unfiltered_propagation(): """재발 방지: safety_signal 이 SAFETY_GATE_KEYS 필터를 거치지 않고 round_gates 전체를 Codex 결정 로직에 전달해야 한다. 특히 allowed_write_path_violation 같이 SAFETY_GATE_KEYS 에 없는 gate 도 auto_execute_allowed 차단에 반영된다. """ result = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["random/oob_path.py"], actions=["allowed_write_path_violation"], ), ], ) assert result["final_verdict"] == "HOLD_FOR_CHAIR" triggered = result["hold_for_chair_packet"]["triggered_gates"] assert "allowed_write_path_violation" in triggered def test_R16_forbidden_target_flag_strict_gate(): """재발 방지: forbidden_write_target_touched 는 오로지 forbidden_write_target gate 가 실제 trigger 된 경우에만 True. out_of_bounds (allowed_write_path_violation) 만으로는 False (★ flag 정합성). """ # out_of_bounds 만 발생 → forbidden_write_target_touched=False result_oob = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["random/oob_path.py"], actions=["oob-only"], ), ], ) assert result_oob["final_verdict"] == "HOLD_FOR_CHAIR" assert result_oob.get("forbidden_write_target_touched") is False # forbidden_write_target 만 발생 → True result_forbidden = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["scripts/finish-task.sh"], actions=["forbidden-only"], ), ], ) assert result_forbidden["final_verdict"] == "HOLD_FOR_CHAIR" assert result_forbidden.get("forbidden_write_target_touched") is True def test_R17_codex_unknown_requires_decision(): """재발 방지: codex_cc_decision_loop.evaluate_codex_decision 은 CODEX_UNKNOWN 결과를 받았을 때 decision_required=True 로 반환해야 한다 (★ ASK_CHAIR 권고와 일관). """ from utils.codex_cc_decision_loop import ( # noqa: E402 evaluate as evaluate_codex_decision, CODEX_UNKNOWN, ) decision_unknown = evaluate_codex_decision( task_id="task-2662", review_round=1, codex_payload={"codex_final_verdict": "GARBAGE_VALUE", "critical_7": False}, safety_signal={}, ) assert decision_unknown.codex_final_verdict == CODEX_UNKNOWN assert decision_unknown.decision_required is True assert decision_unknown.recommended_next_action == "ASK_CHAIR" decision_pass = evaluate_codex_decision( task_id="task-2662", review_round=1, codex_payload={"codex_final_verdict": "PASS", "critical_7": False}, safety_signal={}, ) assert decision_pass.decision_required is False def test_R18_path_normalization_blocks_dotdot_bypass(): """재발 방지: _path_matches_any 가 path normalization (os.path.normpath) 으로 './' '../' redundant segments bypass 를 차단해야 한다. """ forbidden = ["scripts/finish-task.sh"] allowed = ["scripts/finish-task.sh"] # './' redundant prefix → normalize 후 정확 일치 → forbidden_hit (raw path 보존) dotted = enforce_allowed_write_paths( ["./scripts/finish-task.sh"], allowed, forbidden ) assert "./scripts/finish-task.sh" in dotted["forbidden_hit"], ( "redundant './' prefix는 normalize 후 forbidden 매치되어야 함" ) # nested redundant segments → normalize 후 정확 일치 → forbidden_hit (raw path 보존) nested = enforce_allowed_write_paths( ["scripts/./finish-task.sh"], allowed, forbidden ) assert "scripts/./finish-task.sh" in nested["forbidden_hit"], ( "nested './' 는 normalize 후 forbidden 매치되어야 함" ) def test_R19_remaining_findings_preserved_in_go_ready(): """재발 방지: GO_READY 결과에서도 PASS_WITH_RECOMMENDATIONS verdict 의 recommendations 가 remaining_findings 에 보존되어 reviewer 에게 컨텍스트 제공. """ recommendations = [ {"type": "style", "message": "consider docstring"}, {"type": "low", "message": "minor naming"}, ] result = run_micro_refinement( target=_target(), round_outcomes=[ RoundOutcome( codex_payload={ "codex_final_verdict": "PASS_WITH_RECOMMENDATIONS", "critical_7": False, "recommendations": recommendations, }, proposed_changes=[], safety_probe={}, actions=[], ), ], ) assert result["final_verdict"] == "GO_READY" # GO_READY 도 recommendations 보존 (★ HOLD_FOR_CHAIR 만 보존하던 결함 정정) assert len(result.get("remaining_findings", [])) == 2 def test_R20_repeated_disagreement_hold_reason_no_critical_7_false_positive(): """재발 방지: 수렴 실패 (CODEX_FAIL · CODEX_UNKNOWN) + 실제 safety gate 미발사 시 hold_reason='REPEATED_DISAGREEMENT' + critical_7=False (★ false-positive 차단). HOLD packet 안 critical_7 boolean 은 hold_reason 이 'CRITICAL_7' 일 때만 True. """ # CODEX_FAIL · safety gate 미발사 → REPEATED_DISAGREEMENT result_fail = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="FAIL", critical_7=False, proposed=[], actions=("convergence failure",), ), ], ) assert result_fail["final_verdict"] == "HOLD_FOR_CHAIR" packet_fail = result_fail["hold_for_chair_packet"] assert packet_fail["hold_reason"] == "REPEATED_DISAGREEMENT" assert packet_fail["critical_7"] is False assert packet_fail["triggered_gates"] == [] assert result_fail["critical_7"] is False # CODEX_UNKNOWN · safety gate 미발사 → REPEATED_DISAGREEMENT result_unknown = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="GARBAGE", critical_7=False, proposed=[], actions=("unknown verdict",), ), ], ) assert result_unknown["final_verdict"] == "HOLD_FOR_CHAIR" packet_unknown = result_unknown["hold_for_chair_packet"] assert packet_unknown["hold_reason"] == "REPEATED_DISAGREEMENT" assert packet_unknown["critical_7"] is False # CODEX_FAIL + 실제 critical_7=True → 기존 CRITICAL_7 분류 유지 result_critical = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="FAIL", critical_7=True, proposed=[], actions=("real critical-7",), ), ], ) packet_critical = result_critical["hold_for_chair_packet"] assert packet_critical["hold_reason"] == "CRITICAL_7" assert packet_critical["critical_7"] is True def test_R14_aggregate_changed_round_gates_only_check(): """재발 방지: aggregate_changed 추가 조건은 round_gates 단독 검사로 충분. path_check forbidden_hit / out_of_bounds 는 이미 round_gates 에 누적되므로 중복 조건 제거 후에도 의미 동일. """ # forbidden_hit 발생 round → aggregate_changed 에서 제외 result_forbidden = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["scripts/finish-task.sh"], # forbidden target actions=["forbidden-attempt"], ), ], ) assert result_forbidden["final_verdict"] == "HOLD_FOR_CHAIR" assert "scripts/finish-task.sh" not in result_forbidden.get("changed_files", []) # out_of_bounds 발생 round → aggregate_changed 에서 제외 result_oob = run_micro_refinement( target=_target(), round_outcomes=[ _round( verdict="PASS", proposed=["random/unallowed_path.py"], # not in allowed_write_paths actions=["oob-attempt"], ), ], ) assert result_oob["final_verdict"] == "HOLD_FOR_CHAIR" assert "random/unallowed_path.py" not in result_oob.get("changed_files", []) def test_R21_star_no_cross_path_separator(): """ANCHOR-2: `*` (same-dir-only) 는 path separator `/` 를 넘지 않는다. Thread 2 회귀 방지: `schemas/anu_v3_1_*.json` 패턴이 `schemas/anu_v3_1_sub/test.json` 매칭 X.""" allowed = ["schemas/anu_v3_1_*.json"] forbidden: list[str] = [] # same-dir match: O same_dir = enforce_allowed_write_paths( ["schemas/anu_v3_1_micro_refinement_target.json"], allowed, forbidden ) assert same_dir["out_of_bounds"] == [] # nested directory: X (★ Thread 2 핵심) nested = enforce_allowed_write_paths( ["schemas/anu_v3_1_sub/test.json"], allowed, forbidden ) assert "schemas/anu_v3_1_sub/test.json" in nested["out_of_bounds"] def test_R22_double_star_recursive_match(): """ANCHOR-2: `**` 는 recursive match (separator 포함).""" allowed = ["tests/anu_codex_micro_refinement_loop/**"] forbidden: list[str] = [] # depth-1 d1 = enforce_allowed_write_paths( ["tests/anu_codex_micro_refinement_loop/test_basic.py"], allowed, forbidden ) assert d1["out_of_bounds"] == [] # depth-2 (★ recursive) d2 = enforce_allowed_write_paths( ["tests/anu_codex_micro_refinement_loop/sub/test_nested.py"], allowed, forbidden ) assert d2["out_of_bounds"] == [] # depth-3 (★ deep recursive) d3 = enforce_allowed_write_paths( ["tests/anu_codex_micro_refinement_loop/a/b/c/test_deep.py"], allowed, forbidden ) assert d3["out_of_bounds"] == [] def test_R23_cross_platform_separator_normalization(): """ANCHOR-2: Windows `\\\\` separator 가 POSIX `/` 로 정규화되어 매칭. Thread 1 회귀 방지: 패턴/경로 모두 정규화.""" # Windows-style 패턴 (backslash) allowed_win_pattern = ["tests\\\\anu_codex_micro_refinement_loop\\\\**"] forbidden: list[str] = [] # POSIX-style path 와 매칭 posix_path = enforce_allowed_write_paths( ["tests/anu_codex_micro_refinement_loop/test_x.py"], allowed_win_pattern, forbidden ) assert posix_path["out_of_bounds"] == [], ( "Windows-style pattern (backslash) 은 POSIX path 와 매칭되어야 함" ) # 반대로 POSIX 패턴 + Windows path 가능성 - normpath 가 OS-dependent 이므로 # 패턴 정규화만으로 충분히 회귀 방지 (★ Linux test runner 기준) def test_R24_critical_7_flag_set_when_in_triggered_gates(): """Thread 3 회귀 방지: safety_probe 에 의해 `critical_7` 이 triggered_gates 에 들어가지만 다른 gate 가 먼저 hold_reason 을 차지한 경우에도 critical_7_flag=True 보장. """ # safety_probe critical_7=True + permission_expansion=True 동시 발사 # GATE_TO_HOLD_REASON dict 정렬상 critical_7 이 SAFETY_GATE_KEYS 첫번째 이므로 # 실제로는 critical_7 이 primary 가 됨. 그래서 정확한 시나리오: # codex_payload.critical_7=False + safety_probe.critical_7=True 만 발사 # → triggered_gates=["critical_7"], hold_reason="CRITICAL_7" (이미 True) # 더 명확한 회귀 시나리오: triggered_gates 안에 critical_7 포함 + hold_reason != CRITICAL_7 # explicit hold_reason 강제 (test 전용 헬퍼 없으면 직접 함수 호출) from utils.anu_codex_micro_refinement_loop import _build_hold_packet packet = _build_hold_packet( task_id=TASK_ID, triggered_gates=["permission_expansion", "critical_7"], # critical_7 secondary rounds=1, evidence_paths=[], offending_paths=[], decision_items=[], codex_payload={"critical_7": False}, hold_reason="PERMISSION_EXPANSION", # ★ explicit non-CRITICAL_7 ) # ★ Thread 3 핵심: critical_7 in triggered_gates 이면 flag=True assert packet["critical_7"] is True, ( "critical_7 이 triggered_gates 에 있으면 codex_payload/hold_reason 무관하게 flag=True" ) assert packet["hold_reason"] == "PERMISSION_EXPANSION" # hold_reason 은 별개 # 음성 케이스: critical_7 not in triggered_gates → flag=False 유지 packet_neg = _build_hold_packet( task_id=TASK_ID, triggered_gates=["permission_expansion"], rounds=1, evidence_paths=[], offending_paths=[], decision_items=[], codex_payload={"critical_7": False}, hold_reason="PERMISSION_EXPANSION", ) assert packet_neg["critical_7"] is False