"""Run one worker CLI and record what happened.

Shared by every provider: the entrypoint scripts parse arguments and hand over
here. What differs per provider is the command (an ``ExecutionStrategy``); what
differs per surface is the presentation.

Idle is measured from stream arrival, never from the log file's mtime. The
screen deliberately drops thinking events, so an mtime-based watchdog would
SIGTERM a healthy worker in the middle of a long reasoning stretch.
"""
from __future__ import annotations

import json
import os
import selectors
import signal
import subprocess
import time
from pathlib import Path
from typing import Any, Callable, Mapping

from .domain.provider import ServedModelAttestation, ServedModelNormalizer
from .domain.worker_exec import (
    SERVED_MODEL_MISMATCH_EXIT_CODE,
    ExecCommand,
    ExecutionStrategy,
    WorkerExecRequest,
)
from .domain.worker_presentation import JsonEvents, Presentation
from .session_transcript import SessionTranscript
from .json_boundary import JsonBoundaryError, write_owned_object_atomic

LIVE = "live"
QUIET = "quiet"

_SELECT_TIMEOUT_SECONDS = 0.25
_TERM_GRACE_SECONDS = 5
# How long the streams stay open after the worker process itself is gone. A pipe
# holds at most its capacity of already-written output when its writer exits, so
# this only has to cover a drain, never a producer.
_DRAIN_AFTER_EXIT_SECONDS = 2
_TIMEOUT_EXIT_CODE = 124
_READ_SIZE = 8192
_WRITE_SIZE = 8192
_NO_STATUS_EXTRA: Mapping[str, Any] = {}

# The signals that end this process without raising anything Python can catch on
# the way out. SIGINT is absent on purpose: it arrives as KeyboardInterrupt and
# the exception path already closes the sidecar.
_ABNORMAL_SIGNALS = (signal.SIGTERM, signal.SIGHUP)


def run_worker(
    strategy: ExecutionStrategy,
    request: WorkerExecRequest,
    *,
    presentation: str,
    log_path: Path,
    status_path: Path,
    status_extra: Mapping[str, Any] = _NO_STATUS_EXTRA,
    served_model_normalizer: ServedModelNormalizer | None = None,
    served_model_verifier: Callable[[ServedModelAttestation], str] | None = None,
) -> int:
    _validate_status_extra(status_extra)
    _validate_request_write_contract(request, status_extra)
    command = strategy.build_command(request)
    started_monotonic = time.monotonic()
    status = _started_status(status_extra, log_path)
    _write_status(status_path, status)
    guard = _AbnormalExit(status_path, status, started_monotonic)

    try:
        with guard:
            exit_code, timed_out, idle_seconds, raw_model = _launch(
                command,
                log_path,
                presentation=presentation,
                idle_timeout_seconds=request.idle_timeout_seconds,
                on_spawn=guard.watch,
            )
            at_exit = getattr(command.presentation, "served_model_at_exit", None)
            if raw_model is None and at_exit is not None and request.session_id:
                raw_model = at_exit(request.session_id, command.cwd)
    except BaseException as exc:
        # Whatever ended this run — an OS error, a Ctrl-C, a bug in this file —
        # the sidecar has to stop saying `started`. Nothing downstream rewrites
        # it, so `worker_liveness` would read the worker as still working
        # forever. No exit code is invented — its absence is already how every
        # reader spells a run that failed. The exits that never raise at all are
        # the guard's to close.
        guard.close(f"{type(exc).__name__}: {exc}")
        raise

    status = _closed(status, started_monotonic)
    if served_model_normalizer is not None:
        attestation = served_model_normalizer(raw_model)
        status["servedModelAttestation"] = _attestation_payload(attestation)
        if served_model_verifier is not None:
            failure = served_model_verifier(attestation)
            if failure:
                status["failure"] = failure
                exit_code = SERVED_MODEL_MISMATCH_EXIT_CODE
    status["exit_code"] = exit_code
    if timed_out:
        status.update(
            timeout=True,
            idle_at_ts=status["ended_ts"],
            idle_seconds=idle_seconds,
            terminated_by="idle-watchdog",
        )
    _write_status(status_path, status)
    return exit_code


def _validate_status_extra(status_extra: Mapping[str, Any]) -> None:
    schema = status_extra.get("schemaVersion")
    identity_version = status_extra.get("executionIdentityVersion")
    identity_keys = {
        "participantRef",
        "roleExecutionRef",
        "executionLabel",
        "dutyId",
        "invocationRef",
    }
    if schema is None and identity_version is None:
        if set(status_extra) & (
            identity_keys | {"attempt", "servedModelAttestation"}
        ):
            raise ValueError("worker status mixes v1 and v2 execution identity")
        return
    if schema != "2.0" or identity_version != 2:
        raise ValueError("worker status mixes v1 and v2 execution identity")
    if any(
        not isinstance(status_extra.get(key), str) or not status_extra[key]
        for key in identity_keys
    ):
        raise ValueError("worker status v2 execution identity is incomplete")
    attempt = status_extra.get("attempt")
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        raise ValueError("worker status v2 attempt must be positive")
    attestation = status_extra.get("servedModelAttestation")
    if not isinstance(attestation, Mapping) or set(attestation) != {
        "observedModel",
        "normalizedModelRef",
        "level",
        "source",
    }:
        raise ValueError("worker status served model attestation is invalid")


def _validate_request_write_contract(
    request: WorkerExecRequest, status_extra: Mapping[str, Any]
) -> None:
    if status_extra.get("executionIdentityVersion") != 2:
        return
    policy = request.policy.write_policy
    enforcement = request.policy.write_enforcement
    if policy is None or enforcement is None:
        raise ValueError("v2 worker request has no write contract")
    if status_extra.get("writePolicyDigest") != policy.digest:
        raise ValueError("v2 worker request write policy digest changed")
    if status_extra.get("writeEnforcement") != enforcement.to_payload():
        raise ValueError("v2 worker request write enforcement changed")


def _launch(
    command: ExecCommand,
    log_path: Path,
    *,
    presentation: str,
    idle_timeout_seconds: int,
    on_spawn: Callable[[subprocess.Popen[bytes]], None],
) -> tuple[int, bool, int, str | None]:
    live = presentation == LIVE
    transcript = SessionTranscript(log_path, live=live)
    observation = _ServedModelObservation()
    strategy = _with_served_model_observation(command.presentation, observation)
    try:
        # The strategy decided where this provider runs — some CLIs work in the
        # stage tree, others in the project root and reach the tree by flag.
        process = subprocess.Popen(
            list(command.argv),
            cwd=str(command.cwd),
            # 보낼 것이 없으면 DEVNULL 이다. `None` 은 부모의 stdin 을 물려주는
            # 것이라, stdin 을 읽는 CLI 가 터미널에서 실행됐을 때 EOF 를 못 받고
            # 영원히 기다린다 — 워커는 대화형이 아니므로 물려줄 이유가 없다.
            stdin=(
                subprocess.PIPE
                if command.stdin_text is not None
                else subprocess.DEVNULL
            ),
            stdout=subprocess.PIPE,
            stderr=_stderr_target(strategy),
            start_new_session=True,
            env=_child_env(),
        )
        on_spawn(process)
        exit_code, timed_out, idle_seconds = _pump(
            process,
            transcript,
            strategy=strategy,
            presentation=presentation,
            idle_timeout_seconds=idle_timeout_seconds,
            stdin_text=command.stdin_text,
        )
        return exit_code, timed_out, idle_seconds, observation.raw_model
    finally:
        transcript.close()


def _with_served_model_observation(
    presentation: Presentation,
    observation: _ServedModelObservation,
) -> Presentation:
    """JSON 경로의 서빙 모델 관측을 러너가 모아 둔다.

    해석 전략은 이벤트를 보고 모델 문자열만 돌려준다. 그 값을 사이드카에
    적는 일은 러너의 것이라, 여기서 한 번 감싼다.
    """
    if not isinstance(presentation, JsonEvents):
        return presentation
    original = presentation.observe

    def observe(event: Mapping[str, Any]) -> str | None:
        observed = original(event)
        observation.record(observed)
        return observed

    return JsonEvents(
        normalise=presentation.normalise,
        observe=observe,
        join_adjacent_text=presentation.join_adjacent_text,
    )


class _ServedModelObservation:
    def __init__(self) -> None:
        self.raw_model: str | None = None

    def record(self, raw_model: str | None) -> None:
        if self.raw_model is None and isinstance(raw_model, str) and raw_model.strip():
            self.raw_model = raw_model


class _AbnormalExit:
    """Close the status sidecar for the exits that raise nothing at all.

    ``except BaseException`` around the run covers an exception, including the
    ``KeyboardInterrupt`` a SIGINT raises. It does not cover SIGTERM or SIGHUP:
    their default disposition ends the process outright, and nothing in this
    file runs (measured — a bash ``trap … EXIT`` does fire on SIGTERM, which is
    why the shell wrappers needed no equivalent of this class). Those two are
    the common abnormal exits: a pane close, ``okstra team reclaim`` /
    ``okstra team teardown``, session teardown. Without this the sidecar stays at ``started`` and
    ``worker_liveness`` reads a dead worker as a working one.

    SIGKILL and a host crash remain uncovered because nothing can cover them. A
    sidecar still reading ``started`` is the residue they leave, and no reader
    should take it as proof the worker is alive.
    """

    def __init__(
        self, status_path: Path, status: Mapping[str, Any], started_monotonic: float
    ) -> None:
        self._status_path = status_path
        self._status = status
        self._started_monotonic = started_monotonic
        self._process: subprocess.Popen[bytes] | None = None
        self._restore: dict[int, Any] = {}

    def watch(self, process: subprocess.Popen[bytes]) -> None:
        """Adopt the child, so a signal tears down its group rather than orphan it."""
        self._process = process

    def close(self, failure: str) -> None:
        _write_status(
            self._status_path,
            {**_closed(self._status, self._started_monotonic), "failure": failure},
        )

    def __enter__(self) -> _AbnormalExit:
        for number in _ABNORMAL_SIGNALS:
            try:
                self._restore[number] = signal.signal(number, self._handle)
            except ValueError:
                # Handlers install from the main thread only. A caller running
                # the runner off-thread keeps the exception path and nothing
                # more, which is what it had before this class existed.
                break
        return self

    def __exit__(self, *_exception: Any) -> bool:
        for number, previous in self._restore.items():
            signal.signal(number, previous)
        self._restore.clear()
        return False

    def _handle(self, number: int, _frame: Any) -> None:
        if self._process is not None:
            _terminate(self._process)
        self.close(f"signal {signal.Signals(number).name}")
        # Then die the way the sender asked, so the exit code still names the
        # signal instead of reporting a clean stop this run did not make.
        signal.signal(number, signal.SIG_DFL)
        os.kill(os.getpid(), number)


def _closed(status: Mapping[str, Any], started_monotonic: float) -> dict[str, Any]:
    """The sidecar's terminal shape, whatever it was that ended the run.

    ``stage`` is the field every reader keys on — ``wrapper_status.is_terminal``,
    the pane reclaim and the dispatch record all ask whether it reads ``exited``.
    """
    return {
        **status,
        "stage": "exited",
        "ended_ts": int(time.time()),
        "duration_ms": int((time.monotonic() - started_monotonic) * 1000),
    }


def _started_status(status_extra: Mapping[str, Any], log_path: Path) -> dict[str, Any]:
    """The sidecar as it looks before the CLI has produced anything.

    ``status_extra`` carries what only the entrypoint knows — which wrapper the
    caller invoked and which role it dispatched — and is placed where the
    wrapper-written sidecars already carry those keys.
    """
    return {
        "schemaVersion": 1,
        **dict(status_extra),
        "pid": os.getpid(),
        "started_ts": int(time.time()),
        "log_path": str(log_path),
        "stage": "started",
    }


def _stderr_target(presentation: Presentation) -> int:
    """stderr 를 stdout 에 합칠지는 전략이 정한다.

    한 스트림에 진행과 결과가 함께 오는 CLI 는 합쳐 읽어야 순서가 보존되고,
    둘을 갈라 내는 CLI 는 갈라 읽어야 답과 진행이 섞이지 않는다.
    """
    return subprocess.STDOUT if presentation.merges_stderr() else subprocess.PIPE


def _pump(
    process: subprocess.Popen[bytes],
    transcript: SessionTranscript,
    *,
    strategy: Presentation,
    presentation: str,
    idle_timeout_seconds: int,
    stdin_text: str | None = None,
) -> tuple[int, bool, int]:
    selector = selectors.DefaultSelector()
    readers = _register_streams(selector, process, strategy, transcript)
    outgoing = _register_prompt(selector, process, stdin_text)
    closing_text: str | None = None
    last_output = time.monotonic()
    timed_out = False
    idle_seconds = 0
    drain_deadline: float | None = None

    while selector.get_map():
        for key, events in selector.select(timeout=_SELECT_TIMEOUT_SECONDS):
            if events & selectors.EVENT_WRITE:
                outgoing = _push_prompt(selector, key, outgoing)
                continue
            chunk = os.read(key.fd, _READ_SIZE)
            if not chunk:
                selector.unregister(key.fileobj)
                continue
            # Either stream is proof of life: a text CLI reports progress for
            # minutes before a result exists to send.
            last_output = time.monotonic()
            closing_text = readers[key.fd].feed(chunk) or closing_text
        idle_seconds = int(time.monotonic() - last_output)
        if (
            idle_timeout_seconds
            and idle_seconds >= idle_timeout_seconds
            and process.poll() is None
        ):
            timed_out = True
            _terminate(process)
        drain_deadline = _drain_deadline(process, drain_deadline)
        if drain_deadline is not None and time.monotonic() >= drain_deadline:
            _stop_reading(selector)

    for reader in readers.values():
        closing_text = reader.flush() or closing_text
    strategy.flush(transcript)

    exit_code = process.wait()
    if closing_text is None and isinstance(strategy, JsonEvents):
        # JSON 스트림은 결과 이벤트로 끝나야 한다. 텍스트를 흘리는 CLI 에는
        # 놓칠 결과 이벤트가 없다.
        transcript.note("no result event in the CLI's output")
    elif closing_text is not None:
        # Result 는 format_* 가 줄을 만들지 않는다. 아카이브 결론은 러너가 적는다.
        # 상한 밖에 둔다 — 잘린 결론은 사후 분석 전체를 잃게 한다.
        if isinstance(strategy, JsonEvents):
            transcript.write("worker", closing_text, capped=False)
        if presentation == QUIET:
            print(closing_text, flush=True)
    return (_TIMEOUT_EXIT_CODE if timed_out else exit_code), timed_out, idle_seconds


def _drain_deadline(
    process: subprocess.Popen[bytes], deadline: float | None
) -> float | None:
    """When to stop reading, once the worker process itself is gone.

    Without this the loop ends only at EOF on every stream, and a worker that
    leaves a background process behind — a dev server, a file watcher, the
    watcher the container-build skill starts on purpose — leaves that process
    holding the write end of the pipe open for as long as it lives. The shell
    wrappers returned as soon as `wait <pid>` did, so blocking on a grandchild is
    a regression rather than a policy, and the exit code the caller gets stays
    the worker's own.

    The deadline is set once and never pushed back: output arriving after the
    worker exited is the grandchild's, and letting it extend the wait would
    restore exactly the hang this bounds.
    """
    if deadline is not None or process.poll() is None:
        return deadline
    return time.monotonic() + _DRAIN_AFTER_EXIT_SECONDS


def _stop_reading(selector: selectors.BaseSelector) -> None:
    """Drop every remaining stream without signalling whoever still holds it.

    A process the worker deliberately left running is not this runner's to kill.
    The idle watchdog is what ends a run that went wrong; this only ends the
    reading of a run that already finished.
    """
    for key in list(selector.get_map().values()):
        selector.unregister(key.fileobj)


def _register_streams(
    selector: selectors.BaseSelector,
    process: subprocess.Popen[bytes],
    presentation: Presentation,
    transcript: SessionTranscript,
) -> dict[int, _LineReader]:
    """이 자식이 말하는 스트림마다 전략이 지정한 싱크를 붙인다."""
    assert process.stdout is not None
    streams = {"stdout": process.stdout, "stderr": process.stderr}
    readers: dict[int, _LineReader] = {}
    for channel, sink in presentation.sinks(transcript):
        stream = streams[channel]
        assert stream is not None, channel
        selector.register(stream, selectors.EVENT_READ)
        readers[stream.fileno()] = _LineReader(sink)
    return readers


class _LineReader:
    """Whole lines out of one byte stream, handed to that stream's sink.

    One per stream rather than one shared buffer: the two streams of a text CLI
    arrive interleaved, and a shared buffer would splice half a progress line
    onto the front of the result.
    """

    def __init__(self, emit: Callable[[str], str | None]) -> None:
        self._emit = emit
        self._pending = b""

    def feed(self, chunk: bytes) -> str | None:
        self._pending += chunk
        *complete, self._pending = self._pending.split(b"\n")
        return self._drain(complete)

    def flush(self) -> str | None:
        """Whatever the stream ended on without a closing newline."""
        trailing, self._pending = self._pending, b""
        return self._drain([trailing] if trailing else [])

    def _drain(self, raw_lines: list[bytes]) -> str | None:
        closing: str | None = None
        for raw in raw_lines:
            closing = self._emit(raw.decode("utf-8", "replace")) or closing
        return closing


def _register_prompt(
    selector: selectors.BaseSelector,
    process: subprocess.Popen[bytes],
    stdin_text: str | None,
) -> bytes:
    """Queue the prompt for a CLI that reads it from stdin.

    Fed inside the pump rather than written whole before it. A prompt larger
    than the pipe buffer — worker prompts are tens of kilobytes — would
    otherwise block this process while the CLI blocks writing the stdout that
    nobody is draining yet, and the idle watchdog could not even reach that
    stall because it only runs once the pump is looping.
    """
    if stdin_text is None or process.stdin is None:
        return b""
    os.set_blocking(process.stdin.fileno(), False)
    selector.register(process.stdin, selectors.EVENT_WRITE)
    return stdin_text.encode("utf-8")


def _push_prompt(
    selector: selectors.BaseSelector, key: selectors.SelectorKey, payload: bytes
) -> bytes:
    """Hand over as much of the prompt as the pipe will take right now."""
    try:
        written = os.write(key.fd, payload[:_WRITE_SIZE]) if payload else 0
    except BlockingIOError:
        return payload
    except OSError:
        # The CLI exited before reading its prompt. Its exit code is the story;
        # this half-delivered write is not.
        _close_prompt(selector, key)
        return b""
    payload = payload[written:]
    if not payload:
        # EOF is what tells the CLI its prompt is complete.
        _close_prompt(selector, key)
    return payload


def _close_prompt(selector: selectors.BaseSelector, key: selectors.SelectorKey) -> None:
    selector.unregister(key.fileobj)
    key.fileobj.close()


def _attestation_payload(attestation: ServedModelAttestation) -> dict[str, Any]:
    return {
        "observedModel": attestation.observed_model,
        "normalizedModelRef": attestation.normalized_model_ref,
        "level": attestation.level,
        "source": attestation.source,
    }


def _terminate(process: subprocess.Popen[bytes]) -> None:
    """SIGTERM the worker's process group, then SIGKILL whatever survives.

    The group, not the process: a CLI spawns shells and build tools, and
    signalling only the direct child leaves those running with the pipe open.
    ``start_new_session=True`` at spawn is what makes the group addressable by
    the child's own pid.
    """
    try:
        os.killpg(process.pid, signal.SIGTERM)
    except ProcessLookupError:
        return
    try:
        process.wait(timeout=_TERM_GRACE_SECONDS)
    except subprocess.TimeoutExpired:
        try:
            os.killpg(process.pid, signal.SIGKILL)
        except ProcessLookupError:
            pass
        process.wait()


def _child_env() -> dict[str, str]:
    """The worker's environment with git's fsmonitor disabled.

    The main worktree's fsmonitor daemon leaks its IPC socket into task
    worktrees, so git status/commit there intermittently fail with
    `fsmonitor_ipc__send_query`. GIT_CONFIG_* scopes the override to this
    process tree without touching the user's repo config, appending after any
    GIT_CONFIG_* the caller already set.
    """
    env = dict(os.environ)
    index = int(env.get("GIT_CONFIG_COUNT", "0") or "0")
    env[f"GIT_CONFIG_KEY_{index}"] = "core.fsmonitor"
    env[f"GIT_CONFIG_VALUE_{index}"] = "false"
    env["GIT_CONFIG_COUNT"] = str(index + 1)
    return env


def write_host_event(
    path: Path,
    *,
    invocation_ref: str,
    attempt: int,
    event_type: str,
    payload: Mapping[str, Any],
) -> Mapping[str, Any]:
    """Append one orchestrator-owned host event from the runner boundary."""
    from .attempt_evidence import HostEventStreamWriter

    return HostEventStreamWriter(
        path, invocation_ref=invocation_ref, attempt=attempt
    ).append(event_type, payload)


def _write_status(path: Path, status: Mapping[str, Any]) -> None:
    """Best-effort status write; a sidecar failure must not break the run.

    Replaced whole rather than rewritten in place: readers poll this file while
    the worker is still running and must never see a half-written document.
    """
    try:
        write_owned_object_atomic(
            path,
            status,
            artifact="worker wrapper status",
        )
    except (OSError, JsonBoundaryError):
        return