"""불변식 감사 — 에러 로그에 흔적을 안 남긴 채 잘못 끝난 런을 아티팩트로 잡는다. 리드의 자기 보고가 아니라 run-manifest / final-report 가 남긴 사실만 읽는다. 검증 라운드를 빼먹은 리드는 그 라운드를 돌렸는지에 대한 믿을 만한 증인이 아니다. 읽기 전용 — 어떤 타겟의 `.okstra/` 도 쓰거나 옮기거나 지우지 않는다. """ from __future__ import annotations import argparse import datetime as dt import json import re import sys from pathlib import Path from okstra_ctl.error_zip import run_dirs from okstra_ctl.final_report_paths import DATA_JSON_SUFFIX, final_report_data_path from okstra_ctl.paths import okstra_home, resolve_under_root from okstra_ctl.reconcile import NON_TERMINAL_RECENT_STATUSES from okstra_ctl.task_target import project_rel from okstra_ctl.workflow import PHASE_SEQUENCE from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object # 불변식 이름은 여기서만 정의한다. `INVARIANTS` 와 방출부가 같은 상수를 봐야, # 이름을 하나 더할 때 한쪽만 고치고도 스위트가 green 인 상태가 생기지 않는다 — # 그 틈이 벌어지면 소비자의 분류 라우팅이 한 종류를 조용히 흘린다. INVARIANT_VERIFICATION_ROUNDS_RAN = "verification-rounds-ran" INVARIANT_RUN_PRODUCED_ITS_REPORT = "run-produced-its-report" INVARIANT_ROSTER_WAS_FULFILLED = "roster-was-fulfilled" INVARIANT_APPROVAL_NOT_FORGOTTEN = "approval-not-forgotten" INVARIANTS = ( INVARIANT_VERIFICATION_ROUNDS_RAN, INVARIANT_RUN_PRODUCED_ITS_REPORT, INVARIANT_ROSTER_WAS_FULFILLED, INVARIANT_APPROVAL_NOT_FORGOTTEN, ) # 게이트를 "통과"로 선언하는 값. 정본 enum 은 # `schemas/final-report-v2.0.schema.json` 의 PlanBodyVerification.gateResult # (passed / passed-with-dissent / blocked-by-disagreement / aborted-non-result) # 이고 `validators/validate-run.py` 도 같은 두 값을 통과로 취급한다. 실측 run-index # 에서 통과 게이트 23건 중 19건이 passed-with-dissent 라 passed 만 보면 대부분을 # 놓친다. _PASSING_GATES = frozenset({"passed", "passed-with-dissent"}) # 승인 대기가 이 일수를 넘으면 잊힌 것으로 본다. 사람이 일부러 기다리는 상태와 # 잊은 상태를 시각만으로 가를 수는 없어, 창을 넉넉히 잡는다. APPROVAL_STALE_DAYS = 14 # 승인 게이트는 implementation 진입 직전 한 번만 의미를 갖는다 — # `validators/validate-run.py` 가 `current_phase == "implementation"` 인 run 이 # 검증을 통과할 때만 이 플래그를 내린다. 이 phase 부터의 완료 기록이 하나라도 # 있으면 그 태스크는 게이트를 이미 지났고, 남은 플래그는 못 내린 잔재다. _PHASES_AT_OR_PAST_APPROVAL_GATE = frozenset( PHASE_SEQUENCE[PHASE_SEQUENCE.index("implementation"):]) _MANIFEST_GLOB = "run-manifest-*.json" # 워커 결과는 마크다운이다. 같은 디렉터리의 `*.json` 은 결과가 아니라 에러 # 사이드카(`codex-worker-errors--001.json`)라, 그것을 결과로 세면 # 부르다 터진 워커가 정상 참여로 둔갑한다. 확장자를 허용목록으로 못박는다. _WORKER_RESULT_GLOB = "*.md" _SEQ_RESULT_RE = re.compile(r"-(\d{3,})\.md$") # team-state 가 워커의 종결을 기록할 때 쓰는 값. 정본 enum 은 # `schemas/convergence-round-results-v1.0.schema.json` 의 workers[].status 이고, # 실측 team-state 139건의 워커 항목도 이 네 값만 쓴다(completed 385 / not-run 46 # / error 14 / timeout 3). 허용목록으로 둬야 앞으로 생길 비종결 상태(`running` # 류)가 결과 부재의 면죄부로 슬쩍 통과하지 않는다. _RECORDED_WORKER_STATUSES = frozenset( {"completed", "not-run", "error", "timeout"}) def _load_json(path: Path) -> dict: try: loaded = load_owned_object(path, artifact="run audit input") except JsonBoundaryError: return {} return loaded if isinstance(loaded, dict) else {} def _latest_manifest(run_dir: Path) -> tuple[dict, Path | None]: """정본 위치는 `/manifests/run-manifest--.json` (`paths.py` 의 `run_manifests / f"run-manifest{suffixes['manifests']}.json"`). run_dir 바로 아래는 이식된 번들·비정형 트리용 폴백. 한 run_dir 은 한 task-type 이라 파일명 정렬이 곧 seq 정렬이다.""" files = (sorted((run_dir / "manifests").glob(_MANIFEST_GLOB)) or sorted(run_dir.glob(_MANIFEST_GLOB))) if not files: return {}, None return _load_json(files[-1]), files[-1] def _resolve_report(run_dir: Path, project_root: str, expected: str) -> Path | None: """`expectedReportRecordPath` 를 실제 파일로 푼다. 루트 밖 이탈이면 `None`. 이 값은 project_root 기준 상대경로다 — `render.py` 가 `FINAL_REPORT_RECORD_RELATIVE_PATH` 에서 채우고 그 값은 `paths.py` 의 `_rel(project_root, final_report)` 이며, 다른 소비자도 `dispatch_state.resolve_required_path(project_root, ...)` 로 푼다. 글로벌 인덱스에서 온 신뢰 불가 값이라 `resolve_under_root` 로 루트 밖 이탈을 막는다. 이탈이면 `run_dir` 기준으로 폴백하지 않고 판정을 포기한다 — `expected` 가 절대경로면 `run_dir / expected` 가 좌변을 통째로 버려 가드가 무효가 되고, 그렇게 얻은 루트 밖 경로와 그 파일 내용이 감사 결과에 실려 나간다. 루트 안이면서 그 자리에 없을 때만 `run_dir` 기준으로 한 번 더 본다(이식된 번들 방어). 둘 다 없으면 정본 기준인 루트 해석 결과를 돌려준다.""" from_root = resolve_under_root(project_root, expected) if from_root is None: return None if from_root.is_file(): return from_root from_run = run_dir / expected return from_run if from_run.is_file() else from_root def _report_data_path(report_path: Path) -> Path: """구조화 본문이 있는 파일. 정본 `expectedReportRecordPath` 는 마크다운이고 본문은 `.data.json` 형제다. 이미 data.json 을 가리키면 그대로 쓴다 — `final_report_data_path` 는 `.md` 가 아닌 이름에 `with_suffix` 를 걸어 `x.data.json` 을 `x.data.data.json` 으로 바꿔 놓는다.""" if report_path.name.endswith(DATA_JSON_SUFFIX): return report_path return final_report_data_path(report_path) def _run_clock(manifest: dict, manifest_path: Path) -> dt.datetime | None: """이 run 이 스스로 적은 시각. 파일 mtime 이 아니라 manifest 의 `createdAt` 을 먼저 본다 — 번들을 복사하거나 체크아웃하면 mtime 은 현재로 리셋되어 오래된 관측이 전부 조용히 신선해진다. 실측 manifest 139건 전부 tz-aware ISO-8601 `createdAt` 을 갖고 있고 mtime 과의 차는 모두 하루 미만이다. 소비자가 둘이다 — 승인 경과일 계산과, 위반이 언제 관측됐는지(`observedAt`). 후자가 없으면 감사 후보는 시계가 없어 중복 판정에서 에러 로그 후보와 같은 규칙을 쓸 수 없다.""" try: parsed = dt.datetime.fromisoformat( str(manifest.get("createdAt", "")).replace("Z", "+00:00")) except ValueError: parsed = None if parsed is not None: return parsed if parsed.tzinfo else parsed.replace(tzinfo=dt.timezone.utc) try: return dt.datetime.fromtimestamp( manifest_path.stat().st_mtime, dt.timezone.utc) except OSError: return None def _violation(*, invariant, manifest, manifest_path, project_root, detail, source) -> dict: observed = _run_clock(manifest, manifest_path) return { "invariant": invariant, "taskKey": str(manifest.get("taskKey", "")), "taskType": str(manifest.get("taskType", "")), "projectRoot": project_root, "detail": detail, "source": source, # 위반이 관측된 시각. 감사 후보의 중복 판정이 이 값을 `lastSeen` 으로 # 쓴다 — 없으면 "이미 보고한 발생인가"를 물을 수 없어, 감사 경로만 # 자기만의 특수 분기를 갖게 된다. "observedAt": observed.isoformat() if observed is not None else "", } def _as_int(value: object) -> int | None: """감사는 스키마를 지키지 못한 런을 잡으려고 도는데, 그 런의 깨진 필드 하나에 int() 가 터지면 나머지 전부를 못 본다. 못 읽는 값은 건너뛴다.""" try: return int(value or 0) except (TypeError, ValueError): return None def _find_round_counts(node) -> list[tuple[int, object, str]]: """리포트 어디에 있든 roundCount/gateResult 쌍을 찾는다. 리포트 종류마다 블록 이름이 달라 경로를 하드코딩하지 않는다. 판정용 정규화 값과 **파일에 적힌 원값**을 함께 돌려준다 — 위반 본문은 원값을 인용해야 한다. `"roundCount": null` 인 파일을 두고 본문이 `roundCount=0` 이라고 쓰면 근거를 잘못 인용하는 것이다.""" out: list[tuple[int, object, str]] = [] if isinstance(node, dict): if "roundCount" in node and "gateResult" in node: raw = node.get("roundCount") count = _as_int(raw) if count is not None: out.append((count, raw, str(node.get("gateResult", "")))) for value in node.values(): out.extend(_find_round_counts(value)) elif isinstance(node, list): for value in node: out.extend(_find_round_counts(value)) return out def _never_ran(row: dict) -> bool: """아직 돌지 않은(또는 도는 중인) run 은 리포트를 내놓을 차례가 아니다. 상태 목록은 `reconcile.NON_TERMINAL_RECENT_STATUSES` 가 정본 — 여기서 다시 적으면 종료 상태의 정의가 두 벌이 된다.""" return str(row.get("status", "")) in NON_TERMINAL_RECENT_STATUSES def _check_report(run_dir: Path, manifest: dict, manifest_path: Path, project_root: str, row: dict) -> list[dict]: expected = str(manifest.get("expectedReportRecordPath", "")) if not expected: return [] report_path = _resolve_report(run_dir, project_root, expected) if report_path is None: return [] if not report_path.is_file(): if _never_ran(row): return [] # `source` 는 없는 파일이 아니라 그 주장을 만든 manifest 를 가리킨다 — # 위반 본문을 읽는 사람이 열어볼 수 있는 파일이어야 한다. return [_violation( invariant=INVARIANT_RUN_PRODUCED_ITS_REPORT, manifest=manifest, manifest_path=manifest_path, project_root=project_root, detail=(f"expected report " f"{project_rel(report_path, Path(project_root))} is absent"), source=str(manifest_path), )] data_path = _report_data_path(report_path) if not data_path.is_file(): return [] report = _load_json(data_path) out = [] for round_count, raw_count, gate_result in _find_round_counts(report): if round_count == 0 and gate_result.lower() in _PASSING_GATES: out.append(_violation( invariant=INVARIANT_VERIFICATION_ROUNDS_RAN, manifest=manifest, manifest_path=manifest_path, project_root=project_root, detail=f"roundCount={raw_count!r} with gateResult={gate_result!r}", source=str(data_path), )) return out def _names_a_result_for(worker: str, names) -> bool: """결과 파일명 하나가 이 워커의 것인가. `recommendedWorkers` 는 맨 이름(`claude`, `report-writer`)이고 파일은 `claude-worker--001.md` 라 둘은 같은 값이 아니라 접두 관계다. 구분자 `-` 까지 붙여 비교해야 이름 하나가 다른 이름의 접두일 때 서로를 삼키지 않는다.""" return any(str(name).startswith(f"{worker}-") for name in names) def _current_seq_results(results_dir: Path) -> list[str]: """가장 최신 seq 의 결과 파일명만 남긴다. 한 worker-results 디렉터리에 여러 seq 가 공존한다(`paths.py` 가 결과 디렉터리에 manifest 와 **별도의** seq 를 매긴다 — 실측 139건 중 39건이 2개 이상). 전 seq 를 세면 seq 001 에 참여하고 002 에서 사라진 워커가 충족으로 둔갑한다. 같은 함정의 선례가 `context_cost._current_seq_worker_results` 다. 기준은 manifest 의 seq 가 아니라 **결과 파일에 실제로 찍힌 최대 seq** 다. 실측 129건 중 25건에서 manifest seq 가 결과 최대 seq 보다 앞서 있어(예: manifest 003 / 결과 002) manifest 기준으로 자르면 그 25건은 결과 집합이 통째로 비어 로스터 전원이 미보고로 뒤집힌다.""" names = [p.name for p in results_dir.glob(_WORKER_RESULT_GLOB)] by_seq: dict[str, list[str]] = {} for name in names: match = _SEQ_RESULT_RE.search(name) if match: by_seq.setdefault(match.group(1), []).append(name) # seq 를 못 읽는 배치는 전부 살린다. 여기서 빈 목록을 돌려주면 읽지 못한 # 이름 하나 때문에 없는 위반을 만들어 낸다. return by_seq[max(by_seq)] if by_seq else names def _explained_workers(manifest: dict, project_root: str) -> list[str] | None: """team-state 가 종결 상태를 기록해 둔 워커의 결과 파일명. team-state 에 **닿지 못하면** 빈 목록이 아니라 `None` 을 돌려준다. 둘은 다른 사실이다 — 빈 목록은 "확인했고 아무 기록도 없었다"이고 `None` 은 "확인조차 못 했다"이다. 위반 본문이 그 둘을 뭉개면 감사가 하지 않은 확인을 했다고 주장하게 된다. 로스터는 디스패치 대상의 상위집합이라(`dispatch_core._select_workers` 가 `recommended` 를 dispatcher 지원 목록으로 걸러 낸다) 부르지 않은 워커가 정상적으로 생기고, 부르지 않았거나 실패한 워커는 사유와 함께 team-state 에 남는다. 실측 로스터 미보고 9건 전부가 `not-run`(render-only 모드·사용자가 기다리지 말라고 지시) / `error`(Gemini 쿼터 소진, Codex 빈 stdout) / `timeout` 으로 기록돼 있었다. 이 모듈이 겨냥한 것은 "흔적을 안 남긴 채 잘못 끝난" 런이라, 사유가 적힌 부재는 정확히 그 반대다. 워커 식별은 항목의 `agent` 가 아니라 `resultPath` 파일명으로 한다 — report-writer 항목의 `agent` 는 실행 제공자인 `claude` 라 로스터 이름과 다르다.""" state_rel = str(manifest.get("teamStatePath", "") or "") if not state_rel: return None state_path = resolve_under_root(project_root, state_rel) if state_path is None or not state_path.is_file(): return None entries = _load_json(state_path).get("workers") if not isinstance(entries, list): return None return [ Path(str(entry.get("resultPath", ""))).name for entry in entries if isinstance(entry, dict) and str(entry.get("status", "")) in _RECORDED_WORKER_STATUSES ] def _check_roster(manifest: dict, manifest_path: Path, project_root: str, row: dict) -> list[dict]: """로스터에 오른 워커가 결과도 안 남기고 사유도 안 남긴 run 을 잡는다. `workerResultsDirectoryPath` 는 run_dir 이 아니라 **project_root** 기준 상대경로다(`render.py` 가 `WORKER_RESULTS_RELATIVE_PATH` 에서 채운다). 실측 manifest 139건 전부 그랬고, run_dir 기준으로 풀면 한 건도 디렉터리에 닿지 않아 감사가 통째로 빈손이 된다. `expectedReportRecordPath` 와 같은 이유로 `resolve_under_root` 를 거쳐 루트 밖 이탈도 함께 막는다.""" raw_workers = manifest.get("recommendedWorkers") # 문자열이 들어오면 문자 단위로 순회해 `no result from c, l, a, u, d, e` 가 # 나온다. 같은 방어의 선례는 `backfill.py` 의 `raw_workers` 처리. if not isinstance(raw_workers, list): return [] recommended = [str(w) for w in raw_workers] results_rel = str(manifest.get("workerResultsDirectoryPath", "")) if not recommended or not results_rel or _never_ran(row): return [] results_dir = resolve_under_root(project_root, results_rel) if results_dir is None or not results_dir.is_dir(): return [] produced = _current_seq_results(results_dir) explained = _explained_workers(manifest, project_root) missing = [w for w in recommended if not _names_a_result_for(w, produced) and not _names_a_result_for(w, explained or [])] if not missing: return [] # 확인하지 않은 사실을 위반 본문이 주장하면 안 된다. team-state 에 닿지 # 못한 run 에까지 "team-state 가 이들의 결과를 기록하지 않았다"고 쓰면, # 그 문장을 읽는 사람은 감사가 하지 않은 대조를 했다고 믿는다. checked = (", and team-state records no outcome for them" if explained is not None else " (no readable team-state to check for a recorded outcome)") return [_violation( invariant=INVARIANT_ROSTER_WAS_FULFILLED, manifest=manifest, manifest_path=manifest_path, project_root=project_root, detail=(f"no result from {', '.join(missing)} in " f"{project_rel(results_dir, Path(project_root))}{checked}"), source=str(manifest_path), )] def _approval_clock(manifest: dict, manifest_path: Path, now: dt.datetime) -> dt.datetime: """승인 대기가 시작된 시각. 시계를 못 읽으면 `now` 로 떨어져 경과일이 0 이 되고, 읽지 못한 것이 잊힌 승인으로 둔갑하지 않는다.""" return _run_clock(manifest, manifest_path) or now def _approval_is_still_open(manifest: dict, project_root: str) -> bool: """승인 게이트가 지금도 열려 있는가. run-manifest 의 `workflowSnapshot.awaitingApproval` 은 렌더 시점에 얼어붙은 값이다. 그 플래그를 내리는 것은 검증을 통과한 implementation run 이 자기 manifest 에 쓸 때 한 번뿐이라(`validators/validate-run.py` 의 `validation_status == "passed" and current_phase == "implementation"`), 게이트를 세운 implementation-planning manifest 는 승인이 떨어지고 릴리스까지 끝난 뒤에도 영원히 `true` 로 남는다. 현재 값은 task-manifest 의 `workflow.awaitingApproval` 에 있다(같은 파일이 거기에 쓴다). 실측 스냅샷 30건 중 13건이 이미 내려간 게이트였고, 그중에는 implementation → final-verification 까지 지나간 태스크도 있다 — 그대로 두면 이미 배포된 일감에 "N일째 승인 대기" 이슈를 연다. 대조할 현재값에 닿지 못하면 스냅샷을 그대로 믿는다. 없는 근거로 위반을 지우면 이 불변식이 조용히 꺼진다.""" task_rel = str(manifest.get("taskManifestPath", "") or "") if not task_rel: return True task_path = resolve_under_root(project_root, task_rel) if task_path is None or not task_path.is_file(): return True workflow = _load_json(task_path).get("workflow") if not isinstance(workflow, dict): return True if _passed_the_approval_gate(workflow): return False if "awaitingApproval" not in workflow: return True return bool(workflow.get("awaitingApproval")) def _passed_the_approval_gate(workflow: dict) -> bool: """이 태스크가 승인 게이트를 이미 지났는가. 플래그가 열려 있다는 사실만으로는 잊힌 승인이라고 말할 수 없다. 게이트는 implementation 진입 직전에만 의미를 갖는데, 그 뒤로 나아간 태스크에서도 플래그가 열린 채 남는다 — 실측 15건 중 6건이 implementation 을, 3건이 final-verification 까지 끝낸 태스크였다. 그런 태스크에 "N일째 승인 대기"를 붙이면 이미 배포된 일감을 할 일 목록에 올리는 것이고, 이 불변식이 사람에게 보이는 감사 리포트 17행 중 15행을 그 잡음으로 채운다. 완료 기록을 보는 것이지 `currentPhase` 를 보는 것이 아니다 — `currentPhase` 는 다음에 할 일을 가리키므로 게이트 앞에서 멈춘 태스크와 게이트를 지나 되돌아온 태스크를 구분하지 못한다. `phaseStates` 는 phase 이름 → 상태 **문자열** 이다(`render._derive_phase_states` 와 `implementation_outcome._promote_workflow` 가 둘 다 문자열을 넣는다). `completed-awaiting-approval` 은 게이트 그 자체이므로 통과로 세지 않는다.""" states = workflow.get("phaseStates") if not isinstance(states, dict): return False return any( str(states.get(phase, "")) == "completed" for phase in _PHASES_AT_OR_PAST_APPROVAL_GATE ) def _check_approval(manifest: dict, manifest_path: Path, project_root: str, now: dt.datetime) -> list[dict]: """`_never_ran` 을 걸지 않는다 — 실측 승인 대기 30건 중 22건이 `prepared` 다. 승인 대기는 본래 멈춰 선 상태라, 미실행 필터를 얹으면 이 불변식은 잡으려던 것의 대부분을 못 본다.""" snapshot = manifest.get("workflowSnapshot") if not isinstance(snapshot, dict) or not snapshot.get("awaitingApproval"): return [] if not _approval_is_still_open(manifest, project_root): return [] since = _approval_clock(manifest, manifest_path, now) waited = (now - since).days if waited < APPROVAL_STALE_DAYS: return [] return [_violation( invariant=INVARIANT_APPROVAL_NOT_FORGOTTEN, manifest=manifest, manifest_path=manifest_path, project_root=project_root, detail=(f"awaiting approval for {waited} days " f"since {since.date().isoformat()}"), source=str(manifest_path), )] def audit_runs(home: Path, *, now: dt.datetime) -> list[dict]: violations: list[dict] = [] for project_root, run_dir, row in run_dirs(home): if not run_dir.exists(): continue manifest, manifest_path = _latest_manifest(run_dir) if manifest_path is None: continue violations.extend( _check_report(run_dir, manifest, manifest_path, project_root, row)) violations.extend( _check_roster(manifest, manifest_path, project_root, row)) violations.extend( _check_approval(manifest, manifest_path, project_root, now)) return violations def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="okstra run-audit", description="런 아티팩트에서 불변식 위반을 찾는다 (읽기 전용).", ) parser.parse_args(argv) violations = audit_runs(okstra_home(), now=dt.datetime.now(dt.timezone.utc)) by_invariant: dict[str, int] = {} for v in violations: by_invariant[v["invariant"]] = by_invariant.get(v["invariant"], 0) + 1 print(json.dumps({ "violationCount": len(violations), "byInvariant": by_invariant, "violations": violations, }, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))