#!/usr/bin/env python3 # -*- coding: utf-8 -*- """hooks/stop_anu_callback_collector_verifier.py — Stop hook (STAGED · live 미적용). task-2644 ANU_CALLBACK_COLLECTOR_CONTROL_PLANE (회장 verbatim 우선순위 5 + 보강-5). spec: memory/specs/system_anu_callback_collector_control_plane_spec_260524.md spec sha256: b27da557d4245bce476cd63f4ab174aefc8a25d2da07ec2c8d2c83b01ee96153 ★ 본 hook 은 staged template only · ~/.claude/settings.json 에 등록되지 않은 상태. 회장 verbatim 종료 차단 10 조건 (보강-5 반영 후 §14.6): 1. callback_envelope_parsed 없으면 fail 2. context_recovered 없으면 fail 3. terminal_state_classified 없으면 fail 4. next_action_decided 없으면 fail 5. auto_action_dispatched / chair_report_emitted / noop_terminal_recorded / batch_wait_recorded 중 하나 없으면 fail 6. callback_ledger_written 없으면 fail 7. "callback received / 도착 / 수신" 표현 + source attribution 없으면 fail 8. schedule_history 사후 조회를 inbound 수신처럼 표현하면 fail 9. next_action_result 없으면 fail 10. state_version mismatch / .anu_state stale 인데 SAFE_DEGRADED_MODE / HOLD_FOR_CHAIR 둘 다 아니면 fail 추가 보강-5 조건: next_action_result=FAILED + recovery_action 없으면 fail (조건 9 의 보충). Claude Code Stop hook 계약 (참고): - stdin: {"session_id":..., "stop_hook_active":..., "hook_event_name":"Stop"} - stdout: JSON · "decision":"block" + "reason" 반환 시 종료 차단 - exit 0: 일반 · non-zero (2): block 으로도 해석 가능 """ from __future__ import annotations import json import os import sys from pathlib import Path from typing import Any, Dict, List, Optional, Tuple # 동일 worktree 내 utils 를 import 할 수 있도록 경로 보강. _HERE = Path(__file__).resolve().parent _REPO = _HERE.parent if str(_REPO) not in sys.path: sys.path.insert(0, str(_REPO)) from utils import source_attribution_guard # noqa: E402 # type: ignore[import-not-found] from utils.callback_next_action_runner import ( # noqa: E402 # type: ignore[import-not-found] SAFE_DEGRADED_ALLOWED, NextActionResult, validate_branch_invariant, ) SCHEMA = "hooks.stop_anu_callback_collector_verifier.v1" LEDGER_PATH_ENV = "ANU_CALLBACK_LEDGER_PATH" LEDGER_PATH_DEFAULT = "memory/system/.callback_ledger.jsonl" LAST_OUTPUT_ENV = "ANU_LAST_OUTPUT_TEXT" # 종료 직전 user-facing text (선택) COLLECTOR_MODE_ENV = "COKACDIR_MODE" COLLECTOR_MODE_VALUE = "ANU_CALLBACK_COLLECTOR" # 종료 차단 10 조건 ID COND = { 1: "MISSING_CALLBACK_ENVELOPE_PARSED", 2: "MISSING_CONTEXT_RECOVERED", 3: "MISSING_TERMINAL_STATE_CLASSIFIED", 4: "MISSING_NEXT_ACTION_DECIDED", 5: "MISSING_AUTO_OR_CHAIR_OR_NOOP_OR_BATCH_FLAG", 6: "MISSING_CALLBACK_LEDGER_WRITTEN", 7: "RECEIVED_PHRASE_WITHOUT_INBOUND_SOURCE_ATTRIBUTION", 8: "SCHEDULE_HISTORY_LOOKUP_AS_INBOUND", 9: "MISSING_NEXT_ACTION_RESULT", 10: "STATE_STALE_OR_MISMATCH_WITHOUT_SAFE_DEGRADED_OR_HOLD", "9b": "RESULT_FAILED_WITHOUT_RECOVERY_ACTION", "11": "TELEGRAM_OUTSIDE_CHAIR_REQUIRED_BRANCH", # 보강-2 invariant } def _read_stdin() -> Dict[str, Any]: try: if sys.stdin.isatty(): return {} raw = sys.stdin.read() if not raw.strip(): return {} return json.loads(raw) except Exception: return {} def _resolve_ledger_path(workspace_root: Optional[str] = None) -> Path: explicit = os.environ.get(LEDGER_PATH_ENV) if explicit: return Path(explicit) root = workspace_root or os.environ.get("ANU_WORKSPACE_ROOT", "/home/jay/workspace") return Path(root) / LEDGER_PATH_DEFAULT def _load_latest_ledger_entry(ledger_path: Path, callback_id: Optional[str]) -> Optional[Dict[str, Any]]: if not ledger_path.is_file(): return None try: lines = ledger_path.read_text(encoding="utf-8").splitlines() except Exception: return None matched: Optional[Dict[str, Any]] = None for line in reversed(lines): line = line.strip() if not line: continue try: entry = json.loads(line) except Exception: continue if callback_id is None or entry.get("callback_id") == callback_id: matched = entry break return matched def _is_collector_session(stdin_payload: Dict[str, Any]) -> bool: if os.environ.get(COLLECTOR_MODE_ENV, "").strip().upper() == COLLECTOR_MODE_VALUE: return True # additionalContext 가 SessionStart 에서 collector mode 로 들어왔다는 hint 도 허용 if stdin_payload.get("collector_mode") is True: return True return False def evaluate( ledger_entry: Optional[Dict[str, Any]], *, last_output_text: str = "", collector_mode: bool = True, ) -> Tuple[bool, List[str]]: """10 + 2 추가 조건 검증. (block, failures) 반환.""" failures: List[str] = [] if not collector_mode: return False, failures # collector 세션이 아니면 무관 if ledger_entry is None: # ledger 자체가 없으면 1~6, 9 모두 fail (대표로 6 + 1 표시) failures.append(COND[6]) failures.append(COND[1]) return True, failures # 1. envelope_parsed if not ledger_entry.get("envelope_parsed"): failures.append(COND[1]) # 2. context_recovered if not ledger_entry.get("context_recovered"): failures.append(COND[2]) # 3. terminal_state_classified if not ledger_entry.get("terminal_state_classified"): failures.append(COND[3]) # 4. next_action_decided decided = ledger_entry.get("next_action_decided") if not decided: failures.append(COND[4]) # 5. 하나 이상의 행동 flag action_flags = ( ledger_entry.get("auto_action_dispatched"), ledger_entry.get("chair_report_emitted"), ledger_entry.get("noop_terminal_recorded"), ledger_entry.get("batch_wait_recorded"), ) if not any(bool(f) for f in action_flags): failures.append(COND[5]) # 6. ledger_written if not ledger_entry.get("callback_ledger_written"): failures.append(COND[6]) # 7. received phrase + source attribution 검증 source_attr = ledger_entry.get("source_attribution") misuse, _ = source_attribution_guard.detect_received_misuse(last_output_text, source_attr) if misuse: failures.append(COND[7]) # 8. schedule_history 사후 조회 as inbound if source_attribution_guard.detect_schedule_history_as_inbound(last_output_text, source_attr): failures.append(COND[8]) # 9. next_action_result result = ledger_entry.get("next_action_result") if not result: failures.append(COND[9]) # 9b. result=FAILED + recovery 없음 if result == NextActionResult.FAILED.value and not ledger_entry.get("recovery_action"): failures.append(COND["9b"]) # 10. state stale/mismatch 인데 SAFE_DEGRADED_MODE/HOLD_FOR_CHAIR 아님 state_status = ledger_entry.get("state_freshness_status") if state_status in {"STALE", "MISMATCH", "MISSING"}: if decided not in SAFE_DEGRADED_ALLOWED: failures.append(COND[10]) # 11. boost — Telegram = chair-required 한정 invariant (보강-2) invariant = validate_branch_invariant(ledger_entry) if invariant is not None: failures.append(COND["11"]) return bool(failures), failures def run(stdin_payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: payload = stdin_payload if stdin_payload is not None else _read_stdin() collector = _is_collector_session(payload) callback_id = payload.get("callback_id") or os.environ.get("ANU_CALLBACK_ID") last_output_text = ( payload.get("last_output_text") or os.environ.get(LAST_OUTPUT_ENV, "") or "" ) ledger_path = _resolve_ledger_path() entry = _load_latest_ledger_entry(ledger_path, callback_id) if collector else None block, failures = evaluate(entry, last_output_text=last_output_text, collector_mode=collector) out: Dict[str, Any] = { "schema": SCHEMA, "collector_mode": collector, "ledger_path": str(ledger_path), "callback_id": callback_id, "ledger_entry_loaded": entry is not None, "failures": failures, "block": block, } if block: out["decision"] = "block" out["reason"] = ( "ANU_CALLBACK_COLLECTOR_CONTROL_PLANE Stop hook 차단: " + ", ".join(failures) ) return out def main() -> int: try: result = run() except Exception as exc: # pragma: no cover — fail-open 으로 hook 안전 유지 sys.stderr.write(f"[stop_anu_callback_collector_verifier] error: {exc}\n") return 0 sys.stdout.write(json.dumps(result, ensure_ascii=False)) sys.stdout.flush() if result.get("block"): return 2 # Claude Code: non-zero=2 → block return 0 if __name__ == "__main__": raise SystemExit(main())