"""Agent VCS-thrash metrics.

Helpers operate on adapter-normalized tool-call dictionaries. Missing command

text or timing stays unmeasured (None), per ADR-0002; zero means observed zero.

"""

from __future__ import annotations

from collections.abc import Iterable, Mapping

from typing import Any

from oakbench.classify import vcs_info_family, vcs_subcommands

POLLING_LOOP_MIN_CALLS = 3

POLLING_LOOP_WINDOW_MS = 30_000.0

_START_KEYS = ("start_ms", "started_ms", "start_time_ms", "timestamp_ms", "time_ms")

_END_KEYS = ("end_ms", "ended_ms", "end_time_ms")

_ELAPSED_KEYS = ("elapsed_ms", "duration_ms", "wall_ms")

_COMMAND_KEYS = ("command", "cmd", "argv", "display")

def agent_blocked_on_vcs_ms(tool_calls: Iterable[Mapping[str, Any]] | None) -> float | None:

    """Total measured time the agent spent waiting on VCS tool calls."""

    calls = _materialize(tool_calls)

    if calls is None:

        return None

    saw_resolvable_command = False

    saw_vcs = False

    total = 0.0

    for call in calls:

        command = _command_text(call)

        if command is None:

            continue

        saw_resolvable_command = True

        if not vcs_subcommands(command):

            continue

        saw_vcs = True

        duration = _duration_ms(call)

        if duration is None:

            return None

        total += duration

    if not saw_resolvable_command:

        return None

    if not saw_vcs:

        return 0.0

    return round(total, 3)

def vcs_share_of_task_wall(vcs_ms: float | int | None, task_wall_ms: float | int | None) -> float | None:

    vcs_value = _non_negative_float(vcs_ms)

    wall_value = _non_negative_float(task_wall_ms)

    if vcs_value is None or wall_value is None or wall_value == 0:

        return None

    return round(vcs_value / wall_value, 6)

def thrash_events(tool_calls: Iterable[Mapping[str, Any]] | None) -> list[dict[str, Any]] | None:

    """Detect VCS polling loops in adapter-visible tool calls."""

    calls = _materialize(tool_calls)

    if calls is None:

        return None

    observations: list[dict[str, Any]] = []

    saw_resolvable_command = False

    for index, call in enumerate(calls):

        command = _command_text(call)

        if command is None:

            continue

        saw_resolvable_command = True

        if not vcs_subcommands(command):

            continue

        timestamp = _timestamp_ms(call)

        if timestamp is None:

            return None

        observations.append(

            {

                "index": index,

                "command": command,

                "family": vcs_info_family(command),

                "mutates": _vcs_mutates(command),

                "ok": _call_succeeded(call),

                "timestamp_ms": timestamp,

            }

        )

    if not saw_resolvable_command:

        return None

    if not observations:

        return []

    events: list[dict[str, Any]] = []

    open_by_family: dict[str, list[dict[str, Any]]] = {}

    # One event per polling loop, keyed by (family, start index). When the

    # same loop grows by another same-family call, the already-emitted event

    # is updated in place rather than appending a duplicate. The window list

    # is pruned by POLLING_LOOP_WINDOW_MS, so once old calls age out the

    # window's start index changes — which naturally begins a new event for

    # the next, separate burst.

    event_by_loop: dict[tuple[str, int], dict[str, Any]] = {}

    for obs in observations:

        if obs["mutates"]:

            open_by_family.clear()

            continue

        family = obs["family"]

        if family is None or not obs["ok"]:

            continue

        window = [

            item

            for item in open_by_family.get(family, [])

            if obs["timestamp_ms"] - item["timestamp_ms"] <= POLLING_LOOP_WINDOW_MS

        ]

        window.append(obs)

        open_by_family[family] = window

        if len(window) >= POLLING_LOOP_MIN_CALLS:

            key = (family, int(window[0]["index"]))

            event = event_by_loop.get(key)

            if event is None:

                event = {

                    "type": "vcs_polling_loop",

                    "family": family,

                    "count": len(window),

                    "window_ms": round(window[-1]["timestamp_ms"] - window[0]["timestamp_ms"], 3),

                    "start_ms": round(window[0]["timestamp_ms"], 3),

                    "end_ms": round(window[-1]["timestamp_ms"], 3),

                    "commands": [str(item["command"]) for item in window],

                    "threshold": {

                        "min_calls": POLLING_LOOP_MIN_CALLS,

                        "window_ms": POLLING_LOOP_WINDOW_MS,

                    },

                }

                event_by_loop[key] = event

                events.append(event)

            else:

                # Same loop grew: refresh the emitted event in place.

                event["count"] = len(window)

                event["window_ms"] = round(window[-1]["timestamp_ms"] - window[0]["timestamp_ms"], 3)

                event["end_ms"] = round(window[-1]["timestamp_ms"], 3)

                event["commands"] = [str(item["command"]) for item in window]

    return events

def _materialize(tool_calls: Iterable[Mapping[str, Any]] | None) -> list[Mapping[str, Any]] | None:

    if tool_calls is None:

        return None

    try:

        calls = list(tool_calls)

    except TypeError:

        return None

    if any(not isinstance(call, Mapping) for call in calls):

        return None

    return calls

def _command_text(call: Mapping[str, Any]) -> str | None:

    for key in _COMMAND_KEYS:

        value = call.get(key)

        if isinstance(value, str) and value.strip():

            return value.strip()

        if isinstance(value, (list, tuple)) and value:

            parts = [str(part) for part in value]

            if parts:

                return " ".join(parts)

    return None

def _duration_ms(call: Mapping[str, Any]) -> float | None:

    for key in _ELAPSED_KEYS:

        value = _non_negative_float(call.get(key))

        if value is not None:

            return value

    start = _first_number(call, _START_KEYS)

    end = _first_number(call, _END_KEYS)

    if start is None or end is None or end < start:

        return None

    return end - start

def _timestamp_ms(call: Mapping[str, Any]) -> float | None:

    start = _first_number(call, _START_KEYS)

    if start is not None:

        return start

    end = _first_number(call, _END_KEYS)

    duration = _duration_ms(call)

    if end is not None and duration is not None:

        return end - duration

    return None

def _first_number(call: Mapping[str, Any], keys: tuple[str, ...]) -> float | None:

    for key in keys:

        value = _non_negative_float(call.get(key))

        if value is not None:

            return value

    return None

def _non_negative_float(value: Any) -> float | None:

    if value is None or isinstance(value, bool):

        return None

    try:

        parsed = float(value)

    except (TypeError, ValueError):

        return None

    if parsed < 0:

        return None

    return parsed

def _call_succeeded(call: Mapping[str, Any]) -> bool:

    if call.get("failed") is True or call.get("is_error") is True:

        return False

    for key in ("returncode", "exit_code", "exitCode"):

        value = call.get(key)

        if isinstance(value, int) and not isinstance(value, bool):

            return value == 0

    status = call.get("status")

    if isinstance(status, str) and status.lower() in {"failed", "error"}:

        return False

    return True

def _vcs_mutates(command: str) -> bool:

    subcommands = vcs_subcommands(command)

    if not subcommands:

        return False

    return any(vcs_info_family(segment) is None for segment, _ in _vcs_segments(command, subcommands))

def _vcs_segments(command: str, subcommands: list[str]) -> list[tuple[str, str]]:

    from oakbench.classify import vcs_invocations

    invocations = vcs_invocations(command)

    if invocations:

        return invocations

    return [(command, subcommand) for subcommand in subcommands]