"""Collectors: the fetch half of every poll, exercised with a fake transport.

`httpx.MockTransport` means these run at the speed of the parser and still cover what the
live hosts do badly: a 401 from a rotated agent token, a half-degraded agent snapshot, a
gufo server that only answers `/v1/slots`, an engine that counts nothing yet.

NUT is covered here for the *monitor* bookkeeping (payload shape, command gating, status
transitions); the socket protocol itself is in test_nut.py.
"""
from __future__ import annotations

import json
import threading
from http.server import ThreadingHTTPServer
from pathlib import Path
from types import SimpleNamespace
from typing import Any, ClassVar

import httpx
import pytest

from dashboards import certs
from dashboards.app import build_hub
from dashboards.collectors import (
    AgentTls,
    NutMonitor,
    UpstreamError,
    agent_tls_map,
    disks_jobs,
    fetch_disk_snapshot,
    fetch_engine,
    fetch_gateway_snapshot,
    fetch_host_snapshot,
    fetch_snapshot,
    gateway_job,
    host_jobs,
    llm_jobs,
    post_route,
)
from dashboards.config import Settings, Target
from dashboards.poller import Result
from dashboards.prom import Prom
from dashboards.rates import RateEstimator
from dashboards.sources import nut
from dashboards.sources.disks import DisksCollector
from dashboards.sources.host import HostCollector

FIXTURES = Path(__file__).parent / "fixtures"


def client_for(handler: Any) -> httpx.AsyncClient:
    return httpx.AsyncClient(transport=httpx.MockTransport(handler),
                             base_url="http://agent.test")


# --------------------------------------------------------------------------- #
# host agent
# --------------------------------------------------------------------------- #
def agent_body() -> dict[str, Any]:
    return json.loads((FIXTURES / "agent_driveripper.json").read_text())


def advance_cpu(body: dict[str, Any], *, total: float, idle: float) -> None:
    """Move the jiffy counters the way a real /proc/stat does between polls."""
    modes = body["cpu"]["modes"]
    for mode, value in modes.items():
        modes[mode] = value + (total if mode == "total" else idle if mode == "idle" else 0)


async def test_a_real_agent_snapshot_parses() -> None:
    body = agent_body()
    calls: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/snapshot"
        calls.append(request.url.path)
        return httpx.Response(200, json=body)

    target = Target("driveripper", "http://agent.test")
    # min_window=0: these fetches land microseconds apart, far under any real poll
    # cadence; this test pins the Δjiffies window semantics, not the phase-13 floor
    # (tests/test_rates.py pins the floor at a real cadence).
    collector = HostCollector(RateEstimator(min_window=0.0))
    async with client_for(handler) as client:
        first = (await fetch_host_snapshot(client, target, collector)).payload
        # CPU usage is Δjiffies/Δt: one scrape has no window, so it is n/a and not 0 %
        assert first["cpu"]["usage_pct"] is None
        # the counters have to actually move — total 0 delta means "cannot divide", not idle
        same = (await fetch_host_snapshot(client, target, collector)).payload
        assert same["cpu"]["usage_pct"] is None, "no counter movement must read as n/a"
        advance_cpu(body, total=1000, idle=700)
        result = await fetch_host_snapshot(client, target, collector)
    payload = result.payload
    assert payload["cpu"]["usage_pct"] == pytest.approx(30.0, abs=1.0)
    assert payload["name"] == "driveripper"
    assert payload["cpu"]["usage_pct"] is not None
    assert payload["memory"]["total_bytes"] > 0
    # real captures have real GPUs; the legend wants the short name, the tooltip the full one
    assert payload["gpus"], "driveripper's capture must yield at least one GPU"
    assert payload["gpus"][0]["model"] and "model_full" in payload["gpus"][0]
    assert any(s.metric == "cpu.usage_pct" for s in result.samples)


async def test_target_headers_are_sent_to_the_agent() -> None:
    seen: dict[str, str] = {}

    def handler(request: httpx.Request) -> httpx.Response:
        seen["authorization"] = request.headers.get("authorization", "")
        return httpx.Response(200, json=agent_body())

    target = Target("gateway", "http://agent.test", headers={"Authorization": "Bearer tok"})
    async with client_for(handler) as client:
        await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
    assert seen["authorization"] == "Bearer tok"


@pytest.mark.parametrize("status,fragment", [
    (401, "token"),
    (403, "HTTP 403"),
    (500, "HTTP 500"),
])
async def test_agent_rejections_say_which_one_it_was(status, fragment) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, text="nope")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))
    assert fragment in excinfo.value.message
    assert excinfo.value.status == 502


async def test_a_refused_connection_is_an_upstream_error_not_a_traceback() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="unreachable"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


@pytest.mark.parametrize("failure", ["refused", "timeout", "http-500"])
async def test_a_failing_host_never_echoes_its_token_in_the_message(failure: str) -> None:
    """An `UpstreamError` message is not just a log line: the poller keeps it in the health
    record (`/api/health`) and writes it to the `event` table (`/api/events`), both of which
    the browser reads — so the URL it names has to be the redacted one, on every failure path
    (AGENTS.md invariant 6). `parse_targets` normally moves a userinfo token into a header, so
    this is the backstop that keeps the guarantee true for any `Target` that still carries one.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        if failure == "refused":
            raise httpx.ConnectError("connection refused")
        if failure == "timeout":
            raise httpx.ReadTimeout("too slow")
        return httpx.Response(500, text="boom")

    target = Target("gw", "http://SEKRIT-AGENT-TOKEN@agent.test")
    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
    message = excinfo.value.message
    assert "SEKRIT-AGENT-TOKEN" not in message, f"the token leaked through {failure}"
    # …and the message still names the machine it failed to reach
    assert "http://agent.test/snapshot" in message


async def test_a_failing_engine_never_echoes_credentials_in_its_url() -> None:
    """Same rule for the engines: the URL in the message is the redacted one."""
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("connection refused")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="unreachable") as excinfo:
            await fetch_engine(client, "gufo", "http://SEKRIT-ENGINE-TOKEN@agent.test", "",
                               RateEstimator())
    assert "SEKRIT-ENGINE-TOKEN" not in excinfo.value.message
    assert "http://agent.test/metrics" in excinfo.value.message


async def test_a_slow_agent_is_a_gateway_timeout() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("too slow")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))
    assert excinfo.value.status == 504


async def test_html_instead_of_json_is_reported_as_such() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, text="<html><body>502 Bad Gateway</body></html>")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="non-JSON"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


async def test_an_agent_answering_not_ok_is_not_silently_empty() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": False, "detail": "procfs unreadable"})

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="procfs unreadable"):
            await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                      HostCollector(RateEstimator()))


async def test_a_degraded_agent_still_renders_and_says_what_broke() -> None:
    body = agent_body()
    body["errors"] = {"temperatures": "no hwmon sensors"}

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert result.payload["degraded"] == {"temperatures": "no hwmon sensors"}
    assert result.payload["memory"]["total_bytes"] > 0


async def test_smart_addon_errors_do_not_degrade_the_system_source() -> None:
    """A host without smartmontools (or without device read access) is unconfigured, not
    broken (invariant 3): its `smart:` lines must never hold the system page amber. They
    stay in `agent.errors`, where the /disks page explains them in its own panel.
    """
    body = agent_body()
    body["errors"] = ["smart: sdb: Smartctl open device: /dev/sdb failed: Permission denied"]

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert "degraded" not in result.payload
    assert any("smart:" in str(e) for e in result.payload["agent"]["errors"])


async def test_real_degradations_still_show_when_smart_noise_is_alongside() -> None:
    body = agent_body()
    body["errors"] = ["smart: smartctl not found", "temperatures: no hwmon sensors"]

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert result.payload["degraded"] == ["temperatures: no hwmon sensors"]


async def test_a_host_without_a_proxy_config_does_not_degrade_the_system_source() -> None:
    """The proxy add-on's absence lines, same rule and same reason as `smart:` above: two of this
    fleet's three hosts run no haproxy and no nginx, so an agent upgrade must not put a
    `degraded` payload (and therefore an overview alert, `alerts.py` reads this key) on a box that
    never had either. The lines survive in `agent.errors` for the page that owns them — phase 29's
    gateway source reads them the way `/disks` reads `smart_errors`.

    `collectors.ADDON_ERROR_PREFIXES` is the one list of these prefixes; this asserts every
    name by behaviour, so dropping any of them fails here, and `tests/test_agent.py` pins both
    the wording of the lines the agent actually writes and the equality of the two lists.

    The `resolver:` line is phase 32's and the same case with a DNS-shaped name: a host with no
    proxy config has nothing to resolve, and a truncated lookup pass (`resolver: pass budget …`)
    is one route without a count, not a broken box. Both arms belong in this list or every
    upgraded host gets an overview WARN it has no way to clear.
    """
    missing_haproxy = ("haproxy: /etc/haproxy/haproxy.cfg not found — haproxy is not installed "
                       "here, or its config lives elsewhere (--haproxy-config); --no-proxy "
                       "opts out")
    missing_nginx = ("nginx: /etc/nginx/nginx.conf not found — nginx is not installed here, or "
                     "its config lives elsewhere (--nginx-config); --no-proxy opts out")
    no_sockets = ("sockets: /proc/net/tcp not found, /proc/net/tcp6 not found — this host "
                  "publishes no kernel socket table; --no-proxy opts out")
    nothing_to_resolve = ("resolver: nothing to resolve — haproxy not found and nginx not found; "
                          "--no-resolve opts out")
    truncated_lookups = ("resolver: pass budget 3.0s exhausted, 41 of 58 targets looked up")
    body = agent_body()
    body["errors"] = [missing_haproxy, missing_nginx, no_sockets,
                      "smart: smartctl 7.5 not found", nothing_to_resolve, truncated_lookups]

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert "degraded" not in result.payload, "unconfigured is not broken (invariant 3)"
    for prefix in ("haproxy:", "nginx:", "sockets:", "resolver:"):
        assert any(prefix in str(e) for e in result.payload["agent"]["errors"]), \
            "the sentence survives for the page that explains it"


async def test_a_real_degradation_shows_alongside_the_proxy_addon_noise() -> None:
    """Filtering the add-ons must not swallow a genuine half-snapshot beside them."""
    body = agent_body()
    body["errors"] = ["nginx: /etc/nginx/nginx.conf is there but could not be read: boom",
                      "temperatures: no hwmon sensors"]

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        result = await fetch_host_snapshot(client, Target("gw", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert result.payload["degraded"] == ["temperatures: no hwmon sensors"]


# --------------------------------------------------------------------------- #
# physical disks (the /disks source's fetch half)
# --------------------------------------------------------------------------- #
async def test_the_disk_fetch_reads_the_same_agent_snapshot_as_the_system_one(
        disk_snapshot: Any) -> None:
    body = disk_snapshot(mode="healthy")
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/snapshot", "disks has no endpoint of its own"
        seen.append(request.url.path)
        return httpx.Response(200, json=body)

    target = Target("driveripper", "http://agent.test")
    async with client_for(handler) as client:
        result = await fetch_disk_snapshot(client, target, DisksCollector(), timeout=30)
    assert seen == ["/snapshot"]
    assert result.payload["name"] == "driveripper"
    assert result.payload["disks"], "the physical rows come with the snapshot"
    assert any(sample.metric.startswith("disk.temp_c@") for sample in result.samples)


async def test_the_disk_fetch_reports_the_agent_failure_like_the_system_one(
        monkeypatch: pytest.MonkeyPatch) -> None:
    """One dead host must say the same sentence on both pages (shared `fetch_snapshot`)."""
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(500, text="boom")

    target = Target("gw", "http://agent.test")
    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as system_error:
            await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
        with pytest.raises(UpstreamError) as disk_error:
            await fetch_disk_snapshot(client, target, DisksCollector())
    assert disk_error.value.message == system_error.value.message


async def test_an_agent_that_is_not_ok_fails_the_disk_poll_too() -> None:
    """A degraded agent that says `ok: false` has no trustworthy disk rows to hand over."""
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": False, "detail": "procfs unreadable"})

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match="procfs unreadable"):
            await fetch_disk_snapshot(client, Target("gw", "http://agent.test"),
                                      DisksCollector())


class TimeoutRecordingClient(httpx.AsyncClient):
    """An `AsyncClient` that remembers the `timeout=` each request was handed.

    A `MockTransport` replaces httpcore's connection pool, so it does not enforce read
    timeouts and a slow-handler test would prove nothing. The argument is what matters
    anyway: `timeout=None` means *no timeout at all* to httpx, which would hang a poll
    loop forever, and the client default means "the system poll's 4 s".
    """

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self.timeouts: list[Any] = []

    async def get(self, url: Any, **kwargs: Any) -> httpx.Response:  # type: ignore[override]
        self.timeouts.append(kwargs.get("timeout", "<unset>"))
        return await super().get(url, **kwargs)


async def test_the_slow_disk_poll_carries_its_own_timeout_and_never_disables_one() -> None:
    """`HOST_TIMEOUT_SECONDS` (4 s) stays `/system`'s budget; `disks` brings its own 30 s.

    The SMART pass runs inside the agent's snapshot handler, so the two dashboards cannot
    share one budget: too short and `/disks` reads red on every SMART poll, too long and
    `/system` stops noticing a dead host. And the default must stay httpx's sentinel —
    passing a literal `None` silently turns timeout protection off altogether.
    """
    body = {"ok": True, "ts": 1.0, "disk_io": {"devices": []}, "errors": []}

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    client = TimeoutRecordingClient(transport=httpx.MockTransport(handler),
                                    timeout=4.0, base_url="http://agent.test")
    target = Target("gw", "http://agent.test")
    async with client:
        await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))
        assert client.timeouts == [httpx.USE_CLIENT_DEFAULT]
        await fetch_disk_snapshot(client, target, DisksCollector(), timeout=30)
        assert client.timeouts[-1] == 30, "the disk poll waits for the SMART pass"
        await fetch_snapshot(client, target)
        assert client.timeouts[-1] is httpx.USE_CLIENT_DEFAULT
        assert all(timeout is not None for timeout in client.timeouts), (
            "`timeout=None` disables httpx's timeouts entirely — translate it, never pass it")


async def test_smart_addon_errors_reach_the_disk_payload_verbatim(
        disk_snapshot: Any) -> None:
    """`collectors` keeps the `smart:` lines out of `/system`'s `degraded`; `/disks` is
    where they earn their keep — they are the page's reason for an empty card."""
    body = disk_snapshot(mode="denied")

    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=body)

    async with client_for(handler) as client:
        disks = await fetch_disk_snapshot(client, Target("driveripper", "http://agent.test"),
                                          DisksCollector())
        system = await fetch_host_snapshot(client, Target("driveripper", "http://agent.test"),
                                           HostCollector(RateEstimator()))
    assert disks.payload["smart_errors"], "the /disks page gets the agent's own sentence"
    assert "degraded" not in system.payload, "and /system stays unbothered by it"


# --------------------------------------------------------------------------- #
# gateway topology (the /gateway source's fetch half, phase 29)
# --------------------------------------------------------------------------- #
async def test_the_gateway_fetch_reads_the_same_agent_snapshot_as_the_other_two(
        gateway_snapshot: Any) -> None:
    """One endpoint, three readers: the pipes come out of `/snapshot` and nothing else.

    The gateway job has no URL of its own (`GATEWAY_HOST` is a *name*), so this is the pin that
    keeps it that way — and that the projection's own honesty rules survive the trip through the
    collector: gauges under the `gateway` family, **no** events (a count has no transition worth a
    line), and no `window_seconds` for a number that was never diffed (rule 2).
    """
    body = gateway_snapshot("full")
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        assert request.url.path == "/snapshot", "the gateway page has no endpoint of its own"
        seen.append(request.url.path)
        return httpx.Response(200, json=body)

    target = Target("gateway", "http://agent.test")
    async with client_for(handler) as client:
        result = await fetch_gateway_snapshot(client, target, timeout=10)
    assert seen == ["/snapshot"]
    assert result.payload["host"] == "gateway"
    assert result.payload["supported"] is True
    assert result.payload["listeners"], "the configured pipes arrive with the snapshot"
    assert result.samples and all(sample.source == "gateway" for sample in result.samples), (
        "the gauges are their own sample family, not `host`'s")
    assert result.events == [], "a connection count has no transition to report"
    assert "window_seconds" not in result.payload, "a count is not a rate"


async def test_the_gateway_poll_carries_its_own_timeout_and_never_disables_one() -> None:
    """The third budget on one endpoint: the client default until the job asks for its own.

    Same rule the `disks` poll was pinned with — a literal `None` means *no timeout at all* to
    httpx, which would hang the only poller this source has (invariant 4) — and the reason
    `GATEWAY_TIMEOUT_SECONDS` is a setting rather than a constant inherited from `system`.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"ok": True, "ts": 1.0, "errors": []})

    client = TimeoutRecordingClient(transport=httpx.MockTransport(handler),
                                    timeout=4.0, base_url="http://agent.test")
    target = Target("gateway", "http://agent.test")
    async with client:
        await fetch_gateway_snapshot(client, target)
        assert client.timeouts[-1] is httpx.USE_CLIENT_DEFAULT
        await fetch_gateway_snapshot(client, target, timeout=10)
        assert client.timeouts[-1] == 10, "the gateway poll waits on its own budget"
        assert all(timeout is not None for timeout in client.timeouts)


async def test_the_gateway_job_is_one_job_named_after_the_host_it_reads() -> None:
    """`key`/`dashboard`/`label`/`interval` at the factory, and `None` for a name nobody dials.

    The label is what an alert and an event print and what `alerts.py` uses to find the `host:` job
    reading the same agent, so it is the host's name and not the source's — while the key stays the
    single word `gateway` because there is exactly one such job per deployment.
    """
    settings = Settings(host_targets=(Target("gateway", "https://gw.test"),),
                        gateway_interval=45.0, gateway_timeout=7.0)
    entry = AgentTls(verify="/ca.pem", cert=("/c.pem", "/k.pem"), status=lambda: certs.ACTIVE)
    async with httpx.AsyncClient() as client:
        job = gateway_job(settings, client, tls={"gateway": entry})
    assert job is not None
    assert (job.key, job.dashboard, job.label) == ("gateway", "gateway", "gateway")
    assert job.interval == 45.0
    assert job.enabled is True, "a job that exists is a job that runs"
    missing = Settings(host_targets=(Target("deskwork", "http://desk.test"),))
    assert gateway_job(missing, httpx.AsyncClient()) is None, "unconfigured is no task at all"


# --------------------------------------------------------------------------- #
# TLS: the dashboard dials a protected exporter with its own client certificate
# --------------------------------------------------------------------------- #
#
# The certificates come from `dashboards.certs` over a dict store and the server is the real
# `agent/host_agent.py` in the TLS mode task 02 added, so this proves the dial itself: the
# client certificate is presented, the agent is verified against the root this dashboard minted,
# and a host the operator revoked is refused before a socket opens. `httpx.MockTransport` is
# kept for the sentences (a 403 body is the agent's words, not a handshake).


class DictMeta:
    """The three-method `certs.MetaStore` slice, backed by a dict (the `test_certs.py` idiom)."""

    def __init__(self) -> None:
        self.rows: dict[str, str] = {}

    def get_meta(self, key: str) -> str | None:
        return self.rows.get(key)

    def set_meta(self, key: str, value: str) -> None:
        self.rows[key] = value

    def delete_meta(self, key: str) -> int:
        return 1 if self.rows.pop(key, None) is not None else 0


@pytest.fixture
def tls_fleet(tmp_path: Path, host_agent_module: Any) -> Any:
    """One dashboard CA + client pair, one issued agent bundle, and the agent serving it.

    Everything production has: the material written to a private directory (`httpx` takes
    paths), the bundle installed where `install-agent.sh` will put it, and the `AgentTls` the
    jobs get — built by `agent_tls_map`, the same factory `app.build_hub` is handed, so the
    tests dial with the production wiring and not a hand-built guess of it.
    """
    store = DictMeta()
    state = certs.ensure_certs(store)
    state = certs.issue_agent_cert(store, state, "localhost", ["127.0.0.1", "localhost"])
    entry = state.agent("localhost")
    assert entry is not None
    material = certs.write_material(state, tmp_path)
    bundle_path = tmp_path / "dashboards-agent.certs"
    bundle_path.write_text(entry.bundle, encoding="ascii")
    bundle, error = host_agent_module.TlsBundle.load(str(bundle_path))
    assert error is None and bundle is not None
    host_agent_module.Handler.collector = host_agent_module.Collector("/", smart=False)
    host_agent_module.Handler.token = None
    host_agent_module.Handler.bundle = bundle
    httpd = ThreadingHTTPServer(("127.0.0.1", 0), host_agent_module.Handler)
    httpd.socket = bundle.context.wrap_socket(httpd.socket, server_side=True)
    thread = threading.Thread(target=httpd.serve_forever, daemon=True)
    thread.start()
    base = f"https://127.0.0.1:{httpd.server_address[1]}"
    target = Target("localhost", base)
    try:
        yield SimpleNamespace(
            store=store,
            state=state,
            material=material,
            bundle=bundle,
            bundle_rev=entry.bundle_rev,
            base=base,
            target=target,
            tls=agent_tls_map(material, store, [target])["localhost"],
        )
    finally:
        httpd.shutdown()
        httpd.server_close()
        # `Handler` is process-wide state: a later test must not inherit this bundle.
        host_agent_module.Handler.bundle = None
        host_agent_module.Handler.token = None
        certs.remove_material(material)


async def test_an_https_agent_answers_the_dashboard_that_presents_its_certificate(
        tls_fleet: SimpleNamespace) -> None:
    """The whole point of the phase, collector side: our cert opens the agent.

    The two fields the agent adds in TLS mode come back too, because task 05's page compares the
    revision `/certs` issued with the revision the host is actually running.
    """
    async with httpx.AsyncClient(timeout=15.0) as client:
        body = await fetch_snapshot(client, tls_fleet.target, tls=tls_fleet.tls)
        assert body["ok"] is True
        assert body["tls"] is True
        assert body["bundle_rev"] == tls_fleet.bundle_rev
        result = await fetch_host_snapshot(client, tls_fleet.target,
                                           HostCollector(RateEstimator()), tls=tls_fleet.tls)
    assert result.payload["agent"]["tls"] is True
    assert result.payload["agent"]["bundle_rev"] == tls_fleet.bundle_rev
    assert result.payload["memory"]["total_bytes"] > 0, "the snapshot is a real one"


async def test_an_agent_signed_by_another_authority_is_not_trusted(
        tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
    """A root that is not ours verifies nothing here, so the dial fails instead of accepting.

    Trusting any certificate would make the whole phase a no-op with extra steps, and the
    failure has to reach the poller as an `UpstreamError` naming the host rather than as a
    traceback out of `ssl`.
    """
    stranger_base = tmp_path / "stranger"
    stranger_base.mkdir()
    stranger = certs.write_material(certs.ensure_certs(DictMeta()), stranger_base)
    foreign = agent_tls_map(stranger, tls_fleet.store, [tls_fleet.target])["localhost"]
    async with httpx.AsyncClient(timeout=15.0) as client:
        with pytest.raises(UpstreamError) as caught:
            await fetch_snapshot(client, tls_fleet.target, tls=foreign)
    assert "unreachable" in caught.value.message
    assert tls_fleet.base in caught.value.message
    certs.remove_material(stranger)


async def test_dashboard_material_that_cannot_be_read_fails_the_poll(
        tls_fleet: SimpleNamespace, tmp_path: Path) -> None:
    """A missing client-cert file is a red source with a sentence, not a dead poll loop.

    `httpx` loads the CA and the client chain when the client is *built*, so this arrives as a
    `FileNotFoundError` from inside the fetch. A TLS dial that can only fail with a non-HTTP
    exception is the invariant-4 failure mode: the task would die and the source would look
    healthy while its charts stopped.
    """
    broken = AgentTls(verify=str(tls_fleet.material.ca),
                      cert=(str(tmp_path / "gone-cert.pem"), str(tmp_path / "gone-key.pem")),
                      status=lambda: certs.ACTIVE)
    async with httpx.AsyncClient(timeout=15.0) as client:
        with pytest.raises(UpstreamError, match="unreachable") as caught:
            await fetch_snapshot(client, tls_fleet.target, tls=broken)
    assert "FileNotFoundError" in caught.value.message
    assert tls_fleet.base in caught.value.message, "the failure still names the host's URL"


async def test_an_https_agent_is_not_readable_without_the_material(tls_fleet: SimpleNamespace) -> None:
    """A shared client with no client certificate gets nowhere — "only this dashboard can read
    an agent" has to hold for the collector's own client too, not just for a stranger's.
    """
    async with httpx.AsyncClient(timeout=15.0) as client:
        with pytest.raises(UpstreamError) as caught:
            await fetch_snapshot(client, tls_fleet.target)
    assert "unreachable" in caught.value.message


async def test_a_revoked_agent_is_refused_before_anything_is_dialled(
        tls_fleet: SimpleNamespace, monkeypatch: pytest.MonkeyPatch) -> None:
    """The revocation gate is in front of the network, not behind it.

    The spy replaces `httpx.AsyncClient.get` on the class, so it catches the shared client and
    the per-dial TLS client alike: what is being proved is that no request is built at all, not
    that one particular client was unused. A dial-then-refuse would still put a socket on a host
    the operator just struck out.
    """
    calls: list[str] = []

    async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
        calls.append(str(url))
        raise AssertionError(f"a revoked agent was dialled at {url}")

    monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
    certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
    async with httpx.AsyncClient(timeout=15.0) as client:
        with pytest.raises(UpstreamError, match="agent certificate revoked") as caught:
            await fetch_snapshot(client, tls_fleet.target, tls=tls_fleet.tls)
        with pytest.raises(UpstreamError, match="agent certificate revoked"):
            await fetch_disk_snapshot(client, tls_fleet.target, DisksCollector(),
                                      tls=tls_fleet.tls)
    assert calls == [], "a revoked host must not be dialled on either cadence"
    assert "revoked" in caught.value.message


async def test_the_revocation_answer_is_a_live_read_of_the_store(tls_fleet: SimpleNamespace) -> None:
    """`status` is a callable because the operator clicks revoke while this process is running.

    A status string read at startup would keep dialling a struck-out host until the next
    restart, and a dashboard that needs a restart to enforce a security decision does not
    enforce it.
    """
    tls = tls_fleet.tls
    assert tls.status() == certs.ACTIVE
    assert tls.revoked is False
    certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
    assert tls.status() == certs.REVOKED
    assert tls.revoked is True, "the same object, with no rebuild and no restart"


def test_the_revocation_is_per_host_and_the_material_is_shared(tls_fleet: SimpleNamespace) -> None:
    """One identity for every dial, one revocation answer per name.

    `certs.py` gives the dashboard a single client certificate, so `verify`/`cert` are the same
    paths for every host; what differs is whose identity the operator struck out. A name with no
    entry is active rather than refused — never-issued is not the same fact as revoked, and
    answering revoked for it would take a working host red with no operator action behind it
    (invariant 3).
    """
    targets = [tls_fleet.target, Target("deskwork", "https://10.1.0.51:9120")]
    mapping = agent_tls_map(tls_fleet.material, tls_fleet.store, targets)
    assert mapping["localhost"].verify == str(tls_fleet.material.ca)
    assert mapping["localhost"].cert == mapping["deskwork"].cert == (
        str(tls_fleet.material.client_cert), str(tls_fleet.material.client_key))
    certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
    assert mapping["localhost"].status() == certs.REVOKED
    assert mapping["deskwork"].status() == certs.ACTIVE
    assert agent_tls_map(None, tls_fleet.store, targets) == {}, (
        "no material means no TLS dial, exactly the pre-phase-15 path")


def scripted_refusal(body: Any, *, status: int = 403) -> AgentTls:
    """An agent that answers every `/snapshot` with a refusal, in whatever shape it sent the body."""
    def handler(request: httpx.Request) -> httpx.Response:
        if isinstance(body, dict):
            return httpx.Response(status, json=body)
        return httpx.Response(status, text=body)

    return AgentTls(verify="/unused/root.pem", cert=("/unused/cert.pem", "/unused/key.pem"),
                    status=lambda: certs.ACTIVE, transport=httpx.MockTransport(handler))


@pytest.mark.parametrize("body,expected", [
    ({"ok": False, "detail": "client certificate required"}, "client certificate required"),
    ({"ok": False, "detail": "client certificate revoked"}, "client certificate revoked"),
    ("<html><body>403 Forbidden</body></html>", "403 Forbidden"),
    ("", "no reason given"),
])
async def test_a_refusal_from_a_protected_agent_quotes_its_own_words(
        body: Any, expected: str) -> None:
    """The 403 sentence names the refusal the *agent* made, redacted and never blank.

    An `UpstreamError` string goes into the health record and the event table, so it has to say
    which of the agent's two refusals this was (a required certificate and a revoked one are two
    different operator actions) while quoting nothing but the URL it was dialling
    (AGENTS.md invariant 6). The `verify`/`cert` paths here do not exist: a scripted refusal
    proves the wording, and the tests above prove the handshake.
    """
    tls = scripted_refusal(body)
    target = Target("gw", "https://SEKRIT-AGENT-TOKEN@agent.test")
    async with httpx.AsyncClient(timeout=15.0) as client:
        with pytest.raises(UpstreamError) as caught:
            await fetch_snapshot(client, target, tls=tls)
    message = caught.value.message
    assert "agent refused the client certificate" in message
    assert expected in message
    assert "SEKRIT-AGENT-TOKEN" not in message, "the token leaked into a browser-readable string"
    assert "https://agent.test/snapshot" in message, "and it still says where"


@pytest.mark.parametrize("status", [401, 403, 500])
async def test_a_refusal_from_an_unprotected_agent_keeps_the_status_wording(status: int) -> None:
    """The certificate sentence is only for a dial where a certificate was presented.

    A token-mode host answers 401 for a bad token and would answer 403 for something that has
    nothing to do with certificates; calling that "refused the client certificate" would send
    the operator to `/certs` for a token problem.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, text="nope")

    target = Target("gw", "http://agent.test")
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        with pytest.raises(UpstreamError) as caught:
            await fetch_snapshot(client, target,
                                 tls=AgentTls(verify="/unused/root.pem",
                                              cert=("/unused/c.pem", "/unused/k.pem"),
                                              status=lambda: certs.ACTIVE))
    if status == 401:
        assert "token" in caught.value.message
    else:
        assert f"HTTP {status}" in caught.value.message
    assert "client certificate" not in caught.value.message


async def test_a_token_mode_host_still_uses_the_shared_client(tls_fleet: SimpleNamespace) -> None:
    """An `http://` entry is dialled exactly as it was before phase 15.

    The material exists in this deployment (one CA, one client pair, every host), so the scheme
    is the only thing that may decide the path — a rollout is per host, and the plain hosts
    cannot be allowed to pick up a client certificate they never asked for.
    """
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(str(request.url))
        return httpx.Response(200, json=agent_body())

    plain = Target("localhost", "http://agent.test")
    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        body = await fetch_snapshot(client, plain, tls=tls_fleet.tls)
    assert body["ok"] is True
    assert seen == ["http://agent.test/snapshot"], "the shared client, the shared transport"


async def test_a_revoked_agent_fails_both_jobs_with_one_sentence(store: Any,
                                                                 tls_fleet: SimpleNamespace,
                                                                 monkeypatch: pytest.MonkeyPatch) -> None:
    """The phase's criterion, at the level the health record is written: `host:` and `disks:`.

    The two jobs dial one agent on two cadences, so a revocation has to fail both with the same
    sentence (one dead source must not blink the other, and both must stop talking to it).
    """
    calls: list[str] = []

    async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
        calls.append(str(url))
        raise AssertionError(f"a revoked agent was dialled at {url}")

    monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
    certs.revoke(tls_fleet.store, tls_fleet.state, "localhost")
    async with httpx.AsyncClient(timeout=15.0) as client:
        jobs = host_jobs([tls_fleet.target], client, HostCollector(RateEstimator()), 10.0,
                         tls={"localhost": tls_fleet.tls})
        jobs += disks_jobs([tls_fleet.target], client, DisksCollector(), 300.0,
                           tls={"localhost": tls_fleet.tls})
        assert [job.key for job in jobs] == ["host:localhost", "disks:localhost"]
        for job in jobs:
            assert await job.run_once(store) is False
    assert [job.state.error for job in jobs] == [
        "UpstreamError: localhost: agent certificate revoked"] * 2
    assert calls == []


async def test_host_and_disk_jobs_hand_each_host_its_own_material(
        monkeypatch: pytest.MonkeyPatch) -> None:
    """The plumbing, pinned: a host in the map gets its entry, a host outside it gets `None`.

    Two hosts and one wrong-key miss is the minimum that shows the lookup is by name rather than
    "whatever was built first", which is how a revoked host would keep being dialled with
    another host's revocation answer.
    """
    seen: dict[str, Any] = {}

    def record(name: str) -> Any:
        async def scrape(client: Any, target: Target, *_args: Any, **kwargs: Any) -> Result:
            seen[f"{name}:{target.name}"] = kwargs.get("tls")
            return Result(payload={}, samples=[], events=[])

        return scrape

    fetch_host = record("system")
    fetch_disk = record("disks")
    monkeypatch.setattr("dashboards.collectors.fetch_host_snapshot", fetch_host)
    monkeypatch.setattr("dashboards.collectors.fetch_disk_snapshot", fetch_disk)
    entry = AgentTls(verify="/ca.pem", cert=("/c.pem", "/k.pem"),
                     status=lambda: certs.ACTIVE)
    targets = [Target("gw", "https://gw.test"), Target("desk", "http://desk.test")]
    async with httpx.AsyncClient() as client:
        for job in (host_jobs(targets, client, HostCollector(RateEstimator()), 10.0,
                              tls={"gw": entry})
                    + disks_jobs(targets, client, DisksCollector(), 300.0, tls={"gw": entry})):
            await job.fn()
    assert seen == {"system:gw": entry, "system:desk": None,
                    "disks:gw": entry, "disks:desk": None}


async def test_the_tls_self_report_reaches_both_payloads(disk_snapshot: Any) -> None:
    """`tls` + `bundle_rev` travel into `/api/system/snapshot` and `/api/disks/snapshot`.

    Task 05's page has to answer "is the installed bundle the one I just issued?", and both
    dashboards read the same snapshot, so both carry the deployed revision under the same
    `agent` key. Nothing grades it: a revision is an identity counter, not a threshold.
    """
    def stamped(snapshot: dict[str, Any]) -> dict[str, Any]:
        """What a TLS-mode agent's `/snapshot` looks like on the wire."""
        snapshot["tls"] = True
        snapshot["bundle_rev"] = 3
        return snapshot

    target = Target("driveripper", "http://agent.test")
    async with client_for(lambda request: httpx.Response(200,
                                                        json=stamped(agent_body()))) as client:
        system = (await fetch_host_snapshot(client, target,
                                           HostCollector(RateEstimator()))).payload
    assert system["agent"]["tls"] is True
    assert system["agent"]["bundle_rev"] == 3

    # the `/disks` payload is built from a capture that also carries SMART blocks, so it gets
    # its own stamped snapshot rather than sharing the one above
    with_smart = disk_snapshot(mutate=stamped)
    async with client_for(lambda request: httpx.Response(200, json=with_smart)) as client:
        disks = (await fetch_disk_snapshot(client, target, DisksCollector())).payload
    assert disks["agent"] == {"tls": True, "bundle_rev": 3}


async def test_a_token_mode_snapshot_gains_no_tls_fields() -> None:
    """The captures are token-mode agents, and their payloads must stay byte-identical.

    `false` and `0` would be numbers the agent never published (rule 2), and an empty block
    would be a container that means nothing.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json=agent_body())

    target = Target("driveripper", "http://agent.test")
    async with client_for(handler) as client:
        system = (await fetch_host_snapshot(client, target, HostCollector(RateEstimator()))).payload
    assert "tls" not in system["agent"] and "bundle_rev" not in system["agent"]
    assert system["agent"]["version"], "the block is still the block it was"


# --------------------------------------------------------------------------- #
# post_route (the app's second upstream write, phase 33 task 05)
# --------------------------------------------------------------------------- #
def recorded_post(handler: Any) -> tuple[httpx.AsyncClient, list[httpx.Request]]:
    """The `post_route` shape of `client_for`: a scripted answer, every request kept."""
    seen: list[httpx.Request] = []

    def recording(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        return handler(request)

    return httpx.AsyncClient(transport=httpx.MockTransport(recording)), seen


async def test_post_route_posts_the_wire_body_and_returns_the_helpers_object_verbatim() -> None:
    """The wire is task 01's `--serve` contract: `id`, `action`, `dry_run`, and nothing else.

    `danger` is the dashboard's typed confirmation, not a wire key — the agent's writer takes a
    body with no `danger`, and a header or field smuggled in is a contract break the agent would
    answer with its own 422. The answer comes back untouched: an `applied: false` must not
    become a success on the way over (rule 2), so the object is handed over, not rebuilt.
    """
    client, seen = recorded_post(lambda request: httpx.Response(
            200, json={"applied": True, "changed": True, "reloaded": True}))
    async with client:
        answer = await post_route(client, Target("gateway", "http://agent.test"),
                                 {"id": "brain.reeseapps.com", "action": "disable",
                                  "dry_run": False},
                                 timeout=35.0)
    assert answer == {"applied": True, "changed": True, "reloaded": True}
    assert len(seen) == 1
    request = seen[0]
    assert request.method == "POST" and request.url.path == "/route"
    assert json.loads(request.content) == {"id": "brain.reeseapps.com",
                                         "action": "disable", "dry_run": False}
    assert "authorization" not in {key.lower() for key in request.headers}, (
        "the plain path carries only the target's own headers")


async def test_post_route_sends_the_token_on_the_token_path() -> None:
    """A host nobody has bundled dials the compatibility way: the header, exactly as the read sends it.

    `parse_targets` is what moves the userinfo token into the header; a directly built target
    hands `post_route` the same shape, so the pin is the header arriving on the wire.
    """
    client, seen = recorded_post(lambda request: httpx.Response(
            200, json={"applied": False, "changed": False}))
    target = Target("gateway", "http://agent.test",
                   headers={"Authorization": "Bearer SEKRIT-AGENT-TOKEN"})
    async with client:
        await post_route(client, target, {"id": "r", "action": "enable", "dry_run": True},
                        timeout=35.0)
    assert seen[0].headers["Authorization"] == "Bearer SEKRIT-AGENT-TOKEN"


async def test_post_route_sends_no_authorization_on_the_certificate_path() -> None:
    """The bundle is the credential on a `https://` target, and the write is stricter than the read.

    `parse_targets` has already moved any token into the header, so a bundled host's target has a
    clean URL and no header — and the write must add nothing of its own: a header the read would
    have sent (a compatibility host's token) is a second opinion the write refuses, so the
    certificate path sends the body and the certificate and nothing else.
    """
    seen: list[list[tuple[bytes, bytes]]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request.headers.raw)
        return httpx.Response(200, json={"applied": False, "dry_run": True, "changed": True})

    tls = AgentTls(verify="/unused/root.pem", cert=("/unused/cert.pem", "/unused/key.pem"),
                  status=lambda: certs.ACTIVE, transport=httpx.MockTransport(handler))
    async with client_for(lambda request: httpx.Response(200, json={})) as shared:
        answer = await post_route(shared, Target("gateway", "https://agent.test"),
                                 {"id": "r", "action": "disable", "dry_run": True},
                                 timeout=35.0, tls=tls)
    assert answer["dry_run"] is True
    assert len(seen) == 1
    assert not any(key == b"authorization" for key, _ in seen[0]), (
        "a token next to a certificate is the second opinion the write refuses")


class PostTimeoutRecorder(TimeoutRecordingClient):
    """The same seam as `TimeoutRecordingClient`, for the write's own budget argument."""

    async def post(self, url: Any, **kwargs: Any) -> httpx.Response:  # type: ignore[override]
        self.timeouts.append(kwargs.get("timeout", "<unset>"))
        return await super().post(url, **kwargs)


async def test_post_route_carries_the_write_budget_and_never_a_bare_none() -> None:
    """`GATEWAY_WRITE_TIMEOUT_SECONDS` is the write's own 35 s, passed per request.

    A `None` timeout is *no timeout at all* to httpx: a helper stuck on a wedged `systemctl
    reload` would hold a request open for good. The sentinel keeps the client default, a number
    keeps the write's own budget.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(200, json={"applied": False, "changed": False})

    client = PostTimeoutRecorder(transport=httpx.MockTransport(handler),
                                timeout=10.0, base_url="http://agent.test")
    async with client:
        await post_route(client, Target("gw", "http://agent.test"),
                        {"id": "r", "action": "disable", "dry_run": False}, timeout=35.0)
        await post_route(client, Target("gw", "http://agent.test"),
                         {"id": "r", "action": "disable", "dry_run": False})
    assert client.timeouts[-2:] == [35.0, httpx.USE_CLIENT_DEFAULT]


async def test_post_route_timeout_is_a_504_that_names_the_redacted_url() -> None:
    """The message lands in the event table and the browser reads it (invariant 6): the URL is
    the redacted one and the status is the 504 a timeout is, so a slow helper is not a 502.
    """
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("read timed out")

    target = Target("gw", "http://SEKRIT-AGENT-TOKEN@agent.test")
    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as caught:
            await post_route(client, target, {"id": "r", "action": "disable", "dry_run": False})
    assert caught.value.status == 504
    assert "SEKRIT-AGENT-TOKEN" not in caught.value.message
    assert "agent.test/route" in caught.value.message
    assert "timed out" in caught.value.message


async def test_post_route_unreachable_is_a_502_with_the_redacted_url() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("no route to host")

    target = Target("gw", "https://SEKRIT-AGENT-TOKEN@agent.test")
    tls = AgentTls(verify="/unused/root.pem", cert=("/unused/cert.pem", "/unused/key.pem"),
                  status=lambda: certs.ACTIVE, transport=httpx.MockTransport(handler))
    async with client_for(lambda request: httpx.Response(200, json={})) as client:
        with pytest.raises(UpstreamError) as caught:
            await post_route(client, target, {"id": "r", "action": "enable", "dry_run": False},
                            tls=tls)
    assert caught.value.status == 502
    assert "SEKRIT-AGENT-TOKEN" not in caught.value.message


async def test_post_route_401_on_the_token_path_is_about_the_token() -> None:
    """A 401 the token path gets is the rotated-token case: the sentence says so, and the URL
    is still the redacted one."""
    async with client_for(lambda request: httpx.Response(401, json={"detail": "bad token"})) as client:
        with pytest.raises(UpstreamError) as caught:
            await post_route(client, Target("gw", "http://SEKRIT-AGENT-TOKEN@agent.test"),
                            {"id": "r", "action": "disable", "dry_run": False})
    assert "rejected the token (401)" in caught.value.message
    assert "SEKRIT-AGENT-TOKEN" not in caught.value.message


async def test_post_route_refusal_carries_the_helpers_own_word() -> None:
    """A `>= 400` answer is the agent's own refusal — a `403` here is the write gate's own word
    (`writes are not enabled on this agent`), and paraphrasing it would be the dashboard
    claiming to know more than the peer said."""
    async with client_for(lambda request: httpx.Response(
            409, json={"detail": "no such id: brain.reeseapps.com"})) as client:
        with pytest.raises(UpstreamError) as caught:
            await post_route(client, Target("gateway", "http://agent.test"),
                            {"id": "brain.reeseapps.com", "action": "disable", "dry_run": False})
    assert caught.value.status == 502
    assert "HTTP 409" in caught.value.message
    assert "no such id: brain.reeseapps.com" in caught.value.message


async def test_post_route_an_unreadable_answer_is_an_error_with_a_bounded_slice() -> None:
    async with client_for(lambda request: httpx.Response(
            200, text="not json at all " + "x" * 200)) as client:
        with pytest.raises(UpstreamError, match="unreadable") as caught:
            await post_route(client, Target("gateway", "http://agent.test"),
                            {"id": "r", "action": "disable", "dry_run": False})
    assert "x" * 200 not in caught.value.message, "the slice is bounded, not the whole body"
    async with client_for(lambda request: httpx.Response(200, json=[1, 2])) as client:
        with pytest.raises(UpstreamError, match="non-object"):
            await post_route(client, Target("gateway", "http://agent.test"),
                            {"id": "r", "action": "disable", "dry_run": False})


async def test_post_route_refuses_a_revoked_agent_before_any_dial() -> None:
    """The revocation gate is in front of the network, for the write as for the read.

    The scripted transport records what it receives, and a `status` callable that answers
    `REVOKED` is what `certs.agent_status` would answer after the operator's click: the
    refusal must come before the transport sees a request, not after.
    """
    seen: list[httpx.Request] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request)
        return httpx.Response(200, json={"applied": True})

    tls = AgentTls(verify="/unused/root.pem", cert=("/unused/cert.pem", "/unused/key.pem"),
                  status=lambda: certs.REVOKED, transport=httpx.MockTransport(handler))
    async with client_for(lambda request: httpx.Response(200, json={})) as client:
        with pytest.raises(UpstreamError, match="agent certificate revoked") as caught:
            await post_route(client, Target("gateway", "https://agent.test"),
                            {"id": "r", "action": "disable", "dry_run": False}, tls=tls)
    assert seen == [], "a revoked host must not be written"
    assert "revoked" in caught.value.message

# LLM engines
# --------------------------------------------------------------------------- #
GUFO = (FIXTURES / "metrics_gufo.txt").read_text()
VLLM = (FIXTURES / "metrics_vllm.txt").read_text()


def bump(text: str, series: str, amount: float) -> str:
    lines = []
    for line in text.splitlines():
        if line.startswith(series + " "):
            head, _, value = line.rpartition(" ")
            line = f"{head} {float(value) + amount}"
        lines.append(line)
    return "\n".join(lines) + "\n"


async def test_first_gufo_poll_has_no_measured_rate_yet() -> None:
    """One scrape cannot produce Δ/Δt. The cell is n/a, not a stale single-request gauge."""
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=GUFO)
        return httpx.Response(404, text="nope")

    async with client_for(handler) as client:
        result = await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator(),
                                    with_slots=True)
    decode = result.payload["metrics"]["decode_tokens_per_second"]
    assert decode["value"] is None
    assert decode["source"] is None
    # `active` is the /metrics gauge, verbatim the fixture's own number — the capture
    # may have been taken with a request in flight (volatile between re-captures)
    assert result.payload["sessions"]["active"] == \
        Prom(GUFO).scalar(["llamacpp:requests_processing"])
    assert "no /slots answer" in result.payload["sessions"]["detail"]


async def test_second_gufo_poll_measures_throughput_over_a_window() -> None:
    state = {"text": GUFO}

    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=state["text"])
        if request.url.path == "/slots":
            return httpx.Response(200, json=[{"index": 0, "n_ctx": 8192, "predicted": 12}])
        return httpx.Response(404)

    # min_window=0: the two fetches are microseconds apart (no real loop polls this
    # fast); this test pins that the measured cell divides by the window it labels.
    rates = RateEstimator(span=10.0, min_window=0.0)
    async with client_for(handler) as client:
        await fetch_engine(client, "gufo", "http://engine.test", "", rates)
        state["text"] = bump(GUFO, "llamacpp:tokens_predicted_total", 40.0)
        result = await fetch_engine(client, "gufo", "http://engine.test", "", rates,
                                    with_slots=True)
    decode = result.payload["metrics"]["decode_tokens_per_second"]
    assert decode["source"].startswith("computed:Δllamacpp:")
    assert decode["derived"] is True
    assert decode["window"] > 0
    assert decode["value"] == pytest.approx(40.0 / decode["window"], rel=0.05)
    assert result.payload["sessions"]["source"] == "gufo /slots"
    assert result.payload["sessions"]["list"][0]["index"] == 0


async def test_gufo_slots_alias_is_used_when_plain_slots_is_missing() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/metrics":
            return httpx.Response(200, text=GUFO)
        if request.url.path == "/v1/slots":
            return httpx.Response(200, json={"slots": [{"index": 1}]})
        return httpx.Response(404)

    async with client_for(handler) as client:
        result = await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator(),
                                    with_slots=True)
    assert result.payload["sessions"]["list"] == [{"index": 1}]


async def test_vllm_metrics_parse_and_carry_the_api_key() -> None:
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request.headers.get("authorization", ""))
        return httpx.Response(200, text=VLLM)

    async with client_for(handler) as client:
        result = await fetch_engine(client, "vllm", "http://engine.test", "SECRET",
                                    RateEstimator())
    assert seen == ["Bearer SECRET"]
    metrics = result.payload["metrics"]
    assert metrics["num_requests_running"]["value"] is not None
    # vLLM publishes no per-session list: the page is told, rather than shown an empty table
    assert "counts, not sessions" in result.payload["sessions"]["detail"]
    assert any(s.metric.startswith("llm.") for s in result.samples)


@pytest.mark.parametrize("status", [404, 500, 503])
async def test_engine_http_errors_name_the_status(status) -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        return httpx.Response(status, text="")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError, match=f"HTTP {status}"):
            await fetch_engine(client, "vllm", "http://engine.test", "", RateEstimator())


async def test_engine_timeout_is_504() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ReadTimeout("engine busy")

    async with client_for(handler) as client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_engine(client, "gufo", "http://engine.test", "", RateEstimator())
    assert excinfo.value.status == 504


# --------------------------------------------------------------------------- #
# radiance: the /stats sidecar (phase 34, task 04)
# --------------------------------------------------------------------------- #
RADIANCE = (FIXTURES / "metrics_radiance.txt").read_text()


def _radiance_stats() -> dict[str, Any]:
    """The capture's /stats document (the `#` header lines are the script's, not JSON)."""
    body = "\n".join(line for line in
                     (FIXTURES / "stats_radiance.json").read_text().splitlines()
                     if not line.startswith("#"))
    return json.loads(body)


def fake_engine(metrics: httpx.Response | BaseException,
                stats: httpx.Response | BaseException | None = None
                ) -> tuple[httpx.AsyncClient, list[str]]:
    """One fake engine: a scripted `/metrics` answer, and a scripted `/stats` sidecar answer
    if the job dials for it; every requested path is kept, so a retry or an alias shows up.
    """
    seen: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(request.url.path)
        if request.url.path == "/metrics":
            if isinstance(metrics, BaseException):
                raise metrics
            return metrics
        if request.url.path == "/stats":
            assert stats is not None, "the job dialled /stats without one scripted"
            if isinstance(stats, BaseException):
                raise stats
            return stats
        return httpx.Response(404)

    return httpx.AsyncClient(transport=httpx.MockTransport(handler)), seen


async def test_a_configured_radiance_alone_builds_exactly_one_llm_job() -> None:
    """Task 02's registry shape, extended with the job's own fields: a configured
    `RADIANCE_BASE_URL` (gufo and vLLM unset) is exactly one job, with the LLM dashboard,
    the registry's display name and the LLM cadence — the same three facts `llm:gufo` and
    `llm:vllm` already carried."""
    settings = Settings(radiance_url="http://radiance.test:8000")
    async with httpx.AsyncClient() as client:
        jobs = llm_jobs(settings, client, RateEstimator())
    assert len(jobs) == 1
    job = jobs[0]
    assert job.key == "llm:radiance"
    assert job.dashboard == "llm"
    assert job.label == "radiance"
    assert job.interval == settings.llm_interval, "the LLM cadence, not the host's"


async def test_the_hub_wires_the_radiance_job_with_the_llm_cadence_and_estimator(
        store: Any, monkeypatch: pytest.MonkeyPatch) -> None:
    """`app.py` needed no change — this is the check, asserted by running the job through
    its own `fn` rather than by comment: the `llm:radiance` job flows into the hub from
    `llm_jobs` with the LLM interval and the LLM-span estimator (one estimator per
    measuring cadence, phase 23), and the sidecar flag reaches `fetch_engine` from the
    job's call alone."""
    seen: dict[str, Any] = {}

    async def spy_fetch(client: Any, engine: str, base_url: str, api_key: str,
                       rates: RateEstimator, **kwargs: Any) -> Result:
        seen["engine"] = engine
        seen["rates"] = rates
        seen["kwargs"] = kwargs
        return Result(payload={"engine": engine})

    monkeypatch.setattr("dashboards.collectors.fetch_engine", spy_fetch)
    settings = Settings(radiance_url="http://radiance.test:8000")
    async with httpx.AsyncClient() as client:
        hub, _ = build_hub(settings, store, client)
    assert "llm:radiance" in hub.jobs
    job = hub.jobs["llm:radiance"]
    assert job.interval == settings.llm_interval
    assert await job.run_once(store) is True
    assert seen["engine"] == "radiance"
    assert seen["rates"].span == settings.llm_interval, "the LLM-span estimator"
    assert seen["kwargs"] == {"with_slots": False, "with_stats": True}


@pytest.mark.parametrize("settings, expected", [
    (Settings(gufo_url="http://gufo.test:8731"), ["/metrics", "/slots"]),
    (Settings(vllm_url="http://vllm.test:8000"), ["/metrics"]),
    (Settings(radiance_url="http://radiance.test:8000"), ["/metrics", "/stats"]),
])
async def test_each_engine_job_dials_only_its_own_sidecars(
        settings: Settings, expected: list[str]) -> None:
    """The grep criterion, proven by behaviour: `llm_jobs`' flag is what makes
    `fetch_engine` dial `/stats` — the gufo job's `/slots` pass and the vLLM job's bare
    `/metrics` read behave exactly as before (gufo's alias aside, a 200 `/slots` answers
    before `/v1/slots` is tried), and radiance is the only engine that dials the sidecar,
    exactly once and on its unaliased path."""
    text_by_host = {"gufo.test": GUFO, "vllm.test": VLLM, "radiance.test": RADIANCE}
    paths: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        paths.append(request.url.path)
        if request.url.path == "/metrics":
            return httpx.Response(200, text=text_by_host[request.url.host])
        if request.url.path == "/slots":
            return httpx.Response(200, json=[{"index": 0}])
        if request.url.path == "/stats":
            return httpx.Response(200, json=_radiance_stats())
        return httpx.Response(404)

    async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as client:
        jobs = llm_jobs(settings, client, RateEstimator())
        assert len(jobs) == 1
        result = await jobs[0].fn()
    assert isinstance(result, Result)
    assert paths == expected, f"{jobs[0].key}: the job dials only its own sidecars"


@pytest.mark.parametrize("stats_failure", ["http-500", "non-json", "json-list", "connection-error"])
async def test_a_dead_or_partial_stats_sidecar_keeps_the_metrics_poll_healthy(
        stats_failure: str) -> None:
    """Invariant 4, the sidecar's own arm: a `/stats` that is dead, 500, non-JSON, or a
    JSON that is not a dict degrades the session arm (task 03's missing-stats fallback —
    the counts from the /metrics gauges, the cap n/a, never 0) and never takes the
    `/metrics` poll down: one best-effort read, no retry, no second dial, no alias, no
    `UpstreamError`."""
    if stats_failure == "http-500":
        stats: Any = httpx.Response(500, text="sidecar down")
    elif stats_failure == "non-json":
        stats = httpx.Response(200, text="not json at all")
    elif stats_failure == "json-list":
        stats = httpx.Response(200, json=[1, 2])
    else:
        stats = httpx.ConnectError("sidecar down")
    client, seen = fake_engine(httpx.Response(200, text=RADIANCE), stats)
    async with client:
        result = await fetch_engine(client, "radiance", "http://radiance.test:8000",
                                   "", RateEstimator(), with_stats=True)
    assert result.payload["engine"] == "radiance"
    assert result.payload["metrics"]["num_requests_running"]["value"] == \
        Prom(RADIANCE).scalar(["vllm:num_requests_running"]), \
        "the /metrics answer is the poll's one truth, verbatim the fixture's gauge"
    assert "stats" not in result.payload, "an absent or non-dict sidecar carries nothing"
    detail = result.payload["sessions"]["detail"]
    assert "did not answer" in detail, "the session arm names the missing sidecar"
    assert "gauges" in detail, "…and the fallback counts' different provenance"
    assert result.payload["sessions"]["queue_cap"] is None, "n/a, never 0 (rule 2)"
    assert [path for path in seen if path == "/stats"] == ["/stats"], (
        "one best-effort read, no retry, no second dial")


async def test_a_healthy_stats_sidecar_travels_into_the_result() -> None:
    client, seen = fake_engine(httpx.Response(200, text=RADIANCE),
                              httpx.Response(200, json=_radiance_stats()))
    async with client:
        result = await fetch_engine(client, "radiance", "http://radiance.test:8000",
                                   "", RateEstimator(), with_stats=True)
    block = result.payload["stats"]
    assert block["model"] == "juggernaut"
    for key in ("version", "uptime_s", "startup_s", "total_requests", "cards"):
        assert key in block, "the whitelisted subset the page reads (task 03)"
    assert block["cards"][0]["name"] == "AMD Radeon AI PRO R9700"
    assert "vram_used_pct" in block["cards"][0], "the one computed number in the block"
    sessions = result.payload["sessions"]
    assert sessions["source"] == "radiance /stats"
    assert sessions["queue_cap"] == _radiance_stats()["queue_cap"]
    if not _radiance_stats()["requests"]:
        assert "counts, not sessions" in sessions["detail"], "the capture is at rest"
    assert seen == ["/metrics", "/stats"], "one unaliased read, nothing else"


async def test_a_dead_metrics_is_the_upstream_error_and_the_sidecar_never_masks_it() -> None:
    """A 500 `/metrics` beside a perfect `/stats`: the failure is the metrics one, and
    the sidecar is never even consulted — consulted first it could mask the real failure,
    consulted after it would only be wasted work on a poll that is already failed."""
    client, seen = fake_engine(httpx.Response(500, text="boom"),
                              httpx.Response(200, json=_radiance_stats()))
    async with client:
        with pytest.raises(UpstreamError) as excinfo:
            await fetch_engine(client, "radiance", "http://radiance.test:8000",
                               "", RateEstimator(), with_stats=True)
    assert "HTTP 500" in excinfo.value.message
    assert "/metrics" in excinfo.value.message
    assert "/stats" not in seen, "a failed metrics dial never reaches for the sidecar"


async def test_a_failing_radiance_never_echoes_credentials_in_its_url() -> None:
    """Invariant 6 for the radiance dial, the engine-fetch precedent extended: a
    `RADIANCE_BASE_URL` written with a userinfo token fails with the redacted URL, and
    the key-free form leaks nothing either — there is nothing to leak."""
    client, _ = fake_engine(httpx.ConnectError("connection refused"))
    async with client:
        with pytest.raises(UpstreamError, match="unreachable") as excinfo:
            await fetch_engine(client, "radiance",
                               "http://SEKRIT-RADIANCE-TOKEN@radiance.test:8000", "",
                               RateEstimator(), with_stats=True)
    assert "SEKRIT-RADIANCE-TOKEN" not in excinfo.value.message
    assert "http://radiance.test:8000/metrics" in excinfo.value.message


async def test_the_job_list_never_dials_an_unconfigured_engine(
        monkeypatch: pytest.MonkeyPatch) -> None:
    """The registry's gate, spy-pinned at the dial layer (the existing spy pattern): with
    every engine unset, `llm_jobs` builds nothing, and nothing that was built dials
    anything — an unconfigured engine has no job to fail and no dial to hang (invariant 3)."""
    calls: list[str] = []

    async def spy_get(self: Any, url: Any, **kwargs: Any) -> Any:
        calls.append(str(url))
        raise AssertionError(f"an unconfigured engine was dialled at {url}")

    monkeypatch.setattr(httpx.AsyncClient, "get", spy_get)
    settings = Settings()
    async with httpx.AsyncClient() as client:
        jobs = llm_jobs(settings, client, RateEstimator())
        for job in jobs:
            await job.fn()
    assert jobs == []
    assert calls == []


# --------------------------------------------------------------------------- #
# NUT monitor
# --------------------------------------------------------------------------- #
class FakeSimulator:
    """Stands in for NutSimulator: hands back readings we control."""

    def __init__(self, readings: list[nut.UpsReading]) -> None:
        self.readings = readings

    async def read(self) -> list[nut.UpsReading]:
        return list(self.readings)


def reading(status: str, **vars: str) -> nut.UpsReading:
    item = nut.UpsReading(name="cyberpower", description="CP1350PFCLCD", ts=1_700_000_000.0)
    item.vars = {"ups.status": status, "battery.charge": "92", "ups.load": "35",
                 "battery.runtime": "1800", **vars}
    item.commands = ["beeper.mute", "test.panel.start", "test.battery.start.deep"]
    return item


class SettingsStub:
    nut_host, nut_port, nut_timeout = "ups.test", 3493, 5.0
    nut_user = nut_password = ""
    nut_interval = 5.0
    nut_enabled = True
    actions_allowed = True
    read_only = False


async def test_monitor_payload_labels_the_simulator() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    result = await monitor.collect()
    payload = result.payload
    assert payload["simulated"] is True
    assert payload["server"] is None, "a simulator must not claim a server address"
    assert payload["actions_enabled"] is True
    card = payload["ups"][0]
    assert card["name"] == "cyberpower"
    assert card["status"] == "OL"
    assert card["numbers"]["charge_pct"] == 92.0
    # a driver that reports no battery date, temperature, or self-test result is `unknown`,
    # never a silent "healthy" — and never a failure either
    assert card["health"]["overall"] == "unknown", card["health"]
    assert card["health"]["counts"]["fail"] == 0
    assert card["health"]["counts"]["warn"] == 0
    # commands are offered with their risk, and the dangerous one is present but marked
    risks = {item["command"]: item for item in card["commands"]}
    assert risks["beeper.mute"]["enabled"] is True
    assert risks["test.battery.start.deep"]["risk"] == "dangerous"


async def test_a_fully_report_ups_is_called_healthy() -> None:
    full = reading("OL", **{"battery.temperature": "27.4", "battery.mfr.date": "2024-06-01",
                            "battery.test.result": "OK", "input.voltage": "230"})
    full.vars["battery.charge"] = "100"
    card = (await NutMonitor(SettingsStub(), simulator=FakeSimulator([full])).collect()).payload["ups"][0]
    assert card["health"]["overall"] == "healthy", card["health"]
    assert card["health"]["counts"]["unknown"] == 0


async def test_monitor_sends_history_samples_for_every_ups_metric() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OB")]))
    result = await monitor.collect()
    names = {sample.metric for sample in result.samples}
    assert "ups.charge_pct" in names
    assert "ups.load_pct" in names
    assert all(sample.source == "ups" for sample in result.samples)


async def test_status_changes_become_one_event_each() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    assert (await monitor.collect()).events == []
    monitor.simulator = FakeSimulator([reading("OL")])
    assert (await monitor.collect()).events == [], "an unchanged status is not news"
    monitor.simulator = FakeSimulator([reading("OB")])
    events = (await monitor.collect()).events
    assert len(events) == 1
    assert events[0].kind == "status"
    assert "OL → OB" in events[0].message
    assert events[0].severity == "warn", "on battery is a warning, not a failure"
    monitor.simulator = FakeSimulator([reading("OL FUB")])
    events = (await monitor.collect()).events
    assert events[0].severity == "error", "FUB (the UPS decided to power off) is a failure"


async def test_a_self_test_result_is_reported_when_it_changes() -> None:
    monitor = NutMonitor(SettingsStub(), simulator=FakeSimulator([reading("OL")]))
    await monitor.collect()
    # the variable is `ups.test.result` (drivers also use battery.test.result) — see VAR_MAP
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "INPROGRESS"})])
    assert (await monitor.collect()).events == [], "first sighting is not news"
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "FAIL"})])
    events = (await monitor.collect()).events
    assert [event.kind for event in events] == ["test"]
    assert events[0].severity == "warn"
    monitor.simulator = FakeSimulator([reading("OL", **{"ups.test.result": "OK"})])
    assert (await monitor.collect()).events[0].severity == "info"
    # the word this fleet's CyberPower actually answers (phase 12 capture): the feed and
    # the health check grade one outcome, so a passed test is never an amber line
    monitor.simulator = FakeSimulator(
        [reading("OL", **{"ups.test.result": "Done and passed"})])
    assert (await monitor.collect()).events[0].severity == "info"


async def test_gating_is_applied_where_the_commands_are_built() -> None:
    class ReadOnly(SettingsStub):
        actions_allowed = False

    monitor = NutMonitor(ReadOnly(), simulator=FakeSimulator([reading("OL")]))
    card = (await monitor.collect()).payload["ups"][0]
    assert all(item["enabled"] is False for item in card["commands"])
    assert all(item["reason"] for item in card["commands"]), "every refusal must explain itself"
    assert set(card["advertised_commands"]) == {"beeper.mute", "test.panel.start",
                                                "test.battery.start.deep"}, (
        "the driver still advertises them — the refusal is ours, and must say so")


class FakeNutClient:
    """The real upsd branch without a socket: vars work, the optional lists do not."""

    instances: ClassVar[list[FakeNutClient]] = []

    def __init__(self, host: str, port: int, **kwargs: Any) -> None:
        self.host, self.port, self.kwargs = host, port, kwargs
        self.entered = False
        FakeNutClient.instances.append(self)

    async def __aenter__(self) -> FakeNutClient:
        self.entered = True
        return self

    async def __aexit__(self, *exc: Any) -> None:
        self.entered = False

    async def list_ups(self) -> list[dict[str, str]]:
        return [{"name": "cyberpower", "description": "CP1350PFCLCD"}]

    async def list_vars(self, ups: str) -> dict[str, str]:
        return {"ups.status": "OL", "battery.charge": "100", "battery.temperature": "301.1"}

    async def list_cmds(self, ups: str) -> list[str]:
        raise nut.NutError("this driver exports no commands")

    async def list_rw_vars(self, ups: str) -> list[str]:
        raise nut.NutError("LIST RW denied")


async def test_real_upsd_branch_survives_drivers_that_refuse_the_optional_lists(
        monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(nut, "NutClient", FakeNutClient)
    monitor = NutMonitor(SettingsStub())
    result = await monitor.collect()
    client = FakeNutClient.instances[-1]
    assert client.entered is False, "the connection must not outlive one poll"
    assert client.kwargs["timeout"] == 5.0
    card = result.payload["ups"][0]
    assert card["commands"] == [] and card["advertised_commands"] == []
    assert card["numbers"]["charge_pct"] == 100.0
    # a driver reporting Kelvin is normalised, not charted as 301 °C
    assert card["numbers"]["battery_temperature"] == pytest.approx(301.1 - 273.15, abs=0.01)
    samples = {sample.metric: sample.value for sample in result.samples}
    assert samples["ups.battery_temperature_c"] == pytest.approx(27.95, abs=0.01)


async def test_a_real_server_names_itself_in_the_payload(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(nut, "NutClient", FakeNutClient)
    payload = (await NutMonitor(SettingsStub()).collect()).payload
    assert payload["simulated"] is False
    assert payload["server"] == "ups.test:3493"