#!/usr/bin/env python3 # -*- coding: utf-8 -*- """hooks/session_start_anu_callback_collector.py — SessionStart hook (STAGED · live 미적용). task-2644 ANU_CALLBACK_COLLECTOR_CONTROL_PLANE (회장 verbatim 우선순위 1). spec: memory/specs/system_anu_callback_collector_control_plane_spec_260524.md spec sha256: b27da557d4245bce476cd63f4ab174aefc8a25d2da07ec2c8d2c83b01ee96153 ★ 본 hook 은 staged template only · ~/.claude/settings.json 에 등록되지 않은 상태. ★ live 등록은 별도 HARNESS_ENFORCED task 에서 회장 verbatim signature 후 진행. 회장 verbatim flow (spec §1.2): cokacdir 가 ANU collector 세션 spawn → SessionStart hook (본 모듈) 이 ANU_CALLBACK_COLLECTOR 모드 강제 → collector 가 task/spec/.anu_state/frozen anchor 로 context recovery 회장 verbatim 7 단계 (spec §3): 1. envelope parse 2. context recovery (task md / spec md / .anu_state.json / frozen anchor) 3. terminal_state classify 4. next_action decide 5. auto action OR chair telegram 6. .callback ledger write 7. source attribution + exit 본 모듈 책임 (1~2 단계 + 의무 prompt 주입): - ENV COKACDIR_MODE 검사 - envelope 파싱 (stdin JSON 또는 ENV path) - task md / spec md / .anu_state 자동 로드 - frozen anchor 본문 + forbidden/allowed action list + 7 단계 의무 주입 - 3~6 단계는 collector 본체가 utils.callback_adjudicator + callback_next_action_runner 로 수행 · 7 단계 (source attribution + exit) 는 Stop hook 이 검증 Claude Code SessionStart hook 계약 (참고): - stdin: {"session_id":..., "transcript_path":..., "hook_event_name":"SessionStart"} - stdout: additionalContext 로 주입 가능 (JSON) - exit 0: continue · non-zero: block """ from __future__ import annotations import json import os import sys from pathlib import Path from typing import Any, Dict, Optional SCHEMA = "hooks.session_start_anu_callback_collector.v1" COLLECTOR_MODE_ENV = "COKACDIR_MODE" COLLECTOR_MODE_VALUE = "ANU_CALLBACK_COLLECTOR" ENVELOPE_PATH_ENV = "ANU_CALLBACK_ENVELOPE_PATH" ENVELOPE_INLINE_ENV = "ANU_CALLBACK_ENVELOPE_JSON" ANU_STATE_PATH_DEFAULT = "memory/system/.anu_state.json" WORKSPACE_ROOT_ENV = "ANU_WORKSPACE_ROOT" WORKSPACE_ROOT_DEFAULT = "/home/jay/workspace" FROZEN_ANCHORS = [ ("ANCHOR-1", "본 task = ANU collector 자율 처리 + 다음 action 자동 진행 control plane"), ("ANCHOR-2", "Harness 단독 불가능 · Harness + cokacdir + file-state 결합 필수"), ("ANCHOR-3", "8 우선순위 + 15 필수 산출물 1:1 박제"), ("ANCHOR-4", "SessionStart = collector mode 강제 / Stop = 의무 미완·거짓표현 차단"), ("ANCHOR-5", "next_action 11 enum + 자동 허용 8 + 자동 금지 11"), ("ANCHOR-6", "RUNTIME_GUARDED 까지만 본 task · HARNESS_ENFORCED 별도 task"), ("ANCHOR-7", "본 ANU 대화 세션 hook = 이중 안전망 · 주 경로 아님"), ("ANCHOR-8", "PR #146 task-2643 와 격리 · 별도 worktree / 별도 PR"), ("ANCHOR-9", "거짓말 패턴 (수신 vs 사후 조회) Stop hook 사전 차단"), ("ANCHOR-10", "본 ANU 직접 polling 0 · 회장 verbatim dogfood"), ("ANCHOR-11", "source attribution enum 8 · CALLBACK_COLLECTOR_PROCESSED 주 경로"), ("ANCHOR-12", "next_action 3 분기 mutually exclusive · Telegram = chair-required 한정"), ("ANCHOR-13", "merge execution 0 hardcoded · MERGE_READY → REQUEST_CHAIR_MERGE_APPROVAL only"), ("ANCHOR-14", ".anu_state freshness state_version 검증 · stale 시 fail-closed"), ("ANCHOR-15", "Stop hook = decided + attempted + result + evidence_path 4 필드 검증"), ] FORBIDDEN_ACTIONS = [ "merge 실행", "live ~/.claude/settings.json 적용", "live cokacdir 수정", "BOT App token 사용", "chair_authorization 발급", "PR #141 pilot", "production PR lifecycle activation", "expected_files 밖 수정", "credential/permission expansion", "admin override", "destructive git / foreign dirty cleanup", ] ALLOWED_AUTO_ACTIONS = [ "expected_files 내부 non-critical remediation", "phase3 timing race rerun", "allowed OWNER_GEMINI_TRIGGER_ROUTER nudge", "batch sibling callback wait", "all-settled batch adjudication", "follow-up task spec/preflight", "regression rerun", "callback ack/ledger update", ] COLLECTOR_DUTY_7_STEPS = [ "envelope parse", "context recovery (task md / spec md / .anu_state.json / frozen anchor)", "terminal_state classify (utils.callback_adjudicator)", "next_action decide (utils.callback_next_action_runner · 11 enum)", "auto action OR chair telegram (3 분기 mutually exclusive)", ".callback ledger write (schemas/callback_ledger_v1.json)", "source attribution + exit (utils.source_attribution_guard · 8 enum)", ] 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 _load_envelope() -> Optional[Dict[str, Any]]: path = os.environ.get(ENVELOPE_PATH_ENV) if path and Path(path).is_file(): try: return json.loads(Path(path).read_text(encoding="utf-8")) except Exception: return None inline = os.environ.get(ENVELOPE_INLINE_ENV) if inline: try: return json.loads(inline) except Exception: return None return None def _safe_read_text(path: Path, max_bytes: int = 32_000) -> Optional[str]: try: if not path.is_file(): return None data = path.read_bytes() if len(data) > max_bytes: data = data[:max_bytes] return data.decode("utf-8", errors="replace") except Exception: return None def _load_anu_state(workspace_root: Path) -> Optional[Dict[str, Any]]: p = workspace_root / ANU_STATE_PATH_DEFAULT if not p.is_file(): return None try: return json.loads(p.read_text(encoding="utf-8")) except Exception: return None def is_collector_mode() -> bool: mode = os.environ.get(COLLECTOR_MODE_ENV, "").strip().upper() if mode == COLLECTOR_MODE_VALUE: return True return _load_envelope() is not None def build_additional_context( *, envelope: Optional[Dict[str, Any]], task_md_excerpt: Optional[str], spec_md_excerpt: Optional[str], anu_state: Optional[Dict[str, Any]], workspace_root: Path, # noqa: ARG001 — 향후 추가 source loader 확장용 ) -> str: """collector 본체에 주입할 system context (markdown).""" lines = [] lines.append("# ANU_CALLBACK_COLLECTOR_CONTROL_PLANE — SessionStart 주입") lines.append("") lines.append(f"**workspace_root**: `{workspace_root}`") lines.append("**모드**: ANU_CALLBACK_COLLECTOR (envelope 자율 처리)") lines.append("") lines.append("## frozen anchor (15)") for tag, body in FROZEN_ANCHORS: lines.append(f"- **{tag}**: {body}") lines.append("") lines.append("## 자동 진행 허용 8 (회장 verbatim)") for a in ALLOWED_AUTO_ACTIONS: lines.append(f"- {a}") lines.append("") lines.append("## 자동 진행 금지 11 (회장 verbatim)") for a in FORBIDDEN_ACTIONS: lines.append(f"- {a}") lines.append("") lines.append("## collector 의무 7 단계 (Stop hook 검증)") for idx, step in enumerate(COLLECTOR_DUTY_7_STEPS, 1): lines.append(f"{idx}. {step}") lines.append("") lines.append("## envelope (parsed)") if envelope is None: lines.append("- ⚠ envelope 미수신 — SAFE_DEGRADED_MODE / HOLD_FOR_CHAIR 강제") else: cb_id = envelope.get("callback_id", "") task_id = envelope.get("task_id", "") pr = envelope.get("pr_number", None) snap = envelope.get("dispatch_state_snapshot_id") or envelope.get("snapshot_id") or "" lines.append(f"- callback_id: `{cb_id}`") lines.append(f"- task_id: `{task_id}`") lines.append(f"- pr_number: `{pr}`") lines.append(f"- dispatch_state_snapshot_id: `{snap}`") lines.append(f"- canonical_root: `{envelope.get('canonical_root', WORKSPACE_ROOT_DEFAULT)}`") lines.append("") lines.append("## .anu_state (freshness 검증 입력)") if anu_state is None: lines.append("- ⚠ .anu_state 미로드 → state_freshness=MISSING → HOLD_FOR_CHAIR fail-closed") else: lines.append(f"- snapshot_id: `{anu_state.get('snapshot_id')}`") lines.append(f"- state_version: `{anu_state.get('state_version')}`") lines.append(f"- updated_at: `{anu_state.get('updated_at')}`") lines.append("") if task_md_excerpt: lines.append("## task md 발췌 (top 4KB)") lines.append("```markdown") lines.append(task_md_excerpt[:4000]) lines.append("```") if spec_md_excerpt: lines.append("## spec md 발췌 (top 4KB)") lines.append("```markdown") lines.append(spec_md_excerpt[:4000]) lines.append("```") lines.append("") lines.append("## 다음 명령") lines.append( "collector 는 envelope + anchor + state 를 기반으로 callback_adjudicator → " "callback_next_action_runner 를 호출하고 .callback ledger 에 결과를 기록한 뒤 " "source attribution 을 명시하고 종료한다. Stop hook 이 10 조건을 검증한다." ) return "\n".join(lines) def run(stdin_payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: """SessionStart hook 본체. stdout 으로 JSON 반환 (Claude Code 계약).""" _ = stdin_payload if stdin_payload is not None else _read_stdin() workspace_root = Path(os.environ.get(WORKSPACE_ROOT_ENV, WORKSPACE_ROOT_DEFAULT)) if not is_collector_mode(): # 일반 세션 — pass-through (block 하지 않음) return {"schema": SCHEMA, "collector_mode": False, "additionalContext": None} envelope = _load_envelope() task_id = (envelope or {}).get("task_id") or os.environ.get("ANU_TASK_ID") task_md = None if task_id: task_md = _safe_read_text(workspace_root / "memory" / "tasks" / f"{task_id}.md") spec_md = _safe_read_text( workspace_root / "memory" / "specs" / "system_anu_callback_collector_control_plane_spec_260524.md" ) anu_state = _load_anu_state(workspace_root) ctx = build_additional_context( envelope=envelope, task_md_excerpt=task_md, spec_md_excerpt=spec_md, anu_state=anu_state, workspace_root=workspace_root, ) out = { "schema": SCHEMA, "collector_mode": True, "envelope_parsed": envelope is not None, "task_md_loaded": task_md is not None, "spec_md_loaded": spec_md is not None, "anu_state_loaded": anu_state is not None, "anchors_injected": len(FROZEN_ANCHORS), "forbidden_actions_injected": len(FORBIDDEN_ACTIONS), "allowed_auto_actions_injected": len(ALLOWED_AUTO_ACTIONS), "duty_steps_injected": len(COLLECTOR_DUTY_7_STEPS), # Claude Code hook 계약: hookSpecificOutput.additionalContext 로 주입 "hookSpecificOutput": { "hookEventName": "SessionStart", "additionalContext": ctx, }, } return out def main() -> int: try: result = run() except Exception as exc: # pragma: no cover — fail-open for hook safety sys.stderr.write(f"[session_start_anu_callback_collector] error: {exc}\n") return 0 sys.stdout.write(json.dumps(result, ensure_ascii=False)) sys.stdout.flush() return 0 if __name__ == "__main__": raise SystemExit(main())