"""task-2708 P2-A callback pre-registration helper. 회장 인가: CHAIR-AUTH-TASK-2708-P2A-CALLBACK-BEFORE-FAILFAST-PREREGISTRATION-260529 원칙: - idempotent: lock 파일로 1회 결정성 보장 (§5.3/§9.2) - non-blocking: 모든 예외 → try/except → exit 0 (success path 영향 0) - ANU-key only: owner_key 는 ANU key 전용 (c119085addb0f8b7) - helper 수정 0: dispatch.normal_fallback_callback_helper 코드 수정 없이 경유 finish-task.sh 진입 직후 fail-fast(exit 1) 이전에 ANU normal callback 을 1회 결정성 pre-register 하는 헬퍼. dispatch.normal_fallback_callback_helper. launch_callback 을 경유한다 (헬퍼 자체 수정 0). """ from __future__ import annotations import argparse import json import sys from datetime import datetime, timezone from pathlib import Path from typing import Optional import dispatch.normal_fallback_callback_helper as _cb_helper # ─── 상수 ────────────────────────────────────────────────────────────────────── SCHEMA = "callback_preregistration.v1" PROMPT_MAX_BYTES = 3900 def _now_iso() -> str: return datetime.now(timezone.utc).isoformat() def _write_marker(path: Path, payload: dict) -> None: """마커 JSON 을 원자적으로 기록한다.""" path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_suffix(".tmp") tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") tmp.replace(path) def _audit(msg: str) -> None: print(msg, file=sys.stderr) def cmd_launch(args: argparse.Namespace) -> int: """launch subcommand: fail-fast 이전 ANU callback pre-registration.""" task_id: str = args.task_id kind: str = args.kind owner_key: str = args.owner_key chat_id: str = args.chat_id executor_key: str = args.executor_key envelope_path: str = args.envelope_path events_dir: str = args.events_dir workspace: str = args.workspace output: Optional[str] = args.output events = Path(events_dir) # lock 파일 결정 (kind별) if kind == "normal": lock_path = events / f"{task_id}.callback-prereg-lock" else: lock_path = events / f"{task_id}.callback-fallback-lock" # marker 출력 경로 결정 if output: launch_marker_path = Path(output) elif kind == "normal": launch_marker_path = events / f"{task_id}.callback-prereg-launch.json" else: launch_marker_path = events / f"{task_id}.callback-fallback-launch.json" if kind == "normal": not_reg_marker_path = events / f"{task_id}.callback-prereg-not-registered.json" else: not_reg_marker_path = events / f"{task_id}.callback-fallback-not-registered.json" try: # ── Step 1: idempotent lock 확인 ───────────────────────────────────── if lock_path.exists(): _audit( f"[callback_preregistration] [{task_id}] lock exists → silent skip (idempotent §5.3/§9.2): {lock_path}" ) return 0 # ── Step 2: envelope 파일 존재 확인 ────────────────────────────────── if not Path(envelope_path).exists(): _audit( f"[callback_preregistration] [{task_id}] envelope absent → skip (pre-reg not required): {envelope_path}" ) return 0 # ── Step 3: prompt 빌드 ──────────────────────────────────────────── kind_upper = kind.upper() prompt_lines = [ f"task_id={task_id}", f"result_path=memory/events/{task_id}.result.json", f"report_path=memory/reports/{task_id}.md", "collector_role=ANU", f"callback_kind={kind}", f"source_attribution=FINISH_TASK_SH_P2A_PREREGISTRATION_{kind_upper}", f"owner_key={owner_key}", f"canonical_root={workspace}", ] prompt = "\n".join(prompt_lines) # UTF-8 byte 제한 확인 prompt_bytes = prompt.encode("utf-8") if len(prompt_bytes) > PROMPT_MAX_BYTES: prompt = prompt_bytes[:PROMPT_MAX_BYTES].decode("utf-8", errors="ignore") # ── Step 4: launch_callback 호출 ────────────────────────────────── decision = _cb_helper.launch_callback( kind=kind, task_id=task_id, executor_key=executor_key, owner_key=owner_key, chat_id=chat_id, prompt=prompt, at=None, canonical_root=workspace, ) # ── Step 5/6: decision 처리 ──────────────────────────────────────── if decision.ok: # lock 파일 touch lock_path.parent.mkdir(parents=True, exist_ok=True) lock_path.touch() # launch marker 박제 marker_payload = { "schema": SCHEMA, "task_id": task_id, "kind": kind, "status": "ANU_OWNED_READY", "verdict": decision.verdict, "prereg": True, "ts": _now_iso(), } _write_marker(launch_marker_path, marker_payload) # stdout 에 decision.to_json() 출력 print(json.dumps(decision.to_json(), ensure_ascii=False)) return 0 else: # FAIL_CLOSED → non-blocking, exit 0 marker_payload = { "schema": SCHEMA, "task_id": task_id, "kind": kind, "status": "FAIL_CLOSED", "verdict": decision.verdict, "reasons": list(decision.reasons), "prereg": True, "ts": _now_iso(), } _write_marker(not_reg_marker_path, marker_payload) _audit( f"[callback_preregistration] [{task_id}] FAIL_CLOSED verdict={decision.verdict} → non-blocking exit 0" ) return 0 except Exception as exc: # ★ 절대 non-zero exit 금지 _audit( f"[callback_preregistration] [{task_id}] EXCEPTION {type(exc).__name__}: {exc} → non-blocking exit 0" ) # 예외 마커 박제 시도 try: marker_payload = { "schema": SCHEMA, "task_id": task_id, "kind": kind, "status": "EXCEPTION", "error": f"{type(exc).__name__}: {exc}", "prereg": True, "ts": _now_iso(), } _write_marker(not_reg_marker_path, marker_payload) except Exception: pass return 0 def _build_parser() -> argparse.ArgumentParser: parser = argparse.ArgumentParser( prog="callback_preregistration.py", description=( "task-2708 P2-A: fail-fast 이전 ANU callback pre-registration helper. " "CHAIR-AUTH-TASK-2708-P2A-CALLBACK-BEFORE-FAILFAST-PREREGISTRATION-260529" ), ) sub = parser.add_subparsers(dest="subcmd", required=True) launch_p = sub.add_parser("launch", help="ANU callback pre-registration 실행") launch_p.add_argument("--task-id", required=True, help="태스크 ID") launch_p.add_argument( "--kind", choices=["normal", "fallback"], default="normal", help="callback 종류 (default: normal)", ) launch_p.add_argument("--owner-key", required=True, help="ANU owner key") launch_p.add_argument("--chat-id", required=True, help="chat ID") launch_p.add_argument( "--executor-key", default="", help="executor self key (default: '')" ) launch_p.add_argument("--envelope-path", required=True, help="callback envelope 파일 경로") launch_p.add_argument("--events-dir", required=True, help="events 디렉토리 경로") launch_p.add_argument( "--workspace", default="/home/jay/workspace", help="canonical workspace 경로 (default: /home/jay/workspace)", ) launch_p.add_argument( "--output", default=None, help="launch marker 출력 경로 (선택)" ) launch_p.set_defaults(func=cmd_launch) return parser def main(argv=None) -> int: parser = _build_parser() args = parser.parse_args(argv) return args.func(args) if __name__ == "__main__": raise SystemExit(main())