"""Pi coding agent interface — v1's only coding agent.

Runs `pi -p --mode json` and tails its JSONL stdout line by line, forwarding
each event to a callback WHILE the agent works (the streaming crack, solved
by construction). `--session-id` creates-or-continues, so running and
continuing an agent are the same call: same session id = same context window.
"""

from __future__ import annotations

import json
import os
import queue
import signal
import shlex
import socket
import subprocess
import threading
import time
import base64
import hashlib
import re
from enum import Enum
from functools import lru_cache
from pathlib import Path
from typing import Any, Callable, Optional

from .data_types import PiRequest, PiResult, ProviderFailure
from .utils import cleanup_operator_env, now_iso, operator_env

PI_PATH = os.environ.get("PI_PATH", "pi")
MODELS_JSON = os.environ.get("PI_MODELS_PATH", "")

RESULT_SNIPPET_CHARS = 20_000   # tool output rides along whole; clip only guards pathological cases
ARG_VALUE_CHARS = 20_000        # args too — the UI scrolls, it must not be handed cut-off data
LABEL_CHARS = 80                # "bash: <command>" shown as the event name
DEFAULT_AGENT_TIMEOUT_SECONDS = 1800
DEFAULT_AGENT_IDLE_TIMEOUT_SECONDS = 600
AGENT_TIMEOUT_DEFAULTS = {"planner": 5400, "reviewer": 3600}
TERMINATE_GRACE_SECONDS = 2.0
GPT_PROVIDER = "gpt"
GPT_MAX_TOKENS = 16_000
CLIENT_SESSION_PATTERN = re.compile(r"^v1\.[A-Za-z0-9_-]{43}$")

# The arg that identifies a call at a glance, in the order tools tend to use.
PRIMARY_ARGS = ("command", "path", "file_path", "pattern", "query", "url")


class TimeoutLifecycle(str, Enum):
    STARTING = "starting"
    STREAMING = "streaming"
    TERMINATING = "terminating"
    TERMINAL = "terminal"


def _count(value: str) -> int:
    """Parse pi's compact model-list counts (`272K`, `1.0M`)."""
    suffixes = {"K": 1_000, "M": 1_000_000}
    suffix = value[-1:].upper()
    if suffix in suffixes:
        return int(float(value[:-1]) * suffixes[suffix])
    return int(value)


@lru_cache(maxsize=8)
def _pi_catalog(account: str | None = None) -> list[tuple[str, str, int]]:
    """Read pi's merged catalog, including built-in providers and custom models."""
    env = operator_env(account)
    try:
        try:
            result = subprocess.run(
                [PI_PATH, "--list-models"], capture_output=True, text=True,
                timeout=30, env=env, check=False,
            )
        except (OSError, subprocess.TimeoutExpired):
            return []
    finally:
        cleanup_operator_env(env)
    if result.returncode != 0:
        return []
    rows = []
    for line in result.stdout.splitlines()[1:]:
        columns = line.split()
        if len(columns) < 3:
            continue
        try:
            rows.append((columns[0], columns[1], _count(columns[2])))
        except ValueError:
            continue
    return rows


def resolve_model(pattern: str, account: str | None = None) -> tuple[str, str]:
    """Resolve a model pattern to an explicit ``(provider, model_id)`` pair.

    Pi's catalog merges built-in models with ``~/.pi/agent/models.json``. Using
    that same merged view lets SSSF target direct providers such as
    ``openai/gpt-5.6-terra`` without re-registering built-in models locally.
    """
    catalog = [(provider, model_id) for provider, model_id, _ in _pi_catalog(account)]
    if "/" in pattern:
        provider, model_id = pattern.split("/", 1)
        if (provider, model_id) in catalog:
            return provider, model_id
    matches = [(provider, model_id) for provider, model_id in catalog
               if pattern == model_id or pattern in model_id]
    exact = [match for match in matches
             if match[1] == pattern or match[1].endswith("/" + pattern)]
    if len(exact) == 1:
        return exact[0]
    if len(matches) == 1:
        return matches[0]
    if not matches:
        raise ValueError(f"model pattern {pattern!r} not found in pi --list-models — "
                         "authenticate/register it or fix the config")
    raise ValueError(f"model pattern {pattern!r} is ambiguous: {matches}")


def _context_tokens(usage: dict) -> int:
    """Tokens occupying the window after a turn.

    Mirrors pi's own `calculateContextTokens` (coding-agent
    `core/compaction/compaction.ts`), which is what pi compacts against and
    shows in its footer: prefer the provider's `totalTokens`, else sum the
    parts. Cache reads count — cached prompt is still prompt.
    """
    total = usage.get("totalTokens") or 0
    if total:
        return int(total)
    return int(sum(usage.get(part) or 0
                   for part in ("input", "output", "cacheRead", "cacheWrite")))


def context_window(provider: str, model_id: str, account: str | None = None) -> int:
    """The model's context ceiling from pi's merged model catalog."""
    env = operator_env(account)
    try:
        agent_dir = Path(env.get("PI_CODING_AGENT_DIR", str(Path.home() / ".pi" / "agent"))).expanduser()
        try:
            registry = json.loads(Path(MODELS_JSON or agent_dir / "models.json").read_text())
        except FileNotFoundError:
            registry = {}
    finally:
        cleanup_operator_env(env)
    for model in registry.get("providers", {}).get(provider, {}).get("models", []):
        if model.get("id") == model_id:
            return int(model.get("contextWindow") or 0)
    for listed_provider, listed_model, window in _pi_catalog(account):
        if listed_provider == provider and listed_model == model_id:
            return window
    return 0


def authenticated_account(provider: str, env: dict[str, str] | None = None) -> str | None:
    agent_dir = Path((env or operator_env()).get("PI_CODING_AGENT_DIR", str(Path.home() / ".pi" / "agent"))).expanduser()
    auth_path = agent_dir / "auth.json"
    try:
        credentials = json.loads(auth_path.read_text()).get(provider, {})
    except (FileNotFoundError, json.JSONDecodeError, OSError):
        return None
    if not isinstance(credentials, dict):
        return None
    account = credentials.get("accountId")
    if not isinstance(account, str) or not account.strip():
        return None
    account = account.strip()
    parts = auth_path.resolve().parts
    if "accounts" in parts:
        return f"{parts[parts.index('accounts') + 1]} ({account})"
    return account


def _text_of(container: dict) -> str:
    """Join the text blocks of anything pi shapes as {content: [...]} — a
    message or a tool result."""
    return "".join(part.get("text", "") for part in container.get("content", []) or []
                   if isinstance(part, dict) and part.get("type") == "text")


def _clip(text: str, limit: int) -> str:
    return text if len(text) <= limit else text[:limit].rstrip() + "…"


def _label(tool: str, args: dict) -> str:
    """One-line human name for a tool call: `bash: ls -la src`."""
    value = next((args[key] for key in PRIMARY_ARGS
                  if isinstance(args.get(key), str) and args[key].strip()), "")
    if not value:
        value = next((v for v in args.values() if isinstance(v, str) and v.strip()), "")
    value = " ".join(str(value).split())
    return f"{tool}: {_clip(value, LABEL_CHARS)}" if value else tool


def _agent_timeout_seconds(agent_name: str) -> float:
    role_override = f"FACTORY_AGENT_TIMEOUT_SECONDS__{agent_name.upper()}"
    if agent_name and role_override in os.environ:
        return float(os.environ[role_override])
    if "FACTORY_AGENT_TIMEOUT_SECONDS" in os.environ:
        return float(os.environ["FACTORY_AGENT_TIMEOUT_SECONDS"])
    return float(AGENT_TIMEOUT_DEFAULTS.get(agent_name.lower(), DEFAULT_AGENT_TIMEOUT_SECONDS))


def derive_client_session_id(session_id: str) -> str:
    """Stable non-secret Sol client session derived from the pi session id."""
    if not isinstance(session_id, str) or not session_id.strip():
        raise ValueError("pi session id must be a non-empty string")
    try:
        digest = hashlib.sha256(session_id.encode("utf-8")).digest()
    except UnicodeEncodeError as exc:
        raise ValueError("pi session id is not valid UTF-8") from exc
    token = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=")
    derived = f"v1.{token}"
    if not CLIENT_SESSION_PATTERN.fullmatch(derived):
        raise ValueError(f"derived client session id is invalid: {derived!r}")
    return derived


def _provider_failure(value: Any) -> ProviderFailure | None:
    if not isinstance(value, dict):
        return None
    kind = value.get("kind")
    detail = value.get("detail")
    if kind not in {"capped", "unavailable", "protocol", "context_overflow", "timeout", "cancelled"}:
        return None
    if not isinstance(detail, str) or not detail.strip():
        return None
    retry_after = value.get("retry_after_seconds")
    if retry_after is not None:
        try:
            retry_after = int(retry_after)
        except (TypeError, ValueError):
            retry_after = None
    resume_at = value.get("resume_at")
    if resume_at is not None and not isinstance(resume_at, str):
        resume_at = None
    return ProviderFailure(kind=kind, detail=detail.strip(),
                           retry_after_seconds=retry_after, resume_at=resume_at)


def _json_objects(text: str) -> list[dict[str, Any]]:
    objects: list[dict[str, Any]] = []
    for chunk in filter(None, [text.strip(), *[line.strip() for line in text.splitlines()]]):
        try:
            parsed = json.loads(chunk)
        except json.JSONDecodeError:
            continue
        if isinstance(parsed, dict):
            objects.append(parsed)
    return objects


def extract_provider_failure(*texts: str) -> ProviderFailure | None:
    for text in texts:
        if not text:
            continue
        for payload in _json_objects(text):
            for key in ("providerFailure", "provider_failure", "failure"):
                failure = _provider_failure(payload.get(key))
                if failure is not None:
                    return failure
            failure = _provider_failure(payload)
            if failure is not None:
                return failure
    return None


class ToolCallTracker:
    """Folds pi's tool stream into ONE normalized record per completed call.

    pi announces a call as a `toolCall` content block, then emits
    tool_execution_start / _update / _end for it. Only the end carries the
    result, so that is where a record is emitted — one trace event per real
    tool call, the moment it returns, instead of three shapeless ones.

    The record carries the call's real span (`started_at`/`ended_at`), which the
    tracer writes to columns so the UI can lay tool calls on a time axis without
    parsing every payload.
    """

    def __init__(self) -> None:
        self._open: dict[str, dict] = {}
        self._next_seq = 0

    def observe(self, event: dict) -> Optional[dict]:
        """Returns the record for a finished tool call, else None."""
        etype = event.get("type", "")
        if etype in ("agent_attempt_start", "agent_attempt_end"):
            return {"event_type": etype, "tool": "agent_attempt", "tool_call_id": "",
                    "args": {}, "ok": None if etype.endswith("start") else (
                        event.get("returncode") == 0 and not event.get("timedOut")
                        and not event.get("error")),
                    "label": etype, "attempt": event}
        if etype == "message_end":
            for block in event.get("message", {}).get("content", []) or []:
                if isinstance(block, dict) and block.get("type") == "toolCall":
                    self._announce(block.get("id"), block.get("name"),
                                   block.get("arguments"))
            return None
        if etype == "message_update":
            update = event.get("assistantMessageEvent", {}) or {}
            if update.get("type") == "toolcall_end":
                call = update.get("toolCall", {}) or {}
                self._announce(call.get("id"), call.get("name"), call.get("arguments"))
            return None
        if etype == "tool_execution_start":
            self._announce(event.get("toolCallId"), event.get("toolName"),
                           event.get("args"))
            call_id = str(event.get("toolCallId") or "")
            opened = self._open.get(call_id, {})
            if "seq" not in opened:
                opened["seq"] = self._next_seq
                self._next_seq += 1
            args = opened.get("args") or {}
            return {"event_type": "tool_call_start",
                    "tool": opened.get("tool") or "tool", "tool_call_id": call_id,
                    "seq": opened["seq"], "args": args, "ok": None,
                    "attempt_id": event.get("_attempt_id"),
                    "label": _label(opened.get("tool") or "tool", args),
                    "started_at": opened.get("started_at")}
        if etype != "tool_execution_end":
            return None

        call_id = str(event.get("toolCallId") or "")
        opened = self._open.pop(call_id, {})
        tool = str(event.get("toolName") or opened.get("tool") or "tool")
        args = event.get("args") or opened.get("args") or {}
        record = {
            "event_type": "tool_call",
            "tool": tool,
            "tool_call_id": call_id,
            "seq": opened.get("seq"),
            "attempt_id": event.get("_attempt_id"),
            "args": {key: _clip(value, ARG_VALUE_CHARS) if isinstance(value, str) else value
                     for key, value in args.items()},
            "ok": not event.get("isError", False),
            "label": _label(tool, args),
        }
        result_text = _text_of(event.get("result") or {})
        if result_text:
            record["result_snippet"] = _clip(result_text, RESULT_SNIPPET_CHARS)
        record["ended_at"] = now_iso()
        if opened.get("clock"):
            record["duration_ms"] = int((time.monotonic() - opened["clock"]) * 1000)
        if opened.get("started_at"):
            record["started_at"] = opened["started_at"]
        return record

    def _announce(self, call_id, tool, args) -> None:
        """First sighting starts the clock; a later sighting only fills gaps."""
        if not call_id:
            return
        known = self._open.get(str(call_id), {})
        updated = {
            "tool": tool or known.get("tool", ""),
            "args": args or known.get("args", {}),
            "started_at": known.get("started_at") or now_iso(),   # wall clock, for the row
            "clock": known.get("clock") or time.monotonic(),      # monotonic, for duration
        }
        if "seq" in known:
            updated["seq"] = known["seq"]
        self._open[str(call_id)] = updated


def run(request: PiRequest, on_event: Optional[Callable[[dict], None]] = None,
        on_spawn: Optional[Callable[[int, str], None]] = None,
        on_exit: Optional[Callable[[int], None]] = None,
        on_attempt_start: Optional[Callable[[dict], object]] = None,
        on_attempt_end: Optional[Callable[[object, dict], None]] = None,
        timeout_seconds: float | None = None,
        idle_timeout_seconds: float | None = None,
        agent_name: str = "", *,
        clock: Callable[[], float] | None = None,
        sleeper: Callable[[float], None] | None = None,
        popen_factory: Callable[..., Any] | None = None,
        kill_process_group: Callable[[int, int], None] | None = None,
        thread_factory: Callable[..., Any] | None = None,
        spawn_path: str | None = None) -> PiResult:
    """Run one non-interactive pi turn.

    `on_spawn(pid, command_identity)` and `on_exit(pid)` bracket the child
    process so the caller can record it as killable — a hung coding agent is
    otherwise a pid you have to hunt for in `ps` while the run sits there.
    ``command_identity`` is a ``shlex.join`` of the argv passed to ``Popen``.
    """
    client_session_id = derive_client_session_id(request.session_id)
    provider, model_id = resolve_model(request.model, request.account)
    child_env = operator_env(request.account)
    child_env["OVERDECK_PI_CLIENT_SESSION"] = client_session_id
    clock = clock or time.monotonic
    sleeper = sleeper or time.sleep
    popen_factory = popen_factory or subprocess.Popen
    kill_process_group = kill_process_group or os.killpg
    thread_factory = thread_factory or threading.Thread
    cmd = [
        os.environ.get("PI_PATH") or spawn_path or PI_PATH, "-p", "--mode", "json",
        "--provider", provider, "--model", model_id,
        "--thinking", request.thinking,
        "--session-id", request.session_id,
        "--session-dir", request.session_dir,
        "--system-prompt", request.system_prompt,
    ]
    if request.tools:
        cmd += ["--tools", ",".join(request.tools)]
    for extension in request.extensions:
        cmd += ["-e", extension]
    cmd.append(request.prompt)
    command_identity = shlex.join(cmd)

    raw_path = Path(request.raw_output_path)
    raw_path.parent.mkdir(parents=True, exist_ok=True)
    stderr_path = raw_path.with_name("stderr.log")
    timeout = timeout_seconds
    if timeout is None:
        timeout = _agent_timeout_seconds(agent_name or raw_path.parent.name)
    idle_timeout = idle_timeout_seconds
    if idle_timeout is None:
        idle_timeout = float(os.environ.get("FACTORY_AGENT_IDLE_TIMEOUT_SECONDS",
                                            DEFAULT_AGENT_IDLE_TIMEOUT_SECONDS))

    result = PiResult(session_id=request.session_id,
                      provider=provider,
                      model_id=model_id,
                      client_session_id=client_session_id,
                      context_window=context_window(provider, model_id, request.account))
    if provider == GPT_PROVIDER:
        result.max_tokens = GPT_MAX_TOKENS
        result.usage.usage_estimated = True
        result.usage.billing_status = "unavailable"
        result.usage.max_tokens = GPT_MAX_TOKENS
        result.usage.context_window = result.context_window or None
    attempt = None
    process: Any = None
    timed_out = False
    timeout_kind: str | None = None
    error_text: str | None = None
    stderr = ""
    lifecycle = TimeoutLifecycle.STARTING
    terminal_receipt_emitted = False
    account = authenticated_account(provider, child_env)
    resolved_model = f"{provider}/{model_id}"
    if on_attempt_start:
        attempt = on_attempt_start({"session_id": request.session_id,
                                    "client_session_id": client_session_id,
                                    "command": command_identity,
                                    "system_prompt": request.system_prompt,
                                    "user_prompt": request.prompt,
                                    "host": socket.gethostname(),
                                    "account": account,
                                    "provider": provider,
                                    "model": resolved_model,
                                    "model_id": model_id,
                                    "stderr_path": str(stderr_path)})
    if on_event:
        on_event({"type": "agent_attempt_start", "sessionId": request.session_id,
                  "clientSessionId": client_session_id,
                  "attemptId": str(attempt) if attempt is not None else None,
                  "command": command_identity, "stderrPath": str(stderr_path),
                  "account": account, "provider": provider,
                  "model": resolved_model, "modelId": model_id})
    # stdin is DEVNULL, deliberately. The prompt travels in argv, so the child
    # never needs stdin — but inheriting the parent's means pi sees a non-TTY
    # and can sit forever waiting for piped input that will never arrive or
    # EOF. That failure is silent and total: no request goes out, no bytes come
    # back, and the ADW blocks on a read loop with nothing to read. Observed as
    # a run that sat idle at 0% CPU with an empty raw_output.jsonl.
    try:
        process = popen_factory(cmd, stdin=subprocess.DEVNULL,
                                stdout=subprocess.PIPE, stderr=subprocess.PIPE,
                                text=True, bufsize=1, cwd=request.cwd,
                                env=child_env, start_new_session=True)
        if on_spawn:
            on_spawn(process.pid, command_identity)

        stream_records: queue.Queue[dict] = queue.Queue()
        last_output_clock: float | None = None

        def consume_stdout() -> None:
            nonlocal last_output_clock, lifecycle
            with raw_path.open("a") as raw:
                assert process is not None and process.stdout is not None
                for line in process.stdout:
                    last_output_clock = clock()
                    lifecycle = TimeoutLifecycle.STREAMING
                    raw.write(line)
                    raw.flush()
                    try:
                        event = json.loads(line.strip())
                    except json.JSONDecodeError:
                        continue
                    event["_attempt_id"] = str(attempt) if attempt is not None else None
                    if event.get("type") == "message_end":
                        message = event.get("message", {})
                        if message.get("role") == "assistant":
                            text = _text_of(message)
                            if text:
                                result.text = text
                            failure_detail = message.get("errorMessage")
                            if (
                                message.get("stopReason") == "error"
                                and isinstance(failure_detail, str)
                                and failure_detail.strip()
                            ):
                                normalized_failure = failure_detail.lower()
                                failure_kind = (
                                    "context_overflow"
                                    if "context window" in normalized_failure
                                    and "exceed" in normalized_failure
                                    else "protocol"
                                )
                                result.provider_failure = ProviderFailure(
                                    kind=failure_kind,
                                    detail=failure_detail.strip(),
                                )
                            usage = message.get("usage", {}) or {}
                            usage_estimated = bool(usage.get("usage_estimated")) or provider == GPT_PROVIDER
                            billing_status = usage.get("billing_status")
                            if not isinstance(billing_status, str):
                                billing_status = "unavailable" if usage_estimated else None
                            max_tokens = usage.get("max_tokens")
                            if max_tokens is None and provider == GPT_PROVIDER:
                                max_tokens = GPT_MAX_TOKENS
                            normalized_usage = dict(usage)
                            if usage_estimated:
                                normalized_usage["cost"] = {}
                            turn = _context_tokens(usage)
                            result.tokens += turn
                            result.usage.add_turn(normalized_usage, turn)
                            if usage_estimated:
                                result.usage.usage_estimated = True
                                result.usage.billing_status = billing_status
                                result.usage.max_tokens = int(max_tokens) if max_tokens is not None else None
                                result.max_tokens = int(max_tokens) if max_tokens is not None else None
                                result.usage.context_window = result.context_window or None
                            if turn and message.get("stopReason") not in ("aborted", "error"):
                                result.context_tokens = turn
                            if not usage_estimated:
                                result.cost += (usage.get("cost", {}) or {}).get("total", 0.0) or 0.0
                    stream_records.put(event)

        stderr_chunks: list[str] = []

        def consume_stderr() -> None:
            assert process is not None and process.stderr is not None
            for line in process.stderr:
                stderr_chunks.append(line)

        reader = thread_factory(target=consume_stdout, daemon=True)
        errors = thread_factory(target=consume_stderr, daemon=True)
        reader.start()
        errors.start()
        started_clock = clock()

        def forward_stream_records() -> None:
            while True:
                try:
                    event = stream_records.get_nowait()
                except queue.Empty:
                    return
                if on_event:
                    on_event(event)

        while process.poll() is None:
            forward_stream_records()
            now = clock()
            if now - started_clock >= timeout:
                timeout_kind = "wall"
                break
            if (lifecycle is TimeoutLifecycle.STREAMING and last_output_clock is not None
                    and now - last_output_clock >= idle_timeout):
                timeout_kind = "idle"
                break
            sleep_seconds = timeout - (now - started_clock)
            if lifecycle is TimeoutLifecycle.STREAMING and last_output_clock is not None:
                sleep_seconds = min(sleep_seconds, idle_timeout - (now - last_output_clock))
            sleeper(min(0.05, max(0.0, sleep_seconds)))
        if timeout_kind is not None:
            timed_out = True
            lifecycle = TimeoutLifecycle.TERMINATING
            grace_started_clock = clock()
            kill_process_group(process.pid, signal.SIGTERM)
            try:
                process.wait(timeout=TERMINATE_GRACE_SECONDS)
            except subprocess.TimeoutExpired:
                pass
            grace_remaining = TERMINATE_GRACE_SECONDS - (clock() - grace_started_clock)
            if grace_remaining > 0:
                sleeper(grace_remaining)
            try:
                kill_process_group(process.pid, signal.SIGKILL)
            except ProcessLookupError:
                pass
            try:
                result.returncode = process.wait(timeout=TERMINATE_GRACE_SECONDS)
            except subprocess.TimeoutExpired as exc:
                raise TimeoutError("pi process group did not exit after SIGKILL") from exc
        else:
            result.returncode = process.wait()
        lifecycle = TimeoutLifecycle.TERMINAL
        reader.join(timeout=TERMINATE_GRACE_SECONDS)
        errors.join(timeout=TERMINATE_GRACE_SECONDS)
        if reader.is_alive() or errors.is_alive():
            raise TimeoutError("pi stream drain timed out")
        forward_stream_records()
        stderr = "".join(stderr_chunks)
        stderr_path.write_text(stderr)
        if timed_out:
            limit = timeout if timeout_kind == "wall" else idle_timeout
            error_text = f"pi {timeout_kind} timeout: timed out after {limit:g}s"
            result.provider_failure = ProviderFailure(kind="timeout", detail=error_text)
            raise TimeoutError(error_text)
        if result.returncode != 0 and result.provider_failure is None:
            result.provider_failure = extract_provider_failure(stderr)
        if result.provider_failure is not None:
            error_text = (
                f"pi provider failure [{result.provider_failure.kind}]: "
                f"{result.provider_failure.detail}"
            )
            raise RuntimeError(error_text)
        if result.returncode != 0 and not result.text:
            error_text = f"pi exited {result.returncode}: {stderr.strip()[-800:]}"
            raise RuntimeError(error_text)
        return result
    except BaseException as exc:
        error_text = error_text or str(exc)
        raise
    finally:
        lifecycle = TimeoutLifecycle.TERMINAL
        if process is not None and on_exit:
            on_exit(process.pid)
        returncode = result.returncode if process is not None else None
        sig = -returncode if returncode is not None and returncode < 0 else None
        usage = result.usage.model_dump() if result.tokens else None
        provider_failure = result.provider_failure
        if provider_failure is None and not timed_out and sig == signal.SIGTERM:
            detail = error_text or "pi run cancelled"
            provider_failure = ProviderFailure(kind="cancelled", detail=detail)
            result.provider_failure = provider_failure
        cleanup_operator_env(child_env)
        if on_attempt_end:
            on_attempt_end(attempt, {"returncode": returncode, "signal": sig,
                                     "timed_out": timed_out,
                                     "timeout_kind": timeout_kind,
                                     "stderr_path": str(stderr_path),
                                     "tokens": result.tokens or None,
                                     "usage": usage, "error": error_text,
                                     "provider_failure": (provider_failure.model_dump()
                                                          if provider_failure is not None else None)})
        if on_event and not terminal_receipt_emitted:
            terminal_receipt_emitted = True
            on_event({"type": "agent_attempt_end", "returncode": returncode,
                      "attemptId": str(attempt) if attempt is not None else None,
                      "signal": sig, "timedOut": timed_out,
                      "timeoutKind": timeout_kind,
                      "stderrPath": str(stderr_path), "tokens": result.tokens or None,
                      "usage": usage, "error": error_text,
                      "providerFailure": (provider_failure.model_dump()
                                          if provider_failure is not None else None),
                      "lifecycle": lifecycle.value})