"""One backend-owned connection operation per ``manage_connections`` call. Pure data; its only
I/O is one local log line per state change and per settle."""

from __future__ import annotations

import logging
import threading
import time
import uuid
from dataclasses import dataclass, field
from typing import Any, Callable, ClassVar, Dict, List, Optional

from tools.connectors.contract import Actor, SettleReason, TargetState, allowed, resolves

logger = logging.getLogger(__name__)

# Not a config key: a user-tunable wait with clamp rails was a foot-gun (PR1 shipped one, unmerged).
OPERATION_DEADLINE_SECONDS = 300.0


class IllegalTransition(ValueError):
    pass


@dataclass
class Target:
    name: str
    kind: str
    action: str
    state: TargetState = TargetState.pending
    detail: str = ""
    instructions: str = ""
    connect_url: Optional[str] = None
    # The vendor account a managed mint created or observed. Not the desktop transport's connection id.
    connection_id: Optional[str] = None
    # Opaque per-attempt handle when the gateway mints one (absent today; the status route adds it).
    attempt: Optional[str] = None
    # Earliest the watcher may read this target's account again; set from a 429's Retry-After so a
    # rate-limited route is not hammered once per second.
    next_read_at: float = 0.0
    # The credentials an MCP install still needs ({name, prompt, required}); the card draws a
    # field per entry and holds its verb until every required one has text.
    required_env: list[dict[str, Any]] = field(default_factory=list)
    # Fields a transition passes through to the model (``tools`` on a connected MCP target).
    extra: dict[str, Any] = field(default_factory=dict)
    # The install step an ``initiated`` catalog row is in (an ``InstallPhase`` id); drawn only then.
    phase: Optional[str] = None
    # The name of the one hosted account this target connects or repairs; absent for the connector's default account.
    alias: Optional[str] = None
    # The existing account a reconnect repairs, by id: an unnamed account, which no alias can address.
    repair_id: Optional[str] = None

    @property
    def resolved(self) -> bool:
        return resolves(self.kind, self.state)

    def snapshot(self, *, with_url: bool = True) -> dict[str, Any]:
        out: dict[str, Any] = {"name": self.name, "kind": self.kind, "action": self.action, "state": self.state.value,
                               "resolved": self.resolved}
        if self.alias:
            out["alias"] = self.alias
        if self.phase and self.state == TargetState.initiated:
            out["phase"] = self.phase
        if self.detail:
            out["detail"] = self.detail
        if self.instructions:
            out["instructions"] = self.instructions
        if with_url and self.connect_url:
            out["connect_url"] = self.connect_url
        if self.connection_id:
            out["connection_id"] = self.connection_id
        if self.attempt:
            out["attempt"] = self.attempt
        if self.required_env:
            out["required_env"] = self.required_env
        out.update(self.extra)
        return out


@dataclass
class ConnectionOperation:
    # The gateway installs its ``connection.update`` emitter here once; pure data otherwise.
    on_change: ClassVar[
        Optional[Callable[[ConnectionOperation, Optional[dict[str, Any]], dict[str, Any]], None]]
    ] = None

    targets: list[Target]
    session_key: str = ""
    # Stamped by ``live.open``: the profile home the operation was opened under.
    profile_key: str = ""
    # The model's id for the call that opened the operation; the card binds to that tool row only.
    tool_call_id: Optional[str] = None
    # The opening call's ids (a ``CallRef``) for the requests a Try again makes; unset for account ops.
    call: Optional[Any] = None
    op_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    created_at: float = field(default_factory=time.time)
    deadline_at: float = 0.0
    settled_at: Optional[float] = None
    settled_by: Optional[SettleReason] = None
    # Monotonic write counter. Every frame carries the seq of the snapshot it was built from, so a
    # renderer that keeps the highest seq per op can drop a frame that arrives after a newer one.
    seq: int = 0
    # Set on every transition and on settle; the waiting loop sleeps on it.
    wake: threading.Event = field(default_factory=threading.Event, repr=False)
    _settled_snapshot: Optional[dict[str, Any]] = field(default=None, repr=False)
    _lock: threading.Lock = field(default_factory=threading.Lock, repr=False)

    def __post_init__(self) -> None:
        if not self.deadline_at:
            self.deadline_at = self.created_at + OPERATION_DEADLINE_SECONDS

    def target(self, name: str) -> Optional[Target]:
        return next((t for t in self.targets if t.name == name), None)

    def transition(
        self, name: str, to: TargetState, actor: Actor, *, detail: Optional[str] = None,
        connect_url: Optional[str] = None, connection_id: Optional[str] = None, attempt: Optional[str] = None,
        **extra: Any,
    ) -> Optional[dict[str, Any]]:
        """Move one target; the contract decides whether ``actor`` may. Returns the change, or None
        when the target is already in ``to``. Allowed after settlement: the frozen result stays."""
        target = self.target(name)
        if target is None:
            raise IllegalTransition(f"unknown target {name!r}")
        with self._lock:
            if target.state == to:
                return None
            if allowed(target.kind, target.state, to) != actor:
                raise IllegalTransition(f"{target.kind} {name}: {target.state.value} -> {to.value} by {actor.value}")
            change = {"target": name, "from": target.state.value, "to": to.value, "actor": actor.value}
            target.state = to
            if detail is not None:
                target.detail = detail
            change["detail"] = target.detail
            if connect_url is not None:
                target.connect_url = connect_url
            if connection_id is not None:
                target.connection_id = connection_id
            if attempt is not None:
                target.attempt = attempt
            if extra:
                target.extra = dict(extra)
            snapshot = self._bump_locked()
        logger.info("connection op %s: %s %s -> %s by %s", self.op_id, name, change["from"], to.value, actor.value)
        self.wake.set()
        self._changed(change, snapshot)
        return change

    def refresh(self, name: str, *, connect_url: Optional[str], detail: str, actor: Actor = Actor.user,
                phase: Optional[str] = None) -> None:
        """Replace a target's link, detail and install phase without a state change (a repeated
        failure, or the next step of an install).

        ``actor`` says who produced the new text: a second failure of a backend attempt is the
        backend's report, not the user's move, and the frame must not claim otherwise."""
        target = self.target(name)
        if target is None:
            raise IllegalTransition(f"unknown target {name!r}")
        with self._lock:
            target.connect_url = connect_url
            target.detail = detail
            target.phase = phase
            change = {"target": name, "from": target.state.value, "to": target.state.value, "actor": actor.value,
                      "detail": detail}
            snapshot = self._bump_locked()
        self.wake.set()
        self._changed(change, snapshot)

    def _bump_locked(self) -> dict[str, Any]:
        """Advance the write counter and take the snapshot that frame carries. Both happen under
        ``_lock`` so a second writer cannot backdate this frame with its own state.

        After settlement the frozen snapshot is the frame, seq included: a later write changes
        nothing the renderer can see, so it must not leave the operation naming a seq no frame
        carried (the resume snapshot would wait for a frame that never comes)."""
        if self._settled_snapshot is None:
            self.seq += 1
        return self._result_locked()

    def _changed(self, change: Optional[dict[str, Any]], snapshot: dict[str, Any]) -> None:
        hook = type(self).on_change
        if hook is not None:
            hook(self, change, snapshot)

    @property
    def all_resolved(self) -> bool:
        return bool(self.targets) and all(t.resolved for t in self.targets)

    @property
    def settled(self) -> bool:
        return self.settled_at is not None

    def remaining_seconds(self, now: Optional[float] = None) -> float:
        return max(0.0, self.deadline_at - (time.time() if now is None else now))

    def settle(self, by: SettleReason, now: Optional[float] = None) -> bool:
        """Compare-and-set: the first caller freezes the result."""
        with self._lock:
            if self.settled_at is not None:
                return False
            self.settled_at = time.time() if now is None else now
            self.settled_by = by
            for target in self.targets:
                if not target.resolved:
                    target.state = TargetState.not_connected
            self._settled_snapshot = self._bump_locked()
            outcomes = ", ".join(f"{t.name}={t.state.value}" for t in self.targets)
        logger.info("connection op %s settled by %s: %s", self.op_id, by.value, outcomes)
        self.wake.set()
        self._changed(None, self._settled_snapshot)
        return True

    def settle_if_all_resolved(self) -> bool:
        return self.all_resolved and self.settle(SettleReason.all_resolved)

    def _snapshot_locked(self, *, with_urls: bool = True) -> dict[str, Any]:
        return {
            "op_id": self.op_id,
            "seq": self.seq,
            "deadline_at": self.deadline_at,
            "settled_at": self.settled_at,
            "settled_by": self.settled_by.value if self.settled_by else None,
            "targets": [t.snapshot(with_url=with_urls) for t in self.targets],
        }

    def _result_locked(self, *, with_urls: bool = True) -> dict[str, Any]:
        if self._settled_snapshot is not None:
            targets = [dict(t) for t in self._settled_snapshot["targets"]]
            if not with_urls:
                for t in targets:
                    t.pop("connect_url", None)
            return dict(self._settled_snapshot, targets=targets)
        return self._snapshot_locked(with_urls=with_urls)

    def snapshot(self, *, with_urls: bool = True) -> dict[str, Any]:
        with self._lock:
            return self._result_locked(with_urls=with_urls)

    def result(self, *, with_urls: bool = True) -> dict[str, Any]:
        with self._lock:
            result = self._result_locked(with_urls=with_urls)
        for target in result["targets"]:
            target.pop("connection_id", None)
            target.pop("resolved", None)  # the card's flag; the model reads state
        if result.get("settled_at") is None:
            return result
        from tools.registry import registry
        from tools.tool_search_catalog import build_catalog_listing_with_form

        for target in result["targets"]:
            # Setup instructions are the card's text for the user. A catalog entry's notes can be
            # older than this flow ("restart your session to load the tools"), and the model would
            # repeat them after a result that says the tools are callable now.
            target.pop("instructions", None)
            names = target.get("tools") if target.get("kind") == "mcp" else None
            if not names:
                continue
            definitions = registry.get_definitions(set(names), quiet=True)
            listing, _form = build_catalog_listing_with_form(definitions, max_tokens=1500)
            if listing:
                target["tools_listing"] = listing
        return result

    def request_payload(self) -> dict[str, Any]:
        """The ``connection.request`` payload and the resume snapshot: identity, live target snapshots
        (links included, the panel owns them), server-owned deadline."""
        with self._lock:
            targets = [t.snapshot() for t in self.targets]
            seq = self.seq
        payload: dict[str, Any] = {
            "op_id": self.op_id,
            "seq": seq,
            "deadline_at": self.deadline_at,
            "timeout_seconds": OPERATION_DEADLINE_SECONDS,
            "targets": targets,
        }
        if self.tool_call_id:
            payload["tool_call_id"] = self.tool_call_id
        return payload


class DetachedOperation(ConnectionOperation):

    on_change = None