#!/usr/bin/env python3 """taskctl_verify.py — taskctl 상태 전이 직전·직후의 11개 정합성 검사기. task-2459 Phase 2-C 구현. spec: memory/specs/taskctl-verify-spec.md (v1.0). 핵심 정책: - read-only 진단기. git history mutation 금지 (rebase/cherry-pick/reset 0줄). - LLM API 호출 0줄. 외부 네트워크 호출 0줄. - FAIL 시 evidence 만 저장하고 exit ≠ 0 으로 보고. 자동 복구 절대 금지. Exit code: 0 = PASS (모든 검사 PASS or N/A) 1 = FAIL (1건 이상 FAIL) 2 = WARN (FAIL 0건, WARN 1건 이상) 3 = internal error """ from __future__ import annotations import argparse import fnmatch import json import os import re import subprocess import sys import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any VERIFIER_VERSION = "1.0" QC_VERDICT_RE = re.compile( r"^\s*qc_verdict\s*[:\-]\s*(PASS|WARN|FAIL)\s*$", re.IGNORECASE | re.MULTILINE, ) # "## QC Verdict" 섹션 직후 첫 PASS/WARN/FAIL 키워드 QC_SECTION_RE = re.compile( r"##\s*QC\s+Verdict[^\n]*\n+([\s\S]{0,400})", re.IGNORECASE, ) QC_KEYWORD_RE = re.compile(r"\b(PASS|WARN|FAIL)\b") HANDOFF_REQUIRED_FIELDS = ( "handoff_id", "from_bot", "to_bot", "task_id", "timestamp", ) # --------------------------------------------------------------------------- # 유틸리티 # --------------------------------------------------------------------------- def _now_iso() -> str: return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _now_compact() -> str: return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") def _atomic_write_json(path: Path, data: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp_path = tempfile.mkstemp( dir=str(path.parent), prefix=path.name + ".", suffix=".tmp" ) try: with os.fdopen(fd, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) f.write("\n") os.replace(tmp_path, str(path)) except Exception: try: os.unlink(tmp_path) except OSError: pass raise def _git_run( args: list[str], cwd: Path, timeout: int = 30 ) -> subprocess.CompletedProcess: """git read-only 명령. mutation 명령 호출 금지.""" return subprocess.run( ["git"] + args, cwd=str(cwd), capture_output=True, text=True, timeout=timeout, ) def _file_mtime_iso(path: Path) -> str | None: try: ts = path.stat().st_mtime except OSError: return None return datetime.fromtimestamp(ts, tz=timezone.utc).strftime( "%Y-%m-%dT%H:%M:%SZ" ) # --------------------------------------------------------------------------- # 11 검사 — 각 검사는 (status, detail_dict, fail_reason | None) 반환 # --------------------------------------------------------------------------- def check_start_lock(task_id: str, workspace: Path) -> tuple[str, dict, str | None]: lock_path = workspace / ".tasks" / "locks" / f"{task_id}.lock" detail = { "lock_path": str(lock_path), "lock_mtime": _file_mtime_iso(lock_path) if lock_path.exists() else None, } if lock_path.exists(): return "PASS", detail, None return ( "FAIL", detail, f"start_lock: {lock_path} 부재 (start_task_guard 미실행 의심)", ) def check_branch_match( task_id: str, bot: str | None, cwd: Path ) -> tuple[str, dict, str | None]: res = _git_run(["rev-parse", "--abbrev-ref", "HEAD"], cwd=cwd) if res.returncode != 0: return ( "FAIL", { "current_branch": None, "expected_branch": None, "bot_hint": bot, "git_error": res.stderr.strip(), }, "branch_match: git rev-parse 실패", ) current = res.stdout.strip() if current == "HEAD": return ( "FAIL", { "current_branch": "HEAD", "expected_branch": ( f"task/{task_id}-{bot}" if bot else f"task/{task_id}-*" ), "bot_hint": bot, }, "branch_match: detached HEAD 상태", ) if bot: expected = f"task/{task_id}-{bot}" detail = { "current_branch": current, "expected_branch": expected, "bot_hint": bot, } if current == expected: return "PASS", detail, None return ( "FAIL", detail, f"branch_match: 기대 {expected}, 실제 {current}", ) # bot 미지정: prefix 매칭만 prefix = f"task/{task_id}-" detail = { "current_branch": current, "expected_branch": f"{prefix}", "bot_hint": None, } if current.startswith(prefix): return "PASS", detail, None return "FAIL", detail, f"branch_match: {current} 가 {prefix} prefix 와 불일치" def check_worktree_path( task_id: str, bot: str | None, cwd: Path ) -> tuple[str, dict, str | None]: cwd_resolved = cwd.resolve() if bot: marker = f".worktrees/{task_id}-{bot}" else: marker = f".worktrees/{task_id}-" detail = { "cwd": str(cwd_resolved), "cwd_match_pattern": marker, } cwd_str = str(cwd_resolved) if bot: # cwd 가 marker 로 끝나거나 marker 디렉토리 하위 if cwd_str.endswith(marker) or f"{marker}/" in cwd_str + "/": return "PASS", detail, None else: # prefix 검사: '.worktrees/-' 가 path 내 존재 if marker in cwd_str: return "PASS", detail, None return ( "FAIL", detail, f"worktree_path: cwd 가 {marker} worktree 안이 아님", ) def check_cancelled( task_id: str, workspace: Path, cwd: Path ) -> tuple[str, dict, str | None]: candidates = [ workspace / "memory" / "events" / f"{task_id}.cancelled", cwd / "memory" / "events" / f"{task_id}.cancelled", ] seen: set[Path] = set() found: Path | None = None for p in candidates: rp = p.resolve() if p.exists() else p if rp in seen: continue seen.add(rp) if p.exists(): found = p break detail = { "cancelled_marker_path": str(found) if found else None, "cancelled_at": _file_mtime_iso(found) if found else None, } if found is None: return "PASS", detail, None return ( "FAIL", detail, f"cancelled_check: cancel 마커 존재 — {found}", ) def check_dirty_tree(cwd: Path) -> tuple[str, dict, str | None]: res = _git_run(["status", "--porcelain"], cwd=cwd) if res.returncode != 0: return ( "FAIL", {"dirty_files": [], "git_error": res.stderr.strip()}, "dirty_tree: git status 실행 실패", ) lines = [ln for ln in res.stdout.splitlines() if ln.strip()] # porcelain 포맷: 'XY path' 또는 'XY path -> path2' (rename) files = [] for ln in lines[:50]: # 첫 3바이트는 status code + space files.append(ln[3:] if len(ln) > 3 else ln) detail = {"dirty_files": files} if not lines: return "PASS", detail, None return ( "FAIL", detail, f"dirty_tree: {len(lines)} untracked or modified file(s) detected", ) def check_changed_paths( base: str, cwd: Path ) -> tuple[str, dict, str | None]: res = _git_run(["diff", "--name-only", f"{base}..HEAD"], cwd=cwd) if res.returncode != 0: # base 가 없으면 빈 결과로 처리 (정보용 검사이므로 PASS 유지) return ( "PASS", {"changed_paths_list": [], "git_error": res.stderr.strip()}, None, ) paths = sorted({ln.strip() for ln in res.stdout.splitlines() if ln.strip()}) return "PASS", {"changed_paths_list": paths}, None def _extract_allowed_paths(task_md: str) -> list[str] | None: """task md 에서 `allowed_resources.paths` 리스트 추출 (yaml lib 미사용). Returns: - ``None`` : ``allowed_resources`` 블록 자체가 없음 (구버전 task md 호환) - ``[]`` : 블록은 있으나 ``paths`` 리스트가 비어있음 (스펙 위반 — FAIL 처리) - ``[...]``: 추출된 패턴 리스트 """ # ```yaml ... ``` 블록 안의 allowed_resources 추출 in_yaml = False in_allowed = False in_paths = False seen_allowed = False # allowed_resources 블록을 한 번이라도 봤는지 paths: list[str] = [] yaml_indent: int | None = None for raw in task_md.splitlines(): line = raw.rstrip("\n") stripped = line.strip() if stripped.startswith("```"): in_yaml = not in_yaml if not in_yaml: # 블록 종료 시 모든 상태 리셋 in_allowed = False in_paths = False yaml_indent = None continue if not in_yaml: continue # 들여쓰기 측정 leading = len(line) - len(line.lstrip(" ")) if not in_allowed: if stripped.startswith("allowed_resources:"): in_allowed = True seen_allowed = True yaml_indent = leading continue # in_allowed 안에서 if stripped == "" or stripped.startswith("#"): continue # allowed_resources 블록 종료 감지 (들여쓰기가 같거나 더 작은 키 등장) if yaml_indent is not None and leading <= yaml_indent and stripped.endswith(":"): # forbidden_paths 등 다른 키 in_allowed = False in_paths = False continue if stripped.startswith("paths:"): in_paths = True continue if in_paths: if stripped.startswith("- "): val = stripped[2:].strip() # quote 제거 if (val.startswith('"') and val.endswith('"')) or ( val.startswith("'") and val.endswith("'") ): val = val[1:-1] paths.append(val) else: # paths 리스트 종료 in_paths = False # 다음 키로 fall through if stripped.endswith(":"): # 새 키 — paths 종료, allowed_resources 안에 다른 키 pass if not seen_allowed: return None return paths def _glob_match(path: str, pattern: str) -> bool: """`**` 지원하는 glob 매칭.""" # Path.match 는 `**` 를 부분 지원하므로 fnmatch 우선 시도. # `**/x` 는 모든 깊이의 x 매치, `x/**` 는 x 하위 모두 매치. if pattern == path: return True if fnmatch.fnmatchcase(path, pattern): return True # `**` 처리 if "**" in pattern: # `a/**` => a 디렉토리 하위 전부 # `a/**/b.py` => a 와 b.py 사이 임의 디렉토리 # 정규식으로 변환: `**` -> `.*`, 단일 `*` -> `[^/]*` regex_parts: list[str] = [] i = 0 while i < len(pattern): if pattern[i : i + 2] == "**": regex_parts.append(".*") i += 2 # 뒤에 슬래시가 있으면 같이 소비 if i < len(pattern) and pattern[i] == "/": i += 1 # `.*/` 는 비어도 매치돼야 하므로 그룹화 # 마지막 `.*` 를 `(?:.*/)?` 로 교체 regex_parts[-1] = "(?:.*/)?" elif pattern[i] == "*": regex_parts.append("[^/]*") i += 1 elif pattern[i] == "?": regex_parts.append("[^/]") i += 1 elif pattern[i] in ".+()^$|{}\\": regex_parts.append(re.escape(pattern[i])) i += 1 else: regex_parts.append(pattern[i]) i += 1 regex = "^" + "".join(regex_parts) + "$" try: return re.match(regex, path) is not None except re.error: return False return False def check_scope_matrix( task_id: str, workspace: Path, changed_paths: list[str] ) -> tuple[str, dict, str | None]: task_md = workspace / "memory" / "tasks" / f"{task_id}.md" detail: dict[str, Any] = { "allowed_patterns": [], "scope_violations": [], "task_md_path": str(task_md), } if not task_md.exists(): # spec 2.7: task 파일 미존재 = FAIL (task-id 부정확 또는 메타데이터 누락) return ( "FAIL", detail, f"scope_matrix: task md 부재 — {task_md}", ) try: text = task_md.read_text(encoding="utf-8") except OSError as exc: return ( "FAIL", {**detail, "read_error": str(exc)}, f"scope_matrix: task md 읽기 실패 — {exc}", ) patterns = _extract_allowed_paths(text) # allowed_resources 블록 자체가 없는 구버전 task md → WARN (호환) if patterns is None: return ( "WARN", detail, "scope_matrix: allowed_resources 블록 부재 (구버전 task md)", ) detail["allowed_patterns"] = patterns # paths 블록은 있으나 빈 리스트 → 스펙 위반 (FAIL) if not patterns: return ( "FAIL", detail, "scope_matrix: allowed_resources.paths 빈 리스트", ) violations: list[str] = [] for p in changed_paths: if not any(_glob_match(p, pat) for pat in patterns): violations.append(p) detail["scope_violations"] = violations if not violations: return "PASS", detail, None return ( "FAIL", detail, f"scope_matrix: {len(violations)} path(s) outside allowed_resources", ) def check_handoff_chain( task_id: str, workspace: Path ) -> tuple[str, dict, str | None]: handoff_path = workspace / "memory" / "handoffs" / f"{task_id}.json" detail: dict[str, Any] = { "handoff_path": str(handoff_path) if handoff_path.exists() else None, "previous_bot": None, "handoff_schema_errors": [], } if not handoff_path.exists(): return "N/A", detail, None try: data = json.loads(handoff_path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: detail["handoff_schema_errors"] = [f"parse: {exc}"] return ( "FAIL", detail, f"handoff_chain: JSON 파싱 실패 — {exc}", ) # 필수 필드 검사 — 키 부재 + 빈 값(falsy) 모두 누락으로 처리. missing = [f for f in HANDOFF_REQUIRED_FIELDS if not data.get(f)] if missing: detail["handoff_schema_errors"] = [f"missing: {','.join(missing)}"] return ( "FAIL", detail, f"handoff_chain: 필수 필드 누락 — {missing}", ) # task_id mismatch 차단 (handoff JSON 의 task_id 가 인자 task_id 와 일치) if data["task_id"] != task_id: detail["handoff_schema_errors"] = [ f"task_id mismatch: handoff={data['task_id']} arg={task_id}" ] return ( "FAIL", detail, ( f"handoff_chain: task_id mismatch — " f"handoff={data['task_id']} arg={task_id}" ), ) # from_bot(previous_bot) 추적 가능성 — 빈 문자열 / null / 누락이면 차단. previous_bot = data.get("from_bot") if not previous_bot: detail["handoff_schema_errors"] = ["from_bot empty"] return ( "FAIL", detail, "handoff_chain: from_bot(previous_bot) 추적 불가", ) detail["previous_bot"] = previous_bot return "PASS", detail, None def check_mixed_commit( task_id: str, workspace: Path, cwd: Path ) -> tuple[str, dict, str | None]: detector = workspace / "scripts" / "mixed_commit_detector.py" if not detector.exists(): # spec 2.9: 안전 측 FAIL — 안전 검사 자체 부재면 차단. return ( "FAIL", { "mixed_commit_detector_exit": None, "mixed_commit_evidence_path": None, "note": "mixed_commit_detector.py 부재", }, "mixed_commit: mixed_commit_detector.py 부재 — 안전 검증 불가", ) cmd = [ sys.executable, str(detector), task_id, "--json", "--quiet", "--workspace", str(workspace), # git 조회는 task worktree(cwd) 에서 수행해야 함 — 명시적으로 전달. "--git-dir", str(cwd), ] try: res = subprocess.run( cmd, cwd=str(cwd), capture_output=True, text=True, timeout=60, ) except subprocess.TimeoutExpired: # spec 2.9: timeout / 실행 실패도 안전 측 FAIL. return ( "FAIL", { "mixed_commit_detector_exit": None, "mixed_commit_evidence_path": None, "error": "timeout", }, "mixed_commit: detector timeout — 안전 측 FAIL", ) except OSError as exc: return ( "FAIL", { "mixed_commit_detector_exit": None, "mixed_commit_evidence_path": None, "error": str(exc), }, f"mixed_commit: detector 실행 실패 — {exc}", ) detail: dict[str, Any] = { "mixed_commit_detector_exit": res.returncode, "mixed_commit_evidence_path": None, "stdout_tail": res.stdout[-500:] if res.stdout else "", } # detector 는 dry-run(--json) 이므로 evidence 경로 저장 안 함 if res.returncode == 0: return "PASS", detail, None if res.returncode == 1: return "FAIL", detail, "mixed_commit: detector 가 alien commit 감지" # exit 2 (internal_error) 또는 그 외 → 안전 측 FAIL. return ( "FAIL", detail, ( f"mixed_commit: detector internal_error (exit {res.returncode}) — " "안전 측 FAIL" ), ) def check_qc_report_guard( task_id: str, workspace: Path ) -> tuple[str, dict, str | None]: report_path = workspace / "memory" / "reports" / f"{task_id}.md" detail: dict[str, Any] = { "qc_report_path": str(report_path) if report_path.exists() else None, "qc_verdict_value": None, } if not report_path.exists(): return "N/A", detail, None try: text = report_path.read_text(encoding="utf-8") except OSError as exc: return ( "WARN", {**detail, "read_error": str(exc)}, f"qc_report_guard: 읽기 실패 — {exc}", ) # 첫 50줄만 검사 head = "\n".join(text.splitlines()[:50]) verdict: str | None = None m = QC_VERDICT_RE.search(head) if m: verdict = m.group(1).upper() if verdict is None: sec = QC_SECTION_RE.search(head) if sec: kw = QC_KEYWORD_RE.search(sec.group(1)) if kw: verdict = kw.group(1).upper() detail["qc_verdict_value"] = verdict if verdict is None: return ( "WARN", detail, "qc_report_guard: 보고서는 존재하지만 qc_verdict 라인 파싱 실패", ) if verdict in ("PASS", "WARN"): return "PASS", detail, None if verdict == "FAIL": return "FAIL", detail, "qc_report_guard: qc_verdict=FAIL" return ( "WARN", detail, f"qc_report_guard: 알 수 없는 verdict {verdict}", ) def check_guard_sh(workspace: Path, cwd: Path) -> tuple[str, dict, str | None]: guard = workspace / "scripts" / "guard.sh" detail: dict[str, Any] = { "guard_sh_exit": None, "guard_sh_stdout_tail": "", "guard_sh_stderr_tail": "", } if not guard.exists(): return "N/A", detail, None try: res = subprocess.run( ["bash", str(guard)], cwd=str(cwd), capture_output=True, text=True, timeout=60, ) except subprocess.TimeoutExpired: return ( "FAIL", {**detail, "error": "timeout"}, "guard_sh: timeout 60s 초과", ) except OSError as exc: return ( "FAIL", {**detail, "error": str(exc)}, f"guard_sh: 실행 실패 — {exc}", ) detail["guard_sh_exit"] = res.returncode detail["guard_sh_stdout_tail"] = "\n".join( res.stdout.splitlines()[-20:] ) detail["guard_sh_stderr_tail"] = "\n".join( res.stderr.splitlines()[-20:] ) if res.returncode == 0: return "PASS", detail, None return ( "FAIL", detail, f"guard_sh: exit {res.returncode}", ) # --------------------------------------------------------------------------- # 종합 / 출력 # --------------------------------------------------------------------------- def _overall(results: dict[str, str]) -> tuple[str, int]: if any(v == "FAIL" for v in results.values()): return "FAIL", 1 if any(v == "WARN" for v in results.values()): return "WARN", 2 return "PASS", 0 def run_all_checks( task_id: str, bot: str | None, workspace: Path, cwd: Path ) -> dict: results: dict[str, str] = {} details: dict[str, Any] = {} fail_reasons: list[str] = [] def _record(name: str, status: str, detail: dict, reason: str | None) -> None: results[name] = status for k, v in detail.items(): details[k] = v # spec 3.4: fail_reasons 는 FAIL 사유만 누적 (WARN 은 별도 채널). if reason and status == "FAIL": fail_reasons.append(reason) s, d, r = check_start_lock(task_id, workspace) _record("start_lock", s, d, r) s, d, r = check_branch_match(task_id, bot, cwd) _record("branch_match", s, d, r) s, d, r = check_worktree_path(task_id, bot, cwd) _record("worktree_path", s, d, r) s, d, r = check_cancelled(task_id, workspace, cwd) _record("cancelled_check", s, d, r) s, d, r = check_dirty_tree(cwd) _record("dirty_tree", s, d, r) s, d, r = check_changed_paths("origin/main", cwd) _record("changed_paths", s, d, r) changed = details.get("changed_paths_list", []) s, d, r = check_scope_matrix(task_id, workspace, changed) _record("scope_matrix", s, d, r) s, d, r = check_handoff_chain(task_id, workspace) _record("handoff_chain", s, d, r) s, d, r = check_mixed_commit(task_id, workspace, cwd) _record("mixed_commit", s, d, r) s, d, r = check_qc_report_guard(task_id, workspace) _record("qc_report_guard", s, d, r) s, d, r = check_guard_sh(workspace, cwd) _record("guard_sh", s, d, r) overall, exit_code = _overall(results) # 호환성: details.fail_reasons 유지 (구버전 소비자용) + top-level fail_reasons 신규. details["fail_reasons"] = fail_reasons payload = { "task_id": task_id, "verified_at": _now_iso(), "verifier_version": VERIFIER_VERSION, "cwd": str(cwd.resolve()), "bot_hint": bot, "results": results, "details": details, # spec 3.x: top-level fail_reasons 필수 — 모든 FAIL 사유 리스트. "fail_reasons": list(fail_reasons), "overall": overall, "exit_code": exit_code, } return payload def write_evidence_file(task_id: str, workspace: Path, payload: dict) -> Path: ts = _now_compact() path = workspace / ".tasks" / "evidence" / task_id / f"verify-{ts}.json" _atomic_write_json(path, payload) return path # --------------------------------------------------------------------------- # main # --------------------------------------------------------------------------- def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Run 11 task verification checks and emit evidence JSON.", ) parser.add_argument("task_id", help="검증 대상 task id (예: task-2459)") parser.add_argument("--bot", default=None, help="bot 이름 (branch/worktree 정확도 향상)") parser.add_argument( "--workspace", default=os.environ.get("WORKSPACE_ROOT") or os.getcwd(), help="워크스페이스 루트 (기본 cwd / WORKSPACE_ROOT)", ) parser.add_argument( "--json", action="store_true", dest="json_only", help="dry-run: stdout 에 전체 evidence JSON, 파일 저장 안 함", ) parser.add_argument( "--quiet", action="store_true", help="정상 stdout 메시지 억제" ) args = parser.parse_args(argv) workspace = Path(args.workspace).resolve() cwd = Path(os.getcwd()).resolve() if not workspace.exists(): print( f"[taskctl-verify] workspace not found: {workspace}", file=sys.stderr, ) return 3 try: payload = run_all_checks(args.task_id, args.bot, workspace, cwd) except subprocess.TimeoutExpired as exc: print(f"[taskctl-verify] subprocess timeout: {exc}", file=sys.stderr) return 3 except Exception as exc: # noqa: BLE001 print(f"[taskctl-verify] internal error: {exc}", file=sys.stderr) return 3 if args.json_only: # dry-run: stdout 에 full payload, 파일 저장 안 함 if not args.quiet: print(json.dumps(payload, ensure_ascii=False)) return int(payload["exit_code"]) try: evidence_path = write_evidence_file(args.task_id, workspace, payload) except OSError as exc: print( f"[taskctl-verify] failed to write evidence: {exc}", file=sys.stderr, ) # 부분 보고 if not args.quiet: print(json.dumps(payload, ensure_ascii=False)) return 3 if not args.quiet: summary = { "task_id": payload["task_id"], "overall": payload["overall"], "exit_code": payload["exit_code"], "results": payload["results"], "evidence": str(evidence_path.relative_to(workspace)), } print(json.dumps(summary, ensure_ascii=False)) return int(payload["exit_code"]) if __name__ == "__main__": sys.exit(main())