"""Grok Build session collectors.

공식 세션 문서는 대화를 ``~/.grok/sessions/`` 에 둔다. cwd 는 퍼센트 인코딩된
디렉터리 이름이다. ``updates.jsonl`` 의 ``params.update.usage`` 행 하나는
**프롬프트 한 번**의 사용량이다 — ``numTurns`` 는 그 프롬프트 안의 모델 호출
수이고 ``inputTokens`` 는 매번 컨텍스트 전체라 값이 오르내린다(실측
2026-09-08, 대화형 세션 3개 39행: 83k → 798k → 406k → 187k …). 세션 합계는
행의 합이고, 마지막 행은 마지막 프롬프트일 뿐이다. exec 래퍼 워커는 프롬프트가
하나라 행도 하나다.
"""
from __future__ import annotations

import json
import os
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping
from urllib.parse import unquote

from .jsonl_io import iter_jsonl
from .paths import ts_in_window


def grok_sessions_root(
    home: Path | None = None,
    env: Mapping[str, str] | None = None,
) -> Path:
    environ = env if env is not None else os.environ
    configured = str(environ.get("GROK_HOME") or "").strip()
    base = Path(configured).expanduser() if configured else (home or Path.home()) / ".grok"
    return base / "sessions"


def _iso_from_unix(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 _usage_payload(record: dict) -> dict | None:
    params = record.get("params")
    if not isinstance(params, dict):
        return None
    update = params.get("update")
    if not isinstance(update, dict):
        return None
    usage = update.get("usage")
    return usage if isinstance(usage, dict) else None


def _model_snapshot(usage: dict) -> tuple[dict, str | None]:
    models = usage.get("modelUsage")
    if isinstance(models, dict) and len(models) == 1:
        name, payload = next(iter(models.items()))
        if isinstance(payload, dict):
            return payload, name if isinstance(name, str) else None
    return usage, None


_SUM_KEYS = (
    ("totalTokens", "totalTokens"),
    ("inputTokens", "inputTokens"),
    ("outputTokens", "outputTokens"),
    ("cachedInputTokens", "cachedReadTokens"),
    ("reasoningOutputTokens", "reasoningTokens"),
    ("durationMs", "apiDurationMs"),
)


def grok_session_window_total(
    updates_path: Path, since: str | None = None, until: str | None = None,
) -> dict:
    """창 안 usage 행의 합. 창이 없으면 세션 전체.

    한 행이 프롬프트 한 번의 사용량이므로 더한다 — 마지막 행만 읽으면 여러
    프롬프트를 돌린 세션(in-session 리드)은 마지막 프롬프트만 남는다. 창은
    run 보다 먼저 열린 리드 세션에서 다른 task 의 프롬프트를 걸러 낸다.
    """
    sums = {name: 0 for name, _raw in _SUM_KEYS}
    model: str | None = None
    started: str | None = None
    ended: str | None = None
    counted = 0
    for record in iter_jsonl(updates_path):
        usage = _usage_payload(record)
        if usage is None:
            continue
        iso = _iso_from_unix(record.get("timestamp"))
        if iso and not ts_in_window(iso, since, until):
            continue
        snapshot, snapshot_model = _model_snapshot(usage)
        for name, raw in _SUM_KEYS:
            sums[name] += snapshot.get(raw, 0) or 0
        if snapshot_model:
            model = snapshot_model
        if iso and started is None:
            started = iso
        if iso:
            ended = iso
        counted += 1
    if not counted:
        return {"totalTokens": 0, "available": False}
    return {**sums, "model": model, "startedAt": started, "endedAt": ended, "available": True}


def grok_session_total(updates_path: Path) -> dict:
    """세션 전체 — 모든 usage 행의 합."""
    return grok_session_window_total(updates_path)


def _session_started_in_window(
    updates_path: Path, started_at: str, ended_at: str,
) -> bool:
    """첫 usage 시각이 창 안에 있을 때만 이 창의 세션이다.

    창 중간의 갱신만 보면 같은 cwd 의 리드 대화가 워커 창에 섞인다.
    """
    for record in iter_jsonl(updates_path):
        iso = _iso_from_unix(record.get("timestamp"))
        if iso:
            return ts_in_window(iso, started_at, ended_at)
    return False


def _session_active_in_window(
    updates_path: Path, started_at: str, ended_at: str,
) -> bool:
    """창보다 먼저 열렸지만 창 안에도 usage 행이 있는 세션 — in-session 리드."""
    first: str | None = None
    for record in iter_jsonl(updates_path):
        iso = _iso_from_unix(record.get("timestamp"))
        if not iso:
            continue
        if first is None:
            first = iso
            if not ts_in_window(first, None, started_at):
                return False
        if ts_in_window(iso, started_at, ended_at):
            return True
    return False


def find_grok_sessions(
    cwd: Path,
    started_at: str,
    ended_at: str,
    *,
    session_root: Path | None = None,
    active_before_start: bool = False,
) -> list[Path]:
    """cwd 로 인코딩된 세션 중 창 안에서 시작된 updates.jsonl.

    `active_before_start=True` 는 창보다 먼저 시작했지만 창 안에서도 프롬프트를
    돌린 세션을 더한다 — in-session 리드의 모양이고, 토큰은
    `grok_session_window_total` 이 창으로 잘라 센다.
    """
    if not started_at or not ended_at:
        return []
    root = session_root or grok_sessions_root()
    if not root.is_dir():
        return []
    target = str(cwd)
    matches: list[Path] = []
    for child in root.iterdir():
        if not child.is_dir():
            continue
        decoded = unquote(child.name)
        if decoded != target:
            continue
        for session in child.iterdir():
            updates = session / "updates.jsonl"
            if not updates.is_file():
                continue
            if _session_started_in_window(updates, started_at, ended_at) or (
                active_before_start
                and _session_active_in_window(updates, started_at, ended_at)
            ):
                matches.append(updates)
    return sorted(matches)


def grok_session_is_non_interactive(updates_path: Path) -> bool:
    """워커 래퍼 세션은 prompt_context.is_non_interactive 가 true 이다."""
    context_path = updates_path.parent / "prompt_context.json"
    try:
        payload = json.loads(context_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return False
    return payload.get("is_non_interactive") is True