"""Read-side time aggregation for a task. timeline.json 의 runs[] 를 순회하며 각 run 의 team-state(leadUsage/workers usage)를 읽어, task-type·worker 별 cross-run roll-up 과 per-run wall-clock / phase timeline 을 만든다. skill markdown 이 LLM 에게 시키던 집계(byTaskType / perWorker)를 코드 SSOT 로 옮긴 것 — 출력은 raw ms 이고 HH:MM:SS 포맷·Markdown 표 렌더는 호출자(skill)에 남긴다. """ from __future__ import annotations import argparse import datetime as dt import json import sys from pathlib import Path from .fixed_text import line as _text_line from okstra_ctl.paths import resolve_under_root, task_timeline_file from okstra_ctl.json_boundary import load_owned_object from okstra_project import read_task_key from okstra_ctl.task_target import resolve_task_root def time_rows_with_identity(run_root: Path) -> list[dict]: """Project time rows from stored execution labels without reparsing them.""" from okstra_ctl.usage_identity import identity_rows_from_state state_path = run_root / "state" / "team-state.json" try: state = load_owned_object(state_path, artifact="team state") except (OSError, ValueError): return [] return identity_rows_from_state(state if isinstance(state, dict) else {}) def load_runs(task_root: Path) -> tuple[list[dict], bool]: path = task_timeline_file(task_root) try: data = load_owned_object(path, artifact="task timeline") except (OSError, ValueError): return [], True runs = data.get("runs", []) if isinstance(data, dict) else [] if not isinstance(runs, list): return [], True return runs, False def _duration_ms(block) -> int: """usage / leadUsage block 의 durationMs — na_block·missing·None 은 0.""" block = block or {} if block.get("source") == "unavailable": return 0 try: return int(block.get("durationMs", 0) or 0) except (TypeError, ValueError): return 0 def _parse_iso(value: str) -> dt.datetime: return dt.datetime.fromisoformat(value.replace("Z", "+00:00")) def wall_clock_ms(state: dict) -> int: """run 의 실제 경과(벽시계) = max(endedAt) − min(startedAt). lead·worker 윈도가 겹치므로 cpuSum 과 다르다. timestamp 부족하면 0.""" times: list[dt.datetime] = [] blocks = [state.get("leadUsage")] + [ w.get("usage") for w in (state.get("workers") or []) if isinstance(w, dict)] for block in blocks: block = block or {} if block.get("source") == "unavailable": continue for key in ("startedAt", "endedAt"): raw = block.get(key) if not raw: continue try: times.append(_parse_iso(raw)) except (TypeError, ValueError): pass if len(times) < 2: return 0 return int((max(times) - min(times)).total_seconds() * 1000) def _phase_rows(state: dict) -> list[dict]: pt = state.get("phaseTimeline") if not isinstance(pt, dict): return [] return [{"phase": ph.get("phase", ""), "firstAt": ph.get("firstAt", ""), "wallMsToNext": ph.get("wallMsToNext")} for ph in (pt.get("phases") or []) if isinstance(ph, dict)] def _read_state(project_root: Path, rel: str) -> dict | None: target = resolve_under_root(project_root, rel) if target is None: return None try: state = load_owned_object(target, artifact="team state") except (OSError, ValueError): return None return state if isinstance(state, dict) else None def _collect_run(run: dict, project_root: Path) -> dict: """timeline run + team-state → 정규화된 측정. 결과에 `unavailable` reason 또는 측정 필드(leadMs/workers/wallClockMs/phases)를 담는다.""" ts, task_type = run.get("runTimestamp", ""), run.get("taskType", "") base = {"runTimestamp": ts, "taskType": task_type} state = _read_state(project_root, run.get("teamStatePath", "")) if state is None: return {**base, "reason": "team-state missing"} from okstra_ctl.execution_identity import stored_identity from okstra_ctl.usage_identity import identity_rows_from_state lead = _duration_ms(state.get("leadUsage")) workers = [] for worker in (state.get("workers") or []): if not isinstance(worker, dict): continue row = { "workerId": worker.get("workerId", ""), "agent": worker.get("agent", ""), "durationMs": _duration_ms(worker.get("usage")), } row.update(stored_identity(worker)) workers.append(row) if lead == 0 and all(w["durationMs"] == 0 for w in workers): return {**base, "reason": "no durationMs (Phase 7 not reached)"} return {**base, "leadMs": lead, "workers": workers, "wallClockMs": wall_clock_ms(state), "phases": _phase_rows(state), "identityRows": identity_rows_from_state(state)} def _by_task_type(collected: list[dict]) -> list[dict]: order: list[str] = [] acc: dict[str, dict] = {} for run in collected: tt = run["taskType"] if tt not in acc: acc[tt] = {"taskType": tt, "runs": 0, "leadMs": 0, "workersMs": 0} order.append(tt) acc[tt]["runs"] += 1 acc[tt]["leadMs"] += run["leadMs"] acc[tt]["workersMs"] += sum(w["durationMs"] for w in run["workers"]) for tt in order: acc[tt]["cpuSumMs"] = acc[tt]["leadMs"] + acc[tt]["workersMs"] return [acc[tt] for tt in order] def _add_worker(agg: dict, order: list, wid: str, agent: str, ms: int) -> None: if wid not in agg: agg[wid] = {"runs": 0, "totalMs": 0, "agents": set()} order.append(wid) if ms > 0: agg[wid]["runs"] += 1 agg[wid]["totalMs"] += ms if agent: agg[wid]["agents"].add(agent) def _per_worker(collected: list[dict]) -> dict: out: dict[str, list[dict]] = {} for tt in dict.fromkeys(r["taskType"] for r in collected): agg: dict[str, dict] = {} order: list[str] = [] for run in (r for r in collected if r["taskType"] == tt): _add_worker(agg, order, "lead", "", run["leadMs"]) for w in run["workers"]: _add_worker(agg, order, w["workerId"], w["agent"], w["durationMs"]) rows = [] for wid in order: a = agg[wid] if a["runs"] == 0: continue agents = sorted(x for x in a["agents"] if x and x != wid) rows.append({"workerId": wid, "agents": agents, "runs": a["runs"], "totalMs": a["totalMs"], "avgMs": a["totalMs"] // a["runs"]}) out[tt] = rows return out def aggregate_time(runs: list[dict], project_root: Path) -> dict: collected, unavailable = [], [] for run in runs: if not isinstance(run, dict): continue c = _collect_run(run, project_root) (unavailable if "reason" in c else collected).append(c) by_tt = _by_task_type(collected) grand = {k: sum(r[k] for r in by_tt) for k in ("runs", "leadMs", "workersMs", "cpuSumMs")} identity_rows = [] seen: set[str] = set() for collected_run in collected: for row in collected_run.get("identityRows") or []: key = str(row.get("roleExecutionRef") or row.get("executionLabel")) if key in seen: continue seen.add(key) identity_rows.append(row) return { "byTaskType": by_tt, "grandTotal": grand, "perWorker": _per_worker(collected), "perRunWallClock": [{"runTimestamp": c["runTimestamp"], "taskType": c["taskType"], "wallClockMs": c["wallClockMs"]} for c in collected], "phaseTimelines": [{"runTimestamp": c["runTimestamp"], "taskType": c["taskType"], "phases": c["phases"]} for c in collected if c["phases"]], "identityRows": identity_rows, "unavailable": unavailable, } def _worker_text(result: dict) -> str: parts: list[str] = [] workers = result.get("perWorker") if not isinstance(workers, dict): return "" for task_type, rows in workers.items(): for row in rows if isinstance(rows, list) else []: if isinstance(row, dict): parts.extend(("\n## Worker\n\n", _text_line("Task type", task_type), _text_line("Worker ID", row.get("workerId")), _text_line("Agents", ",".join(row.get("agents", [])) if isinstance(row.get("agents"), list) else None), _text_line("Runs", row.get("runs")), _text_line("Total ms", row.get("totalMs")), _text_line("Average ms", row.get("avgMs")))) return "".join(parts) def _timeline_text(result: dict) -> str: parts: list[str] = [] for row in result.get("perRunWallClock", []): if isinstance(row, dict): parts.extend(("\n## Run wall clock\n\n", _text_line("Run timestamp", row.get("runTimestamp")), _text_line("Task type", row.get("taskType")), _text_line("Wall clock ms", row.get("wallClockMs")))) for timeline in result.get("phaseTimelines", []): if not isinstance(timeline, dict): continue for phase in timeline.get("phases", []): if isinstance(phase, dict): parts.extend(("\n## Phase\n\n", _text_line("Run timestamp", timeline.get("runTimestamp")), _text_line("Task type", timeline.get("taskType")), _text_line("Phase", phase.get("phase")), _text_line("First at", phase.get("firstAt")), _text_line("Wall ms to next", phase.get("wallMsToNext")))) return "".join(parts) def render_result_text(result: dict) -> str: parts = ["# Okstra Time Input\n\n", _text_line("Task key", result.get("taskKey"))] total = result.get("grandTotal") if isinstance(result.get("grandTotal"), dict) else {} for key, label in (("runs", "Total runs"), ("leadMs", "Total lead ms"), ("workersMs", "Total workers ms"), ("cpuSumMs", "Total CPU sum ms")): parts.append(_text_line(label, total.get(key))) for row in result.get("byTaskType", []): if not isinstance(row, dict): continue parts.extend(("\n## Task type\n\n", _text_line("Task type", row.get("taskType")), _text_line("Runs", row.get("runs")), _text_line("Lead ms", row.get("leadMs")), _text_line("Workers ms", row.get("workersMs")), _text_line("CPU sum ms", row.get("cpuSumMs")))) parts.extend((_worker_text(result), _timeline_text(result))) for row in result.get("unavailable", []): if isinstance(row, dict): parts.extend(("\n## Unavailable run\n\n", _text_line("Run timestamp", row.get("runTimestamp")), _text_line("Task type", row.get("taskType")), _text_line("Reason", row.get("reason")))) return "".join(parts) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( prog="okstra time-report", description="Aggregate elapsed work time across a task's runs (raw ms).") parser.add_argument("target", help="task root path or task-key") 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) task_root, project_root = resolve_task_root(args.target, args.project_root, args.cwd) runs, _ = load_runs(task_root) result = {"ok": True, "taskKey": read_task_key(task_root)} result.update(aggregate_time(runs, project_root)) output = render_result_text(result) if args.text else json.dumps(result, ensure_ascii=False, indent=2) print(output, end="" if args.text else "\n") return 0 if __name__ == "__main__": raise SystemExit(main(sys.argv[1:]))