""" delegation_watcher.py — ANU delegation completion watcher (workspace module) task-2773: 완료감지 축만 구현 (merge/activation 무관) 순수 표준 라이브러리만 사용 (self-contained stdlib only). """ import json from datetime import datetime, timezone from pathlib import Path from typing import Optional # --------------------------------------------------------------------------- # 기본 경로 계산: utils/ 의 부모 = workspace 루트 # --------------------------------------------------------------------------- _DEFAULT_WORKSPACE: Path = Path(__file__).resolve().parents[1] _DEFAULT_LEDGER_PATH: Path = _DEFAULT_WORKSPACE / "memory" / "state" / "pending_delegations.jsonl" # 허용 close 상태 값 _VALID_CLOSE_STATUSES = {"done", "failure", "essence_pass", "merged"} def _now_iso() -> str: """현재 UTC 시각을 ISO8601 형식으로 반환.""" return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _resolve_ledger(ledger_path: Optional[str | Path]) -> Path: """ledger_path 인자가 None 이면 기본 경로를 반환.""" if ledger_path is None: return _DEFAULT_LEDGER_PATH return Path(ledger_path) def record_delegation( task_id: str, bot: str, team: str, branch: Optional[str], expected_result_paths: list, dispatch_time: str, ledger_path=None, ) -> dict: """ledger 에 1줄 append (status='pending'). 중복 task_id 는 append (audit 목적, 최신 우선). 반환: append 된 레코드 dict. """ lp = _resolve_ledger(ledger_path) lp.parent.mkdir(parents=True, exist_ok=True) record = { "task_id": task_id, "bot": bot, "team": team, "branch": branch, "expected_result_paths": [str(p) for p in expected_result_paths], "dispatch_time": dispatch_time, "status": "pending", "record_type": "record", "ts": _now_iso(), } with open(lp, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") return record def check_pending_delegations(workspace=None, ledger_path=None) -> list: """ledger 의 status=='pending' 항목마다 디스크 확인 후 판정 dict 리스트 반환. 동일 task_id 가 여러 번 append 됐으면 최신(마지막) record 우선. 이후 close 레코드(record_type=='close')가 있으면 해당 task_id 는 결과에서 제외(종결됨). ledger_path 미지정 + workspace 지정 시 해당 workspace 기준 ledger 경로 사용. """ if ledger_path is None and workspace is not None: ledger_path = Path(workspace) / "memory" / "state" / "pending_delegations.jsonl" lp = _resolve_ledger(ledger_path) if not lp.exists(): return [] # ledger 파싱 (깨진 줄은 skip) all_records: list = [] with open(lp, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line: continue try: all_records.append(json.loads(line)) except json.JSONDecodeError: continue # 방어적 skip # task_id 별 lifecycle 이벤트를 순서대로 추적: # - record_type=='record' 면 active=True + latest_records 갱신 (재오픈 포함) # - record_type=='close' 면 active=False # 최종적으로 active==True 인 task_id 만 후보 (마지막 이벤트가 record 인 것만) latest_records: dict = {} # task_id -> record dict (마지막 record 타입 레코드) active_ids: dict = {} # task_id -> bool (True=마지막 이벤트가 record) for rec in all_records: tid = rec.get("task_id") if tid is None: continue rtype = rec.get("record_type") if rtype == "record": # 뒤에 나올수록 최신 → 덮어쓰기 (재오픈 포함) latest_records[tid] = rec active_ids[tid] = True elif rtype == "close": active_ids[tid] = False results: list = [] for task_id, rec in latest_records.items(): # 마지막 lifecycle 이벤트가 close 이면 종결된 항목 — 결과에서 제외 if not active_ids.get(task_id, False): continue # status=='pending' 인 항목만 검사 if rec.get("status") != "pending": continue expected_paths: list = rec.get("expected_result_paths", []) # --- 파일 존재 여부 확인 --- has_done = False has_failure_envelope = False has_result_or_pr = False has_crash = False for p_item in expected_paths: p_str = str(p_item) # Path 객체도 안전하게 처리 p = Path(p_str) if p_str.endswith(".done") and p.exists(): has_done = True if p_str.endswith(".failure-envelope.json") and p.exists(): has_failure_envelope = True if (p_str.endswith("result.json") or p_str.endswith(".result.json")) and p.exists(): has_result_or_pr = True # crash marker 직접 경로 지정 확인 if ".supervisor-crash-marker.json" in p_str and p.exists(): has_crash = True # crash marker: record 의 crash_marker_path 확인 crash_marker_path = rec.get("crash_marker_path") if crash_marker_path and Path(crash_marker_path).exists(): has_crash = True # crash marker: result.json 부모 디렉토리에서 찾기 if not has_crash: for p_item in expected_paths: p_str = str(p_item) # Path 객체도 안전하게 처리 if p_str.endswith("result.json") or p_str.endswith(".result.json"): parent_dir = Path(p_str).parent # '.supervisor-crash-marker.json' 또는 '.supervisor-crash-marker.json' for marker_name in ( f"{task_id}.supervisor-crash-marker.json", ".supervisor-crash-marker.json", ): if (parent_dir / marker_name).exists(): has_crash = True break if has_crash: break # crash marker: expected_result_paths 의 부모 디렉토리에서도 확인 if not has_crash: checked_dirs: set = set() for p_item in expected_paths: p_str = str(p_item) # Path 객체도 안전하게 처리 parent_dir = Path(p_str).parent if parent_dir in checked_dirs: continue checked_dirs.add(parent_dir) for marker_name in ( f"{task_id}.supervisor-crash-marker.json", ".supervisor-crash-marker.json", ): if (parent_dir / marker_name).exists(): has_crash = True break if has_crash: break # --- status_detected 판정 --- if has_done: status_detected = "done" elif has_failure_envelope: status_detected = "failure" elif has_result_or_pr: # .done 부재만으로 실패 단정 금지 status_detected = "ready_for_adjudication" else: status_detected = "pending" # detected: 산출물이 하나라도 있으면 True (status_detected != 'pending') detected = status_detected != "pending" # deliverable_ok: result.json/.done/PR deliverable 존재 deliverable_ok = has_done or has_result_or_pr # crash=True 라도 deliverable_ok=True 이면 분리 (실패 단정 금지) # needs_adjudication needs_adjudication = (status_detected == "ready_for_adjudication") or ( has_crash and deliverable_ok ) results.append( { "task_id": task_id, "status_detected": status_detected, "detected": detected, "crash": has_crash, "deliverable_ok": deliverable_ok, "needs_adjudication": needs_adjudication, "expected_result_paths": expected_paths, } ) return results def close_delegation(task_id: str, status: str, ledger_path=None) -> dict: """ledger 에 close 레코드 append. status in {'done','failure','essence_pass','merged'} 검증 (그 외 ValueError). 반환: append 된 레코드 dict. """ if status not in _VALID_CLOSE_STATUSES: raise ValueError( f"close_delegation: 유효하지 않은 status '{status}'. " f"허용값: {sorted(_VALID_CLOSE_STATUSES)}" ) lp = _resolve_ledger(ledger_path) lp.parent.mkdir(parents=True, exist_ok=True) record = { "task_id": task_id, "status": status, "record_type": "close", "ts": _now_iso(), } with open(lp, "a", encoding="utf-8") as f: f.write(json.dumps(record, ensure_ascii=False) + "\n") return record def render_pending_briefing(items: list) -> str: """check_pending_delegations 결과 리스트를 hook 주입용 짧은 텍스트로 렌더. items 가 비어있으면 '' 반환. 비어있지 않으면 pending/ready_for_adjudication 을 강조하는 멀티라인 텍스트 반환. """ if not items: return "" lines = ["=== PENDING DELEGATIONS ==="] for item in items: tid = item.get("task_id", "?") sd = item.get("status_detected", "?") na = item.get("needs_adjudication", False) crash = item.get("crash", False) flags = [] if na: flags.append("NEEDS_ADJUDICATION") if crash: flags.append("CRASH_DETECTED") flag_str = " [" + ", ".join(flags) + "]" if flags else "" lines.append(f" - {tid}: {sd}{flag_str}") lines.append("===========================") return "\n".join(lines)