""" v2_13_gate.py — ANU v2.13 Phase 1 얇은 gate 최소 구현 (순수 함수, stdlib only) ★ non-matching thin gate 분리 구성 (task-2758, 회장 결정 2026-06-17): matching 축(allowlist axis 텍스트 검출)은 4회차 종료·보류되어 이 모듈에서 완전히 제거되었다. diff/tool_calls 비정형 텍스트에서 위험 축을 추출하던 detect_allowlist_axis 및 그 보조 함수(_extract_touched_files/_glob_match/ _match_forbidden_paths/_match_forbidden_ops/_is_word_form_op/_phrase_in_tokens/ _normalize/_flatten_tool_calls/_AXIS_PATTERNS)는 전부 삭제됐다. gate 판정은 이제 runtime_state(rs) 의 명시적 flag 와 contract 메타데이터, 그리고 Step5 goal expansion 의 구조화 입력(rs.touched_files vs contract.expected_files) set membership 비교에만 의존한다. ESCALATION_PACKET_FIELDS 결정 근거: doctrine §13.1 에스컬레이션 패킷 15 필드 + §13 budget 카운터 2 필드(same_signature_retry_count, same_goal_correction_count) = 총 17 필드. 절대 제약: - daemon/socket/API/external probe/key custody/DB permission/systemd 변경 0건. - budget reset / promotion APPROVED 부여 / goal_id 생성 금지 (Chair 권한). - 코드는 검사만 수행함. """ import re # --------------------------------------------------------------------------- # (a) Verdict 상수 # --------------------------------------------------------------------------- class GateVerdict: GOAL_ENABLING_CORRECTION_ALLOWED = "GOAL_ENABLING_CORRECTION_ALLOWED" GOAL_EXPANSION_BLOCKED = "GOAL_EXPANSION_BLOCKED" SEMANTIC_RESIDUE_CHAIR_DEFAULT = "SEMANTIC_RESIDUE_CHAIR_DEFAULT" # ALLOWLIST_AXIS_CHAIR_REQUIRED — dead trigger (matching 축 보류, task-2758). # enum 상수는 호환/이력 보존을 위해 유지하나, 이 verdict 를 발생시키던 # detect_allowlist_axis 기반 trigger 경로는 classify_gate 에서 제거됐다. # 현재 classify_gate 는 이 verdict 를 절대 반환하지 않는다. ALLOWLIST_AXIS_CHAIR_REQUIRED = "ALLOWLIST_AXIS_CHAIR_REQUIRED" BUDGET_EXCEEDED_CHAIR_REQUIRED = "BUDGET_EXCEEDED_CHAIR_REQUIRED" ROLLBACK_COST_FLOOR_VIOLATION = "ROLLBACK_COST_FLOOR_VIOLATION" INVALID_CONTRACT_PROVENANCE = "INVALID_CONTRACT_PROVENANCE" EXTERNAL_PROBE_REQUIRED = "EXTERNAL_PROBE_REQUIRED" CONTROL_DB_DIRECT_WRITE_BLOCKED = "CONTROL_DB_DIRECT_WRITE_BLOCKED" PROMOTION_SELF_APPROVAL_BLOCKED = "PROMOTION_SELF_APPROVAL_BLOCKED" GATE_VERDICTS = frozenset([ GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED, GateVerdict.GOAL_EXPANSION_BLOCKED, GateVerdict.SEMANTIC_RESIDUE_CHAIR_DEFAULT, GateVerdict.ALLOWLIST_AXIS_CHAIR_REQUIRED, GateVerdict.BUDGET_EXCEEDED_CHAIR_REQUIRED, GateVerdict.ROLLBACK_COST_FLOOR_VIOLATION, GateVerdict.INVALID_CONTRACT_PROVENANCE, GateVerdict.EXTERNAL_PROBE_REQUIRED, GateVerdict.CONTROL_DB_DIRECT_WRITE_BLOCKED, GateVerdict.PROMOTION_SELF_APPROVAL_BLOCKED, ]) # --------------------------------------------------------------------------- # (b) normalize_path — Step5 goal expansion 전용 안전 경로 정규화 # --------------------------------------------------------------------------- def normalize_path(p): """파일 경로 형식 정규화 (Step5 goal expansion set membership 비교 전용). 안전 정규화만 수행한다: - OS separator(``\\``) → posix(``/``) 통일 - leading ``./`` 반복 제거 - 중복 slash 축약 - 대소문자는 **보존** (대소문자 구분 파일시스템 — 경로 소문자화 금지) ★ matching 보류(task-2758)에 따른 안전 제약: - ``a/``·``b/`` 접두사를 **무조건 strip 하지 않는다**. 실제 repo 경로가 ``a/...``·``b/...`` 일 수 있으므로 보존한다. - substring 매칭에 쓰지 않는다. rs.touched_files 와 contract.expected_files 양쪽을 동일하게 normalize 한 뒤 **set membership** 비교에만 사용한다. - 소문자화 금지. """ if not p: return "" s = str(p).replace("\\", "/") s = re.sub(r"/+", "/", s) # 중복 slash 축약 while s.startswith("./"): # leading ./ 반복 제거 s = s[2:] return s.strip("/") # --------------------------------------------------------------------------- # (c) ROLLBACK_FLOOR_TABLE # --------------------------------------------------------------------------- ROLLBACK_FLOOR_TABLE = { "docs_report_only": "LOW", "expected_files_reversible_code": "LOW", "test_fixture_only": "LOW", "local_runner_wiring": "MEDIUM", "runtime_config_touch": "MEDIUM", "control_db_write": "IRREVERSIBLE", # HIGH 또는 IRREVERSIBLE → 보수적 "heartbeat_direct_update": "HIGH", "promotion_approved": "IRREVERSIBLE", "systemd_unit_edit": "HIGH", "systemd_enable_install": "IRREVERSIBLE", # HIGH 또는 IRREVERSIBLE → 보수적 "credential_owner_key_secret": "IRREVERSIBLE", "production_write": "IRREVERSIBLE", "merge_push_release": "IRREVERSIBLE", "external_send": "IRREVERSIBLE", # HIGH 또는 IRREVERSIBLE → 보수적 "forbidden_path": "HIGH", } # --------------------------------------------------------------------------- # (d) COST_ORDER + _cost_rank # --------------------------------------------------------------------------- COST_ORDER = ["LOW", "MEDIUM", "HIGH", "IRREVERSIBLE"] def _cost_rank(c): """비용 문자열 → 정수 순위. 알 수 없으면 0(LOW).""" try: # 앞뒤 공백/줄바꿈(" LOW ", "HIGH\n")이 있으면 index() 가 실패해 잘못 LOW(0) # 로 떨어진다 → strip() 정규화로 정상 순위를 찾도록 한다 (MEDIUM). return COST_ORDER.index(str(c).strip().upper()) except (ValueError, AttributeError): return 0 # --------------------------------------------------------------------------- # (e) rollback_floor # --------------------------------------------------------------------------- def rollback_floor(touched_axes): """ touched_axes(list[str]) 중 ROLLBACK_FLOOR_TABLE에 있는 축들의 floor 중 최대값 반환. 비어있거나 매칭 없으면 "LOW". ★ task-2758: classify_gate 는 touched_axes 를 항상 [] 로 전달한다 (matching 축 검출 제거). 단 이 순수 함수는 명시적 축 리스트(예: contract 가 제공하는 축, 테스트)로 직접 호출될 수 있어 테이블 로직을 유지한다. """ max_rank = 0 for axis in (touched_axes or []): floor_val = ROLLBACK_FLOOR_TABLE.get(axis) if floor_val is not None: rank = _cost_rank(floor_val) if rank > max_rank: max_rank = rank return COST_ORDER[max_rank] # --------------------------------------------------------------------------- # (f) max_rollback_cost # --------------------------------------------------------------------------- def max_rollback_cost(reported, floor): """COST_ORDER 기준 max(reported, floor). 잘못된 값은 LOW로 간주.""" r = _cost_rank(reported) f = _cost_rank(floor) return COST_ORDER[max(r, f)] def effective_floor(touched_axes, contract, forced=None): """축 테이블 floor / contract.rollback_floor / forced floor 중 최댓값. classify_gate 가 contract 의 rollback_floor 를 무시하던 결함을 닫는다. - touched_axes → ROLLBACK_FLOOR_TABLE 기반 floor (task-2758: 항상 [] 전달) - contract["rollback_floor"] → Chair 가 contract 에 명시한 하한 - forced → 분기별 강제 하한 (control_db=HIGH, promotion=IRREVERSIBLE) 셋 중 COST_ORDER 최댓값을 반환. """ max_rank = _cost_rank(rollback_floor(touched_axes)) if isinstance(contract, dict): cf = contract.get("rollback_floor") if cf: max_rank = max(max_rank, _cost_rank(cf)) if forced: max_rank = max(max_rank, _cost_rank(forced)) return COST_ORDER[max_rank] # --------------------------------------------------------------------------- # (g) build_escalation_packet + ESCALATION_PACKET_FIELDS # --------------------------------------------------------------------------- ESCALATION_PACKET_FIELDS = ( "goal_id", "requested_decision", "gate_verdict", "semantic_residue", "trust_roots_touched", "rollback_cost", "blast_radius", "budget_remaining", "diff_summary", "expected_files", "outside_expected_files", "forbidden_operations_detected", "external_probe_status", "recommended_action", "safe_default", "same_signature_retry_count", "same_goal_correction_count", ) def build_escalation_packet( goal_id, requested_decision, gate_verdict, semantic_residue, trust_roots_touched, rollback_cost, blast_radius, budget_remaining, diff_summary, expected_files, outside_expected_files, forbidden_operations_detected, external_probe_status, recommended_action, safe_default, same_signature_retry_count=0, same_goal_correction_count=0, ): """ 에스컬레이션 패킷 생성. 정확히 17개 키를 가진 dict 반환. doctrine §13.1 (15필드) + §13 budget 카운터 2필드 = 17. 코드는 패킷을 생성만 하며 budget reset / APPROVED 부여 불가 (Chair 전용). """ return { "goal_id": goal_id, "requested_decision": requested_decision, "gate_verdict": gate_verdict, "semantic_residue": semantic_residue, "trust_roots_touched": trust_roots_touched, "rollback_cost": rollback_cost, "blast_radius": blast_radius, "budget_remaining": budget_remaining, "diff_summary": diff_summary, "expected_files": expected_files, "outside_expected_files": outside_expected_files, "forbidden_operations_detected": forbidden_operations_detected, "external_probe_status": external_probe_status, "recommended_action": recommended_action, "safe_default": safe_default, "same_signature_retry_count": same_signature_retry_count, "same_goal_correction_count": same_goal_correction_count, } # --------------------------------------------------------------------------- # (h) classify_gate # --------------------------------------------------------------------------- _CONTRACT_REQUIRED = [ "goal_id", "scope", "blast_radius", "allowed_capability_delta", "expected_files", "allowed_runtime_changes", "forbidden_operations", "rollback_floor", "budget", "exit_criteria", ] _VALID_PROVENANCES = ("chair_injected", "strict_template") def classify_gate(contract, diff, tool_calls, runtime_state): """ gate 판정 (결정론 — 위→아래 첫 매칭 반환). ★ non-matching thin gate (task-2758): allowlist 축 텍스트 검출 (detect_allowlist_axis) 이 제거됐다. control_db / promotion / external_probe 분기는 runtime_state 의 명시적 flag 단독으로만 판정한다. diff/tool_calls 는 더 이상 위험 축 추출에 쓰이지 않으며(시그니처 호환을 위해 인자만 유지), Step5 goal expansion 만 구조화 입력 rs.touched_files 를 사용한다. Parameters ---------- contract : dict — goal_contract diff : str — (호환용 인자; 위험 축 추출에 쓰지 않음) tool_calls : list — (호환용 인자; 위험 축 추출에 쓰지 않음) runtime_state : dict Returns ------- dict {"verdict": str, "reason": str, "touched_axes": list, "rollback_cost": str} """ rs = runtime_state or {} # --- 1. contract provenance 검증 --- if not isinstance(contract, dict): return { "verdict": GateVerdict.INVALID_CONTRACT_PROVENANCE, "reason": "contract is not a dict", "touched_axes": [], "rollback_cost": "LOW", } missing = [f for f in _CONTRACT_REQUIRED if f not in contract] if missing: return { "verdict": GateVerdict.INVALID_CONTRACT_PROVENANCE, "reason": f"contract missing required fields: {missing}", "touched_axes": [], "rollback_cost": "LOW", } if contract.get("provenance") not in _VALID_PROVENANCES: return { "verdict": GateVerdict.INVALID_CONTRACT_PROVENANCE, "reason": ( f"invalid provenance: {contract.get('provenance')!r}; " f"must be one of {_VALID_PROVENANCES}" ), "touched_axes": [], "rollback_cost": "LOW", } # --- 2. flag 기반 위험 분기 --- # ★ task-2758: matching 축 검출 제거 → touched_axes 는 항상 빈 리스트. # ALLOWLIST_AXIS_CHAIR_REQUIRED trigger 경로도 함께 제거됐다(dead trigger). # control_db / promotion / external_probe 는 rs flag 단독 판정. touched_axes = [] external_probe_status = rs.get("external_probe_status", "NOT_REQUIRED") reported = rs.get("reported_rollback_cost", "LOW") if rs.get("control_db_op"): # control DB write 는 비가역 통제 자산 — floor 를 HIGH 이상으로 강제. floor_val = effective_floor(touched_axes, contract, forced="HIGH") return { "verdict": GateVerdict.CONTROL_DB_DIRECT_WRITE_BLOCKED, "reason": "control_db_op flagged in runtime_state", "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } if rs.get("promotion_self_approval"): # promotion APPROVED 전환은 비가역 — floor 를 IRREVERSIBLE 로 강제. floor_val = effective_floor(touched_axes, contract, forced="IRREVERSIBLE") return { "verdict": GateVerdict.PROMOTION_SELF_APPROVAL_BLOCKED, "reason": "promotion_self_approval flagged in runtime_state", "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } if rs.get("external_probe_required") and external_probe_status != "PASS": floor_val = effective_floor(touched_axes, contract) return { "verdict": GateVerdict.EXTERNAL_PROBE_REQUIRED, "reason": ( f"external probe required but status={external_probe_status!r}" ), "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } # --- 3. rollback floor --- # contract.rollback_floor 를 floor 산정에 반영 → ROLLBACK_COST_FLOOR_VIOLATION # 이 classify_gate 를 통해서도 도달 가능하다. floor_val = effective_floor(touched_axes, contract) if _cost_rank(reported) < _cost_rank(floor_val): return { "verdict": GateVerdict.ROLLBACK_COST_FLOOR_VIOLATION, "reason": ( f"reported rollback cost {reported!r} < floor {floor_val!r}" ), "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } # --- 4. budget --- budget_remaining = rs.get("budget_remaining") # budget_remaining 이 비숫자 str/타입이면 `<= 0` 에서 TypeError → 숫자 변환 가드. # 숫자형(또는 숫자 문자열 "0")만 비교하고, 변환 불가하면 budget 미설정으로 본다 (MEDIUM). budget_val = None if budget_remaining is not None: try: budget_val = float(budget_remaining) except (TypeError, ValueError): budget_val = None if budget_val is not None and budget_val <= 0: return { "verdict": GateVerdict.BUDGET_EXCEEDED_CHAIR_REQUIRED, "reason": f"budget_remaining={budget_remaining} <= 0", "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } # --- 5. goal expansion (C1 — 구조화 입력만) --- # rs.touched_files(구조화 list) 와 contract.expected_files 를 normalize_path # 로 정규화한 뒤 set membership 비교만 수행한다. diff/tool_calls 비정형 텍스트 # 추출 기반 matching 은 사용하지 않는다(task-2758). expected_files = contract.get("expected_files") or [] touched_files = rs.get("touched_files") or [] allowed_cap_delta = contract.get("allowed_capability_delta") or [] capability_delta = rs.get("capability_delta") or [] expected_norm = {normalize_path(f) for f in expected_files} outside_files = [f for f in touched_files if normalize_path(f) not in expected_norm] outside_caps = [c for c in capability_delta if c not in allowed_cap_delta] if outside_files or outside_caps: return { "verdict": GateVerdict.GOAL_EXPANSION_BLOCKED, "reason": ( f"outside_files={outside_files}, outside_caps={outside_caps}" ), "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } # --- 6. semantic residue --- if rs.get("semantic_residue"): return { "verdict": GateVerdict.SEMANTIC_RESIDUE_CHAIR_DEFAULT, "reason": "semantic_residue flagged in runtime_state", "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), } # --- 7. 전부 통과 --- return { "verdict": GateVerdict.GOAL_ENABLING_CORRECTION_ALLOWED, "reason": "all checks passed", "touched_axes": touched_axes, "rollback_cost": max_rollback_cost(reported, floor_val), }