"""ANU v3.1 - ANU-Codex Micro Refinement Convergence Loop (Phase 1 Core). Reference: ANU v3 master spec section 5.4-5.8. Task: task-2662 (CHAIR-AUTH-V3-1-CODEX-MICRO-LOOP-20260525-JJONGS-START-001). Scope (allowed, per task-2662 md): - document / contract / fixture / test layer refinement only - GO_READY / HOLD_FOR_CHAIR packet generation - safe round repetition with NO hard cap (section 5.4) - safety_gates 8-flag enforcement (immediate HOLD on any trigger) - allowed_write_paths enforcement on changed_files Scope (forbidden, per task-2662 md 14-item ban list): 1. dev bot automatic dispatch 2. branch push without chair approval 3. automatic GitHub write 4. merge 5. auto-merge 6. production mutation 7. real write mode 8. OWNER PAT manipulation 9. credential change 10. Axis 1/2/3 runtime change 11. live settings.json change 12. dispatch.py change 13. HARNESS_ENFORCED full declaration 14. ANU-Work production deployment declaration This module is a pure-logic refinement loop driver. It does NOT perform any of the forbidden operations. It only: (a) accepts a `micro_refinement_target` payload + a sequence of round outcomes (each carrying a Codex verdict, proposed changes, and a safety probe); (b) enforces the 8 safety_gates and the allowed_write_paths boundary; (c) collapses the round history into a `micro_refinement_result`; (d) emits a GO_READY or HOLD_FOR_CHAIR packet shaped per sections 5.7/5.8. """ from __future__ import annotations import functools import json import os import re from dataclasses import dataclass, field from datetime import datetime, timezone from typing import Any, Callable, Dict, Iterable, List, Mapping, Optional, Sequence from utils.codex_cc_decision_loop import ( CODEX_FAIL, CODEX_PASS, CODEX_PASS_WITH_RECOMMENDATIONS, CODEX_UNKNOWN, evaluate as evaluate_codex_decision, ) VERDICT_GO_READY = "GO_READY" VERDICT_HOLD_FOR_CHAIR = "HOLD_FOR_CHAIR" SAFETY_GATE_KEYS = ( "critical_7", "security_high_or_critical", "permission_expansion", "github_write_required", "dev_bot_reactivation_required", "real_write_required", "forbidden_write_target", "scope_expansion", ) GATE_TO_HOLD_REASON: Dict[str, str] = { "critical_7": "CRITICAL_7", "security_high_or_critical": "SECURITY_HIGH", "permission_expansion": "PERMISSION_EXPANSION", "github_write_required": "GITHUB_WRITE_REQUIRED", "dev_bot_reactivation_required": "DEV_BOT_REACTIVATION_REQUIRED", "real_write_required": "REAL_WRITE_REQUIRED", "forbidden_write_target": "FORBIDDEN_WRITE_TARGET", "scope_expansion": "SCOPE_EXPANSION", "allowed_write_path_violation": "ALLOWED_WRITE_PATH_VIOLATION", } class MicroRefinementTargetError(ValueError): """Raised when a micro_refinement_target payload is malformed.""" @dataclass class RoundOutcome: """One ANU-Codex refinement round. `codex_payload` carries the Codex verdict for this round. `proposed_changes` is the list of file paths the round wants to write to (allowed_write_paths boundary is enforced against this list). `safety_probe` is the 8-gate snapshot for this round. `actions` is a human-readable list of refinement actions taken. """ codex_payload: Mapping[str, Any] proposed_changes: Sequence[str] = field(default_factory=tuple) safety_probe: Mapping[str, bool] = field(default_factory=dict) actions: Sequence[str] = field(default_factory=tuple) @dataclass class MicroRefinementResult: """Internal representation; serialize via `.as_dict()`.""" task_id: str rounds: int final_verdict: str codex_final_verdict: str critical_7: bool permission_expansion: bool github_write_required: bool dev_bot_reactivation_required: bool real_write_required: bool forbidden_write_target_touched: bool scope_expansion_detected: bool security_high_or_critical: bool changed_files: List[str] remaining_findings: List[Dict[str, Any]] round_history: List[Dict[str, Any]] evidence_paths: List[str] triggered_gates: List[str] offending_paths: List[str] decision_items: List[Dict[str, Any]] chair_command: Optional[str] = None ready_for: Optional[str] = None go_ready_packet_path: Optional[str] = None hold_packet_path: Optional[str] = None def as_dict(self) -> Dict[str, Any]: return { "schema": "anu_v3.micro_refinement_result.v1", "task_id": self.task_id, "rounds": self.rounds, "final_verdict": self.final_verdict, "codex_final_verdict": self.codex_final_verdict, "critical_7": self.critical_7, "permission_expansion": self.permission_expansion, "github_write_required": self.github_write_required, "dev_bot_reactivation_required": self.dev_bot_reactivation_required, "real_write_required": self.real_write_required, "forbidden_write_target_touched": self.forbidden_write_target_touched, "scope_expansion_detected": self.scope_expansion_detected, "security_high_or_critical": self.security_high_or_critical, "changed_files": list(self.changed_files), "remaining_findings": list(self.remaining_findings), "round_history": list(self.round_history), "evidence_paths": list(self.evidence_paths), "go_ready_packet_path": self.go_ready_packet_path, "hold_packet_path": self.hold_packet_path, } def validate_target(target: Mapping[str, Any]) -> None: """Lightweight schema check against schemas/anu_v3_1_micro_refinement_target.json.""" if target.get("schema") != "anu_v3.micro_refinement_target.v1": raise MicroRefinementTargetError("schema mismatch") task_id = target.get("task_id") if not isinstance(task_id, str) or not task_id.startswith("task-"): raise MicroRefinementTargetError("task_id must look like 'task-XXXX'") if target.get("goal_condition") != "CODEX_PASS_OR_PASS_WITH_RECOMMENDATIONS": raise MicroRefinementTargetError("goal_condition must be CODEX_PASS_OR_PASS_WITH_RECOMMENDATIONS") if target.get("round_limit_policy") != "NO_HARD_CAP_FOR_MICRO_REFINEMENT": raise MicroRefinementTargetError( "round_limit_policy must be NO_HARD_CAP_FOR_MICRO_REFINEMENT (ANCHOR-3)" ) if target.get("max_rounds") not in (None, 0): raise MicroRefinementTargetError( "max_rounds must be null or 0 for micro refinement (hard cap 0)" ) gates = target.get("safety_gates") or {} missing = [k for k in SAFETY_GATE_KEYS if k not in gates] if missing: raise MicroRefinementTargetError(f"safety_gates missing keys: {missing}") for key, val in gates.items(): if val != "hold": raise MicroRefinementTargetError( f"safety_gates[{key!r}] must be 'hold' (got {val!r})" ) if not isinstance(target.get("allowed_write_paths"), list): raise MicroRefinementTargetError("allowed_write_paths must be a list") if not isinstance(target.get("forbidden_write_targets"), list): raise MicroRefinementTargetError("forbidden_write_targets must be a list") @functools.lru_cache(maxsize=256) def _glob_to_regex(pattern: str) -> re.Pattern[str]: """Convert a glob pattern to a compiled regex. Semantic rules (POSIX glob, ANCHOR-2): * → [^/]* (same-directory-only, does NOT cross path separators) ** → .* (recursive, crosses path separators) ? → [^/] (single char, does NOT cross path separator) other chars → re.escape'd literally Separator normalisation: both pattern and path use POSIX ``/``. """ # Normalise separators in pattern (cross-platform: Windows \\ → /) # Use re.sub to collapse one-or-more consecutive backslashes into a single / pattern = re.sub(r"\\+", "/", pattern) # Collapse duplicate slashes that may result from double-backslash sequences while "//" in pattern: pattern = pattern.replace("//", "/") # Tokenise: split on "**" first so we can treat it specially DOUBLE_STAR_SENTINEL = "\x00" pattern = pattern.replace("**", DOUBLE_STAR_SENTINEL) parts: list[str] = [] for ch in pattern: if ch == DOUBLE_STAR_SENTINEL: parts.append(".*") elif ch == "*": parts.append("[^/]*") elif ch == "?": parts.append("[^/]") elif ch in r"\.+^${}[]|()" : parts.append(re.escape(ch)) else: parts.append(ch) regex_str = "^" + "".join(parts) + "$" return re.compile(regex_str) def _path_matches_any(path: str, patterns: Iterable[str]) -> bool: # Normalise path: OS-specific separator → POSIX / norm_path = os.path.normpath(path).replace(os.sep, "/") for pattern in patterns: if _glob_to_regex(pattern).match(norm_path): return True return False def enforce_allowed_write_paths( changed_files: Sequence[str], allowed_write_paths: Sequence[str], forbidden_write_targets: Sequence[str], ) -> Dict[str, List[str]]: """Per ANCHOR-4: changed_files must stay inside allowed_write_paths.""" out_of_bounds: List[str] = [] forbidden_hit: List[str] = [] for path in changed_files: if _path_matches_any(path, forbidden_write_targets): forbidden_hit.append(path) continue if not _path_matches_any(path, allowed_write_paths): out_of_bounds.append(path) return {"out_of_bounds": out_of_bounds, "forbidden_hit": forbidden_hit} def _safety_probe_truthy_keys(probe: Mapping[str, Any]) -> List[str]: triggered: List[str] = [] for key in SAFETY_GATE_KEYS: if bool(probe.get(key, False)): triggered.append(key) return triggered def _isoformat_now() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _build_go_ready_packet( *, task_id: str, codex_verdict: str, rounds: int, changed_files: Sequence[str], evidence_paths: Sequence[str], chair_command: str, ready_for: str, ) -> Dict[str, Any]: """Section 5.7. ANCHOR-5: must carry chair_command.""" if codex_verdict not in (CODEX_PASS, CODEX_PASS_WITH_RECOMMENDATIONS): raise ValueError( f"GO_READY requires PASS or PASS_WITH_RECOMMENDATIONS, got {codex_verdict!r}" ) if not chair_command: raise ValueError("chair_command must be non-empty (ANCHOR-5)") return { "schema": "anu_v3.go_ready_packet.v1", "task_id": task_id, "ready_for": ready_for, "codex_verdict": codex_verdict, "critical_7": False, "permission_expansion": False, "forbidden_action": False, "required_chair_decision": "FINAL_GO_ONLY", "chair_command": chair_command, "rounds": rounds, "changed_files": list(changed_files), "evidence_paths": list(evidence_paths), "generated_at": _isoformat_now(), } def _build_hold_packet( *, task_id: str, triggered_gates: Sequence[str], rounds: int, evidence_paths: Sequence[str], offending_paths: Sequence[str], decision_items: Sequence[Mapping[str, Any]], codex_payload: Mapping[str, Any], hold_reason: Optional[str] = None, ) -> Dict[str, Any]: """Section 5.8. ANCHOR-5: must name trigger gate via hold_reason + triggered_gates.""" if not triggered_gates and hold_reason is None: raise ValueError( "HOLD packet requires ≥1 triggered_gate or an explicit hold_reason" ) if hold_reason is None: primary = triggered_gates[0] hold_reason = GATE_TO_HOLD_REASON.get(primary, "CRITICAL_7") critical_7_flag = ( "critical_7" in triggered_gates or bool(codex_payload.get("critical_7", False)) or hold_reason == "CRITICAL_7" ) return { "schema": "anu_v3.hold_for_chair_packet.v1", "task_id": task_id, "hold_reason": hold_reason, "critical_7": critical_7_flag, "triggered_gates": list(triggered_gates), "decision_items": list(decision_items)[:3], "recommended_next_action": "ASK_CHAIR", "rounds": rounds, "evidence_paths": list(evidence_paths), "offending_paths": list(offending_paths), "generated_at": _isoformat_now(), } def _decision_item_for_gate(gate: str, offending: Sequence[str]) -> Dict[str, Any]: return { "gate": gate, "reason": GATE_TO_HOLD_REASON.get(gate, "CRITICAL_7"), "offending_paths": list(offending), } def run_micro_refinement( *, target: Mapping[str, Any], round_outcomes: Sequence[RoundOutcome], chair_command_builder: Optional[Callable[[Mapping[str, Any], str], str]] = None, ready_for: str = "DISPATCH", evidence_paths: Optional[Sequence[str]] = None, ) -> Dict[str, Any]: """Drive a micro refinement convergence loop end-to-end. Returns a `micro_refinement_result.v1` dict that embeds the GO_READY or HOLD_FOR_CHAIR packet under `go_ready_packet` / `hold_for_chair_packet`. """ validate_target(target) if not round_outcomes: raise ValueError("round_outcomes must contain ≥1 RoundOutcome") task_id = target["task_id"] allowed_write_paths: Sequence[str] = target["allowed_write_paths"] forbidden_write_targets: Sequence[str] = target["forbidden_write_targets"] evidence_list: List[str] = list(evidence_paths or []) round_history: List[Dict[str, Any]] = [] aggregate_changed: List[str] = [] triggered_gates: List[str] = [] offending_paths: List[str] = [] decision_items: List[Dict[str, Any]] = [] last_codex_verdict: str = CODEX_UNKNOWN last_codex_payload: Mapping[str, Any] = {} hold_triggered = False hold_round_index: Optional[int] = None for idx, outcome in enumerate(round_outcomes, start=1): codex_payload = dict(outcome.codex_payload or {}) safety_probe = dict(outcome.safety_probe or {}) proposed = list(outcome.proposed_changes or []) path_check = enforce_allowed_write_paths( proposed, allowed_write_paths, forbidden_write_targets ) round_gates: List[str] = [] if path_check["forbidden_hit"]: round_gates.append("forbidden_write_target") offending_paths.extend(path_check["forbidden_hit"]) if path_check["out_of_bounds"]: round_gates.append("allowed_write_path_violation") offending_paths.extend(path_check["out_of_bounds"]) for gate in _safety_probe_truthy_keys(safety_probe): if gate not in round_gates: round_gates.append(gate) if bool(codex_payload.get("critical_7", False)) and "critical_7" not in round_gates: round_gates.append("critical_7") decision = evaluate_codex_decision( task_id=task_id, review_round=idx, codex_payload=codex_payload, safety_signal={k: True for k in round_gates}, ) last_codex_verdict = decision.codex_final_verdict last_codex_payload = codex_payload round_history.append( { "round": idx, "codex_verdict": decision.codex_final_verdict, "actions": list(outcome.actions), "safety_gates_triggered": list(round_gates), } ) if not round_gates: for p in proposed: if p not in aggregate_changed: aggregate_changed.append(p) if round_gates: hold_triggered = True hold_round_index = idx triggered_gates = round_gates # 각 trigger된 gate별 individual decision_item 생성. # path-bound gates 는 해당 path 만 매핑 · 그 외 gates 는 empty paths. gate_to_paths = { "forbidden_write_target": list(path_check["forbidden_hit"]), "allowed_write_path_violation": list(path_check["out_of_bounds"]), } for gate in round_gates: gate_paths = gate_to_paths.get(gate, []) decision_items.append(_decision_item_for_gate(gate, gate_paths)) break if decision.codex_final_verdict in (CODEX_PASS, CODEX_PASS_WITH_RECOMMENDATIONS): break if not hold_triggered and last_codex_verdict not in ( CODEX_PASS, CODEX_PASS_WITH_RECOMMENDATIONS, ): # 수렴 실패 (CODEX_FAIL · CODEX_UNKNOWN) 시 triggered_gates 빈 list 유지 # — safety gate 실제 trigger 없으면 critical_7 false-positive 차단 triggered_gates = [] final_verdict = ( VERDICT_HOLD_FOR_CHAIR if hold_triggered or last_codex_verdict not in (CODEX_PASS, CODEX_PASS_WITH_RECOMMENDATIONS) else VERDICT_GO_READY ) rounds_executed = ( hold_round_index if hold_round_index is not None else len(round_history) ) gate_flags = {key: (key in triggered_gates) for key in SAFETY_GATE_KEYS} gate_flags["critical_7"] = gate_flags["critical_7"] or bool( last_codex_payload.get("critical_7", False) ) result = MicroRefinementResult( task_id=task_id, rounds=rounds_executed, final_verdict=final_verdict, codex_final_verdict=last_codex_verdict, critical_7=gate_flags["critical_7"], permission_expansion=gate_flags["permission_expansion"], github_write_required=gate_flags["github_write_required"], dev_bot_reactivation_required=gate_flags["dev_bot_reactivation_required"], real_write_required=gate_flags["real_write_required"], forbidden_write_target_touched=gate_flags["forbidden_write_target"], scope_expansion_detected=gate_flags["scope_expansion"], security_high_or_critical=gate_flags["security_high_or_critical"], changed_files=aggregate_changed, remaining_findings=list(last_codex_payload.get("recommendations", []) or []), round_history=round_history, evidence_paths=evidence_list, triggered_gates=triggered_gates, offending_paths=offending_paths, decision_items=decision_items, ) packet: Dict[str, Any] if final_verdict == VERDICT_GO_READY: builder = chair_command_builder or _default_chair_command_builder chair_command = builder(target, last_codex_verdict) packet = _build_go_ready_packet( task_id=task_id, codex_verdict=last_codex_verdict, rounds=rounds_executed, changed_files=aggregate_changed, evidence_paths=evidence_list, chair_command=chair_command, ready_for=ready_for, ) result.chair_command = chair_command result.ready_for = ready_for result.go_ready_packet_path = None result.hold_packet_path = None result_dict = result.as_dict() result_dict["go_ready_packet"] = packet else: hold_reason: Optional[str] = None packet_gates = list(triggered_gates) if not packet_gates and last_codex_verdict in (CODEX_FAIL, CODEX_UNKNOWN): hold_reason = "REPEATED_DISAGREEMENT" packet = _build_hold_packet( task_id=task_id, triggered_gates=packet_gates, rounds=rounds_executed, evidence_paths=evidence_list, offending_paths=offending_paths, decision_items=decision_items or ([_decision_item_for_gate(packet_gates[0], offending_paths)] if packet_gates else []), codex_payload=last_codex_payload, hold_reason=hold_reason, ) result_dict = result.as_dict() result_dict["hold_for_chair_packet"] = packet return result_dict def _default_chair_command_builder( target: Mapping[str, Any], codex_verdict: str ) -> str: """Section 5.7 + ANCHOR-5: chair_command is what 회장 copy-pastes.""" task_id = target["task_id"] return ( f"APPROVE FINAL_GO {task_id} verdict={codex_verdict} " "policy=v3_1_codex_micro_refinement_loop_core_pr_only_no_auto_merge" ) def serialize_result(result_dict: Mapping[str, Any]) -> str: """JSON-serialize a result dict deterministically for evidence writeback.""" return json.dumps(result_dict, ensure_ascii=False, sort_keys=True, indent=2)