# -*- coding: utf-8 -*- """utils.ci_watch_handoff_runner — task-2642 CI_WATCH_HANDOFF state machine. 회장 verbatim (2026-05-23 19:38 KST) 1:1 박제: 'ANU 는 CI/Gemini 를 직접 기다리지 않는다. PR open 이후 대기/감시/자동수렴은 반드시 bot 또는 watcher task 에 위임한다.' 본 runner 책임 (spec §4 state machine): PR_OPEN (handoff received) ↓ poll loop · interval=poll_interval_seconds · timeout=max_watch_minutes ├─ CI 미완료 (PENDING) → polling continue ├─ CI FAIL 자동수렴 가능 → AUTO_REMEDIATE → re-push → re-poll ├─ CI FAIL 자동수렴 불가 → CI_FAILED_NON_REMEDIABLE ├─ forbidden_path 수정 감지 → CHAIR_REQUIRED (Critical7) └─ CI PASS → router 호출 (Gemini freshness) ├─ FRESH → MERGE_READY ├─ NUDGE_POSTED → polling continue (fresh review 대기) ├─ NUDGE_DEDUPED → polling continue (직전 nudge active) ├─ GEMINI_EXTERNAL_TRIGGER_STALE → terminal 동명 enum ├─ NUDGE_PERMISSION_DENIED / │ CHAIR_UI_FALLBACK_REQUIRED / │ NUDGE_FAILED / │ NOT_GEMINI_TRIGGER → CHAIR_REQUIRED (NUDGE_403 / permission) ↓ TERMINAL_REACHED → CALLBACK_FIRE (envelope UTF-8 ≤3900 bytes) ↓ WATCHER_EXIT PR #144 OWNER_GEMINI_TRIGGER_ROUTER stack 재사용 (gemini_router_call_fn 으로 inject). forbidden 15종 + owner_trigger 4종 + owner_gemini_trigger_router 3종 무수정. one-way isolation: utils/ 외부 import 0 (PR #144 router 는 caller 가 inject). live cokacdir / gh CLI / merge / push 호출 0 (regression mock 가능). frozen anchor: ANCHOR-1: state machine 1 차 — handoff 정규화 후 poll loop 진입 ANCHOR-2: router final_state 8종 → terminal_states 5종 매핑 (CHAIR_UI_FALLBACK / NUDGE_PERMISSION_DENIED / NUDGE_FAILED / NOT_GEMINI_TRIGGER → CHAIR_REQUIRED) ANCHOR-3: auto_remediation_fn 반환 enum 4종 (APPLIED / NON_REMEDIABLE / FORBIDDEN_HIT / LOOP_BOUNDARY) → terminal classification ANCHOR-4: callback envelope UTF-8 ≤3900 bytes hard limit (task-2612+3 박제) — 초과 시 RunnerContractError fail-closed ANCHOR-5: handoff.terminal_states subset 외부의 분류는 CHAIR_REQUIRED 로 escalate ANCHOR-6: real auto-merge 0 / push 0 / PR open 0 / live cokacdir 0 — caller 가 inject 한 callable 만 호출 """ from __future__ import annotations from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Final from utils.ci_watch_handoff_audit import ( CiWatchHandoffAudit, EVENT_AUTO_REMEDIATE, EVENT_CALLBACK_FIRED, EVENT_HANDOFF_RECEIVED, EVENT_OWNER_NUDGE, EVENT_POLL_TICK, EVENT_TERMINAL_REACHED, ) from utils.ci_watch_handoff_schema import ( ALL_TERMINAL_STATES, TERMINAL_CHAIR_REQUIRED, TERMINAL_CI_FAILED_NON_REMEDIABLE, TERMINAL_GEMINI_EXTERNAL_TRIGGER_STALE, TERMINAL_LOOP_BOUNDARY, TERMINAL_MERGE_READY, validate_handoff, ) # CI status enum (ci_status_fn 반환) CI_STATUS_PENDING: Final[str] = "PENDING" CI_STATUS_PASS: Final[str] = "PASS" CI_STATUS_FAIL: Final[str] = "FAIL" ALL_CI_STATUSES: Final[frozenset[str]] = frozenset( {CI_STATUS_PENDING, CI_STATUS_PASS, CI_STATUS_FAIL} ) # auto_remediation_fn 반환 enum REMEDIATE_APPLIED: Final[str] = "APPLIED" REMEDIATE_NON_REMEDIABLE: Final[str] = "NON_REMEDIABLE" REMEDIATE_FORBIDDEN_HIT: Final[str] = "FORBIDDEN_HIT" REMEDIATE_LOOP_BOUNDARY: Final[str] = "LOOP_BOUNDARY" ALL_REMEDIATE_OUTCOMES: Final[frozenset[str]] = frozenset( { REMEDIATE_APPLIED, REMEDIATE_NON_REMEDIABLE, REMEDIATE_FORBIDDEN_HIT, REMEDIATE_LOOP_BOUNDARY, } ) # PR #144 RouterResult.final_state 값 (재사용만 — 본 모듈은 string 매칭). ROUTER_FRESH: Final[str] = "FRESH" ROUTER_NUDGE_POSTED: Final[str] = "NUDGE_POSTED" ROUTER_NUDGE_DEDUPED: Final[str] = "NUDGE_DEDUPED" ROUTER_STALE: Final[str] = "GEMINI_EXTERNAL_TRIGGER_STALE" ROUTER_CHAIR_UI_FALLBACK: Final[str] = "CHAIR_UI_FALLBACK_REQUIRED" ROUTER_NUDGE_PERMISSION_DENIED: Final[str] = "NUDGE_PERMISSION_DENIED" ROUTER_NUDGE_FAILED: Final[str] = "NUDGE_FAILED" ROUTER_NOT_GEMINI_TRIGGER: Final[str] = "NOT_GEMINI_TRIGGER" DEFAULT_LOOP_BOUNDARY_ATTEMPTS: Final[int] = 3 # callback envelope hard limit (task-2612+3 박제) CALLBACK_ENVELOPE_BYTE_LIMIT: Final[int] = 3900 # ANU collector key 단일 출처 (task md §finalize §5) ANU_COLLECTOR_KEY: Final[str] = "c119085addb0f8b7" CANONICAL_ROOT: Final[str] = "/home/jay/workspace" class RunnerContractError(RuntimeError): """runner 입력 / inject callable / envelope 위반.""" def _default_no_op_remediation( _handoff: dict, _ci_snap: "CIStatusSnapshot" ) -> str: """fallback auto_remediation_fn — 항상 NON_REMEDIABLE (회장 verbatim §8 보호). caller 가 explicit 하게 auto_remediation_fn 을 inject 하지 않으면 watcher 는 어떤 자동수렴도 하지 않고 CI_FAILED_NON_REMEDIABLE 로 분류 (회장 verbatim §8 watcher 책임 명시화 보호). """ return REMEDIATE_NON_REMEDIABLE @dataclass(frozen=True) class RouterCallResult: """gemini_router_call_fn 반환 (PR #144 RouterResult 의 subset).""" final_state: str permission_diagnostics_present: bool = False reason: str = "" @dataclass(frozen=True) class CIStatusSnapshot: """ci_status_fn 반환 — caller 가 한 poll tick 의 CI 상태를 집약.""" status: str # CI_STATUS_PENDING / PASS / FAIL failing_checks: tuple[str, ...] = () severity: str = "" # "medium" / "style" / "quality" / "non-critical-high" / "high" / "critical" forbidden_path_touched: bool = False same_function_high_repeated: bool = False @dataclass(frozen=True) class TerminalDecision: """terminal state 분류 결과 + 진단 컨텍스트.""" terminal_state: str reason: str router_final_state: str | None = None ci_status: str | None = None auto_remediation_attempts: int = 0 loop_iterations: int = 0 @dataclass(frozen=True) class RunResult: """runner.run() 의 최종 반환 — terminal decision + callback 결과.""" decision: TerminalDecision callback_fired: bool callback_prompt_bytes: int class CiWatchHandoffRunner: """CI_WATCH_HANDOFF state machine 단일 진입점 (spec §4). side-effect 추상화 (test injection 가능): - ``ci_status_fn``: ``Callable[[dict], CIStatusSnapshot]`` — 한 poll tick 에서 caller 가 CI 상태를 회수해서 반환. regression 은 sequence 형태로 호출되어 PENDING → PASS / FAIL 등을 차례대로 반환. - ``gemini_router_call_fn``: ``Callable[[dict], RouterCallResult]`` — PR #144 OWNER_GEMINI_TRIGGER_ROUTER stack 호출 wrapper (재사용만). 본 runner 는 router 자체를 import 하지 않고 wrapper 시그니처로 inject. - ``auto_remediation_fn``: ``Callable[[dict, CIStatusSnapshot], str]`` — expected_files 내부 medium/style/quality/non-critical HIGH 자동수렴. 반환은 REMEDIATE_* enum. dev bot 위임 (ANU 직접 코드 0). - ``callback_send_fn``: ``Callable[[str], int]`` — envelope text 를 받아 실제 cokacdir --cron 호출 (또는 mock). 반환 무관 (audit 는 byte 수만 기록). runner 자체는 sleep 0 / live shell 0. cadence 는 caller 가 흉내내고 본 클래스는 state classification 만 책임. 회귀는 ci_status_fn 을 generator 스타일로 시퀀스 반환하도록 만들어 검증. """ def __init__( self, *, workspace_root: str | Path, ci_status_fn: Callable[[dict], CIStatusSnapshot] | None = None, gemini_router_call_fn: Callable[[dict], RouterCallResult] | None = None, auto_remediation_fn: Callable[[dict, CIStatusSnapshot], str] | None = None, callback_send_fn: Callable[[str], Any] | None = None, audit: CiWatchHandoffAudit | None = None, max_polls: int = 60, loop_boundary_attempts: int = DEFAULT_LOOP_BOUNDARY_ATTEMPTS, ) -> None: if ci_status_fn is None: raise RunnerContractError("ci_status_fn callable must be injected") if gemini_router_call_fn is None: raise RunnerContractError( "gemini_router_call_fn callable must be injected " "(PR #144 OWNER_GEMINI_TRIGGER_ROUTER wrapper)" ) if not isinstance(max_polls, int) or max_polls <= 0: raise RunnerContractError("max_polls must be positive int") if ( not isinstance(loop_boundary_attempts, int) or loop_boundary_attempts <= 0 ): raise RunnerContractError("loop_boundary_attempts must be positive int") self._workspace_root = Path(workspace_root).resolve() self._ci_status_fn = ci_status_fn self._gemini_router_call_fn = gemini_router_call_fn self._auto_remediation_fn = auto_remediation_fn or _default_no_op_remediation self._callback_send_fn = callback_send_fn self._audit = ( audit if audit is not None else CiWatchHandoffAudit(self._workspace_root) ) self._max_polls = max_polls self._loop_boundary_attempts = loop_boundary_attempts # ─── public API ────────────────────────────────────────────────────────── def run( self, handoff: dict, *, task_id: str = "", watcher_schedule_id: str | None = None, ) -> RunResult: """state machine 한 번 실행. terminal decision + callback 결과 반환.""" normalized = validate_handoff(handoff) pr = normalized["pr_number"] head = normalized["head_sha"] watcher_owner = normalized["watcher_owner"] terminal_subset = set(normalized["terminal_states"]) callback_on_terminal = bool(normalized["callback_on_terminal_state"]) self._audit.append( { "task_id": task_id, "pr_number": pr, "head_sha": head, "watcher_owner": watcher_owner, "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_HANDOFF_RECEIVED, "reason": ( f"PR #{pr} handoff received; watcher_owner={watcher_owner}; " f"branch={normalized['branch']}" ), } ) auto_remediation_attempts = 0 loop_iterations = 0 for tick in range(self._max_polls): loop_iterations = tick + 1 ci_snap = self._ci_status_fn(normalized) if not isinstance(ci_snap, CIStatusSnapshot): raise RunnerContractError( f"ci_status_fn must return CIStatusSnapshot, got " f"{type(ci_snap).__name__}" ) if ci_snap.status not in ALL_CI_STATUSES: raise RunnerContractError( f"ci_status_fn returned invalid status {ci_snap.status!r}; " f"allowed={sorted(ALL_CI_STATUSES)}" ) self._audit.append( { "task_id": task_id, "pr_number": pr, "head_sha": head, "watcher_owner": watcher_owner, "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_POLL_TICK, "ci_status": ci_snap.status, "loop_iterations": loop_iterations, "reason": ( f"poll tick {loop_iterations}: CI={ci_snap.status}; " f"severity={ci_snap.severity or 'n/a'}; " f"forbidden_touched={ci_snap.forbidden_path_touched}" ), } ) # forbidden_path 직접 감지 → CHAIR_REQUIRED 즉시 escalate if ci_snap.forbidden_path_touched: return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_CHAIR_REQUIRED, reason=( "forbidden_paths 수정 감지 — Critical7 " "(spec §3.2 / 회장 verbatim §3.2)" ), ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) if ci_snap.status == CI_STATUS_PENDING: continue if ci_snap.status == CI_STATUS_FAIL: # loop boundary 선검사 (same-function HIGH 반복 + attempts >= N) if ( ci_snap.same_function_high_repeated and auto_remediation_attempts >= self._loop_boundary_attempts ): return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_LOOP_BOUNDARY, reason=( f"auto_remediation attempts=" f"{auto_remediation_attempts} >= " f"{self._loop_boundary_attempts} + same-function " "HIGH 반복 — LOOP_BOUNDARY (spec §3.5)" ), ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) outcome = self._auto_remediation_fn(normalized, ci_snap) if outcome not in ALL_REMEDIATE_OUTCOMES: raise RunnerContractError( f"auto_remediation_fn returned invalid outcome " f"{outcome!r}; allowed={sorted(ALL_REMEDIATE_OUTCOMES)}" ) auto_remediation_attempts += 1 self._audit.append( { "task_id": task_id, "pr_number": pr, "head_sha": head, "watcher_owner": watcher_owner, "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_AUTO_REMEDIATE, "ci_status": ci_snap.status, "auto_remediation_attempts": auto_remediation_attempts, "loop_iterations": loop_iterations, "reason": ( f"auto_remediation outcome={outcome} " f"(severity={ci_snap.severity or 'n/a'})" ), } ) if outcome == REMEDIATE_APPLIED: continue # re-poll if outcome == REMEDIATE_FORBIDDEN_HIT: return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_CHAIR_REQUIRED, reason=( "auto_remediation refused — forbidden_paths hit " "/ Critical7 escalation (spec §3.2)" ), ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) if outcome == REMEDIATE_LOOP_BOUNDARY: return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_LOOP_BOUNDARY, reason=( "auto_remediation declared LOOP_BOUNDARY " "(same-function HIGH 반복) — spec §3.5" ), ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) # outcome == REMEDIATE_NON_REMEDIABLE return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_CI_FAILED_NON_REMEDIABLE, reason=( f"CI failure non-remediable — " f"severity={ci_snap.severity!r} " f"failing_checks={list(ci_snap.failing_checks)} " "(spec §3.4)" ), ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) # ci_snap.status == CI_STATUS_PASS → Gemini freshness check via router router_result = self._gemini_router_call_fn(normalized) if not isinstance(router_result, RouterCallResult): raise RunnerContractError( f"gemini_router_call_fn must return RouterCallResult, got " f"{type(router_result).__name__}" ) self._audit.append( { "task_id": task_id, "pr_number": pr, "head_sha": head, "watcher_owner": watcher_owner, "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_OWNER_NUDGE, "router_final_state": router_result.final_state, "ci_status": ci_snap.status, "loop_iterations": loop_iterations, "reason": ( f"router → {router_result.final_state}; " f"perm_diag={router_result.permission_diagnostics_present}; " f"{router_result.reason}" ), } ) mapped = self._map_router_state(router_result.final_state) if mapped is None: # FRESH / NUDGE_POSTED / NUDGE_DEDUPED → polling continue # (NUDGE_POSTED 는 fresh review 도착 polling, DEDUPED 는 직전 nudge active) if router_result.final_state == ROUTER_FRESH: return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_MERGE_READY, reason=( "CI PASS + Gemini fresh + 0 unresolved + CLEAN " "— MERGE_READY (spec §3.1)" ), router_final_state=router_result.final_state, ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) # NUDGE_POSTED / NUDGE_DEDUPED → polling continue continue return self._finalize( decision=TerminalDecision( terminal_state=mapped, reason=self._terminal_reason_for_router( router_result.final_state ), router_final_state=router_result.final_state, ci_status=ci_snap.status, auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=normalized, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) # max_polls 도달 → CHAIR_REQUIRED (timeout) return self._finalize( decision=TerminalDecision( terminal_state=TERMINAL_CHAIR_REQUIRED, reason=( f"max_polls ({self._max_polls}) 도달 — terminal 미분류 " "(max_watch_minutes hard timeout 정합)" ), auto_remediation_attempts=auto_remediation_attempts, loop_iterations=loop_iterations, ), handoff=handoff, task_id=task_id, watcher_schedule_id=watcher_schedule_id, terminal_subset=terminal_subset, callback_on_terminal=callback_on_terminal, ) # ─── internal helpers ──────────────────────────────────────────────────── @staticmethod def _map_router_state(state: str) -> str | None: """router final_state → CI_WATCH_HANDOFF terminal_state 매핑. 반환 None 이면 polling continue 또는 MERGE_READY (caller 분기). """ if state in (ROUTER_FRESH, ROUTER_NUDGE_POSTED, ROUTER_NUDGE_DEDUPED): return None if state == ROUTER_STALE: return TERMINAL_GEMINI_EXTERNAL_TRIGGER_STALE if state in ( ROUTER_CHAIR_UI_FALLBACK, ROUTER_NUDGE_PERMISSION_DENIED, ROUTER_NUDGE_FAILED, ROUTER_NOT_GEMINI_TRIGGER, ): return TERMINAL_CHAIR_REQUIRED return TERMINAL_CHAIR_REQUIRED # unknown → fail-closed escalate @staticmethod def _terminal_reason_for_router(state: str) -> str: if state == ROUTER_STALE: return ( "OWNER nudge 1회 hard limit 후 fresh review 미도착 — " "GEMINI_EXTERNAL_TRIGGER_STALE (spec §3.3)" ) if state == ROUTER_NUDGE_PERMISSION_DENIED: return ( "router NUDGE_PERMISSION_DENIED (403 X-Accepted-* headers 기록) " "— CHAIR_REQUIRED (회장 verbatim §8 / NUDGE_403=permission)" ) if state == ROUTER_CHAIR_UI_FALLBACK: return ( "router CHAIR_UI_FALLBACK_REQUIRED — CHAIR_REQUIRED " "(회장 UI 최후수단 회피, 본 정책 §2.10)" ) if state == ROUTER_NUDGE_FAILED: return ( "router NUDGE_FAILED (transient/unknown) — CHAIR_REQUIRED " "(자동 retry 회피, 회장 보고)" ) if state == ROUTER_NOT_GEMINI_TRIGGER: return ( "router NOT_GEMINI_TRIGGER (PR Review 오인 차단 — task-2640 박제) " "— CHAIR_REQUIRED" ) return f"unknown router state {state!r} — CHAIR_REQUIRED (fail-closed)" def _finalize( self, *, decision: TerminalDecision, handoff: dict, task_id: str, watcher_schedule_id: str | None, terminal_subset: set, callback_on_terminal: bool, ) -> RunResult: """terminal 도달 audit + (옵션) callback envelope 발사.""" if decision.terminal_state not in ALL_TERMINAL_STATES: raise RunnerContractError( f"terminal_state {decision.terminal_state!r} not in enum" ) # handoff.terminal_states 가 enum 일부만 허용한 경우 → CHAIR_REQUIRED escalate if decision.terminal_state not in terminal_subset: decision = TerminalDecision( terminal_state=TERMINAL_CHAIR_REQUIRED, reason=( f"computed terminal={decision.terminal_state} 이 handoff " f"terminal_states subset {sorted(terminal_subset)} 외부 — " "CHAIR_REQUIRED escalate (ANCHOR-5)" ), router_final_state=decision.router_final_state, ci_status=decision.ci_status, auto_remediation_attempts=decision.auto_remediation_attempts, loop_iterations=decision.loop_iterations, ) self._audit.append( { "task_id": task_id, "pr_number": handoff["pr_number"], "head_sha": handoff["head_sha"], "watcher_owner": handoff["watcher_owner"], "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_TERMINAL_REACHED, "terminal_state": decision.terminal_state, "router_final_state": decision.router_final_state, "ci_status": decision.ci_status, "auto_remediation_attempts": decision.auto_remediation_attempts, "loop_iterations": decision.loop_iterations, "reason": decision.reason, } ) callback_fired = False callback_bytes = 0 if callback_on_terminal: envelope = self._build_callback_envelope( handoff=handoff, decision=decision, task_id=task_id, watcher_schedule_id=watcher_schedule_id, ) byte_count = len(envelope.encode("utf-8")) # ANCHOR-4: envelope ≤3900 bytes hard limit if byte_count > CALLBACK_ENVELOPE_BYTE_LIMIT: raise RunnerContractError( f"callback envelope exceeds {CALLBACK_ENVELOPE_BYTE_LIMIT} " f"bytes ({byte_count}) — task-2612+3 박제 hard limit" ) callback_bytes = byte_count if self._callback_send_fn is not None: # caller 가 cokacdir --cron 호출 (또는 mock). 반환값은 무시. self._callback_send_fn(envelope) callback_fired = True self._audit.append( { "task_id": task_id, "pr_number": handoff["pr_number"], "head_sha": handoff["head_sha"], "watcher_owner": handoff["watcher_owner"], "watcher_schedule_id": watcher_schedule_id or "", "event": EVENT_CALLBACK_FIRED, "terminal_state": decision.terminal_state, "callback_prompt_bytes": callback_bytes, "reason": ( f"ANU normal callback fired ({callback_bytes} bytes UTF-8 " f"<= {CALLBACK_ENVELOPE_BYTE_LIMIT})" ), } ) return RunResult( decision=decision, callback_fired=callback_fired, callback_prompt_bytes=callback_bytes, ) @staticmethod def _build_callback_envelope( *, handoff: dict, decision: TerminalDecision, task_id: str, watcher_schedule_id: str | None, ) -> str: """ANU normal callback envelope (spec §4.3). envelope 5축 + canonical_root + terminal_state + watcher_owner + schedule_id 명시. UTF-8 ≤3900 bytes hard limit (caller wc -c 검증). 상세 보고는 result.json / report.md 에 위임 (envelope 만 포함). """ head_prefix = handoff["head_sha"][:12] lines = [ f"[CI_WATCH_HANDOFF_TERMINAL] task={task_id or 'n/a'}", f"pr=#{handoff['pr_number']} head={head_prefix}", f"branch={handoff['branch']}", f"watcher_owner={handoff['watcher_owner']}", f"watcher_schedule_id={watcher_schedule_id or ''}", f"terminal_state={decision.terminal_state}", f"router_final_state={decision.router_final_state or 'n/a'}", f"ci_status={decision.ci_status or 'n/a'}", f"auto_remediation_attempts={decision.auto_remediation_attempts}", f"loop_iterations={decision.loop_iterations}", f"canonical_root={CANONICAL_ROOT}", f"collector_role=ANU owner_key={ANU_COLLECTOR_KEY}", f"reason={decision.reason}", ] return "\n".join(lines) __all__ = [ "CI_STATUS_PENDING", "CI_STATUS_PASS", "CI_STATUS_FAIL", "ALL_CI_STATUSES", "REMEDIATE_APPLIED", "REMEDIATE_NON_REMEDIABLE", "REMEDIATE_FORBIDDEN_HIT", "REMEDIATE_LOOP_BOUNDARY", "ALL_REMEDIATE_OUTCOMES", "ROUTER_FRESH", "ROUTER_NUDGE_POSTED", "ROUTER_NUDGE_DEDUPED", "ROUTER_STALE", "ROUTER_CHAIR_UI_FALLBACK", "ROUTER_NUDGE_PERMISSION_DENIED", "ROUTER_NUDGE_FAILED", "ROUTER_NOT_GEMINI_TRIGGER", "DEFAULT_LOOP_BOUNDARY_ATTEMPTS", "CALLBACK_ENVELOPE_BYTE_LIMIT", "ANU_COLLECTOR_KEY", "CANONICAL_ROOT", "RunnerContractError", "RouterCallResult", "CIStatusSnapshot", "TerminalDecision", "RunResult", "CiWatchHandoffRunner", ]