"""Read-side cross-task roll-up for a task-group (or the whole project). 단일 task 집계기(time-report / error-report / recap)는 task-root 하나만 받는다. 이 모듈은 그 위에서 catalog 를 task-group 단위로 fan-out 해 task 별 run 수·소요 시간·에러 수·report 경로를 모으고, group 차원의 합계/분포를 deterministic 하게 roll-up 한다. report 본문 종합(자연어 요약)은 LLM(skill)에 남긴다 — 여기서는 report 경로만 노출한다. time-report 와 동일하게 raw ms 를 출력하고 HH:MM:SS 포맷·Markdown 표 렌더는 호출자(skill)에 위임한다. """ from __future__ import annotations import argparse import json import sys from pathlib import Path from okstra_ctl import next_phase from okstra_ctl.error_log_core import glob_error_logs, parse_records from okstra_ctl.time_report import aggregate_time, load_runs from okstra_ctl.json_boundary import JsonBoundaryError, load_owned_object from okstra_ctl.fixed_text import line from okstra_project import ( ResolverError, list_project_tasks, resolve_project_root, ) _CRITIC_COUNTERS = ( "gapsProposed", "gapsMerged", "gapsRejected", "gapsUnverified", ) def _collect_critic(task_root: Path) -> dict: """task 의 모든 run 에서 커버리지 크리틱 결과를 합산. 크리틱은 opt-in 이라 대부분의 run 에 `config.critic` 이 없다. 그런 run 은 `runsWithCritic` 에 세지 않는다 — "쓰지 않았다" 와 "썼는데 0건" 은 정책 판단이 갈리는 서로 다른 사실이다. 읽기 전용이므로 깨진 state 파일은 건너뛴다. """ totals = {"runsWithCritic": 0} | {name: 0 for name in _CRITIC_COUNTERS} for state_path in sorted(task_root.glob("runs/*/state/convergence-*.json")): try: payload = load_owned_object(state_path, artifact="convergence state") except JsonBoundaryError: continue critic = (payload.get("config") or {}).get("critic") if not isinstance(critic, dict): continue totals["runsWithCritic"] += 1 for name in _CRITIC_COUNTERS: value = critic.get(name) if isinstance(value, int) and not isinstance(value, bool): totals[name] += value return totals def _collect_task(entry: dict, project_root: Path) -> dict: """list_project_tasks entry → roll-up 한 줄. catalog 상태 필드 + time/error 측정. runCount 는 timeline 의 전체 run 수(usable durationMs 없는 run 도 포함)이고 cpuSumMs/wallClockMs 는 durationMs 가 있는 run 만 반영하므로 0 일 수 있다. """ task_root = Path(entry["_resolvedTaskRoot"]) runs, _ = load_runs(task_root) time_agg = aggregate_time(runs, project_root) records, _ = parse_records(glob_error_logs(task_root)) critic = _collect_critic(task_root) return { "criticRuns": critic["runsWithCritic"], "criticGapsProposed": critic["gapsProposed"], "criticGapsMerged": critic["gapsMerged"], "criticGapsRejected": critic["gapsRejected"], "criticGapsUnverified": critic["gapsUnverified"], "taskKey": entry.get("taskKey", ""), "taskGroup": entry.get("taskGroup", ""), "taskId": entry.get("taskId", ""), "taskType": entry.get("taskType", ""), "workCategory": entry.get("workCategory", "") or "unknown", "workStatus": entry.get("workStatus", "") or "unknown", "currentPhase": entry.get("currentPhase", ""), "currentPhaseState": entry.get("currentPhaseState", ""), "nextRecommendedPhase": next_phase.promote( entry.get("nextRecommendedPhase") ), "latestRunStatus": entry.get("latestRunStatus", ""), "updatedAt": entry.get("updatedAt", ""), "reportPath": entry.get("latestReportRecordPath", ""), "runCount": len(runs), "cpuSumMs": time_agg["grandTotal"]["cpuSumMs"], "wallClockMs": sum(r["wallClockMs"] for r in time_agg["perRunWallClock"]), "errorCount": len(records), } def _tally(tasks: list[dict], key: str) -> dict: """task 들의 한 필드를 값별 개수로. 빈 값은 'unknown' 으로 접는다.""" counts: dict[str, int] = {} for task in tasks: bucket = task.get(key) or "unknown" counts[bucket] = counts.get(bucket, 0) + 1 return dict(sorted(counts.items())) def _critic_totals(tasks: list[dict]) -> dict: return { "runsWithCritic": sum(t["criticRuns"] for t in tasks), "gapsProposed": sum(t["criticGapsProposed"] for t in tasks), "gapsMerged": sum(t["criticGapsMerged"] for t in tasks), "gapsRejected": sum(t["criticGapsRejected"] for t in tasks), "gapsUnverified": sum(t["criticGapsUnverified"] for t in tasks), } def build_rollup(project_root: Path, task_group: str | None) -> dict: entries = list_project_tasks(project_root, task_group=task_group or None) tasks = [_collect_task(e, project_root) for e in entries] totals = { "critic": _critic_totals(tasks), "runs": sum(t["runCount"] for t in tasks), "cpuSumMs": sum(t["cpuSumMs"] for t in tasks), "wallClockMs": sum(t["wallClockMs"] for t in tasks), "errors": sum(t["errorCount"] for t in tasks), "byWorkStatus": _tally(tasks, "workStatus"), "byWorkCategory": _tally(tasks, "workCategory"), "byCurrentPhase": _tally(tasks, "currentPhase"), "byTaskType": _tally(tasks, "taskType"), } return { "ok": True, "projectRoot": str(project_root), "taskGroup": task_group or None, "taskCount": len(tasks), "tasks": tasks, "totals": totals, } def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="okstra rollup", description="Roll up run results across a task-group's tasks (raw ms).") parser.add_argument("--task-group", default="", help="scope to this task-group (default: whole project catalog)") parser.add_argument("--project-root", default="") parser.add_argument("--cwd", default=".") parser.add_argument("--json", action="store_true", help="emit JSON (always on)") parser.add_argument("--text", action="store_true", help="emit fixed text fields") args = parser.parse_args(argv) try: project_root = Path( resolve_project_root(explicit_root=args.project_root, cwd=args.cwd) ).resolve() except ResolverError as exc: payload = {"ok": False, "stage": "resolve", "reason": str(exc)} print(render_rollup_text(payload) if args.text else json.dumps(payload)) return 2 result = build_rollup(project_root, args.task_group) print( render_rollup_text(result) if args.text else json.dumps(result, ensure_ascii=False, indent=2) ) return 0 def render_rollup_text(payload: dict) -> str: """rollup 모델 표면에 승인된 필드만 고정 순서로 투영한다.""" rows = ["Okstra rollup\n"] rows.append(line("Status", "ready" if payload.get("ok") else "error")) rows.append(line("Task group", payload.get("taskGroup") or "all")) rows.append(line("Task count", payload.get("taskCount"))) totals = payload.get("totals") if isinstance(payload.get("totals"), dict) else {} rows.append(line("Total runs", totals.get("runs"))) rows.append(line("Total errors", totals.get("errors"))) rows.append(line("CPU sum ms", totals.get("cpuSumMs"))) rows.append(line("Wall clock ms", totals.get("wallClockMs"))) rows.extend(_distribution_lines("Work status", totals.get("byWorkStatus"))) rows.extend(_distribution_lines("Work category", totals.get("byWorkCategory"))) rows.extend(_distribution_lines("Current phase", totals.get("byCurrentPhase"))) rows.extend(_distribution_lines("Task type", totals.get("byTaskType"))) tasks = payload.get("tasks") if isinstance(payload.get("tasks"), list) else [] for index, task in enumerate(tasks, 1): if not isinstance(task, dict): continue rows.extend(_task_lines(index, task)) if not payload.get("ok"): rows.append(line("Failure stage", payload.get("stage"))) rows.append(line("Failure reason", payload.get("reason"))) return "".join(rows) def _distribution_lines(title: str, raw_values: object) -> list[str]: values = raw_values if isinstance(raw_values, dict) else {} rows: list[str] = [] for index, (name, count) in enumerate(sorted(values.items()), 1): rows.append(line(f"{title} {index} name", name)) rows.append(line(f"{title} {index} count", count)) return rows def _task_lines(index: int, task: dict) -> list[str]: fields = ( ("key", "taskKey"), ("type", "taskType"), ("category", "workCategory"), ("status", "workStatus"), ("phase", "currentPhase"), ("phase state", "currentPhaseState"), ("latest run status", "latestRunStatus"), ("updated at", "updatedAt"), ("runs", "runCount"), ("CPU sum ms", "cpuSumMs"), ("wall clock ms", "wallClockMs"), ("errors", "errorCount"), ("report path", "reportPath"), ) rows = [line(f"Task {index} {label}", task.get(key)) for label, key in fields] next_phase = task.get("nextRecommendedPhase") values = next_phase if isinstance(next_phase, dict) else {} for label, key in (("next phase", "phase"), ("next phase status", "status"), ("next phase rationale", "rationale")): rows.append(line(f"Task {index} {label}", values.get(key))) return rows if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))