"""Top-level orchestrator that walks team-state and gathers all worker usage.""" from __future__ import annotations import json import sys import os from datetime import datetime, timezone from pathlib import Path from typing import Any from okstra_project.dirs import OKSTRA_RELATIVE from .blocks import accounting_workers, na_block, usage_block from .claude import ( claude_code_status_total, claude_session_totals, find_claude_agent_sessions, find_claude_team_sessions, ) from .codex import ( codex_session_ids, codex_session_is_worker, codex_session_total, codex_session_window_total, codex_wrapper_session_ids, find_codex_sessions, ) from .antigravity import ( antigravity_session_total, antigravity_status_total, find_antigravity_sessions, status_carries_usage, ) from .grok import ( find_grok_sessions, grok_session_is_non_interactive, grok_session_total, grok_session_window_total, ) from .paths import claude_project_dir, find_session_jsonl, utc_now from .pricing import antigravity_cost_usd, provider_billable_equivalent, provider_cost_usd from okstra_ctl.usage_identity import ( unrostered_usage_workers, usage_dispatch_records, usage_session_ids, ) from okstra_ctl.models import provider_wrappers from okstra_ctl.wrapper_status import ( log_path_for_prompt, read_wrapper_status, status_path_for_prompt, ) def match_prefixes(worker_id: str) -> list[str]: """Return the agentName prefixes that should be attributed to ``worker_id``. The Agent harness records the `name` arg on every dispatch as `agentName` in the subagent jsonl. Lead frequently appends suffixes (`-002`, `-reverify-r1`, `-impl`, `-2`) when it dispatches the same role multiple times or in different sub-flows. We treat every `agentName` matching one of these prefixes — either exactly or as `-` — as belonging to this worker so its tokens get aggregated. For implementation / final-verification runs the role variants `-executor` and `-verifier` are also attributed back to the matching provider worker (the bare `` prefix's `-` match covers any role suffix Lead assigns). """ if not worker_id: return [] if worker_id == "report-writer": return ["report-writer"] prefixes = [worker_id] if not worker_id.endswith("-worker"): prefixes.append(f"{worker_id}-worker") prefixes.append(f"{worker_id}-executor") return prefixes def agent_matches(agent_name: str, prefixes: list[str]) -> bool: if not agent_name: return False for prefix in prefixes: if agent_name == prefix or agent_name.startswith(f"{prefix}-"): return True return False def _aggregate_totals(items: list[dict]) -> dict: """Sum token + tool counters across multiple session totals dicts. `startedAt` / `endedAt` collapse to the union window; `durationMs` is recomputed from that window so re-tries and convergence rounds count against a single contiguous span. `model` and `agentName` keep the first non-empty value (the canonical role identity). """ aggregate: dict = { "totalTokens": 0, "inputTokens": 0, "outputTokens": 0, "cacheCreationTokens": 0, "cacheCreation5mTokens": 0, "cacheCreation1hTokens": 0, "cacheReadTokens": 0, "cachedInputTokens": 0, "reasoningOutputTokens": 0, "cachedTokens": 0, "thoughtsTokens": 0, "toolTokens": 0, "toolUses": 0, "durationMs": 0, "agentName": None, "model": None, "startedAt": None, "endedAt": None, } for t in items: for k in ("totalTokens", "inputTokens", "outputTokens", "cacheCreationTokens", "cacheCreation5mTokens", "cacheCreation1hTokens", "cacheReadTokens", "cachedInputTokens", "reasoningOutputTokens", "cachedTokens", "thoughtsTokens", "toolTokens", "toolUses"): aggregate[k] += t.get(k, 0) or 0 if aggregate["agentName"] is None and t.get("agentName"): aggregate["agentName"] = t["agentName"] if aggregate["model"] is None and t.get("model"): aggregate["model"] = t["model"] s, e = t.get("startedAt"), t.get("endedAt") if s and (aggregate["startedAt"] is None or s < aggregate["startedAt"]): aggregate["startedAt"] = s if e and (aggregate["endedAt"] is None or e > aggregate["endedAt"]): aggregate["endedAt"] = e wall = _wall_ms(aggregate["startedAt"], aggregate["endedAt"]) if wall is not None: aggregate["durationMs"] = wall return aggregate def run_artifact_suffix(team_state_path: Path) -> str | None: """``team-state--.json`` → ``-``. task 디렉토리 한 곳에 여러 run(재시도·이전 phase·레거시 타임스탬프)의 산출물이 섞여 있어, glob 으로 아무거나 집으면 엉뚱한 run 의 시각을 쓰게 된다(관측: 가장 오래된 레거시 manifest 의 createdAt 을 집어 윈도우가 한 달로 벌어짐). **접미사가 곧 매니페스트 이름은 아니다.** okstra 는 카테고리별로 seq 를 따로 매기므로 한 run 이 `manifests: 025` 와 `state: 023` 을 동시에 가질 수 있다. run-manifest 는 `_run_manifest_for_team_state` 가 매니페스트의 `teamStatePath` 역참조로 짚는다. status 산출물은 team-state 와 같은 카테고리라 이 접미사로 계속 짚는다.""" name = team_state_path.name if not (name.startswith("team-state-") and name.endswith(".json")): return None return name[len("team-state-"):-len(".json")] def _read_json_or_none(path: Path) -> dict | None: try: data = json.loads(path.read_text()) except (OSError, json.JSONDecodeError): return None return data if isinstance(data, dict) else None def _run_manifest_for_team_state( run_dir: Path, suffix: str, team_state_path: Path ) -> dict | None: """이 team-state 가 속한 run 의 매니페스트. 접미사가 곧 매니페스트 이름이라는 전제는 카테고리별 seq 가 갈리는 순간 깨진다. 실측(2026-08-27, fontsninja-nlpvibe `nestjs-migration-…`): 한 run 의 `manifests` 가 025, `state`·`prompts`·`worker-results` 가 023, `reports` 가 017 이었다. team-state 이름에서 뽑은 `023` 으로 매니페스트를 집으면 **12시간 전 run** 의 `createdAt`(04:45:34Z) 이 창의 시작이 되고, 그 사이 같은 세션이 돈 이전 run 들이 통째로 창에 들어온다 — `relax_start=False` 가 막으려던 바로 그 오탐이다(그 run 에서 로스터에 없는 워커 5명이 이 run 의 위반으로 보고됐다). 매니페스트는 자기 team-state 를 `teamStatePath` 에 적으므로 그것으로 맞춘다. 맞는 것이 없으면 접미사 방식으로 돌아간다 — 그 필드가 없던 옛 매니페스트다. """ direct = run_dir / "manifests" / f"run-manifest-{suffix}.json" # `Path.glob` 는 디렉터리가 없어도 빈 결과를 낸다 — 감쌀 예외가 없다. candidates = [ direct, *sorted((run_dir / "manifests").glob("run-manifest-*.json")), ] for path in candidates: data = _read_json_or_none(path) if data is None: continue recorded = str(data.get("teamStatePath") or "") if recorded and Path(recorded).name == team_state_path.name: return data return _read_json_or_none(direct) def _created_at_by_suffix(run_dir: Path, suffix: str) -> str | None: """접미사가 곧 매니페스트 이름인 조회. **다른** run 을 접미사로 지목할 때만 쓴다(`_previous_run_end`). 이번 run 은 team-state 로 맞춰야 한다 — 두 접미사가 갈리는 run 이 실재한다. """ data = _read_json_or_none(run_dir / "manifests" / f"run-manifest-{suffix}.json") return None if data is None else data.get("createdAt") def relaxation_floor(run_dir: Path, suffix: str, manifest: dict | None) -> str | None: """since 완화가 내려갈 수 있는 하한. in-session 리드(`entryMode: current-session`)의 세션은 run 을 위해 태어난 것이 아니다 — 한 세션이 여러 날에 걸쳐 여러 task 의 run 을 돌린다(관측 2026-09-02, dev-10626 error-analysis r04: 세션 첫 ts 08-17T08:00Z, run createdAt 09-02T20:03Z, 리드 소요 398h / $292 로 보고). 그 세션의 첫 ts 는 이 run 의 시작에 대해 아무것도 말하지 않으므로 하한은 매니페스트 createdAt 자체다 — 완화가 일어나지 않는다. 프로세스를 새로 띄운 리드는 세션이 prep 직전에 태어나므로 종전대로 직전 run 의 종료가 하한이다(`_previous_run_end`). """ if manifest is not None and manifest.get("entryMode") == "current-session": created = manifest.get("createdAt") if created: return str(created) return _previous_run_end(run_dir, suffix) def window_start_is_pinned(since: str | None, floor: str | None) -> bool: """하한이 시작점 이상이면 완화할 여지가 없다 — 세션 jsonl 을 훑을 이유도 없다.""" return bool(since and floor and floor >= since) def relax_window_start( since: str | None, earliest: str | None, floor: str | None ) -> str | None: """`earliest` 가 `since` 보다 앞서면 하한 안에서 시작점을 앞당긴다.""" if not earliest: return since if floor and earliest < floor: earliest = floor if since is None or earliest < since: return earliest return since def run_window_relaxation_floor(team_state_path: Path) -> str | None: """collect 의 found-sessions 완화가 쓰는 하한 — `resolve_run_window` 와 같은 규칙.""" suffix = run_artifact_suffix(team_state_path) if not suffix: return None run_dir = team_state_path.parent.parent manifest = _run_manifest_for_team_state(run_dir, suffix, team_state_path) return relaxation_floor(run_dir, suffix, manifest) def _recorded_run_end(state: dict, manifest: dict | None) -> str | None: """완료된 run 이 지난 수집에서 기록한 창의 끝 — 리드·워커 usage 블록의 endedAt 최댓값. run 의 끝을 적는 코드가 없다(`runEndedAt` 은 읽기만, status 파일은 안 쓰인다). 그래서 완료된 run 을 다시 수집하면 창의 끝이 지금이 되어, 같은 세션이 그 뒤에 돌린 다른 run 까지 이 run 의 리드 창에 들어온다(관측 2026-09-03: dev-10626 error-analysis r04 를 하루 뒤 재finalize 하면 오늘 run 이 들어옴). 매니페스트가 `completed` 인 run 은 처음 완료됐을 때 수집한 창이 곧 run 의 끝이므로 그 값으로 고정한다. 아직 완료 전이면 None — 리드가 서술문을 고쳐 Phase 7 을 다시 도는 동안은 창이 지금까지 늘어나는 것이 맞다. """ if not isinstance(manifest, dict) or manifest.get("status") != "completed": return None ends: list[str] = [] blocks = [state.get("leadUsage")] + [ worker.get("usage") for worker in accounting_workers(state) ] for block in blocks: if not isinstance(block, dict) or block.get("source") == "unavailable": continue ended = block.get("endedAt") if isinstance(ended, str) and ended: ends.append(ended) return max(ends) if ends else None def _run_end_estimate(run_dir: Path, suffix: str) -> str | None: """run 종료 근사 — 같은 run 의 status 산출물 mtime(reconcile 후 고정, Phase 7 재렌더로도 바뀌지 않음). 완료 전(status 부재)이면 None.""" p = run_dir / "status" / f"final-{suffix}.status" try: mtime = p.stat().st_mtime except OSError: return None return datetime.fromtimestamp(mtime, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def _project_root_from_team_state(team_state_path: Path) -> Path | None: """team-state path(`/.okstra/tasks/.../state/team-state-*.json`)에서 프로젝트 루트(`.okstra` 를 담은 디렉토리)를 되짚는다. 못 찾으면 None.""" for parent in team_state_path.resolve().parents: if parent.name == OKSTRA_RELATIVE.name: return parent.parent return None def _session_first_ts(jsonl_path: Path) -> str | None: """세션 jsonl 의 최소 timestamp — reconstruct_needles_from_workers 와 동일한 top-level `timestamp` 규칙. 파일 부재/파싱 실패 레코드는 건너뛴다.""" first_ts: str | None = None try: with jsonl_path.open(encoding="utf-8") as fh: for raw in fh: try: rec = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError): continue ts = rec.get("timestamp") if ts and (first_ts is None or ts < first_ts): first_ts = ts except OSError: return None return first_ts def _earliest_lead_session_ts(state: dict, team_state_path: Path) -> str | None: """`state["leadSessionIds"]` 각 lead 세션 jsonl 첫 ts 중 최소값. 부재하거나 어느 세션에서도 ts 를 못 찾으면 None. proj_dir 은 team-state path 에서 되짚은 프로젝트 루트에 `claude_project_dir` 규칙을 적용해 해소한다.""" lead_ids = [sid for sid in (state.get("leadSessionIds") or []) if sid] if not lead_ids: return None project_root = _project_root_from_team_state(team_state_path) if project_root is None: return None proj_dir = claude_project_dir(project_root) earliest: str | None = None for sid in lead_ids: ts = _session_first_ts(proj_dir / f"{sid}.jsonl") if ts and (earliest is None or ts < earliest): earliest = ts return earliest def _previous_run_suffix(run_dir: Path, suffix: str) -> str | None: """같은 task-type 에서 이 run 직전에 실행된 run 의 접미사, 없으면 None. seq 는 연속이 아닐 수 있으므로(실패한 prep 이 번호를 소모한다) manifest 를 실제로 스캔해 현재 seq 미만 중 최대를 고른다.""" task_type, _, raw_seq = suffix.rpartition("-") if not task_type or not raw_seq.isdigit(): return None current = int(raw_seq) prefix, width = f"run-manifest-{task_type}-", len(raw_seq) earlier = [] try: names = [p.name for p in (run_dir / "manifests").iterdir()] except OSError: return None for name in names: if not (name.startswith(prefix) and name.endswith(".json")): continue candidate = name[len(prefix):-len(".json")] if candidate.isdigit() and int(candidate) < current: earlier.append(int(candidate)) if not earlier: return None return f"{task_type}-{max(earlier):0{width}d}" def _previous_run_end(run_dir: Path, suffix: str) -> str | None: """직전 run 의 종료 시각 — 종료 산출물이 없으면 그 run 의 시작 시각. `since` 완화의 하한이다. 한 세션에서 같은 task-type 을 두 번 돌리면 `leadSessionIds` 의 세션 jsonl 첫 ts 는 *첫 run* 의 시작을 가리키므로, 완화를 그대로 두면 seq002 의 윈도우가 seq001 구간을 통째로 삼킨다 — seq001 의 PROGRESS 체크포인트가 seq002 검증에 섞여 순서 오탐을 낸다 (관측: dev-10172 — seq002 검증이 seq001 의 phase-6 타임스탬프를 인용).""" prev = _previous_run_suffix(run_dir, suffix) if prev is None: return None try: prev_state = json.loads( (run_dir / "state" / f"team-state-{prev}.json").read_text() ) ended_at = prev_state.get("runEndedAt") if ended_at: return str(ended_at) except OSError: # expected-miss: 직전 run 이 team-state 를 남기지 않고 끝났으면 파일이 # 없다. 아래 두 추정 경로가 바로 그 경우를 위한 폴백이다. pass except json.JSONDecodeError as exc: # 파일은 있는데 깨진 것은 폴백이 상정한 상황이 아니다. print(f"token-usage: team-state for the previous run ({prev}) is not " f"valid JSON, estimating instead ({exc})", file=sys.stderr) return _run_end_estimate(run_dir, prev) or _created_at_by_suffix(run_dir, prev) def resolve_run_window( team_state_path: Path, state: dict, *, relax_start: bool = True ) -> tuple[str | None, str | None]: """이 run 의 [시작, 종료] ISO 윈도우. in-session lead 는 자기 run 을 사용자의 *세션 전체* jsonl 에 기록하므로, 윈도우 없이 합산하면 무관한 모든 턴(다른 작업·대화)이 lead 토큰·시간에 섞여 폭증한다(관측: requirements-discovery 한 run 에 lead 1.7억 토큰 / $416 / 3h). 토큰 집계를 이 윈도우로 스코핑해 그 run 분만 센다. 시작 = 이 run 의 run-manifest createdAt, 종료 = team-state.runEndedAt → 이 run 의 status mtime → 완료된 run 이 지난 수집에서 기록한 창의 끝(`_recorded_run_end`) → 현재 시각(아직 진행 중) 순으로 해소한다. 접미사를 못 뽑으면 (None, None) — 윈도우 없이 전체를 세는 기존 동작으로 안전 폴백. 시작 완화(`relax_start`, 기본 True — 토큰 수집용): run-manifest createdAt 이 이 run 의 실제 lead/worker 세션보다 늦게 찍히면(관측: dev-9902 — createdAt 11:43Z 가 05:34Z 시작 세션보다 늦음) since 가 앞선 세션을 잘라낸다. Task 3 이 기록한 `leadSessionIds[]` 세션의 첫 ts 최소값이 createdAt 보다 앞서면 그 값으로 since 를 앞당긴다. `leadSessionIds` 부재(legacy)면 createdAt 그대로 — 기존 동작 불변. 완화 하한: 직전 run 의 종료 시각 이전으로는 내려가지 않는다 (`_previous_run_end`). 세션 jsonl 은 run 마다 새로 생기지 않으므로 하한이 없으면 같은 세션의 이전 run 이 이 윈도우에 통째로 들어온다. in-session 리드(`entryMode: current-session`)는 완화하지 않는다 — 그 세션은 run 보다 먼저 태어났고 다른 task 의 run 도 돌렸으므로 첫 ts 는 세션 탄생일이다 (`relaxation_floor`). `relax_start=False` (세션 스코프 검증기 전용 — session-conformance / forbidden-actions): 완화를 건너뛰고 since 를 createdAt 에 고정한다. createdAt 은 prep 시각이라 이 run 의 lead·worker 활동보다 반드시 앞서므로 건전한 하한이고, 재사용 세션에 섞인 같은 세션의 이전 run·다른 task 활동을 창에서 배제한다 (dev-10172 D-3: 이전 planning run 의 phase-6/7 앵커, 다른 task 의 git push 를 이 run 의 위반으로 오탐하던 근본 원인). 완화는 토큰을 놓치지 않으려고 세션 birth 까지 내려가지만, 검증기에는 그 관대함이 곧 오탐이다.""" suffix = run_artifact_suffix(team_state_path) if not suffix: return None, None run_dir = team_state_path.parent.parent manifest = _run_manifest_for_team_state(run_dir, suffix, team_state_path) since = None if manifest is None else manifest.get("createdAt") if relax_start: floor = relaxation_floor(run_dir, suffix, manifest) if not window_start_is_pinned(since, floor): earliest = _earliest_lead_session_ts(state, team_state_path) since = relax_window_start(since, earliest, floor) until = ( state.get("runEndedAt") or _run_end_estimate(run_dir, suffix) or _recorded_run_end(state, manifest) or utc_now() ) return since, until def _wall_ms(start_iso: str | None, end_iso: str | None) -> int | None: if not start_iso or not end_iso: return None try: a = datetime.fromisoformat(start_iso.replace("Z", "+00:00")) b = datetime.fromisoformat(end_iso.replace("Z", "+00:00")) except ValueError: return None return max(0, int((b - a).total_seconds() * 1000)) def phase_timeline(markers: list[dict]) -> dict: """lead 의 PROGRESS 체크포인트(prompts/lead/okstra-lead-contract.md "Progress reporting")로 run 내부 단계 경계 wall-clock 을 복원한다 (perf plan v2 P0 계측). phase id = marker 의 첫 토큰(`phase-5.5-convergence` 등). 같은 phase 의 반복 마커(poll/collect/dispatch)는 first/last 로 접는다. `wallMsToNext` 는 다음 phase 의 firstAt 까지 — 마지막 phase 는 None(run 종료 신호가 마커에 없으므로 추정하지 않는다). 마커가 하나도 없으면 phases 가 빈 채로 남아 "이 run 은 측정 불가"를 명시적으로 표현한다. """ phases: list[dict] = [] by_id: dict[str, dict] = {} for m in markers: at = m.get("at") phase_id = (m.get("marker") or "").split(None, 1)[0] if not phase_id: continue entry = by_id.get(phase_id) if entry is None: entry = {"phase": phase_id, "firstAt": at, "lastAt": at, "markerCount": 0, "wallMsToNext": None} by_id[phase_id] = entry phases.append(entry) entry["markerCount"] += 1 if at: if entry["firstAt"] is None or at < entry["firstAt"]: entry["firstAt"] = at if entry["lastAt"] is None or at > entry["lastAt"]: entry["lastAt"] = at for current, nxt in zip(phases, phases[1:]): current["wallMsToNext"] = _wall_ms(current["firstAt"], nxt["firstAt"]) return {"source": "lead-progress-markers", "phases": phases} def resolve_team_name(state: dict) -> str: """team-state 에서 이 run 의 team name 을 해소한다. Phase 3 TeamCreate 성공 시 lead 가 기록한 값을 우선한다. 기록 위치는 계약 버전에 따라 둘 중 하나다: - nested: state.team.teamName (현재 문서화된 스키마) - root: state.teamName (v0.24 이전 관행; 실 run 에 여전히 흔함) 둘 다 비어 있으면 `okstra-` 관례로 폴백 — task-id 만 쓰므로 multi-segment task key 에서 빈번히 mis-match 하는 최후 수단이다. """ state_team = state.get("team") or {} team_name = state_team.get("teamName") or state.get("teamName") or "" if not team_name: task_key = state.get("taskKey", "") task_id = task_key.rsplit(":", 1)[-1] if task_key else "" team_name = f"okstra-{task_id}" if task_id else "" return team_name def resolve_team_needles(state: dict) -> list[str]: """이 run 에 속한 harness teamName(`session-`) 전체. lead 재발급마다 세대가 갈리므로 team-state.teamName 단일값으로는 후속 세대 워커를 놓친다. 축1 이 기록한 `observedTeamNames[]` 를 정본으로 쓰고, 수동 정정으로 teamName 에 직접 박힌 `session-*` 값도 흡수한다. audit label(`okstra-`) 은 harness needle 이 아니므로 제외한다. """ needles: set[str] = set() for name in state.get("observedTeamNames") or []: if isinstance(name, str) and name.startswith("session-"): needles.add(name) label = resolve_team_name(state) if label.startswith("session-"): needles.add(label) return sorted(needles) def resolve_team_needles_with_source( state: dict, cwd, since, until, projects_root=None, ) -> tuple[list[str], str]: """세 소비처(collect + 두 validator) 공통 needle 조립 규칙. needle 은 `resolve_team_needles(state)` 만으로 조립한다 — 축1 의 `observedTeamNames[]` 세대들 + `session-*` 형태의 수동 정정 라벨. audit label(`okstra-`)은 needle 로 절대 쓰지 않는다: 하네스 jsonl 은 `session-*` teamName 만 실어(okstra-lead-contract.md), 라벨은 여러 run 이 공유하는 task-key 파생값이라 needle 로 쓰면 같은 task-key 의 다른 run 세션을 끌어와 cross-run mis-attribution 을 일으킨다(spec §2.1 이 배제하는 바로 그 실패). `resolve_team_name` 은 표시 라벨(`usageSummary.teamName`) 전용. `resolve_team_needles` 가 비면(observedTeamNames 미기록 legacy run) 축4 의 `reconstruct_needles_from_workers` 로 run window 안 워커 jsonl 의 `session-*` teamName 을 best-effort 복구하고 `reconstructed-from-workers` provenance 를 돌려준다 — window 로 스코핑되므로 타 run 을 안전하게 배제한다. 그마저 비면 `none`. 반환한 source 는 각 소비처가 자기 관례대로 표면화한다 (collect → usageSummary.needleSource). """ needles = resolve_team_needles(state) if needles: return needles, "observed-team-names" reconstructed = reconstruct_needles_from_workers( cwd, since, until, projects_root=projects_root ) if reconstructed: return reconstructed, "reconstructed-from-workers" return [], "none" def _resolve_project_path(project_root: Path, raw_path: str) -> Path | None: if not raw_path: return None path = Path(raw_path) return path if path.is_absolute() else project_root / path def _load_lead_event_records(project_root: Path, state: dict) -> list[dict]: path = _resolve_project_path(project_root, state.get("leadEventsPath", "")) if path is None or not path.is_file(): return [] records = [] for line in path.read_text(encoding="utf-8").splitlines(): if not line.strip(): continue try: record = json.loads(line) except json.JSONDecodeError: continue if isinstance(record, dict): records.append(record) return records def _codex_worker_windows(project_root: Path, state: dict) -> dict[str, list[tuple[str, str]]]: active: dict[tuple[str, str], str] = {} windows: dict[str, list[tuple[str, str]]] = {} for event in _load_lead_event_records(project_root, state): details = event.get("details") or {} if not isinstance(details, dict): continue worker_id = str(details.get("workerId") or "").strip() if not worker_id: continue attempt = str(details.get("attempt") or "") key = (worker_id, attempt) event_type = event.get("eventType") timestamp = str(event.get("timestamp") or "").strip() if not timestamp: continue if event_type == "worker-dispatched": active[key] = timestamp elif event_type in {"worker-result-collected", "worker-failed"}: started_at = active.pop(key, "") if started_at: windows.setdefault(worker_id, []).append((started_at, timestamp)) running_workers = { str(worker.get("workerId") or "").strip() for worker in accounting_workers(state) if isinstance(worker, dict) and worker.get("status") in {"running", "in-progress"} } if running_workers: open_until = utc_now() for (worker_id, _attempt), started_at in active.items(): if worker_id in running_workers: windows.setdefault(worker_id, []).append((started_at, open_until)) return windows # Claude 의 트랜스크립트는 세션 jsonl 이고, 그것은 이 모듈의 Claude 경로가 # 읽는다. 별도의 CLI 트랜스크립트가 없으므로 CLI 경로에 태우면 매번 # `no transcript was found` 가 붙는다 — 세션을 찾아 토큰을 붙인 행에도. _SESSION_JSONL_PROVIDERS = frozenset({"claude"}) def _cli_assignment_provider(worker: dict) -> str: worker_id = str(worker.get("workerId") or "").strip() explicit_provider = str(worker.get("provider") or "").strip() provider = explicit_provider or str(worker.get("agent") or "").strip() if provider in _SESSION_JSONL_PROVIDERS: return "" runner = str(worker.get("runner") or "").strip() if runner and runner != "cli-wrapper": return "" if not explicit_provider and worker_id == "report-writer": provider = "codex" return provider if provider in provider_wrappers() else "" def _cli_sessions_for_windows( provider: str, project_root: Path, windows: list[tuple[str, str]], ) -> list[Path]: sessions: list[Path] = [] seen: set[Path] = set() for started_at, ended_at in windows: if provider == "codex": matches = find_codex_sessions(project_root, started_at, ended_at) elif provider == "antigravity": matches = find_antigravity_sessions(project_root, started_at, ended_at) elif provider == "grok": matches = find_grok_sessions(project_root, started_at, ended_at) else: matches = [] for path in matches: if path not in seen: seen.add(path) sessions.append(path) return sessions def _cli_session_totals( provider: str, session_paths: list[Path], *, window: tuple[str, str] | None = None, ) -> list[dict]: """세션별 합계. `window` 는 리드 전용 — run 보다 먼저 열린 세션을 창으로 자른다.""" totals = [] for session_path in session_paths: if provider == "codex" and window is not None: total = codex_session_window_total(session_path, *window) elif provider == "codex": total = codex_session_total(session_path) elif provider == "grok" and window is not None: total = grok_session_window_total(session_path, *window) elif provider == "antigravity": total = ( antigravity_status_total(session_path) if status_carries_usage(session_path) else antigravity_session_total(session_path) ) elif provider == "grok": total = grok_session_total(session_path) elif provider in {"zai", "claude"}: total = claude_code_status_total(session_path) else: continue if total.get("available"): totals.append(total) return totals _THREE_RATE_PROVIDERS = frozenset({"codex", "grok", "kimi"}) def _cli_usage_block(provider: str, totals: dict, session_paths: list[Path]) -> dict: """CLI 공급자 세션 합계를 보고서가 읽는 usage 블록으로. codex·grok·kimi 는 캐시 재읽기를 입력 토큰 안에 넣어 보고한다(codex `cached_input_tokens ⊂ input_tokens`, grok `cachedReadTokens ⊂ inputTokens`). 보고서 표는 원시 토큰을 "캐시 재읽기를 뺀 값" 으로 정의하고 캐시 재읽기·과금 환산 토큰에 자기 칸을 두므로, claude-jsonl 블록과 같은 키(`cacheReadTokens`, `billableEquivalentTokens`)로 같은 정의의 값을 싣는다 — 종전엔 `cachedInputTokens` 키에만 남아 표의 두 칸이 `--` 로 찍혔다(실측 2026-09-09, dev-10627-2 planning 002: grok 890,880 / codex 3,864,448 토큰이 빠짐). 비용은 `cliEstimatedCostUsd` 그대로다 — 합계표가 CLI 비용을 별도 행으로 둔다. """ block = usage_block(totals, source=f"{provider}-cli") cached_input = totals.get("cachedInputTokens", 0) or 0 if provider in _THREE_RATE_PROVIDERS and cached_input: block["totalTokens"] = max(0, block["totalTokens"] - cached_input) block["cacheReadTokens"] = cached_input block["cliTotalTokens"] = block["totalTokens"] block["cliSessionPaths"] = [str(path) for path in session_paths] if totals.get("model"): block["model"] = totals["model"] block["cliModel"] = totals["model"] if provider in _THREE_RATE_PROVIDERS or provider == "zai": input_tokens = totals.get("inputTokens", 0) or 0 if provider == "zai": # GLM 상태는 캐시 토큰을 입력과 별도로 보고한다. 공통 단가는 # 캐시 포함 입력을 받으며, 캐시 생성은 일반 입력 단가로 계산한다. cached_input = totals.get("cacheReadTokens", 0) or 0 input_tokens += (totals.get("cacheCreationTokens", 0) or 0) + cached_input rates = ( totals.get("model"), input_tokens, cached_input, totals.get("outputTokens", 0) or 0, ) cost = provider_cost_usd(provider, *rates) billable = provider_billable_equivalent(provider, *rates) if billable is not None: block["billableEquivalentTokens"] = billable elif provider == "antigravity": cost = antigravity_cost_usd( totals.get("model"), totals.get("inputTokens", 0) or 0, totals.get("outputTokens", 0) or 0, ) else: cost = None block["cliNote"] = "Token usage recorded; provider billing cost is unavailable." if cost is not None: block["cliEstimatedCostUsd"] = cost return block def _epoch_iso(value: object) -> str | None: if not isinstance(value, (int, float)) or isinstance(value, bool): return None return datetime.fromtimestamp(value, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def wrapper_execution(status_path: Path | None) -> dict: """Read wrapper execution evidence without inferring token availability.""" if status_path is None or not status_path.is_file(): return {"status": "not-started"} status = read_wrapper_status(status_path) if status is None: return {"status": "failed", "statusPath": str(status_path)} if status.timeout: execution_status = "timeout" elif status.stage == "started": execution_status = "started" elif status.stage == "exited" and status.exit_code == 0: execution_status = "exited" else: execution_status = "failed" execution = { "status": execution_status, "statusPath": str(status_path), "startedAt": _epoch_iso(status.raw.get("started_ts")), "endedAt": _epoch_iso(status.raw.get("ended_ts")), } if status.exit_code is not None: execution["exitCode"] = status.exit_code duration = _wrapper_duration_ms(status.raw.get("duration_ms"), execution) if duration is not None: execution["durationMs"] = duration return execution def _wrapper_duration_ms(raw_duration: object, execution: dict) -> int | None: if isinstance(raw_duration, (int, float)) and not isinstance(raw_duration, bool): return max(0, int(raw_duration)) return _wall_ms(execution.get("startedAt"), execution.get("endedAt")) def _execution_note(execution: dict) -> str: status = execution["status"] if status == "exited": return f"wrapper exited {execution.get('exitCode', 0)}" if status == "timeout": return "wrapper timed out" if status == "failed": exit_code = execution.get("exitCode") if exit_code is not None: return f"wrapper failed with exit code {exit_code}" return "wrapper status is invalid" if status == "started": return "wrapper started and has not recorded an exit" return "wrapper status sidecar is unavailable" def collect_cli_usage( *, provider: str, status_path: Path | None, sessions: list[Path], fallback_window_used: bool = False, ) -> dict: """Collect attributable CLI usage while retaining execution evidence.""" execution = wrapper_execution(status_path) execution_note = _execution_note(execution) totals = _cli_session_totals(provider, sessions) if not sessions: block = na_block( f"{execution_note}; CLI usage attribution unavailable because no transcript was found" ) block["cliNote"] = block["note"] elif not totals: block = na_block( f"{execution_note}; transcript found but no final token snapshot was recorded" ) block["cliNote"] = block["note"] block["cliSessionPaths"] = [str(path) for path in sessions] else: block = _cli_usage_block(provider, _aggregate_totals(totals), sessions) block["cliExecutionStatus"] = execution["status"] if "durationMs" in execution: block["durationMs"] = execution["durationMs"] if fallback_window_used: fallback_note = "wrapper status sidecar unavailable; used aggregate wrapper window fallback" prior_note = block.get("cliNote") block["cliNote"] = f"{prior_note}; {fallback_note}" if prior_note else fallback_note return block def _prompt_cli_window( project_root: Path, prompt_path_raw: str, ) -> tuple[tuple[str, str] | None, Path | None]: """한 dispatch 의 래퍼 창 — 프롬프트 옆 status 사이드카의 started/ended.""" prompt_path = _resolve_project_path(project_root, prompt_path_raw) status_path = status_path_for_prompt(prompt_path) if prompt_path is not None else None execution = wrapper_execution(status_path) started_at = execution.get("startedAt") if not started_at: return None, status_path return (started_at, execution.get("endedAt") or utc_now()), status_path def _worker_prompt_paths(worker: dict, records: list[dict]) -> list[str]: """이 워커가 띄운 dispatch 의 프롬프트 경로 전부 — 원장 행이 없으면(v1) 명부 행의 하나.""" paths: list[str] = [] for record in records: raw = str(record.get("promptPath") or "").strip() if raw and raw not in paths: paths.append(raw) if not paths: raw = str(worker.get("promptPath") or "").strip() if raw: paths.append(raw) return paths def _worker_session_cwds(project_root: Path, records: list[dict]) -> list[Path]: """세션 디렉터리가 인코딩된 cwd 후보 — grok·kimi 는 워크트리 안에서 돈다. `providers/grok/adapter.py` 는 `request.worktree_path or request.project_root` 를 cwd 로 넘기고 세션 디렉터리는 그 경로를 퍼센트 인코딩한 이름이다. 프로젝트 루트로만 찾으면 워크트리 안에서 돈 세션은 0건이다(실측 2026-09-08 jobs implementation-planning r01: grok critic 세션이 워크트리 이름 아래 있었다). codex·claude 는 루트에서 돌므로 워크트리 후보는 빈 결과로 끝난다. """ cwds: list[Path] = [] for record in records: raw = str(record.get("worktreePath") or "").strip() if raw and Path(raw) not in cwds: cwds.append(Path(raw)) if project_root not in cwds: cwds.append(project_root) return cwds def _antigravity_usage_sources( project_root: Path, worker: dict, status_path: Path | None, ) -> list[Path]: """agy 의 토큰 스냅샷이 있는 곳. 홈 세션 디렉터리가 아니다. 정본은 래퍼 status 사이드카의 `usage` 다 — 러너가 스트림의 마지막 `result.usage` 를 종료 시점에 적는다(`worker_runner`). 워커 로그는 러너가 사람이 읽는 줄로 옮겨 적은 것이라 stream-json 이 아니고, 그 로그에서 usage 를 찾던 동안 antigravity 워커 전부가 `transcript found but no final token snapshot was recorded` 였다(관측 2026-09-02, dev-10626 r01·r02·r04). 로그를 읽는 경로는 러너 이전의 raw stream-json 로그를 위해 남긴다 — 스냅샷이 있는 run 의 로그는 옮겨 적은 텍스트라 트랜스크립트로 나열하지 않는다. """ if status_path is not None and status_carries_usage(status_path): return [status_path] paths: list[Path] = [] prompt_path = _resolve_project_path(project_root, str(worker.get("promptPath") or "")) if prompt_path is not None: log_path = log_path_for_prompt(prompt_path) if log_path.is_file(): paths.append(log_path) status = read_wrapper_status(status_path) if status_path else None if status is not None and status.log_path is not None and status.log_path.is_file(): if status.log_path not in paths: paths.append(status.log_path) return paths def _worker_cli_usage_block( *, provider: str, project_root: Path, worker: dict, records: list[dict], fallback_windows: list[tuple[str, str]], ) -> dict: """공급자 프로세스 트랜스크립트와 실행 증거로 워커 사용량 블록을 만든다. dispatch 원장 행마다 래퍼 창을 하나씩 잡아 그 창의 세션을 전부 합산한다 — 재검증·critic-gap·plan-verify 로 다시 띄운 실행이 각각 새 세션이므로 첫 프롬프트 하나만 보면 나머지는 통째로 빠진다. 실행 상태는 마지막 dispatch 의 것, 소요 시간은 래퍼 창의 합이다. """ windows: list[tuple[str, str]] = [] prompt_paths = _worker_prompt_paths(worker, records) status_paths: list[Path | None] = [] for raw in prompt_paths: window, status_path = _prompt_cli_window(project_root, raw) status_paths.append(status_path) if window is not None: windows.append(window) used_fallback = not windows if used_fallback: windows = fallback_windows session_paths, attribution = _worker_cli_sessions( provider, project_root, worker, records, windows, status_paths, ) status_path = status_paths[-1] if status_paths else None if provider == "zai": session_paths = list(dict.fromkeys( path for path in status_paths if path is not None and path.is_file() )) if provider == "antigravity": for raw, source_status in zip(prompt_paths, status_paths): for source in _antigravity_usage_sources( project_root, {"promptPath": raw}, source_status, ): if source not in session_paths: session_paths.append(source) block = collect_cli_usage( provider=provider, status_path=status_path, sessions=session_paths, fallback_window_used=used_fallback, ) block.update(attribution) durations = [ wrapper_execution(path).get("durationMs") for path in status_paths if path is not None ] measured = [value for value in durations if isinstance(value, int)] if len(measured) > 1: block["durationMs"] = sum(measured) return block def _worker_cli_sessions( provider: str, project_root: Path, worker: dict, records: list[dict], windows: list[tuple[str, str]], status_paths: list[Path | None], ) -> tuple[list[Path], dict]: candidates = list(dict.fromkeys( path for cwd in _worker_session_cwds(project_root, records) for path in _cli_sessions_for_windows(provider, cwd, windows) )) if provider != "codex": return candidates, {} wanted = {str(record["sessionId"]) for record in records if record.get("sessionId")} for raw in _worker_prompt_paths(worker, records): prompt = _resolve_project_path(project_root, raw) if prompt is not None: wanted.update(codex_wrapper_session_ids(log_path_for_prompt(prompt))) for path in status_paths: status = read_wrapper_status(path) if path is not None else None if status is not None and status.raw.get("sessionId"): wanted.add(str(status.raw["sessionId"])) if wanted: selected = [path for path in candidates if codex_session_ids(path) & wanted] found = {session_id for path in selected for session_id in codex_session_ids(path)} return selected, {"attribution": "session-id", "requestedSessionIds": sorted(wanted), "missingSessionIds": sorted(wanted - found)} unique_windows = all( len({path for cwd in _worker_session_cwds(project_root, records) for path in _cli_sessions_for_windows(provider, cwd, [window])}) <= 1 for window in windows ) if unique_windows: return candidates, {"attribution": "legacy-window"} return [], { "attribution": "ambiguous", "unattributedCliSessionPaths": [str(path) for path in candidates], "cliNote": "Several Codex sessions match the wrapper window; no session identity was recorded.", } def _remove_shared_codex_usage(state: dict) -> None: """같은 세션의 시간대 추정을 확정된 소유자보다 우선하지 않는다.""" rows = [row for row in accounting_workers(state) if (row.get("usage") or {}).get("source") == "codex-cli"] owners: dict[str, list[dict]] = {} for row in rows: for path in row["usage"].get("cliSessionPaths") or []: owners.setdefault(path, []).append(row) exclusions = [] for row in rows: block = row["usage"] rejected = [] for path in block.get("cliSessionPaths") or []: shared = owners[path] if len(shared) < 2: continue exact = [item for item in shared if item["usage"].get("attribution") == "session-id"] if len(exact) != 1 or exact[0] is not row: rejected.append(path) if not rejected: continue exclusions.append((row, rejected)) for row, rejected in exclusions: block = row["usage"] kept = [Path(path) for path in block.get("cliSessionPaths") or [] if path not in rejected] replacement = collect_cli_usage(provider="codex", status_path=None, sessions=kept) replacement.update({key: block[key] for key in ("durationMs", "cliExecutionStatus") if key in block}) replacement.update(attribution="ambiguous", unattributedCliSessionPaths=rejected, cliNote="Codex session matched more than one worker; duplicate attribution excluded.") row["usage"] = replacement def _unattributed_usage(state: dict, claude_usage: dict | None) -> dict | None: workers = accounting_workers(state) assigned = {path for row in workers for path in (row.get("usage") or {}).get("cliSessionPaths") or []} assigned.update((state.get("leadUsage") or {}).get("cliSessionPaths") or []) ambiguous = {path for row in workers for path in (row.get("usage") or {}).get("unattributedCliSessionPaths") or []} paths = [Path(path) for path in sorted(ambiguous - assigned)] totals = _cli_session_totals("codex", paths) if not totals: return claude_usage cli_usage = _cli_usage_block("codex", _aggregate_totals(totals), paths) blocks = [cli_usage] + ([claude_usage] if claude_usage is not None else []) combined = _aggregate_totals(blocks) for key in ("billableEquivalentTokens", "estimatedCostUsd", "cliEstimatedCostUsd"): combined[key] = sum(block.get(key, 0) or 0 for block in blocks) combined.update(source="unattributed", cliSessionPaths=[str(path) for path in paths], note="Measured session usage counted once; worker identity is ambiguous.") if claude_usage is not None: combined["sessionIds"] = claude_usage.get("sessionIds") or [] return combined def _wrapper_claude_worker(worker: dict) -> bool: """호스트가 claude 가 아닐 때 래퍼로 띄운 claude 워커.""" provider = str(worker.get("provider") or worker.get("agent") or "").strip() runner = str(worker.get("runner") or "").strip() return provider in _SESSION_JSONL_PROVIDERS and runner in {"", "cli-wrapper"} def _claude_wrapper_usage_block( *, project_root: Path, worker_id: str, state: dict, window: tuple[str | None, str | None], incremental: bool, ) -> dict: """래퍼로 띄운 claude 워커의 세션 jsonl — dispatch 가 발급한 세션 id 로 찾는다. `okstra-claude-exec.sh` 는 `claude -p --session-id ` 로 돌고 그 id 는 `workerDispatches[].sessionId` 에 적힌다(`_dispatch_record`). 트랜스크립트는 `~/.claude/projects/<루트 인코딩>/.jsonl` 에 있는데, claude 가 아닌 호스트의 수집기는 이 워커를 "세션 jsonl 공급자" 라며 건너뛰어 항상 `unavailable` 이었다(실측 2026-09-08 jobs implementation-planning r01: claude planner 5회·report-writer 2회 세션이 전부 디스크에 있었다). claude 호스트의 `collect_claude_runtime_usage` 가 같은 id 로 하는 일을 여기서 한다. """ since, until = window session_ids = usage_session_ids(state, worker_id) if not session_ids: return _claude_wrapper_status_usage( project_root, usage_dispatch_records(state, worker_id), "claude wrapper worker has no dispatch session id recorded in workerDispatches", ) totals: list[dict] = [] paths: list[Path] = [] attributed: list[str] = [] for session_id in session_ids: path = find_session_jsonl(session_id, project_root) if path is None: continue session_totals = claude_session_totals( path, since=since, until=until, incremental=incremental, ) # 창 밖 세션은 0 토큰 totals 가 되어 허위 0 으로 보고된다 — claude 호스트 # 경로와 같은 `startedAt` 검사로 거른다. if not session_totals.get("startedAt"): continue totals.append(session_totals) paths.append(path) attributed.append(session_id) if not totals: return _claude_wrapper_status_usage( project_root, usage_dispatch_records(state, worker_id), "claude session jsonl not found under " f"{claude_project_dir(project_root)} for dispatch session ids {session_ids}" ) block = usage_block(_aggregate_totals(totals), source="claude-jsonl") block["sessionIds"] = attributed block["sessionPaths"] = [str(path) for path in paths] return block def _claude_wrapper_status_usage(project_root: Path, records: list, reason: str) -> dict: paths = [] for raw in _worker_prompt_paths({}, records): prompt = _resolve_project_path(project_root, raw) if prompt is not None: paths.append(status_path_for_prompt(prompt)) totals = _cli_session_totals("claude", paths) if not totals: return na_block(reason) block = usage_block(_aggregate_totals(totals), source="claude-jsonl") block["attribution"] = "wrapper-status" block["statusPaths"] = [str(path) for path in paths] return block def _attach_cli_usage(block: dict, cli: dict) -> None: """Claude 호스트 사용량에 이미 귀속한 공급자 CLI 비용을 붙인다.""" for key in ( "cliTotalTokens", "cliEstimatedCostUsd", "cliModel", "cliSessionPaths", "cliNote", "cliExecutionStatus", ): if key in cli: block[key] = cli[key] def _collect_cli_runtime_usage( state: dict, project_root: Path, team_state_path: Path | None = None, *, incremental: bool = True, ) -> dict: state["additionalWorkerUsage"] = unrostered_usage_workers(state) windows_by_worker = _codex_worker_windows(project_root, state) run_window: tuple[str | None, str | None] = (None, None) if team_state_path is not None: run_window = resolve_run_window(team_state_path, state) for worker in accounting_workers(state): if not isinstance(worker, dict): continue worker_id = str(worker.get("workerId") or "").strip() records = [ dict(record) for record in usage_dispatch_records(state, worker_id) ] if worker_id else [] provider = _cli_assignment_provider(worker) if provider: worker["usage"] = _worker_cli_usage_block( provider=provider, project_root=project_root, worker=worker, records=records, fallback_windows=windows_by_worker.get(worker_id, []), ) elif worker_id and _wrapper_claude_worker(worker): worker["usage"] = _claude_wrapper_usage_block( project_root=project_root, worker_id=worker_id, state=state, window=run_window, incremental=incremental, ) else: worker["usage"] = na_block( "worker usage is not read from a provider CLI transcript " "(host-native or no registered CLI provider): " f"{worker.get('provider') or worker.get('agent') or worker.get('workerId')}" ) state["leadUsage"] = _cli_lead_usage(state, project_root, team_state_path) _remove_shared_codex_usage(state) _populate_usage_summary(state, team_name=resolve_team_name(state), sessions_found=0, needle_source="none") return state def _cli_lead_provider(state: dict) -> str: lead = state.get("lead") if isinstance(state.get("lead"), dict) else {} return str( lead.get("provider") or lead.get("agent") or state.get("leadRuntime") or "" ).strip() _CLI_LEAD_PROVIDERS = frozenset({"grok", "codex"}) def _cli_lead_usage( state: dict, project_root: Path, team_state_path: Path | None, ) -> dict: """호스트 리드 세션은 워커가 가져간 경로를 뺀 뒤 같은 트랜스크립트에서 읽는다.""" missing = ( "Host lead token accounting is unavailable; CLI worker usage is " "collected only from attributable provider logs." ) provider = _cli_lead_provider(state) if team_state_path is None: return na_block(missing) if provider == "antigravity": return na_block( "Antigravity host lead has no registered session transcript; " "worker stream-json logs are collected only." ) if provider not in _CLI_LEAD_PROVIDERS: return na_block(missing) run_since, run_until = resolve_run_window(team_state_path, state) if not run_since or not run_until: return na_block( f"{provider} lead usage accounting is unavailable because the run window is missing." ) worker_paths = { Path(path) for worker in accounting_workers(state) if isinstance(worker, dict) for path in ((worker.get("usage") or {}).get("cliSessionPaths") or []) } # in-session 리드(codex·grok)는 run 보다 먼저 열린 세션이다 — 창 안에서 # 시작한 세션만 보면 없다고 나오고, 세션 전체를 더하면 다른 task 의 턴이 # 섞인다. 창 안에서 활동한 세션을 후보에 넣고 토큰은 창으로 잘라 센다. window = (run_since, run_until) if provider == "codex": candidates = find_codex_sessions( project_root, run_since, run_until, active_before_start=True, ) else: candidates = find_grok_sessions( project_root, run_since, run_until, active_before_start=True, ) sessions = _select_cli_lead_sessions( provider, [path for path in candidates if path not in worker_paths], state, window=window, ) totals = _cli_session_totals(provider, sessions, window=window) if not totals: return na_block( f"{provider} lead usage accounting is unavailable because no host session " "was active in the run window." ) return _cli_usage_block(provider, _aggregate_totals(totals), sessions) def _cli_worker_session(provider: str, path: Path) -> bool: if provider == "grok": return grok_session_is_non_interactive(path) if provider == "codex": return codex_session_is_worker(path) return False def _select_cli_lead_sessions( provider: str, sessions: list[Path], state: dict, *, window: tuple[str, str] | None = None, ) -> list[Path]: """워커 래퍼 세션을 빼고, 아이디가 있으면 그 세션, 없으면 벽시계가 가장 긴 대화.""" candidates = [ path for path in sessions if not _cli_worker_session(provider, path) ] if not candidates: return [] pinned = _pinned_cli_lead_session(provider, candidates, state) if pinned is not None: return [pinned] if len(candidates) == 1: return candidates best = candidates[0] best_ms = -1 for path in candidates: totals = _cli_session_totals(provider, [path], window=window) total = totals[0] if totals else {} wall = _wall_ms(total.get("startedAt"), total.get("endedAt")) if total else None if wall is None: wall = -1 if wall > best_ms: best = path best_ms = wall return [best] def _pinned_cli_lead_session( provider: str, candidates: list[Path], state: dict, ) -> Path | None: lead = state.get("lead") if isinstance(state.get("lead"), dict) else {} wanted = {str(lead.get("sessionId") or "").strip()} if provider == "grok": wanted.add(str(os.environ.get("GROK_SESSION_ID") or "").strip()) wanted.discard("") for path in candidates: names = {path.parent.name, path.name, path.stem} if provider == "codex": names.update(codex_session_ids(path)) if wanted & names: return path if provider == "codex" and any( session_id and session_id in path.name for session_id in wanted ): return path return None def _usage_summary_totals(lead: dict, workers: list[dict], unattributed_usage: dict | None) -> dict: lead_total = lead.get("totalTokens", 0) or 0 lead_cache_read = lead.get("cacheReadTokens", 0) or 0 lead_billable = lead.get("billableEquivalentTokens", 0) or 0 lead_cost = lead.get("estimatedCostUsd") or lead.get("cliEstimatedCostUsd") or 0 worker_total = sum((w.get("usage") or {}).get("totalTokens", 0) or 0 for w in workers) worker_cache_read = sum((w.get("usage") or {}).get("cacheReadTokens", 0) or 0 for w in workers) worker_billable = sum((w.get("usage") or {}).get("billableEquivalentTokens", 0) or 0 for w in workers) worker_cost = sum((w.get("usage") or {}).get("estimatedCostUsd", 0) or 0 for w in workers) cli_cost = sum((w.get("usage") or {}).get("cliEstimatedCostUsd", 0) or 0 for w in workers) if unattributed_usage is not None: worker_total += unattributed_usage.get("totalTokens", 0) or 0 worker_cache_read += unattributed_usage.get("cacheReadTokens", 0) or 0 worker_billable += unattributed_usage.get("billableEquivalentTokens", 0) or 0 worker_cost += unattributed_usage.get("estimatedCostUsd", 0) or 0 cli_cost += unattributed_usage.get("cliEstimatedCostUsd", 0) or 0 return { "leadTotalTokens": lead_total, "workerTotalTokens": worker_total, "grandTotalTokens": lead_total + worker_total, "leadCacheReadTokens": lead_cache_read, "workerCacheReadTokens": worker_cache_read, "grandCacheReadTokens": lead_cache_read + worker_cache_read, "leadBillableEquivalentTokens": lead_billable, "workerBillableEquivalentTokens": worker_billable, "grandBillableEquivalentTokens": lead_billable + worker_billable, "estimatedCostUsd": { "lead": round(lead_cost, 4), "claudeWorkers": round(worker_cost, 4), "cliWorkers": round(cli_cost, 4), "grandTotal": round(lead_cost + worker_cost + cli_cost, 4), }, } def _unmatched_usage_models(lead: dict, workers: list[dict]) -> list[str]: unmatched_models: list[str] = [] if ( lead.get("model") and lead.get("estimatedCostUsd") is None and lead.get("cliEstimatedCostUsd") is None and (lead.get("totalTokens") or 0) > 0 ): unmatched_models.append(lead["model"]) for w in workers: u = w.get("usage") or {} # CLI 블록의 가격은 `cliEstimatedCostUsd` 에 붙는다 — 그 키를 안 보면 # 가격이 붙은 grok 도 미매칭으로 찍힌다. if ( u.get("source") not in {"codex-cli", "agy-cli"} and u.get("model") and u.get("estimatedCostUsd") is None and u.get("cliEstimatedCostUsd") is None and (u.get("totalTokens") or 0) > 0 ): unmatched_models.append(u["model"]) if u.get("cliModel") and u.get("cliEstimatedCostUsd") is None and (u.get("cliTotalTokens") or 0) > 0: unmatched_models.append(u["cliModel"]) return sorted(set(unmatched_models)) def _populate_usage_summary( state: dict, *, team_name: str, sessions_found: int, unattributed_sessions: list[str] | None = None, unattributed_usage: dict[str, Any] | None = None, needle_source: str = "observed-team-names", ) -> None: workers = accounting_workers(state) unattributed_usage = _unattributed_usage(state, unattributed_usage) lead = state.get("leadUsage") or {} state["usageSummary"] = { **_usage_summary_totals(lead, workers, unattributed_usage), "collectedAt": utc_now(), "teamName": team_name, "needleSource": needle_source, "sessionsFound": sessions_found, "unmatchedModels": _unmatched_usage_models(lead, workers), "unattributedTeamSessions": unattributed_sessions or [], "unattributedWorkerUsage": unattributed_usage, "definitions": { "totalTokens": "Sum of input + output + cache_creation tokens — the volume the session put through the model once. cache_read is excluded and reported separately as cacheReadTokens: a session re-reads its whole context from cache every turn, so folding it in here would count the same tokens once per turn.", "cacheReadTokens": "Context re-read from cache. Billed at 0.1x base input, so it lands in billableEquivalentTokens and in the cost even though it is not part of totalTokens.", "billableEquivalentTokens": "Tokens normalized to base-input-price units (cache_creation_5m x1.25, cache_creation_1h x2.0, cache_read x0.1, output x5). 5m vs 1h is split from usage.cache_creation when the API breakdown is present; otherwise all cache_creation falls into 5m.", "estimatedCostUsd": "USD cost using public list pricing for the model recorded in the session. cliWorkers covers attributable registered-provider CLI calls.", }, } def _claude_sessions_for_run( state: dict, team_state_path: Path, cwd: Path, run_since: str | None, run_until: str | None, incremental: bool, ) -> tuple: team_name = resolve_team_name(state) lead_sid = (state.get("lead") or {}).get("sessionId") # 1) Claude sessions (lead + claude-side workers). Cache totals at scan # time so we don't re-read the jsonl when a worker matches multiple # sessions. needle 집합(observedTeamNames 여러 세대)으로 스캔하고, 없으면 # 워커 jsonl 에서 복구한다 — team_name 은 표시 라벨 전용. team_needles, needle_source = resolve_team_needles_with_source( state, cwd, run_since, run_until ) claude_sessions = find_claude_team_sessions(cwd, team_needles, lead_sid, incremental=incremental) # Task 9: 축1 기록(observed-team-names)으로 발견한 세션들의 min first-ts 로 # run_since 를 완화한다. 불확실한 reconstruct/none source 는 완화하지 # 않는다(오귀속 방지). 하한은 `resolve_run_window` 와 같다 — 발견 세션에는 # 리드 세션 자신이 들어 있고, in-session 리드의 세션 첫 ts 는 run 이 아니라 # 세션의 탄생일이다(관측: 16일 전). 하한 없이 두던 동안 리드 소요 시간과 # 비용이 세션 수명 전체로 보고됐다. if needle_source == "observed-team-names": floor = run_window_relaxation_floor(team_state_path) if not window_start_is_pinned(run_since, floor): found_earliest = None for path in claude_sessions.values(): ts = _session_first_ts(path) if ts and (found_earliest is None or ts < found_earliest): found_earliest = ts run_since = relax_window_start(run_since, found_earliest, floor) return team_name, lead_sid, claude_sessions, run_since, needle_source def _claude_session_evidence( state: dict, cwd: Path, claude_sessions: dict, lead_sid: str | None, run_since: str | None, run_until: str | None, incremental: bool, ) -> tuple: by_agent: dict[str, list[tuple[str, Path, dict]]] = {} lead_path: Path | None = None # Team-tagged non-lead sessions that carry no agentName. These are almost # always a worker dispatched without the Agent `name` arg (so the harness # recorded no agentName) — the session exists and is team-tagged, but there # is nothing to match it to a workerId by. Surfacing them in usageSummary # gives the "unavailable" worker a visible cause instead of vanishing # silently (observed in dev-9692 error-analysis: claude/codex workers # dispatched without `name` → both unavailable, report-writer named → fine). unattributed_sessions: list[str] = [] unattributed_totals: list[dict] = [] for sid, path in claude_sessions.items(): if sid == lead_sid: lead_path = path continue totals = claude_session_totals(path, since=run_since, until=run_until, incremental=incremental) agent = totals.get("agentName") if agent: by_agent.setdefault(agent, []).append((sid, path, totals)) else: unattributed_sessions.append(sid) unattributed_totals.append(totals) # implicit-team run (CC v2.1.178+: no TeamCreate, status "implicit") or # legacy no-team run (skipped/concurrent-run, error fallback) — the implicit # team tags worker jsonls with the harness team name (`session-`), # NOT okstra's `teamName` label, so the needle scan above only finds the lead. # agentName 기반 발견으로 worker 세션을 보강한다. run 윈도우 밖 세션(in-window # 이벤트 없음 = startedAt 부재)은 같은 agentName 의 타 run 세션이므로 버린다. team_create_status = str((state.get("teamCreate") or {}).get("status", "")).strip() if team_create_status in ("implicit", "skipped", "error"): worker_prefix_pool = [ prefix for w in accounting_workers(state) for prefix in match_prefixes(w.get("workerId") or "") ] subagent_parent_sids = [lead_sid] if lead_sid else [] agent_sessions = find_claude_agent_sessions( cwd, worker_prefix_pool, incremental=incremental, subagent_parent_sids=subagent_parent_sids, ) for sid, path in agent_sessions.items(): if sid == lead_sid or sid in claude_sessions: continue totals = claude_session_totals(path, since=run_since, until=run_until, incremental=incremental) if not totals.get("startedAt"): continue agent = totals.get("agentName") if agent: by_agent.setdefault(agent, []).append((sid, path, totals)) return by_agent, lead_path, unattributed_sessions, unattributed_totals def _attach_claude_lead_usage( state: dict, cwd: Path, lead_path: Path | None, lead_sid: str | None, run_since: str | None, run_until: str | None, incremental: bool, ) -> None: # Lead. if lead_path is not None: totals = claude_session_totals(lead_path, since=run_since, until=run_until, incremental=incremental) state["leadUsage"] = usage_block(totals, source="claude-jsonl") state["leadUsage"]["sessionId"] = lead_sid state["phaseTimeline"] = phase_timeline(totals.get("progressMarkers") or []) else: state["leadUsage"] = na_block( f"lead session jsonl not found under {claude_project_dir(cwd)} (sessionId={lead_sid})" ) def _matching_claude_worker_sessions( worker: dict, state: dict, cwd: Path, by_agent: dict, run_since: str | None, run_until: str | None, incremental: bool, ) -> tuple: sessions_dir = claude_project_dir(cwd) worker_id = worker.get("workerId") prefixes = match_prefixes(worker_id) if worker_id else [] # pane 워커는 별도 `claude -p` 프로세스라 jsonl 에 agentName 도 teamName # 도 안 남긴다 — 아래 prefix 경로로는 영원히 매칭되지 않는다. dispatch 가 # 발급해 적어 둔 이 id 가 그 세션을 짚는 유일한 결정적 단서다. 재시도는 # attempt 마다 새 세션이므로 기록된 id 를 전부 합산한다. # 헬퍼에서 worker_id=None 은 "run 전체"라, 이름 없는 워커에 그대로 넘기면 # 그 워커가 run 의 모든 세션을 흡수한다. dispatched_sids = usage_session_ids(state, worker_id) if worker_id else [] matched: list[tuple[str, Path, dict]] = [] matched_sids: set[str] = set() for sid in dispatched_sids: path = sessions_dir / f"{sid}.jsonl" if not path.is_file(): continue totals = claude_session_totals(path, since=run_since, until=run_until, incremental=incremental) # 창 밖 세션은 여기서 버린다 — agentName 발견 경로가 위에서 같은 # `startedAt` 검사로 거르는 것과 같은 이유다. 넣으면 0 토큰 totals 가 # usage_block 을 타고 "이 워커는 0 을 썼다"로 보고되어, unavailable 로 # 남아야 할 상태를 허위 0 이 덮는다. if not totals.get("startedAt"): continue matched.append((sid, path, totals)) matched_sids.add(sid) # 조건부 폴백이 아니라 합집합이다. 한 워커의 attempt 1 이 pane(sid 로만 # 찾힌다), attempt 2 가 in-process 서브에이전트(agentName 으로만 찾힌다)일 # 수 있고, `if not matched:` 로 두면 sid 가 하나라도 맞는 순간 attempt 2 의 # 토큰이 통째로 누락된다. 양쪽에서 온 같은 세션은 sid 로 한 번만 센다. for agent_name, entries in by_agent.items(): if not agent_matches(agent_name, prefixes): continue for entry in entries: if entry[0] in matched_sids: continue # team-needle 경로는 창 검사 없이 by_agent 를 채운다. 같은 리드 # 세션의 다음 run 이 띄운 워커도 같은 needle 에 걸리는데, 창 밖이라 # startedAt 이 없어 아래 정렬에서 맨 앞에 서고 `sessionId` 를 # 차지한다 — 토큰은 0 이라 합계는 안 틀리고 리포트가 가리키는 # 세션만 남의 것이 된다. if not entry[2].get("startedAt"): continue matched.append(entry) matched_sids.add(entry[0]) return matched, dispatched_sids, prefixes def _collect_claude_workers( state: dict, cwd: Path, by_agent: dict, run_since: str | None, run_until: str | None, incremental: bool, ) -> set[str]: # Workers — dispatch 가 기록한 세션 id 로 짚은 세션과 agentName prefix 로 # 찾은 세션의 합집합을 합산한다(재배치 `-002`, convergence `-reverify-r1`, # implementation `-executor`, report-writer `-impl` / `-2` 등). # sid 로 귀속한 세션 id 전체 — 아래 unattributed 폴드에서 빼는 데 쓴다. attributed_sids: set[str] = set() cli_windows_by_worker = _codex_worker_windows(cwd, state) for worker in accounting_workers(state): worker_id = worker.get("workerId") matched, dispatched_sids, prefixes = _matching_claude_worker_sessions( worker, state, cwd, by_agent, run_since, run_until, incremental, ) attributed_sids.update(sid for sid, _path, _totals in matched) if not matched: provider = _cli_assignment_provider(worker) if provider and str(worker.get("runner") or "") == "cli-wrapper": worker["usage"] = _worker_cli_usage_block( provider=provider, project_root=cwd, worker=worker, records=[dict(record) for record in usage_dispatch_records(state, worker_id)] if worker_id else [], fallback_windows=cli_windows_by_worker.get(worker_id, []), ) else: worker["usage"] = _claude_wrapper_status_usage( cwd, usage_dispatch_records(state, worker_id) if worker_id else [], "no Claude session jsonl in the run window for dispatched " f"sessionIds {dispatched_sids} or agentName prefixes {prefixes}", ) continue # Stable order by startedAt so the "primary" session is the first one. matched.sort(key=lambda x: x[2].get("startedAt") or "") primary_sid, _primary_path, _primary_totals = matched[0] aggregate = _aggregate_totals([t for _, _, t in matched]) block = usage_block(aggregate, source="claude-jsonl") block["sessionId"] = primary_sid if len(matched) > 1: block["additionalSessionIds"] = [sid for sid, _, _ in matched[1:]] block["matchedAgentNames"] = sorted({t.get("agentName") for _, _, t in matched if t.get("agentName")}) # For CLI-wrapper workers, retain execution evidence and attach any # attributable provider transcript usage without treating absence as zero. provider = _cli_assignment_provider(worker) if provider: cli = _worker_cli_usage_block( provider=provider, project_root=cwd, worker=worker, records=[dict(record) for record in usage_dispatch_records(state, worker_id)], fallback_windows=cli_windows_by_worker.get(worker_id, []), ) _attach_cli_usage(block, cli) worker["usage"] = block return attributed_sids def _unattributed_claude_usage( unattributed_sessions: list[str], unattributed_totals: list[dict], attributed_sids: set[str], ) -> tuple[list[str], dict | None]: # Fold team-tagged worker sessions that carry no agentName into the worker # pool. They cannot be mapped to a specific workerId (so each named worker # row above stays `unavailable`), but the tokens are real team-worker spend — # most often an in-process teammate whose work is commingled in a team-tagged # session the harness never tagged with `name`. Without this, the run-level # Worker total reads 0 and the report validator hard-fails a legitimate run. # Attribution is aggregate, not per-worker; usageSummary records it openly. # # 먼저 sid 로 귀속된 세션을 뺀다. 이 두 집합은 `agentName` 유무로 배타적이었지만 # (by_agent 는 있어야 들어가고 unattributed 는 없어야 들어간다) sid 경로는 # agentName 을 안 보므로 그 배타성이 더는 성립하지 않는다. team needle 에 # 걸리면서 agentName 이 없는 세션이 동시에 기록된 dispatch sid 이기도 하면, # 빼지 않을 경우 그 토큰이 워커 usage 와 이 폴드 양쪽에 들어가 `_populate_usage_summary` # 의 workerTotalTokens 를 부풀린다. 두 리스트는 인덱스가 대응하므로 함께 거른다. kept = [ (sid, totals) for sid, totals in zip(unattributed_sessions, unattributed_totals) if sid not in attributed_sids ] unattributed_sessions = [sid for sid, _ in kept] unattributed_totals = [totals for _, totals in kept] unattributed_usage = None if unattributed_totals: unattributed_usage = usage_block( _aggregate_totals(unattributed_totals), source="claude-jsonl" ) unattributed_usage["sessionIds"] = unattributed_sessions unattributed_usage["note"] = ( "Team-tagged worker session(s) with no agentName (dispatched without " "the Agent `name` arg, or an in-process teammate commingled with the " "lead). Folded into the worker pool as an aggregate because they " "cannot be mapped to a specific workerId." ) return unattributed_sessions, unattributed_usage def collect_claude_runtime_usage( team_state_path: Path, project_root: Path | None = None, *, incremental: bool = True, ) -> dict: state = json.loads(team_state_path.read_text()) state["additionalWorkerUsage"] = unrostered_usage_workers(state) cwd = project_root or _infer_project_root(team_state_path, state) run_since, run_until = resolve_run_window(team_state_path, state) team_name, lead_sid, sessions, run_since, needle_source = _claude_sessions_for_run( state, team_state_path, cwd, run_since, run_until, incremental, ) by_agent, lead_path, unattributed_sessions, unattributed_totals = _claude_session_evidence( state, cwd, sessions, lead_sid, run_since, run_until, incremental, ) _attach_claude_lead_usage(state, cwd, lead_path, lead_sid, run_since, run_until, incremental) attributed = _collect_claude_workers(state, cwd, by_agent, run_since, run_until, incremental) unattributed_sessions, unattributed_usage = _unattributed_claude_usage( unattributed_sessions, unattributed_totals, attributed, ) _remove_shared_codex_usage(state) _populate_usage_summary( state, team_name=team_name, sessions_found=len(sessions), unattributed_sessions=unattributed_sessions, unattributed_usage=unattributed_usage, needle_source=needle_source, ) return state def collect_cli_runtime_usage( team_state_path: Path, project_root: Path | None = None, *, incremental: bool = True, ) -> dict: state = json.loads(team_state_path.read_text()) cwd = project_root or _infer_project_root(team_state_path, state) return _collect_cli_runtime_usage( state, cwd, team_state_path, incremental=incremental, ) def collect( team_state_path: Path, project_root: Path | None = None, *, incremental: bool = True, ) -> dict: from okstra_ctl.application.collect_usage import collect_usage from okstra_ctl.ports.usage_accounting import UsageRequest from okstra_ctl.registry.host_registry import default_host_registry state = json.loads(team_state_path.read_text()) cwd = project_root or _infer_project_root(team_state_path, state) adapter = default_host_registry().resolve(str(state.get("leadRuntime") or "")) request = UsageRequest(team_state_path, cwd, incremental) return collect_usage(request, adapter.usage_accounting()).payload def _infer_project_root(team_state_path: Path, state: dict) -> Path: rel = state.get("runDirectoryPath") or "" p = team_state_path.resolve().parent while p != p.parent: if rel and (p / rel).is_dir(): return p if (p / OKSTRA_RELATIVE).is_dir(): return p p = p.parent raise SystemExit(f"could not infer project root from {team_state_path}") def reconstruct_needles_from_workers(cwd, since, until, projects_root=None) -> list[str]: """observedTeamNames 가 없는 과거 run 의 best-effort 복구: run window 안 워커 jsonl(agentName 존재)이 달고 있는 session-* teamName variant 집합. """ from okstra_token_usage.paths import claude_project_dir, ts_in_window proj_dir = claude_project_dir(cwd, projects_root) found: set[str] = set() if not proj_dir.is_dir(): return [] for p in proj_dir.glob("*.jsonl"): agent = team = first_ts = None try: with p.open(encoding="utf-8") as fh: for raw in fh: try: rec = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError): continue if agent is None and rec.get("agentName"): agent = rec["agentName"] if team is None and rec.get("teamName"): team = rec["teamName"] ts = rec.get("timestamp") if ts and (first_ts is None or ts < first_ts): first_ts = ts except OSError: continue # 의도적 divergence: 이 untrusted best-effort 경로에서는 ts 를 못 구한 # 워커를 제외한다 (ts_in_window 의 파싱 실패→포함 정책과 반대). if agent and team and str(team).startswith("session-") \ and first_ts and ts_in_window(first_ts, since, until): found.add(team) return sorted(found)