# -*- coding: utf-8 -*- """utils.callback_next_action_runner_v2 — 11 enum next_action decision + 6 result enum (v2). task-2644+1 ANU_CALLBACK_COLLECTOR_CONTROL_PLANE_CLEAN_REPLACEMENT task md: memory/tasks/task-2644+1.md spec (read-only): memory/specs/system_anu_callback_collector_control_plane_spec_260524.md v2 변경: - schema name suffix v2 - helper_integration 가용성 명시 (필수 · 미가용 시 HOLD_FOR_CHAIR fail-closed) - dispatch_via_helper(): DISPATCH_AUTO_REMEDIATION 실행 시 task-2646 registration helper 단일 진입점 사용 강제 (자체 cron CLI 호출 금지) - replacement_verdict 필드 추가 """ from __future__ import annotations from enum import Enum from typing import Any, Dict, Optional from utils import callback_collector_helper_integration as _integration SCHEMA = "utils.callback_next_action_runner.v2" REPLACEMENT_OF = "utils.callback_next_action_runner.v1" class NextActionV2(str, Enum): REQUEST_CHAIR_MERGE_APPROVAL = "REQUEST_CHAIR_MERGE_APPROVAL" REPORT_CRITICAL7 = "REPORT_CRITICAL7" REPORT_PERMISSION_OR_CREDENTIAL_EXPANSION = "REPORT_PERMISSION_OR_CREDENTIAL_EXPANSION" DISPATCH_AUTO_REMEDIATION = "DISPATCH_AUTO_REMEDIATION" RERUN_ALLOWED_GATE = "RERUN_ALLOWED_GATE" RUN_OWNER_GEMINI_TRIGGER_ROUTER = "RUN_OWNER_GEMINI_TRIGGER_ROUTER" CREATE_FOLLOWUP_TASK_SPEC = "CREATE_FOLLOWUP_TASK_SPEC" WAIT_FOR_BATCH_SIBLINGS = "WAIT_FOR_BATCH_SIBLINGS" BATCH_ADJUDICATE = "BATCH_ADJUDICATE" NOOP_TERMINAL = "NOOP_TERMINAL" HOLD_FOR_CHAIR = "HOLD_FOR_CHAIR" class NextActionBranchV2(str, Enum): AUTO_EXECUTABLE = "auto-executable" CHAIR_REQUIRED = "chair-required" TERMINAL_NOOP = "terminal_noop" class NextActionResultV2(str, Enum): DISPATCH_REGISTERED = "DISPATCH_REGISTERED" TELEGRAM_SENT = "TELEGRAM_SENT" LEDGER_ONLY = "LEDGER_ONLY" HOLD_PACKET_CREATED = "HOLD_PACKET_CREATED" BATCH_WAIT_RECORDED = "BATCH_WAIT_RECORDED" FAILED = "FAILED" BRANCH_OF: Dict[str, str] = { NextActionV2.DISPATCH_AUTO_REMEDIATION.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.RERUN_ALLOWED_GATE.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.RUN_OWNER_GEMINI_TRIGGER_ROUTER.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.CREATE_FOLLOWUP_TASK_SPEC.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.WAIT_FOR_BATCH_SIBLINGS.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.BATCH_ADJUDICATE.value: NextActionBranchV2.AUTO_EXECUTABLE.value, NextActionV2.REQUEST_CHAIR_MERGE_APPROVAL.value: NextActionBranchV2.CHAIR_REQUIRED.value, NextActionV2.REPORT_CRITICAL7.value: NextActionBranchV2.CHAIR_REQUIRED.value, NextActionV2.REPORT_PERMISSION_OR_CREDENTIAL_EXPANSION.value: NextActionBranchV2.CHAIR_REQUIRED.value, NextActionV2.HOLD_FOR_CHAIR.value: NextActionBranchV2.CHAIR_REQUIRED.value, NextActionV2.NOOP_TERMINAL.value: NextActionBranchV2.TERMINAL_NOOP.value, } SAFE_DEGRADED_ALLOWED = frozenset( {NextActionV2.NOOP_TERMINAL.value, NextActionV2.HOLD_FOR_CHAIR.value} ) FORBIDDEN_AUTO_FLAGS = ( "merge_execution", "live_settings_modification", "live_cokacdir_modification", "bot_app_token_use", "chair_authorization_issue", "pr_141_pilot", "production_pr_lifecycle_activation", "admin_override", "destructive_git", "foreign_dirty_cleanup", ) def _truthy(value: Any) -> bool: if isinstance(value, bool): return value if isinstance(value, (int, float)): return value != 0 if isinstance(value, str): return value.strip().lower() in {"true", "1", "yes", "hit"} return bool(value) def branch_of(action: str) -> str: return BRANCH_OF.get(action, NextActionBranchV2.CHAIR_REQUIRED.value) def validate_state_freshness( envelope: Dict[str, Any], anu_state: Optional[Dict[str, Any]], ) -> str: env_snapshot = envelope.get("dispatch_state_snapshot_id") or envelope.get("snapshot_id") if anu_state is None: return "MISSING" cur_snapshot = anu_state.get("snapshot_id") if cur_snapshot is None: return "MISSING" if env_snapshot is None: return "MISMATCH" if env_snapshot != cur_snapshot: return "STALE" return "FRESH" def has_forbidden_auto_flag(envelope: Dict[str, Any]) -> Optional[str]: for f in FORBIDDEN_AUTO_FLAGS: if _truthy(envelope.get(f)): return f return None _SENTINEL = object() def decide( adjudication: Dict[str, Any], envelope: Optional[Dict[str, Any]] = None, anu_state: Any = _SENTINEL, batch_state: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """v2 11-enum next_action decision (보강-2/3/4 통합 fail-closed + helper gate).""" envelope = envelope or {} integ_status = _integration.integration_status() if not integ_status.available: return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason="HELPER_INTEGRATION_BYPASS_FAIL_CLOSED", freshness="MISSING", replacement_verdict="REPLACEMENT_FAIL", integration_status=integ_status.to_json(), ) terminal_state = adjudication.get("terminal_state") policy = adjudication.get("policy_class") if anu_state is _SENTINEL: freshness = adjudication.get("state_freshness_status") or "FRESH" else: freshness = validate_state_freshness(envelope, anu_state) # type: ignore[arg-type] if freshness in {"STALE", "MISMATCH", "MISSING"}: return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason=f"STATE_FRESHNESS_{freshness}_FAIL_CLOSED", freshness=freshness, integration_status=integ_status.to_json(), ) forbidden = has_forbidden_auto_flag(envelope) if forbidden: return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason=f"FORBIDDEN_AUTO_FLAG_{forbidden}", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "NOT_CONTROL_PLANE_COMPLIANT": return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason="NOT_CONTROL_PLANE_COMPLIANT_REQUIRES_CHAIR", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "MERGE_READY" or policy == "MERGE_POLICY_LOCK": return _wrap( NextActionV2.REQUEST_CHAIR_MERGE_APPROVAL, reason="MERGE_POLICY_LOCK_HARDCODED", freshness=freshness, merge_lock=True, integration_status=integ_status.to_json(), ) if terminal_state == "CRITICAL7_HIT" or policy == "CRITICAL7_REPORT": return _wrap( NextActionV2.REPORT_CRITICAL7, reason="CRITICAL7_HIT_CHAIR_REQUIRED", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state in {"PERMISSION_EXPANSION", "CREDENTIAL_EXPANSION"} or policy == "PERMISSION_OR_CREDENTIAL_EXPANSION": return _wrap( NextActionV2.REPORT_PERMISSION_OR_CREDENTIAL_EXPANSION, reason="PERMISSION_OR_CREDENTIAL_EXPANSION_CHAIR_REQUIRED", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "AUTO_REMEDIATION_CANDIDATE": if policy == "AUTO_REMEDIATION_INSIDE_EXPECTED_FILES": return _wrap( NextActionV2.DISPATCH_AUTO_REMEDIATION, reason="AUTO_REMEDIATION_INSIDE_EXPECTED_FILES", freshness=freshness, integration_status=integ_status.to_json(), ) return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason="AUTO_REMEDIATION_OUTSIDE_EXPECTED_FILES_FORBIDDEN", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "OWNER_GEMINI_TRIGGER": return _wrap( NextActionV2.RUN_OWNER_GEMINI_TRIGGER_ROUTER, reason="OWNER_GEMINI_NUDGE_ALLOWED", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "SIBLING_INCOMPLETE" or (batch_state and batch_state.get("all_settled") is False): return _wrap( NextActionV2.WAIT_FOR_BATCH_SIBLINGS, reason="BATCH_SIBLINGS_NOT_ALL_SETTLED", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state == "SIBLING_FINAL" or (batch_state and batch_state.get("all_settled") is True): return _wrap( NextActionV2.BATCH_ADJUDICATE, reason="BATCH_ALL_SETTLED", freshness=freshness, integration_status=integ_status.to_json(), ) if policy == "PHASE3_TIMING_RACE": return _wrap( NextActionV2.RERUN_ALLOWED_GATE, reason="PHASE3_TIMING_RACE_RERUN", freshness=freshness, integration_status=integ_status.to_json(), ) if policy == "REGRESSION_RERUN": return _wrap( NextActionV2.RERUN_ALLOWED_GATE, reason="REGRESSION_RERUN_ALLOWED", freshness=freshness, integration_status=integ_status.to_json(), ) if policy == "FOLLOWUP_TASK_SPEC": return _wrap( NextActionV2.CREATE_FOLLOWUP_TASK_SPEC, reason="FOLLOWUP_TASK_REQUIRED", freshness=freshness, integration_status=integ_status.to_json(), ) if terminal_state in {"TERMINAL_PASS", "TERMINAL_FAIL"} and policy == "LEDGER_ONLY_NOOP": return _wrap( NextActionV2.NOOP_TERMINAL, reason="TERMINAL_LEDGER_ONLY", freshness=freshness, integration_status=integ_status.to_json(), ) return _wrap( NextActionV2.HOLD_FOR_CHAIR, reason="DEFAULT_FAIL_CLOSED", freshness=freshness, integration_status=integ_status.to_json(), ) def _wrap( action: NextActionV2, *, reason: str, freshness: str = "FRESH", merge_lock: bool = False, replacement_verdict: str = "AUTHORITATIVE_CLEAN_REPLACEMENT", integration_status: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: branch = branch_of(action.value) return { "schema": SCHEMA, "replacement_verdict": replacement_verdict, "next_action_decided": action.value, "next_action_branch": branch, "reason": reason, "merge_policy_lock_applied": merge_lock, "state_freshness_status": freshness, "telegram_emitted_allowed": branch == NextActionBranchV2.CHAIR_REQUIRED.value, "helper_integration_status": integration_status, } def record_result( decision: Dict[str, Any], *, attempted: bool, result: str, evidence_path: str, recovery_action: Optional[str] = None, ) -> Dict[str, Any]: if result not in {r.value for r in NextActionResultV2}: raise ValueError(f"invalid next_action_result: {result}") out = dict(decision) out.update({ "next_action_attempted": bool(attempted), "next_action_result": result, "next_action_evidence_path": evidence_path, "recovery_action": recovery_action, }) return out def validate_branch_invariant(decision: Dict[str, Any]) -> Optional[str]: action_raw = decision.get("next_action_decided") branch = decision.get("next_action_branch") result = decision.get("next_action_result") if not isinstance(action_raw, str): return None expected_branch = BRANCH_OF.get(action_raw) if expected_branch and branch != expected_branch: return f"BRANCH_MISMATCH (expected={expected_branch}, got={branch})" if result == NextActionResultV2.TELEGRAM_SENT.value and branch != NextActionBranchV2.CHAIR_REQUIRED.value: return f"TELEGRAM_OUTSIDE_CHAIR_REQUIRED_BRANCH (branch={branch})" return None def dispatch_via_helper( *, task_id: str, executor_key: str, chat_id: str, prompt: str, at: str, canonical_root: str = "/home/jay/workspace", require_envelope: bool = True, ) -> Dict[str, Any]: """DISPATCH_AUTO_REMEDIATION 등 auto-executable 분기에서 cron 등록 시 사용. Helper integration → task-2646 register_callback 단일 진입점. 자체 cron CLI 호출 금지 (task md 원칙 4 + 10 / ANCHOR-2). """ return _integration.register_normal_callback( task_id=task_id, executor_key=executor_key, chat_id=chat_id, prompt=prompt, at=at, canonical_root=canonical_root, require_envelope=require_envelope, dispatch_path=True, direct_cron_path=False, ) __all__ = [ "SCHEMA", "REPLACEMENT_OF", "NextActionV2", "NextActionBranchV2", "NextActionResultV2", "BRANCH_OF", "SAFE_DEGRADED_ALLOWED", "FORBIDDEN_AUTO_FLAGS", "branch_of", "validate_state_freshness", "has_forbidden_auto_flag", "decide", "record_result", "validate_branch_invariant", "dispatch_via_helper", ]