"""Codex CLI rollout collectors.""" from __future__ import annotations import json import os import re from datetime import datetime, timezone from pathlib import Path from .jsonl_io import iter_jsonl from .paths import CODEX_SESSIONS, codex_session_roots, ts_in_window _DEFAULT_CODEX_SESSIONS = CODEX_SESSIONS def codex_wrapper_session_ids(log_path: Path) -> set[str]: """사용자 프롬프트 앞 실행기 헤더에서만 세션 ID를 읽는다.""" try: with log_path.open(encoding="utf-8", errors="replace") as stream: for line in stream: clean = re.sub(r"\x1b\[[0-9;]*m", "", line).strip() if clean == "user" or clean.endswith("[worker]user"): break match = re.search(r"(?:^|\[worker\])session id:\s*([\w-]+)\s*$", clean) if match: return {match.group(1)} except OSError: return set() return set() def codex_session_total(jsonl_path: Path) -> dict: """Return last token_count snapshot from a codex rollout jsonl.""" last: dict | None = None cwd_val: str | None = None model_val: str | None = None started: str | None = None ended: str | None = None for rec in iter_jsonl(jsonl_path): if rec.get("type") == "session_meta": payload = rec.get("payload") or {} cwd_val = payload.get("cwd") started = payload.get("timestamp") elif rec.get("type") == "turn_context": payload = rec.get("payload") or {} if model_val is None and payload.get("model"): model_val = payload["model"] elif rec.get("type") == "event_msg": payload = rec.get("payload") or {} if payload.get("type") == "token_count": last = payload ended = rec.get("timestamp") if last is None: return {"totalTokens": 0, "cwd": cwd_val, "model": model_val, "available": False} info = last.get("info") or {} total = info.get("total_token_usage") or {} return { "totalTokens": total.get("total_tokens", 0) or 0, "inputTokens": total.get("input_tokens", 0) or 0, "cachedInputTokens": total.get("cached_input_tokens", 0) or 0, "outputTokens": total.get("output_tokens", 0) or 0, "reasoningOutputTokens": total.get("reasoning_output_tokens", 0) or 0, "cwd": cwd_val, "model": model_val, "startedAt": started, "endedAt": ended, "available": True, } def codex_session_window_total(jsonl_path: Path, since: str, until: str) -> dict: """창 안에서 이 세션이 쓴 토큰 — token_count 이벤트의 `last_token_usage` 합. in-session 리드(`entryMode: current-session`)의 rollout 은 run 보다 먼저 열려 다른 task 도 돌린 세션이라 마지막 `total_token_usage` 스냅샷은 세션 전체다. 누적치는 이벤트마다 `last_token_usage` 만큼 늘고 컨텍스트 압축 뒤 되돌아가므로(실측 2026-09-08 jobs 리드 세션: 601 이벤트 중 2회 감소), 창 끝·시작 누적치의 차가 아니라 창 안 이벤트의 `last_token_usage` 를 더한다. `last_token_usage` 가 없는 옛 기록은 직전 누적치와의 차(0 이상)로 센다. 창 안 이벤트가 없으면 `available: False`. """ keys = ( ("totalTokens", "total_tokens"), ("inputTokens", "input_tokens"), ("cachedInputTokens", "cached_input_tokens"), ("outputTokens", "output_tokens"), ("reasoningOutputTokens", "reasoning_output_tokens"), ) sums = {name: 0 for name, _raw in keys} previous: dict = {} model_val: str | None = None cwd_val: str | None = None started: str | None = None first: str | None = None last: str | None = None for rec in iter_jsonl(jsonl_path): kind = rec.get("type") payload = rec.get("payload") or {} if kind == "session_meta": cwd_val = payload.get("cwd") started = payload.get("timestamp") elif kind == "turn_context": if model_val is None and payload.get("model"): model_val = payload["model"] elif kind == "event_msg" and payload.get("type") == "token_count": info = payload.get("info") or {} cumulative = info.get("total_token_usage") or {} turn = info.get("last_token_usage") timestamp = str(rec.get("timestamp") or "") if timestamp and ts_in_window(timestamp, since, until): for name, raw in keys: if isinstance(turn, dict): sums[name] += turn.get(raw, 0) or 0 else: sums[name] += max( 0, (cumulative.get(raw, 0) or 0) - (previous.get(raw, 0) or 0) ) if first is None: first = timestamp last = timestamp previous = cumulative if first is None: return {"totalTokens": 0, "cwd": cwd_val, "model": model_val, "available": False} # 창 안에서 열린 세션은 세션 시작이 곧 창 안 활동의 시작이다. 창보다 먼저 # 열린 세션(in-session 리드)은 창 안 첫 이벤트부터 센다. if started and ts_in_window(started, since, until): first = started return {**sums, "cwd": cwd_val, "model": model_val, "startedAt": first, "endedAt": last, "available": True} def find_codex_session(cwd: Path, started_at: str, ended_at: str) -> Path | None: """Find the latest codex rollout jsonl in the requested window.""" sessions = find_codex_sessions(cwd, started_at, ended_at) return sessions[-1] if sessions else None def _session_meta_payload(path: Path) -> dict | None: try: with path.open() as fh: first = fh.readline() except OSError: return None if not first: return None try: record = json.loads(first) except json.JSONDecodeError: return None if record.get("type") != "session_meta": return None payload = record.get("payload") if not isinstance(payload, dict): payload = {} timestamp = payload.get("timestamp") or record.get("timestamp") or "" return {**payload, "timestamp": timestamp} def _session_metadata(path: Path) -> tuple[str, str] | None: payload = _session_meta_payload(path) if payload is None: return None return str(payload.get("cwd") or ""), str(payload.get("timestamp") or "") def codex_session_is_worker(path: Path) -> bool: """exec 래퍼 세션. 대화형 리드는 originator=codex-tui / source=cli 이다.""" payload = _session_meta_payload(path) if not payload: return False originator = str(payload.get("originator") or "").strip() source = str(payload.get("source") or "").strip() return originator == "codex_exec" or source == "exec" def codex_session_ids(path: Path) -> set[str]: ids = {path.name, path.stem} payload = _session_meta_payload(path) or {} for key in ("session_id", "id"): value = str(payload.get(key) or "").strip() if value: ids.add(value) return {item for item in ids if item} def _modified_in_window(path: Path, started_at: str, until: str) -> bool: try: mtime = path.stat().st_mtime except OSError: return False modified = datetime.fromtimestamp(mtime, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") return ts_in_window(modified, started_at, None) def find_codex_sessions( cwd: Path, started_at: str, ended_at: str, *, session_roots: tuple[Path, ...] | None = None, active_before_start: bool = False, ) -> list[Path]: """Find codex rollout jsonls whose meta.cwd matches the window. 기본은 창 안에서 *시작한* 세션이다 — exec 래퍼 워커는 dispatch 마다 새 rollout 을 열므로 그것으로 충분하다. `active_before_start=True` 는 창보다 먼저 시작했지만 창 안에서도 기록이 이어진 세션(파일 mtime 이 창 시작 이후)을 더한다. in-session 리드가 그 모양이다: 세션은 run 보다 먼저 태어났으니 시작 시각으로 거르면 `no host session started in the run window` 로 빠지고, 토큰은 `codex_session_window_total` 이 창으로 잘라 센다. """ if not started_at or not ended_at: return [] if session_roots is None: if CODEX_SESSIONS != _DEFAULT_CODEX_SESSIONS: session_roots = (CODEX_SESSIONS,) else: session_roots = codex_session_roots(Path.home(), os.environ) target_cwd = str(cwd) candidates: list[tuple[str, Path]] = [] seen: set[Path] = set() for root in session_roots: if not root.is_dir(): continue for p in root.rglob("rollout-*.jsonl"): identity = p.resolve() if identity in seen: continue seen.add(identity) metadata = _session_metadata(p) if metadata is None: continue session_cwd, ts = metadata if session_cwd != target_cwd: continue if not ts_in_window(ts, started_at, ended_at): if not ( active_before_start and ts_in_window(ts, None, started_at) and _modified_in_window(p, started_at, ended_at) ): continue candidates.append((ts, p)) if not candidates: return [] candidates.sort() return [path for _ts, path in candidates]