"""Bundled Grok provider catalog."""
import json
from pathlib import Path
from typing import Any, Mapping
from urllib.parse import quote

from okstra_ctl.domain.provider import (
    LeadLaunchSpec,
    ModelSpec,
    ProviderSpec,
    ServedModelAttestation,
)
from okstra_ctl.domain.role import ALL_ROLE_TOKENS
from okstra_ctl.domain.worker_exec import (
    ExecCommand,
    PolicySupport,
    WorkerExecRequest,
)
from okstra_ctl.domain.worker_presentation import JsonEvents
from okstra_ctl.domain.worker_stream import StreamEvent, Text, ToolCall, ToolResult


GROK = {
    "grok-4.6": ModelSpec("grok-4.6", "grok-4.6", "grok-4.6", pricing=(2.00, 0.50, 6.00)),
    # picker 에서는 감춘다(사유는 codex 의 gpt-5.4-mini 와 동일).
    "grok-build-0.1": ModelSpec("grok-build-0.1", "grok-build-0.1", "grok-build-0.1", pricing=(1.00, 0.20, 2.00), selectable=False),
}


def _catalog_model_id(observed: str) -> str:
    """Undo the `-build` suffix grok reports its served model under.

    Measured on grok 1.x: asking for `grok-4.6` closes with
    `modelUsage: {"grok-4.6-build": ...}`. The catalog holds the requested ids, so reading the
    reported one as-is would fail the served-model gate on the very model
    okstra asked for. A catalog entry that already carries the suffix
    (`grok-build-0.1`) is matched first and left alone.
    """
    if observed in GROK:
        return observed
    bare = observed.removesuffix("-build")
    return bare if bare != observed and bare in GROK else observed


def normalise_served_model(raw_model: str | None) -> ServedModelAttestation:
    if not raw_model or not raw_model.strip():
        return ServedModelAttestation.unknown()
    model_id = raw_model.strip().lower().removeprefix("grok/")
    return ServedModelAttestation(
        raw_model,
        f"grok/{_catalog_model_id(model_id)}",
        "exact",
        "provider-output",
    )


def served_model_from_session(
    session_id: str, cwd: Path, *, home: Path | None = None
) -> str | None:
    """세션 기록에서 실제로 서빙된 모델을 읽는다.

    grok 은 일하는 동안 어느 이벤트에도 모델을 적지 않고, 닫는 기록에만
    남긴다. 스트림을 해석하지 않는 경로에서는 여기가 유일한 관측 지점이다.
    디렉터리 이름은 cwd 를 퍼센트 인코딩한 것이다.
    """
    root = (home or Path.home() / ".grok") / "sessions"
    encoded = quote(str(cwd), safe="")
    history = root / encoded / session_id / "chat_history.jsonl"
    if not history.is_file():
        return None
    try:
        for line in history.read_text(encoding="utf-8", errors="replace").splitlines():
            if not line.strip():
                continue
            record = json.loads(line)
            model = record.get("model_id") if isinstance(record, Mapping) else None
            if isinstance(model, str) and model.strip():
                return model
    except (OSError, ValueError):
        return None
    return None


def observe_served_model(event: Mapping[str, Any]) -> str | None:
    """Read the model grok actually served from its closing `end` event.

    grok names no model on the events it streams as it works; the only place
    it states one is `end.modelUsage`, keyed by model. Looking for a top-level
    `model` field — Claude's shape — found nothing on every event, so the run
    recorded `servedModelAttestation: unknown` and the served-model gate had
    nothing to check.
    """
    model = event.get("model")
    if isinstance(model, str) and model.strip():
        return model
    message = event.get("message")
    if isinstance(message, Mapping):
        model = message.get("model")
        if isinstance(model, str) and model.strip():
            return model
    if event.get("type") == "end":
        usage = event.get("modelUsage")
        if isinstance(usage, Mapping):
            names = [name for name in usage if isinstance(name, str) and name.strip()]
            # More than one model in a single turn means the CLI switched
            # mid-run; naming either one would be a guess, so report neither.
            if len(names) == 1:
                return names[0]
    return None


def run_concluded(event: Mapping[str, Any]) -> bool:
    """이 어휘의 정상 종결 — `end` 이벤트.

    grok 은 Result 이벤트를 내지 않는다. 실측(status 사이드카 100건): 건강한
    run 71건이 `end.modelUsage` 에서만 나오는 observedModel 을 갖고, result
    이벤트 없이 exit 0 으로 끝난 실패 run 들에는 `end` 가 없다.
    """
    return event.get("type") == "end"


def stream_events(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]:
    """grok `--output-format streaming-json` 한 줄을 공통 이벤트로 옮긴다.

    이 CLI 의 `-p` 기본 `plain` 은 종료 시에만 stdout 에 쓴다. 파이프에 붙은
    워커는 그 동안 pane 과 로그가 0바이트다(실측 2026-08-22). 같은 호출에
    `streaming-json` 을 주면 ACP 세션 업데이트가 NDJSON 으로 흐른다. 키는
    `type=text|tool_call|tool_call_update|end` 이고 Claude 의
    `message.content` 가 아니다 — 그 스키마로 읽으면 이벤트는 전부 버려진다.
    """
    kind = event.get("type")
    if kind == "text":
        data = event.get("data")
        # 공백만 있는 토큰도 살린다. 버리면 붙인 문장에서 단어가 붙는다.
        return (Text(body=data),) if isinstance(data, str) and data else ()
    if kind == "tool_call":
        name = str(event.get("toolName") or event.get("title") or "tool")
        return (ToolCall(name=name, detail=_call_detail(event)),)
    if kind == "tool_call_update":
        return _completed_tool(event)
    if kind == "error":
        message = event.get("message")
        return (Text(body=message),) if isinstance(message, str) and message.strip() else ()
    return ()


def _call_detail(event: Mapping[str, Any]) -> str:
    payload = event.get("rawInput")
    if isinstance(payload, Mapping):
        for key in ("command", "file_path", "path", "pattern", "query"):
            value = payload.get(key)
            if value:
                return str(value)
        return next(
            (str(value) for value in payload.values() if isinstance(value, str) and value),
            "",
        )
    title = event.get("title")
    return str(title) if isinstance(title, str) else ""


# `rawOutput` 의 첫 키는 도구 판별자다 — `ReadFile`, `ListDir`, `GrepSearch`.
# 값 중 첫 문자열을 고르던 동안에는 언제나 이 이름표가 잡혀, 46KB 를 읽은
# 호출도 로그와 pane 에 `← done (8 bytes)` 한 줄로만 남았다.
_TOOL_DISCRIMINATOR = "type"
_FINISHED = ("completed", "failed")


def _completed_tool(event: Mapping[str, Any]) -> tuple[StreamEvent, ...]:
    """끝난 도구 호출 하나를 본문과 성패까지 담아 옮긴다.

    `failed` 를 실제로 읽는다. 상태를 무시하고 `completed` 만 통과시키던
    동안에는 존재하지 않는 파일을 읽은 호출이 스트림에서 통째로 사라져,
    워커가 왜 다음 수를 바꿨는지 기록에 남지 않았다(실측 2026-08-26,
    grok 1.x: 완료 76건 중 `failed` 2건).
    """
    status = event.get("status")
    if status not in _FINISHED:
        return ()
    body = _finished_body(event)
    return (
        ToolResult(
            body=body,
            size_bytes=len(body.encode("utf-8")),
            failed=status == "failed",
        ),
    )


def _finished_body(event: Mapping[str, Any]) -> str:
    """이 호출이 실제로 돌려준 것.

    ACP `content` 블록이 정본이다 — 실측한 완료 76건 중 70건이 도구 출력을
    그대로 담았다. 남은 6건(`list_dir`)은 블록 없이 `rawOutput` 만 보내므로
    그쪽을 폴백으로 둔다.
    """
    blocks = event.get("content")
    if isinstance(blocks, list):
        rendered = [text for block in blocks if (text := _content_text(block))]
        if rendered:
            return "\n".join(rendered)
    return _output_body(event.get("rawOutput"))


def _content_text(block: Any) -> str:
    if not isinstance(block, Mapping):
        return ""
    kind = block.get("type")
    if kind == "content":
        inner = block.get("content")
        return str(inner.get("text", "")) if isinstance(inner, Mapping) else ""
    if kind == "diff":
        # 파일을 고친 호출은 텍스트가 아니라 diff 로 닫는다. 이 호출이 남긴
        # 것은 새 본문이므로 경로와 함께 그것을 적는다.
        return f"--- {block.get('path', '')}\n{block.get('newText', '')}"
    return ""


def _output_body(output: Any) -> str:
    """`rawOutput` 에서 사람이 읽을 것을 꺼낸다.

    판별자 키를 건너뛴 다음, 한 겹 안쪽까지 문자열을 찾는다 — `list_dir` 은
    `Content.content`, `read_file` 은 `FileContent.content` 에 본문을 둔다.
    """
    if isinstance(output, str):
        return output
    if not isinstance(output, Mapping):
        return "" if output is None else str(output)
    for key, value in output.items():
        if key == _TOOL_DISCRIMINATOR:
            continue
        if isinstance(value, str) and value:
            return value
        if isinstance(value, Mapping):
            nested = next(
                (item for item in value.values() if isinstance(item, str) and item), ""
            )
            if nested:
                return nested
    return json.dumps(output, ensure_ascii=False)

GROK_LEAD_LAUNCH = LeadLaunchSpec(
    executable="grok",
    model_flag="--model",
    resume_session_id_flag="--resume",
)


class GrokExecution:
    """grok CLI invocation.

    Runs inside the stage tree when there is one. That is where this provider's
    production path already puts it, and it is why the working directory is part
    of the returned command rather than something the runner picks.
    """

    def build_command(self, request: WorkerExecRequest) -> ExecCommand:
        cwd = request.worktree_path or request.project_root
        # `-p` 기본 `plain` 은 종료 시에만 stdout 에 쓴다. 러너는 파이프라
        # TUI 도 열리지 않고, `--no-alt-screen` 도 그 침묵을 바꾸지 않는다.
        return ExecCommand(
            argv=(
                "grok",
                "-p",
                request.prompt_text,
                "-m",
                request.model,
                "--cwd",
                str(cwd),
                "--output-format",
                "streaming-json",
            ),
            stdin_text=None,
            cwd=cwd,
            presentation=JsonEvents(
                normalise=stream_events,
                observe=observe_served_model,
                join_adjacent_text=True,
                is_conclusion=run_concluded,
            ),
        )

    def policy_support(self) -> PolicySupport:
        return PolicySupport(
            can_auto_approve=False,
            can_bound_write_scope=False,
            note="this CLI exposes no approval or sandbox flag",
        )


def create_provider() -> ProviderSpec:
    return ProviderSpec(
        provider="grok",
        display_label="Grok",
        models=GROK,
        default_models={
            role: "grok-4.6"
            for role in (
                "lead",
                "analyser",
                "critic",
                "designer",
                "planner",
                "executor",
                "verifier",
                "report-writer",
                "translator",
            )
        },
        wrapper="okstra-grok-exec.sh",
        supported_roles=ALL_ROLE_TOKENS,
        execution_capabilities=frozenset(
            {
                "lead-session",
                "worker-artifact-io",
                "source-readonly",
                "extended-artifact-authoring",
                "project-mutation",
            }
        ),
        lead_launch=GROK_LEAD_LAUNCH,
        exec_strategy=GrokExecution(),
        served_model_normalizer=normalise_served_model,
    )