"""Observability (#251/#253): best-effort, fire-and-forget events into
`rstack-agents observe` so the Business Hub mirrors Tau activity. Never on the
tool-call critical path; can never disrupt a Tau session.

owner: RStack developed by Richardson Gunde; Tau adapter contributed by Jeomon
"""
from __future__ import annotations

import asyncio
import json
import os

from .proc import _communicate_or_kill, _own_process_group_kwargs
from .resolve import _resolve_bin_argv

# Pending observations awaiting the drainer, as (payload, cwd) tuples, plus
# the background drainer task — kept referenced so the event loop can't GC it
# mid-flight; discarded on completion. (#253)
_OBSERVE_QUEUE: list = []
_OBSERVE_TASKS: set = set()


async def _emit_observations(payloads: list, cwd: str) -> None:
    """Feed a batch of observability events to ONE `rstack-agents observe`
    process as JSONL stdin (#251, G-09 #552 — observe accepts newline-separated
    JSON objects, one event per line, so a burst of tool calls costs one
    process instead of one per event).

    Best-effort and completely non-disruptive: `observe` always exits 0, no-ops
    when there is no active run, and redacts secrets. We swallow every error and
    bound the subprocess with a hard timeout so a slow/hung write can never
    stall — this coroutine only ever runs inside the fire-and-forget drainer
    (see `_observe_bg`), so it is off the tool-call critical path entirely.
    """
    # G-09 (#552): local binary preferred over `npx --yes` — see resolve.py.
    argv_prefix, _needs_network = _resolve_bin_argv("rstack-agents", cwd)
    if argv_prefix is None:
        return
    try:
        proc = await asyncio.create_subprocess_exec(
            *argv_prefix, "observe", "--source", "tau", "--project", cwd,
            stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.DEVNULL,
            env=os.environ, cwd=cwd,
            **_own_process_group_kwargs(),
        )
        # #487: this already had a timeout, but on expiry the broad `except
        # Exception: pass` below swallowed asyncio.TimeoutError WITHOUT ever
        # killing the process — routed through the shared helper, which kills
        # AND reaps before this outer except ever runs. The bound scales with
        # batch size (each event is a locked ledger append) but stays hard.
        data = "\n".join(json.dumps(p) for p in payloads).encode("utf-8")
        await _communicate_or_kill(proc, data, timeout=min(30.0, 5.0 + len(payloads)))
    except Exception:
        pass  # observability is additive — never let it break a session


async def _drain_observations() -> None:
    """Single background drainer: while events keep arriving, take everything
    queued so far and ship each cwd's events through one observe process.
    Events that arrive while a flush is in flight simply join the next batch —
    no artificial latency when the queue is quiet (an empty-queue drainer
    exits immediately and the next _observe_bg starts a fresh one)."""
    while _OBSERVE_QUEUE:
        batch = list(_OBSERVE_QUEUE)
        _OBSERVE_QUEUE.clear()
        groups: dict = {}
        for payload, cwd in batch:
            groups.setdefault(cwd, []).append(payload)
        for cwd, payloads in groups.items():
            await _emit_observations(payloads, cwd)


def _observe_bg(payload: dict, cwd: str) -> None:
    """Queue an observation WITHOUT awaiting it — keeps observe off the
    tool-call critical path so a slow write never adds latency to a Tau call.
    Falls back to a no-op if there is no running event loop. (#253)

    G-09 (#552): observations are queued and drained in batches (see
    _drain_observations) so a burst of tool calls spawns one observe process,
    not one per event.

    The no-loop check must be get_running_loop(), not "ensure_future raises"
    (Qodo review, PR #553 — and CI proved Qodo's exact wording right): on
    Pythons older than ~3.12, ensure_future with no RUNNING loop falls back
    to get_event_loop(), which silently CREATES a loop that will never run —
    the task attaches to it and the payload leaks into the module-lifetime
    queue forever. get_running_loop() raises deterministically on every
    supported version, so a loop-less call is a true no-op. The drainer is
    also scheduled before the append (race-free: the created task cannot run
    until this synchronous function returns), so no failure mode between the
    two calls can leak an entry either."""
    try:
        asyncio.get_running_loop()
    except RuntimeError:
        return  # no running event loop: the documented no-op
    try:
        if not any(not task.done() for task in _OBSERVE_TASKS):
            task = asyncio.ensure_future(_drain_observations())
            _OBSERVE_TASKS.add(task)
            task.add_done_callback(_OBSERVE_TASKS.discard)
        _OBSERVE_QUEUE.append((payload, cwd))
    except Exception:
        pass  # never let scheduling failure surface in a session