"""Top-level orchestrator that walks team-state and gathers all worker usage."""
from __future__ import annotations

import json
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 na_block, usage_block
from .claude import (
    claude_session_totals,
    find_claude_agent_sessions,
    find_claude_team_sessions,
)
from .codex import (
    codex_session_ids,
    codex_session_is_worker,
    codex_session_total,
    find_codex_sessions,
)
from .antigravity import antigravity_session_total, find_antigravity_sessions
from .grok import (
    find_grok_sessions,
    grok_session_is_non_interactive,
    grok_session_total,
)
from .paths import claude_project_dir, utc_now
from .pricing import antigravity_cost_usd, provider_cost_usd
from okstra_ctl.dispatch_state import worker_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 `<prefix>-<suffix>` — as belonging
    to this worker so its tokens get aggregated. For implementation /
    final-verification runs the role variants `<provider>-executor` and
    `<provider>-verifier` are also attributed back to the matching provider
    worker (the bare `<provider>` prefix's `<prefix>-<suffix>` 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-<task-type>-<seq>.json`` → ``<task-type>-<seq>``.

    이 접미사로 *같은 run* 의 run-manifest / status 를 정확히 짚는다. task 디렉토리
    한 곳에 여러 run(재시도·이전 phase·레거시 타임스탬프)의 산출물이 섞여 있어,
    glob 으로 아무거나 집으면 엉뚱한 run 의 시각을 쓰게 된다(관측: 가장 오래된
    레거시 manifest 의 createdAt 을 집어 윈도우가 한 달로 벌어짐)."""
    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 _run_manifest_created_at(run_dir: Path, suffix: str) -> str | None:
    p = run_dir / "manifests" / f"run-manifest-{suffix}.json"
    try:
        return json.loads(p.read_text()).get("createdAt")
    except (OSError, json.JSONDecodeError):
        return 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(`<root>/.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, json.JSONDecodeError):
        pass
    return _run_end_estimate(run_dir, prev) or _run_manifest_created_at(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 → 현재 시각(아직 진행 중) 순으로 해소한다. 접미사를 못 뽑으면
    (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 이 이 윈도우에 통째로 들어온다.

    `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
    since = _run_manifest_created_at(run_dir, suffix)
    if relax_start:
        earliest = _earliest_lead_session_ts(state, team_state_path)
        if earliest and (since is None or earliest < since):
            since = earliest
        floor = _previous_run_end(run_dir, suffix)
        if floor and since and since < floor:
            since = floor
    until = state.get("runEndedAt") or _run_end_estimate(run_dir, suffix) 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>` 관례로 폴백 — 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-<leadSid-prefix>`) 전체.

    lead 재발급마다 세대가 갈리므로 team-state.teamName 단일값으로는 후속
    세대 워커를 놓친다. 축1 이 기록한 `observedTeamNames[]` 를 정본으로 쓰고,
    수동 정정으로 teamName 에 직접 박힌 `session-*` 값도 흡수한다.
    audit label(`okstra-<task-key>`) 은 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-<task-key>`)은 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 state.get("workers", [])
        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


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()
    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]) -> list[dict]:
    totals = []
    for session_path in session_paths:
        if provider == "codex":
            total = codex_session_total(session_path)
        elif provider == "antigravity":
            total = antigravity_session_total(session_path)
        elif provider == "grok":
            total = grok_session_total(session_path)
        else:
            continue
        if total.get("available"):
            totals.append(total)
    return totals


def _cli_usage_block(provider: str, totals: dict, session_paths: list[Path]) -> dict:
    block = usage_block(totals, source=f"{provider}-cli")
    block["cliTotalTokens"] = totals.get("totalTokens", 0) or 0
    block["cliSessionPaths"] = [str(path) for path in session_paths]
    if totals.get("model"):
        block["model"] = totals["model"]
        block["cliModel"] = totals["model"]
    if provider in {"codex", "grok", "kimi"}:
        cost = provider_cost_usd(
            provider,
            totals.get("model"),
            totals.get("inputTokens", 0) or 0,
            totals.get("cachedInputTokens", 0) or 0,
            totals.get("outputTokens", 0) or 0,
        )
    else:
        cost = antigravity_cost_usd(
            totals.get("model"),
            totals.get("inputTokens", 0) or 0,
            totals.get("outputTokens", 0) or 0,
        )
    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 _worker_cli_windows(
    project_root: Path,
    worker: dict,
    fallback_windows: list[tuple[str, str]],
) -> tuple[list[tuple[str, str]], Path | None, bool]:
    prompt_path = _resolve_project_path(project_root, str(worker.get("promptPath") or ""))
    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")
    ended_at = execution.get("endedAt")
    if started_at:
        return [(started_at, ended_at or utc_now())], status_path, False
    return fallback_windows, status_path, True


def _antigravity_log_paths(
    project_root: Path,
    worker: dict,
    status_path: Path | None,
) -> list[Path]:
    """agy stream-json 은 워커 로그에 남는다. 홈 세션 디렉터리가 아니다."""
    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,
    fallback_windows: list[tuple[str, str]],
) -> dict:
    """공급자 프로세스 트랜스크립트와 실행 증거로 워커 사용량 블록을 만든다."""
    windows, status_path, used_fallback = _worker_cli_windows(
        project_root,
        worker,
        fallback_windows,
    )
    session_paths = _cli_sessions_for_windows(provider, project_root, windows)
    if provider == "antigravity":
        for log_path in _antigravity_log_paths(project_root, worker, status_path):
            if log_path not in session_paths:
                session_paths.append(log_path)
    return collect_cli_usage(
        provider=provider,
        status_path=status_path,
        sessions=session_paths,
        fallback_window_used=used_fallback,
    )


def _attach_cli_usage(
    block: dict,
    provider: str,
    project_root: Path,
    windows: list[tuple[str, str]],
    status_path: Path | None,
    fallback_window_used: bool,
) -> None:
    """Layer aggregated CLI token/cost onto a Claude-side worker ``block``.

    A CLI-wrapper worker re-dispatched within one aggregated wrapper
    window produces several rollout jsonls; pricing only the latest (the old
    ``find_codex_session`` -> ``sessions[-1]`` behavior) dropped the earlier
    attempts' tokens. We sum every in-window session here so the redispatched
    CLI spend is fully reported. antigravity 는 워커 로그의 stream-json
    ``result.usage`` 를 같은 합산 경로에 붙인다.
    """
    session_paths = _cli_sessions_for_windows(provider, project_root, windows)
    cli = collect_cli_usage(
        provider=provider,
        status_path=status_path,
        sessions=session_paths,
        fallback_window_used=fallback_window_used,
    )
    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,
) -> dict:
    windows_by_worker = _codex_worker_windows(project_root, state)
    for worker in state.get("workers", []):
        if not isinstance(worker, dict):
            continue
        provider = _cli_assignment_provider(worker)
        if not provider:
            worker["usage"] = na_block(
                "worker is host-native or has no registered CLI provider: "
                f"{worker.get('provider') or worker.get('agent') or worker.get('workerId')}"
            )
            continue
        worker_id = str(worker.get("workerId") or "").strip()
        worker["usage"] = _worker_cli_usage_block(
            provider=provider,
            project_root=project_root,
            worker=worker,
            fallback_windows=windows_by_worker.get(worker_id, []),
        )
    state["leadUsage"] = _cli_lead_usage(state, project_root, team_state_path)
    _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 state.get("workers") or []
        if isinstance(worker, dict)
        for path in ((worker.get("usage") or {}).get("cliSessionPaths") or [])
    }
    sessions = _select_cli_lead_sessions(
        provider,
        [
            path
            for path in _cli_sessions_for_windows(
                provider, project_root, [(run_since, run_until)],
            )
            if path not in worker_paths
        ],
        state,
    )
    totals = _cli_session_totals(provider, sessions)
    if not totals:
        return na_block(
            f"{provider} lead usage accounting is unavailable because no host session "
            "started 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,
) -> 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])
        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 _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 = state.get("workers", [])
    lead = state.get("leadUsage") or {}
    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

    unmatched_models: list[str] = []
    if lead.get("model") and lead.get("estimatedCostUsd") is None and (lead.get("totalTokens") or 0) > 0:
        unmatched_models.append(lead["model"])
    for w in workers:
        u = w.get("usage") or {}
        if (
            u.get("source") not in {"codex-cli", "agy-cli"}
            and u.get("model")
            and u.get("estimatedCostUsd") 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"])
    state["usageSummary"] = {
        "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),
        },
        "collectedAt": utc_now(),
        "teamName": team_name,
        "needleSource": needle_source,
        "sessionsFound": sessions_found,
        "unmatchedModels": sorted(set(unmatched_models)),
        "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 collect_claude_runtime_usage(
    team_state_path: Path,
    project_root: Path | None = None,
    *,
    incremental: bool = True,
) -> dict:
    # incremental: 세션 jsonl 스캔에 byte cursor 캐시 사용 (P6). 캐시는 윈도우
    # 적용 전 이벤트를 저장하므로 결과는 전체 스캔과 동일 — False 는 캐시 경로를
    # 완전히 우회하는 정확성 폴백(CLI --no-cache).
    state = json.loads(team_state_path.read_text())
    cwd = project_root or _infer_project_root(team_state_path, state)
    run_since, run_until = resolve_run_window(team_state_path, state)
    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 를 완화한다. session-* needle 은 세션 고유라 발견 세션은 이 run
    # 것이므로 안전. 불확실한 reconstruct/none source 는 완화하지 않는다(오귀속 방지).
    if needle_source == "observed-team-names":
        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
        if found_earliest and (run_since is None or found_earliest < run_since):
            run_since = found_earliest
    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-<leadSid>`),
    # 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 state.get("workers", [])
            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))

    # 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})"
        )

    # Workers — dispatch 가 기록한 세션 id 로 짚은 세션과 agentName prefix 로
    # 찾은 세션의 합집합을 합산한다(재배치 `-002`, convergence `-reverify-r1`,
    # implementation `-executor`, report-writer `-impl` / `-2` 등).
    sessions_dir = claude_project_dir(cwd)
    # sid 로 귀속한 세션 id 전체 — 아래 unattributed 폴드에서 빼는 데 쓴다.
    attributed_sids: set[str] = set()
    cli_windows_by_worker = _codex_worker_windows(cwd, state)
    for worker in state.get("workers", []):
        worker_id = worker.get("workerId")
        agent = worker.get("agent")
        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 = worker_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)
            attributed_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])

        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,
                    fallback_windows=cli_windows_by_worker.get(
                        str(worker_id or "").strip(), []
                    ),
                )
            else:
                worker["usage"] = na_block(
                    "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:
            windows, status_path, used_fallback = _worker_cli_windows(
                cwd,
                worker,
                [(aggregate.get("startedAt") or "", aggregate.get("endedAt") or "")],
            )
            _attach_cli_usage(
                block,
                provider,
                cwd,
                windows,
                status_path,
                used_fallback,
            )
        worker["usage"] = block

    # 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."
        )

    _populate_usage_summary(
        state,
        team_name=team_name,
        sessions_found=len(claude_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)


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)