"""Which enforcement tier the INSTALLED Tau actually supports (#575).

Tau's `tool_call` hook is the universal pre-execution gate: it fires for every
tool the engine executes, so one handler can guard MCP and extension tools that
built-in shadowing structurally cannot see (the G-07 residue #552 recorded).

The catch is that the commit which made an extension's block actually work
(`a8b6459`) is in NO released version. On released 0.9.2 the handler still runs,
but `tool_call` is absent from `ExtensionRuntime._INTERCEPTABLE_EVENTS`, so its
return value goes to the catch-all subscriber and is DISCARDED — the gate
prompts, the operator denies, and the call proceeds. Registering a blocking
handler there would manufacture the ILLUSION of enforcement, which is strictly
worse than not registering one.

A version check cannot tell the two apart, and this is measured, not assumed:
Tau HEAD and the v0.9.2 tag both declare `version = "0.9.2"` in pyproject.toml,
`grep -rn "__version__" tau/` matches nothing, and `git tag --contains a8b6459`
is empty. So the tier is decided by BEHAVIOR, offline, before the first tool
call — and only ever from Tau's public API (`tau.extensions.__all__` /
`tau.hooks.__all__`), never a private symbol like `_INTERCEPTABLE_EVENTS`.

The probe builds a throwaway hooks bus nobody else is listening to, bridges a
sentinel handler through the REAL `ExtensionRuntime`, and checks whether a block
survives `Hooks.emit()`. Anything unexpected — a missing symbol, a changed
constructor, any exception — resolves to UNKNOWN, which behaves exactly like
SHADOW_ONLY. That is #371's fail-closed principle one level up: a probe that
cannot determine capability must never answer "capable".

owner: RStack developed by Richardson Gunde
"""
from __future__ import annotations

import asyncio
import threading

# Hook honored end-to-end: gate every tool AND keep the shadows.
TIER_UNIVERSAL = "universal"
# Known fail-open Tau (released 0.9.2 shape): shadows only, and say so.
TIER_SHADOW_ONLY = "shadow-only"
# The probe could not decide. Enforced exactly like SHADOW_ONLY, reported apart.
TIER_UNKNOWN = "unknown"

_PROBE_ID = "rstack-capability-probe"
# The probe is pure in-memory dispatch on a private bus; a second is already
# generous. Exceeding it means something is deeply wrong → UNKNOWN.
_PROBE_TIMEOUT_S = 5.0


class _NullRuntimeRef:
    """Stands in for Tau's `_RuntimeRef`. The bridge only dereferences
    `.runtime` when building an ExtensionContext, and a None runtime is the
    documented "no context available" case — so the probe never needs a live
    session to answer the question."""

    runtime = None


def _emit_off_loop(coro):
    """Run one coroutine to completion on a private loop in a worker thread.

    `register()` may be called with or without a running event loop, and a probe
    must not care which: `asyncio.run` would raise inside a live loop, and
    scheduling onto the caller's loop would make a synchronous answer impossible.
    A dedicated thread sidesteps both.
    """
    box = {}

    def runner():
        try:
            box["value"] = asyncio.run(coro)
        except BaseException:  # noqa: BLE001 - surfaced as UNKNOWN by the caller
            box["failed"] = True

    thread = threading.Thread(target=runner, name="rstack-capability-probe", daemon=True)
    thread.start()
    thread.join(timeout=_PROBE_TIMEOUT_S)
    if thread.is_alive() or box.get("failed"):
        return None
    return box.get("value")


def detect_tool_call_tier() -> str:
    """Probe whether THIS Tau honors an extension's `tool_call` block."""
    try:
        from tau.extensions import Extension, ExtensionRuntime, LoadExtensionsResult
        from tau.hooks import Hooks, ToolCallEvent, ToolCallEventResult
    except Exception:  # noqa: BLE001 - an older/newer Tau without these symbols
        return TIER_UNKNOWN

    try:
        sentinel = ToolCallEventResult(block=True, reason=_PROBE_ID)

        async def _probe_handler(event, ctx):  # noqa: ANN001 - Tau's (event, ctx) shape
            return sentinel

        bus = Hooks()
        ExtensionRuntime(
            LoadExtensionsResult(
                extensions=[Extension(path=f"<{_PROBE_ID}>", handlers={"tool_call": [_probe_handler]})],
                errors=[],
            ),
            bus,
            _NullRuntimeRef(),
        )

        # Stage 1 — did the bridge register the handler as INTERCEPTABLE at all?
        # On a fail-open Tau it lands on the catch-all subscriber instead, and
        # the bus has no tool_call handler to collect from.
        if bus.handler_count("tool_call") <= 0:
            return TIER_SHADOW_ONLY

        # Stage 2 — registration is not the contract; SURVIVING emit() is. Prove
        # the block comes back out, rather than trusting that it would.
        results = _emit_off_loop(
            bus.emit(ToolCallEvent(tool_call_id=_PROBE_ID, tool_name="__rstack_probe__", input={}))
        )
        if results is None:
            return TIER_UNKNOWN
        for res in results:
            if isinstance(res, ToolCallEventResult) and res.block:
                return TIER_UNIVERSAL
        return TIER_SHADOW_ONLY
    except Exception:  # noqa: BLE001 - a probe crash must degrade the tier, never break load
        return TIER_UNKNOWN