"""Session-scoped roster and bounded live transcript snapshots for shared clients.

Async projection adapted from JoaoMarcos44's PR #70899; controls reuse the
existing subagent.steer RPC rather than introducing a second steering runtime.
"""

from .method_ctx import HandlerRegistry, bind_module

_registry = HandlerRegistry()
method = _registry.method

_SUBAGENT_SNAPSHOT_FIELDS = (
    "subagent_id", "parent_id", "depth", "goal", "delegation_id", "model",
    "started_at", "status", "tool_count", "last_tool", "accepting_steer",
)
_SUBAGENT_TAIL_BYTES = 16384


def _owned_subagent_records(session_id, transport, owner):
    from tools.delegate_tool_registry import _active_subagents, _active_subagents_lock, _subagent_transport_matches

    with _active_subagents_lock:
        return [dict(r) for r in _active_subagents.values()
                if r.get("owner_session_id") == session_id
                and _subagent_transport_matches(r, transport)
                and r.get("owner_session_record") is owner]


def _visible_subagent_records(session_id, transport, owner):
    """Read-only roster for ``subagent.list``: the exact-owner records PLUS children whose durable
    conversation lineage (``owner_agent_session_id`` resolved to its compression tip) is this
    session's agent — the same spine ``delegate_task(action="list")`` walks in-process.

    Spawn freezes ``owner_session_id`` to the UI session id of that moment; a Desktop reconnect /
    resume remints the id and rebuilds the session record, and compression rotates the durable key,
    so the exact match alone hid every still-running child from the panel for good (#114909).
    Control RPCs (steer / interrupt / tail) keep the exact generation authority."""
    from tools.delegate_tool_registry import (
        _active_subagents, _active_subagents_lock, _owns_subagent_record, _subagent_transport_matches,
    )

    with _active_subagents_lock:
        records = [dict(r) for r in _active_subagents.values()]
    agent = owner.get("agent")
    # Lineage resolution may read the session DB — evaluated outside the registry lock.
    return [r for r in records
            if (r.get("owner_session_id") == session_id
                and _subagent_transport_matches(r, transport)
                and r.get("owner_session_record") is owner)
            or _owns_subagent_record(r, agent)]


@method("subagent.list")
def _(rid, params):
    session_id = _str_param(params, "session_id")
    transport, owner = _current_session_steer_authority(session_id)
    if transport is None or owner is None:
        return _err(rid, 4001, "session not found or not owned by this transport")
    live = _visible_subagent_records(session_id, transport, owner)
    return _ok(rid, {
        "subagents": [{key: r.get(key) for key in _SUBAGENT_SNAPSHOT_FIELDS} for r in live],
        "delegations": _failed_delegations(session_id, owner),
    })


def _failed_delegations(session_id, owner):
    """Recently failed async delegation tasks for this session from the durable store (the live
    roster forgets ended children and dies with a renderer reload, #97202). Read under the session's
    profile home, where its delegations were persisted; a store error degrades to no rows."""
    from tools.async_delegation import failed_delegations_for_session

    agent_session_id = str(getattr(owner.get("agent"), "session_id", "") or "")
    try:
        with _session_home_scope(owner):
            return failed_delegations_for_session(session_id, agent_session_id)
    except Exception:
        logger.debug("subagent.list: failed-delegation read failed for %s", session_id, exc_info=True)
        return []


@method("subagent.interrupt")
def _(rid, params):
    from agent.interrupt_compat import request_hard_interrupt

    subagent_id = _str_param(params, "subagent_id")
    if not subagent_id:
        return _err(rid, 4000, "subagent_id required")
    session_id = _str_param(params, "session_id")
    transport, owner = _current_session_steer_authority(session_id)
    if transport is None or owner is None:
        return _err(rid, 4001, "session not found or not owned by this transport")
    record = next((r for r in _owned_subagent_records(session_id, transport, owner)
                   if r.get("subagent_id") == subagent_id), None)
    agent = record.get("agent") if record else None
    # Interrupt the authorized object, never re-resolve a globally recyclable id.
    found = False
    if agent is not None:
        try:
            found = bool(request_hard_interrupt(agent, f"Interrupted via TUI ({subagent_id})"))
        except Exception:
            logger.debug("subagent interrupt failed", exc_info=True)
    return _ok(rid, {"found": found, "subagent_id": subagent_id})


@method("subagent.tail")
def _(rid, params):
    session_id = _str_param(params, "session_id")
    subagent_id = _str_param(params, "subagent_id")
    if not subagent_id:
        return _err(rid, 4000, "subagent_id required")
    transport, owner = _current_session_steer_authority(session_id)
    if transport is None or owner is None:
        return _err(rid, 4001, "session not found or not owned by this transport")
    result = {"subagent_id": subagent_id, "available": False, "text": "", "truncated": False}
    record = next((r for r in _owned_subagent_records(session_id, transport, owner)
                   if r.get("subagent_id") == subagent_id), None)
    path = getattr(record.get("agent"), "_live_transcript_path", None) if record else None
    if not path:
        return _ok(rid, result)
    try:
        with open(path, "rb") as stream:
            size = stream.seek(0, 2)
            stream.seek(max(0, size - _SUBAGENT_TAIL_BYTES))
            text = stream.read(_SUBAGENT_TAIL_BYTES).decode("utf-8", errors="ignore")
    except OSError:
        # Creation/cleanup races are normal while a child starts or ends.
        return _ok(rid, result)
    return _ok(rid, {**result, "available": True, "text": text, "truncated": size > _SUBAGENT_TAIL_BYTES})


def register(server):
    bind_module(globals(), server)