phase: 29_gateway_dashboard_page

Phase 29 completion pass is done. Final state:

**Verification (all re-run this pass, no task files touched)**
- `uv run pytest` → **1746 passed, 16 skipped**; `uv run pytest --cov` → **99.05 %** (gate 94 %); `bash .agents/validate.sh` → **rc=0** (ruff clean, pyright `0 errors, 0 warnings`, `node --check` clean); `uv run pytest -m live --live` → **12 passed**.
- Live, read-only: dev server **10/10 sources `ok`** (incl. `gateway`); `data/logs/29-task06-live.cjs` → **49 PASS / 0 FAIL** with per-port inbound counts reconciled against a direct `/proc/net/tcp{,6}` read (divergence 0); `data/logs/verify-29-task06-predates.py` → 0 failed (1.1.0 arm renders `WHY_PREDATES`, no pipes, every tally `n/a`, job still `ok`); phase-27 mobile audit re-run over all nine served pages → **371 PASS / 0 FAIL / 0 FINDING** (`data/logs/29-final-mobile-audit.out`).
- Criteria: (1) unconfigured `GATEWAY_HOST` → `404 {configured:false}` with dial spies + `#gate-empty` sentence ✅; (2) 1.1.0/`--no-proxy` → `supported:false`, no `0`s, never red ✅; (3) live gateway 1.2.0 → pipes animate, totals reconcile ✅; (4) no `severity`/verdict colour anywhere, validate green ✅; (5) docs updated + gap 13 logged ✅; (6) the five existing dashboards, jobs and `/api/summary` unchanged (live control leg) ✅.

**Docs (consistency pass, incl. pre-existing drift)**
- `README.md`: architecture diagram now lists `sources/gateway.py`, the 30 s gateway job and the third `/snapshot` reader; `GATEWAY_HOST` row says *a name, never a URL*; the `POLL_MS_*` row states `/gateway`'s fixed 15 s browser poll; "nine pollers" → per-source cadences; phase-28 "until phase 29" wording → current state; new Testing paragraph for `tests/test_gateway_source.py` (join pinned from the captures, reconciliation identity, `getaddrinfo` spy).
- `AGENTS.md`: phase-28 stop-point wording replaced with the shipped phase-29 state; suite/coverage figures → 1746 / 16 skipped / 99.08 %→99.05 % band and `~99 %`; window-chips paragraph records that `/gateway` has **no** chips (fixed 24 h); new testing-policy bullet for the join; stale page counts fixed (seven → eight tabbed pages / `MARKUP_PAGES`, clock-order and badge-count transcripts annotated); gap 12 now records the phase-29 nine-page audit re-run.
- `tests/test_api.py`: comment/docstring-only fixes for the "seven pages" wording that phase 29 made eight.

**Deviations / notes**
- Nothing implemented: all six task files were already in `complete/`, so this run was the final verification + docs pass; no git index or phase file was touched.
- Coverage wobbles 99.05–99.08 % between runs on a pre-existing timing-dependent line (`collectors.py` `/slots`-alias `except`/`break`), not a phase-29 path (`sources/gateway.py` is 100 %).
- `uv pip list --outdated` still lists upgrades; dependencies did not move in this phase, so they were left alone rather than upgraded mid-verification.
- Historical browser-transcript counts (phase 18/20/25 script runs) were left as recorded history and annotated instead of rewritten.

**Next phase:** none — `.agents/phases/todo/` is empty after 29; further work needs a new phase authored first.

tests/conftest.py

@@ -243,6 +243,76 @@ def disk_snapshot(agent_snapshots: dict[str, dict[str, Any]], smart_shape: Any)
     return build
 
 
+# ---- phase 29: the gateway page (the /gateway source, API and page) -------- #
+
+#: The captured gateway tree, as the paths the agent's own `--root` resolution expects.
+#: These are the verbatim phase-28 captures (`scripts/capture-gateway-config.sh`), so the
+#: config half of a `/gateway` test is the file the real ingress runs, not a memory of it.
+GATEWAY_TREE = {
+    "etc/haproxy/haproxy.cfg": "gateway_haproxy.cfg",
+    "etc/nginx/nginx.conf": "gateway_nginx_main.conf",
+    "etc/nginx/conf.d/http.conf": "gateway_nginx_conf_d_http.conf",
+    "etc/nginx/stream.d/stream.conf": "gateway_nginx_stream_d_stream.conf",
+    "proc/net/tcp": "gateway_proc_net_tcp.txt",
+    "proc/net/tcp6": "gateway_proc_net_tcp6.txt",
+}
+
+GATEWAY_MODES = ("full", "predates", "no_haproxy", "no_nginx", "sockets_silent")
+
+
+@pytest.fixture
+def gateway_snapshot(host_agent_module: Any, tmp_path: Path) -> Any:
+    """`build(mode="full")` -> a 1.2.0-style agent snapshot carrying the three proxy sections.
+
+    Not a hand-written dict: the captured configs and `/proc` tables are laid into a fake
+    `--root` tree and read by the **real agent `Collector`**, so the sections handed to
+    `sources/gateway.py` are the bytes agent 1.2.0 serves. A field rename on either side of the
+    wire therefore fails a source test instead of drawing an empty pipe on the page (rule 7).
+
+    Modes are the degradation arms the page has to render:
+
+    * `full` — all three sections `available: true`, live sockets from the capture
+    * `predates` — none of the three keys at all (an agent older than 1.2.0, or `--no-proxy`)
+    * `no_haproxy` / `no_nginx` — that one config file is absent: `available: false` plus the
+      agent's own **not found** line, the other half of the page untouched
+    * `sockets_silent` — neither `/proc` table readable: `available: null`, so every live count
+      must come back `None` and never `0`
+    """
+    def build(mode: str = "full") -> dict[str, Any]:
+        if mode not in GATEWAY_MODES:
+            raise AssertionError(f"unknown gateway snapshot mode {mode!r}")
+        root = tmp_path / f"root-{mode}"
+        for relative, fixture in GATEWAY_TREE.items():
+            if not fixture:
+                continue
+            target = root / relative
+            target.parent.mkdir(parents=True, exist_ok=True)
+            target.write_text((FIXTURES / fixture).read_text())
+        if mode == "no_haproxy":
+            (root / "etc/haproxy/haproxy.cfg").unlink()
+        if mode == "no_nginx":
+            for path in (root / "etc/nginx/nginx.conf", root / "etc/nginx/conf.d/http.conf",
+                         root / "etc/nginx/stream.d/stream.conf"):
+                path.unlink()
+        if mode == "sockets_silent":
+            (root / "proc/net/tcp").unlink()
+            (root / "proc/net/tcp6").unlink()
+
+        errors: list[str] = []
+        collector = host_agent_module.Collector(root)
+        snapshot: dict[str, Any] = {
+            "ok": True, "schema": 1, "ts": 1_760_000_000.0, "host": "gateway",
+            "agent_version": host_agent_module.AGENT_VERSION, "errors": errors,
+        }
+        if mode != "predates":
+            snapshot["haproxy"] = collector.haproxy(errors)
+            snapshot["nginx"] = collector.nginx(errors)
+            snapshot["sockets"] = collector.sockets(errors)
+        return snapshot
+
+    return build
+
+
 @pytest.fixture
 def metrics_text() -> Any:
     """metrics_text('gufo') / metrics_text('vllm') -> the captured exposition verbatim."""

tests/test_alerts.py

@@ -14,9 +14,10 @@ are the ones that make the landing page trustworthy rather than merely busy:
 
 from __future__ import annotations
 
+import inspect
 from typing import Any
 
-from dashboards import alerts
+from dashboards import alerts, severity
 from dashboards.config import Settings
 from dashboards.sources import nut
 from dashboards.sources.nut_sim import NutSimulator
@@ -481,6 +482,83 @@ def test_a_dead_upsd_is_one_alert_and_not_its_stale_cards_too() -> None:
     assert [item["source"] for item in out] == ["NUT server"]
 
 
+# --------------------------------------------------------------------------- #
+# the gateway page: reachability is the only voice, and a dead agent is still one line
+# --------------------------------------------------------------------------- #
+def gateway_state(**kwargs: Any) -> dict[str, Any]:
+    """The health record of the one `gateway` job (the `nut` shape: key `gateway`, label = host)."""
+    return state("gateway", "gateway", "gateway", **kwargs)
+
+
+def test_a_stale_gateway_job_is_a_warn_measured_in_its_own_thirty_second_poll() -> None:
+    """The pipes are drawn from a 30 s poll, so 60 s of silence is two healthy polls.
+
+    Judging this source by the 10 s system interval would amber the landing page every minute for
+    a job behaving exactly as configured — the same mistake the `disks` rule exists to avoid, one
+    cadence later.
+    """
+    settings = Settings(gateway_interval=30.0)
+    assert alerts.build(settings, [gateway_state(age=60.0)]) == []
+    out = alerts.build(settings, [gateway_state(age=100.0)])
+    assert out[0]["severity"] == "warn"
+    assert out[0]["href"] == "/gateway", "the overview's row must link to the page it describes"
+    assert "every 30s" in out[0]["message"], "the message prints the gateway's own interval"
+
+
+def test_a_gateway_job_that_fails_while_its_host_answers_is_the_only_voice_it_has() -> None:
+    """A wedged proxy/socket section is a fault nothing else can report (the `disks` rule, once more)."""
+    out = alerts.build(SETTINGS, [
+        gateway_state(ok=False, error="UpstreamError: gateway: timed out"),
+        state("host:gateway", "system", "gateway"),
+    ], now=2_000.0)
+    assert len(out) == 1
+    assert out[0]["source"] == "gateway"
+    assert out[0]["href"] == "/gateway"
+    assert "timed out" in out[0]["message"]
+
+
+def test_a_gateway_job_that_failed_because_its_agent_is_down_is_one_line_not_two() -> None:
+    """Three jobs dial one agent; the operator has one problem, so the overview says it once."""
+    down = {"ok": False,
+            "error": "UpstreamError: gateway: http://h:9120/snapshot unreachable (ConnectError)"}
+    out = alerts.build(SETTINGS, [
+        gateway_state(**down),
+        state("disks:gateway", "disks", "gateway", **down),
+        state("host:gateway", "system", "gateway", **down),
+    ], now=2_000.0)
+    assert len(out) == 1, "the echo of one outage must not read as three problems"
+    assert out[0]["id"].startswith("gateway:host:"), out[0]["id"]
+
+
+def test_an_unconfigured_gateway_source_says_nothing_even_when_its_job_says_down() -> None:
+    """No `HOST_TARGETS` entry named `GATEWAY_HOST` means no job — and a reported one stays quiet."""
+    out = alerts.build(Settings(gateway_interval=30.0),
+                       [gateway_state(ok=False, configured=False, error="not configured")],
+                       now=2_000.0)
+    assert out == []
+
+
+def test_a_gateway_payload_with_a_fabricated_verdict_alerts_nothing() -> None:
+    """The phase's no-verdict rule, asserted as a rule and not only as an absence.
+
+    Three ways, because deleting a rule satisfies an absence-only assertion:
+
+    * `alerts.build` takes no gateway payload at all — its signature is the whole surface, so a
+      gateway count can reach an alert only through a threshold added to this module;
+    * `severity.THRESHOLDS` has no `gateway.*` row, so no such threshold could be written without
+      editing the one file that owns thresholds (and a busy proxy is the job, the `net.*` case);
+    * the module's `HREF` does carry `/gateway`, so reachability still reaches the overview.
+    """
+    assert not {"gateway", "gateway_payload"} & set(inspect.signature(alerts.build).parameters)
+    assert not [name for name in severity.THRESHOLDS if name.startswith("gateway.")], (
+        "a connection count is not a graded number")
+    assert alerts.HREF["gateway"] == "/gateway"
+    assert alerts.AGENT_ECHO_DASHBOARDS == ("disks", "gateway")
+    # and an answering gateway with nothing else going on is the all clear
+    assert alerts.build(Settings(gateway_interval=30.0), [gateway_state()], [], [], None, [],
+                        now=2_000.0) == []
+
+
 # --------------------------------------------------------------------------- #
 # configuration
 # --------------------------------------------------------------------------- #

tests/test_auth_api.py

@@ -56,7 +56,12 @@ CURL_HEADERS = {"accept": "*/*", "user-agent": "curl/8.5.0"}
 PROTECTED_API = (
     "/api/config", "/api/events", "/api/metrics", "/api/history", "/api/summary",
     "/api/system/snapshot", "/api/system/history", "/api/disks/snapshot",
-    "/api/disks/history", "/api/llm/snapshot", "/api/llm/history", "/api/ups/snapshot",
+    "/api/disks/history",
+    # Phase 29: the ingress's pipes. `…/snapshot` carries the gateway's whole routing map — every
+    # domain it serves and every address it dials — so it is listed here for the same reason the
+    # bundle download is, whatever the payload happens to hold today.
+    "/api/gateway/snapshot", "/api/gateway/history",
+    "/api/llm/snapshot", "/api/llm/history", "/api/ups/snapshot",
     "/api/ups/history", "/api/ups/commands", "/api/auth/status",
     # Phase 15: the certificate authority's own surface. `/api/certs/<name>/bundle` is every
     # agent's private key, so "it needs a session" is the whole point of listing it here.
@@ -68,7 +73,11 @@ PROTECTED_API = (
 #: thing that must never regress is a new route answering anonymously. `/certs` is the sharper
 #: case of the two — an anonymous caller reaching it would see the CA's serials and every host
 #: the dashboard dials.
-PROTECTED_PAGES = ("/", "/system", "/disks", "/llm", "/ups", "/settings", "/certs")
+#: `/gateway` (phase 29 task 04) is the sixth dashboard: it prints this fleet's ingress map —
+#: every listening port, every `server_name`, every internal target — so it is gated by being a
+#: page and nothing else.
+PROTECTED_PAGES = ("/", "/system", "/disks", "/llm", "/ups", "/gateway", "/settings",
+                   "/certs")
 
 
 #: The one write this app has, with the fields the route needs (the UPS is unconfigured, so a
@@ -1598,8 +1607,8 @@ def test_the_served_route_table_is_the_surface_the_gate_describes(anon: TestClie
     pages = sorted(str(route.path) for route in served_routes(anon)
                    if isinstance(getattr(route, "path", None), str)
                    and not str(route.path).startswith(("/api", "/static")))
-    assert pages == ["/", "/certs", "/disks", "/llm", "/login", "/settings", "/system",
-                     "/ups"]
+    assert pages == ["/", "/certs", "/disks", "/gateway", "/llm", "/login", "/settings",
+                     "/system", "/ups"]
     for path in pages:
         response = anon.get(path, headers=BROWSER_HEADERS)
         if path == "/login":

tests/test_collectors.py

@@ -28,11 +28,13 @@ from dashboards.collectors import (
     disks_jobs,
     fetch_disk_snapshot,
     fetch_engine,
+    fetch_gateway_snapshot,
     fetch_host_snapshot,
     fetch_snapshot,
+    gateway_job,
     host_jobs,
 )
-from dashboards.config import Target
+from dashboards.config import Settings, Target
 from dashboards.poller import Result
 from dashboards.rates import RateEstimator
 from dashboards.sources import nut
@@ -409,6 +411,80 @@ async def test_smart_addon_errors_reach_the_disk_payload_verbatim(
     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["inbound"], "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
 # --------------------------------------------------------------------------- #

tests/test_config.py

@@ -87,6 +87,54 @@ def test_settings_default_to_no_sources_configured() -> None:
     assert settings.any_source_enabled is False
 
 
+def test_the_gateway_source_names_the_ingress_host_and_its_own_cadence() -> None:
+    """Phase 29: one `HOST_TARGETS` name is the gateway, and it is polled on its own clock.
+
+    `gateway_host` defaults to the fleet's real name so installing agent 1.2.0 on a host called
+    `gateway` is the whole configuration — but the **default is not a target**: with no
+    `HOST_TARGETS` entry of that name the source is unconfigured, `/api/gateway/*` 404s without
+    dialling, and the page says so (invariant 3). The interval sits between the system poll and
+    SMART because a connection count is the shape of half a minute, not of ten seconds.
+    """
+    settings = Settings()
+    assert settings.gateway_host == "gateway"
+    assert settings.gateway_interval == 30.0
+    assert settings.gateway_timeout == 10.0
+    assert settings.gateway_enabled is False
+    assert settings.any_source_enabled is False
+
+
+def test_gateway_enabled_is_true_only_for_a_name_that_is_a_target() -> None:
+    """A name with no agent behind it is unconfigured, not broken — the `nut_enabled` shape."""
+    def with_targets(raw: str, name: str = "gateway") -> bool:
+        targets, _problems = parse_targets(raw)
+        return Settings(host_targets=tuple(targets), gateway_host=name).gateway_enabled
+
+    assert with_targets("gateway=https://gw.example.net:9120") is True
+    assert with_targets(
+        "deskwork=https://d.example.net:9120 gateway=https://gw.example.net:9120") is True
+    assert with_targets("deskwork=https://d.example.net:9120") is False
+    assert with_targets("router=https://gw.example.net:9120", name="router") is True
+    assert with_targets("Gateway=https://gw.example.net:9120") is False, ("names are lowercase")
+    assert with_targets("") is False
+
+
+def test_the_gateway_settings_come_from_the_environment(monkeypatch: pytest.MonkeyPatch) -> None:
+    """`GATEWAY_HOST=router` is how a deployment whose ingress is called something else opts in."""
+    monkeypatch.setenv("GATEWAY_HOST", "router")
+    monkeypatch.setenv("GATEWAY_INTERVAL_SECONDS", "15")
+    monkeypatch.setenv("GATEWAY_TIMEOUT_SECONDS", "20")
+    reloaded = importlib.reload(config)
+    try:
+        assert reloaded.SETTINGS.gateway_host == "router"
+        assert reloaded.SETTINGS.gateway_interval == 15.0
+        assert reloaded.SETTINGS.gateway_timeout == 20.0
+    finally:
+        for name in ("GATEWAY_HOST", "GATEWAY_INTERVAL_SECONDS", "GATEWAY_TIMEOUT_SECONDS"):
+            monkeypatch.delenv(name)
+        importlib.reload(config)
+
+
 def test_the_disk_source_defaults_to_a_five_minute_cadence() -> None:
     """SMART is not a per-second signal (phase 09), and the fetch waits for the agent's
     own 20 s smartctl budget. Both are env-only, like every other interval."""

tests/test_gateway_source.py

@@ -0,0 +1,718 @@
+"""`sources/gateway.py`: the config + socket join that becomes the `/gateway` pipes.
+
+The shape half of these tests runs the **captured** gateway configs and `/proc` tables through
+the real agent `Collector` (the `gateway_snapshot` fixture), so a rename of a section key on
+either side of the wire fails here rather than drawing an empty pipe on a live page. The
+arithmetic half uses synthetic sections, because the point of those tests is which socket lands
+in which pipe and what the source refuses to decide — facts a capture states once and a
+re-capture may lawfully change (the same rule `tests/test_agent.py` holds for the socket tables).
+"""
+from __future__ import annotations
+
+import json
+import socket as socket_module
+from collections.abc import Mapping, Sequence
+from typing import Any
+
+from dashboards.sources import gateway
+
+# --------------------------------------------------------------------------- #
+# synthetic builders: a section exactly as the agent serves it, minus the file
+# --------------------------------------------------------------------------- #
+
+def bind(port: int | None, addr: str = "::") -> dict[str, Any]:
+    return {"addr": addr, "port": port}
+
+
+def haproxy_section(frontends: Sequence[dict[str, Any]],
+                    backends: Sequence[dict[str, Any]] = (), *,
+                    available: bool | None = True) -> dict[str, Any]:
+    return {"available": available, "config": "/etc/haproxy/haproxy.cfg",
+            "frontends": list(frontends), "backends": list(backends)}
+
+
+def frontend(name: str, ports: Sequence[int | None], *, default: str | None = None,
+             use: Sequence[str] = ()) -> dict[str, Any]:
+    return {"name": name, "binds": [bind(port) for port in ports],
+            "default_backend": default, "backends": list(use)}
+
+
+def backend(name: str, servers: Sequence[dict[str, Any]]) -> dict[str, Any]:
+    return {"name": name, "servers": list(servers)}
+
+
+def server(name: str, target: str | None, port: int | None = None) -> dict[str, Any]:
+    return {"name": name, "target": target, "port": port}
+
+
+def nginx_section(servers: Sequence[dict[str, Any]] = (), maps: Sequence[dict[str, Any]] = (), *,
+                  available: bool | None = True) -> dict[str, Any]:
+    return {"available": available, "config": "/etc/nginx/nginx.conf", "files": [],
+            "servers": list(servers), "upstreams": [], "maps": list(maps)}
+
+
+def nginx_server(ports: Sequence[int | None], names: Sequence[str] = (),
+                 passes: Sequence[dict[str, Any]] = ()) -> dict[str, Any]:
+    return {"listen": [bind(port, addr="") for port in ports], "server_names": list(names),
+            "proxy_passes": list(passes)}
+
+
+def proxy_pass(target: tuple[str, int] | None, *, variable: str | None = None,
+               location: str | None = "/") -> dict[str, Any]:
+    host, port = target if target else (None, None)
+    return {"location": location, "scheme": "http" if target else None,
+            "target": None if host is None else {"host": host, "port": port},
+            "variable": variable}
+
+
+def mapping_block(variable: str,
+                  entries: Sequence[tuple[str, tuple[str, int | None] | None]]) -> dict[str, Any]:
+    return {"variable": variable,
+            "entries": [{"key": key,
+                         "target": None if target is None else {"host": target[0],
+                                                                "port": target[1]}}
+                        for key, target in entries]}
+
+
+def sockets_section(listens: Sequence[dict[str, Any]] = (),
+                    established: Sequence[dict[str, Any]] = (), *,
+                    available: bool | None = True) -> dict[str, Any]:
+    return {"available": available, "listens": list(listens), "established": list(established)}
+
+
+def listen(port: int, addr: str = "0.0.0.0", family: str = "v4") -> dict[str, Any]:
+    return {"family": family, "addr": addr, "port": port}
+
+
+def conn(local: tuple[str, int], remote: tuple[str, int], *,
+         family: str = "v4") -> dict[str, Any]:
+    return {"family": family, "local": {"addr": local[0], "port": local[1]},
+            "remote": {"addr": remote[0], "port": remote[1]}}
+
+
+def snapshot(*, haproxy: dict[str, Any] | None = None, nginx: dict[str, Any] | None = None,
+             sockets: dict[str, Any] | None = None, errors: Sequence[str] = (),
+             agent_version: str | None = "1.2.0") -> dict[str, Any]:
+    """The document one agent `/snapshot` would answer, with only what a test names."""
+    doc: dict[str, Any] = {"ok": True, "schema": 1, "ts": 1_760_000_000.0, "host": "gateway",
+                           "errors": list(errors)}
+    if agent_version is not None:
+        doc["agent_version"] = agent_version
+    if haproxy is not None:
+        doc["haproxy"] = haproxy
+    if nginx is not None:
+        doc["nginx"] = nginx
+    if sockets is not None:
+        doc["sockets"] = sockets
+    return doc
+
+
+def pipe_by_port(payload: Mapping[str, Any], port: int) -> dict[str, Any]:
+    for pipe in payload["inbound"]:
+        if pipe["port"] == port:
+            return pipe
+    raise AssertionError(f"no inbound pipe on port {port}: {[p['port'] for p in payload['inbound']]}")
+
+
+def pipe_to(payload: Mapping[str, Any], host: str, port: int | None) -> dict[str, Any]:
+    for pipe in payload["outbound"]:
+        if pipe["target"]["host"] == host and pipe["target"]["port"] == port:
+            return pipe
+    raise AssertionError(f"no outbound pipe to {host}:{port}")
+
+
+def walk_keys(node: Any) -> set[str]:
+    """Every key name anywhere in a payload — the device the "no severity / no window" pins use."""
+    if isinstance(node, Mapping):
+        found = set(node.keys())
+        for value in node.values():
+            found |= walk_keys(value)
+        return found
+    if isinstance(node, Sequence) and not isinstance(node, str | bytes):
+        return set().union(*(walk_keys(item) for item in node)) if node else set()
+    return set()
+
+
+# --------------------------------------------------------------------------- #
+# the captured gateway: every configured pipe, from the real agent code path
+# --------------------------------------------------------------------------- #
+
+def test_the_captured_configs_yield_every_configured_pipe(gateway_snapshot: Any) -> None:
+    """Six haproxy frontends, four nginx listen ports, the stream map's fan-out — all present."""
+    payload = gateway.parse(gateway_snapshot("full"), "gateway")
+    assert payload["configured"] is True
+    assert payload["supported"] is True
+    assert payload["why"] is None, "a gateway whose every section answered gets no why-line"
+    assert payload["sections"] == {"haproxy": True, "nginx": True, "sockets": True}
+
+    haproxy_ports = [row["port"] for row in payload["haproxy"]["frontends"]]
+    assert haproxy_ports == [22, 2000, 5201, 8080, 25565, 25566], "config order, one row per bind"
+    assert {row["backend"] for row in payload["haproxy"]["frontends"]} == {
+        "gitea-2222", "sec-gitea-2000", "iperf3-5201", "unifi-external-8080",
+        "testcraft-25565", "nimcraft-25566"}
+
+    assert [row["port"] for row in payload["nginx"]["servers"]] == [80, 443, 4443, 8443]
+    names = {row["port"]: row["server_names"] for row in payload["nginx"]["servers"]}
+    assert names[80] == ["_"], "the catch-all is the config's own text, kept verbatim"
+    assert names[443] == [], "the stream server routes by SNI, so it names no domain of its own"
+    assert "nextcloud.reeseapps.com" in names[4443] and "gitea.reeseapps.com" in names[4443], (
+        "five vhosts share 4443 and their domains belong to the one pipe for that port")
+
+    assert payload["nginx"]["map_targets"], "the stream map is the fan-out the page draws"
+    assert len(payload["nginx"]["map_targets"]) == 17
+    homepage = next(row for row in payload["nginx"]["map_targets"]
+                    if row["key"] == "homepage.reeseapps.com")
+    assert homepage == {"key": "homepage.reeseapps.com",
+                        "host": "reeseapps-homepage.reeselink.com", "port": 443}
+
+
+def test_the_source_reads_the_keys_the_agent_actually_publishes(gateway_snapshot: Any,
+                                                                host_agent_module: Any) -> None:
+    """The coupling is pinned: `binds`/`port`, `servers`/`target`, `listen`/`server_names`.
+
+    The fixture runs the real parser, so this compares the source's output against the agent's
+    own document rather than against a copy of its field names — which is what makes a rename
+    fail here instead of printing `n/a` beside a healthy gateway (AGENTS.md rule 7).
+    """
+    snap = gateway_snapshot("full")
+    payload = gateway.parse(snap, "gateway")
+    agent_bind_ports = {bind["port"] for frontend in snap["haproxy"]["frontends"]
+                        for bind in frontend["binds"]}
+    assert {row["port"] for row in payload["haproxy"]["frontends"]} == agent_bind_ports
+    agent_servers = {(row["name"], row["target"], row["port"])
+                     for block in snap["haproxy"]["backends"] for row in block["servers"]}
+    assert {(pipe["target"]["name"], pipe["target"]["host"], pipe["target"]["port"])
+            for pipe in payload["outbound"] if pipe["source"] == "haproxy"} == agent_servers
+    agent_listen_ports = {listen["port"] for server in snap["nginx"]["servers"]
+                          for listen in server["listen"]}
+    assert {row["port"] for row in payload["nginx"]["servers"]} == agent_listen_ports
+    # and the three section names themselves, as the snapshot publishes them
+    assert {"haproxy", "nginx", "sockets"} <= set(snap)
+    assert host_agent_module.AGENT_VERSION == "1.2.0", "the sections these keys belong to"
+
+
+def test_the_live_counts_reconcile_with_the_socket_table(gateway_snapshot: Any) -> None:
+    """Every ESTABLISHED row is counted exactly once — in a pipe, or in `unmatched`.
+
+    This is the property phase 29 task 06 checks against a direct `/proc/net/tcp` read; pinning
+    it on the capture means the arithmetic is already right before anything is dialled.
+    """
+    snap = gateway_snapshot("full")
+    payload = gateway.parse(snap, "gateway")
+    established = len(snap["sockets"]["established"])
+    in_pipes = sum(pipe["connections"] for pipe in payload["inbound"])
+    out_pipes = sum(pipe["connections"] or 0 for pipe in payload["outbound"])
+    unmatched = payload["unmatched"]
+    assert payload["sockets"] == {"listens": len(snap["sockets"]["listens"]),
+                                  "established": established}
+    assert in_pipes + out_pipes + unmatched["inbound"] + unmatched["outbound"] == established
+    # the ambiguous fan-out is inside `unmatched.outbound`, not on top of it
+    assert sum(group["connections"] for group in unmatched["groups"]) <= unmatched["outbound"]
+
+
+def test_a_listening_port_no_config_claims_stays_visible(gateway_snapshot: Any) -> None:
+    """Phase 28's "an unclaimed socket stays visible as one", counted, not silently dropped."""
+    snap = gateway_snapshot("full")
+    payload = gateway.parse(snap, "gateway")
+    live = {row["port"] for row in snap["sockets"]["listens"]}
+    configured = {pipe["port"] for pipe in payload["inbound"]}
+    assert payload["unmatched"]["other"] == len(live - configured)
+    # the agent's own port is dialled by nobody in either config, so it is always the example
+    assert 9120 in live - configured
+
+
+def test_the_map_fanout_becomes_pipes_labelled_by_domain(gateway_snapshot: Any) -> None:
+    """Every map entry is a pipe the page can draw, and `$var` never becomes a host."""
+    payload = gateway.parse(gateway_snapshot("full"), "gateway")
+    map_pipes = [pipe for pipe in payload["outbound"] if pipe["source"] == "map"]
+    assert len(map_pipes) == len(payload["nginx"]["map_targets"])
+    assert not [pipe for pipe in payload["outbound"]
+                if pipe["target"]["host"] is None or "$" in str(pipe["target"]["host"])], (
+        "a proxy_pass $var carries no address, so it must not become a pipe of its own")
+    brain = pipe_to(payload, "reeseapps-brain-of-reese.reeselink.com", 443)
+    assert brain["target"]["by_port"] is True, "a name can only ever be matched on its port"
+
+
+# --------------------------------------------------------------------------- #
+# the inbound join
+# --------------------------------------------------------------------------- #
+
+def test_a_socket_lands_in_the_frontend_that_owns_its_local_port() -> None:
+    snap = snapshot(
+        haproxy=haproxy_section([frontend("ssh", [22], default="ssh-be")]),
+        nginx=nginx_section([nginx_server([443], ["a.example.test"])]),
+        sockets=sockets_section(
+            listens=[listen(22), listen(443)],
+            established=[conn(("10.1.1.27", 443), ("203.0.113.7", 51000)),
+                         conn(("10.1.1.27", 443), ("203.0.113.9", 51001)),
+                         conn(("10.1.1.27", 22), ("198.51.100.4", 33000))]))
+    payload = gateway.parse(snap, "gateway")
+    https = pipe_by_port(payload, 443)
+    assert https["source"] == "nginx"
+    assert https["connections"] == 2
+    assert https["clients"] == ["203.0.113.7", "203.0.113.9"], "the client is the remote end"
+    assert https["domains"] == ["a.example.test"]
+    assert https["metric"] == "gateway.in_conns@443"
+    assert pipe_by_port(payload, 22)["source"] == "haproxy"
+    assert payload["unmatched"] == {"inbound": 0, "outbound": 0, "other": 0, "groups": []}
+
+
+def test_a_frontend_that_is_idle_is_still_a_pipe_with_zero() -> None:
+    """A configured port with no connections is `0`, not `n/a` — the socket table did answer."""
+    snap = snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
+                    sockets=sockets_section(listens=[listen(22)]))
+    payload = gateway.parse(snap, "gateway")
+    assert pipe_by_port(payload, 22)["connections"] == 0
+
+
+def test_a_connection_on_a_port_no_config_names_is_reported_not_dropped() -> None:
+    """`unmatched.inbound`: this host is serving something neither file mentions."""
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([443], ["a.example.test"])]),
+        sockets=sockets_section(
+            listens=[listen(443), listen(8081)],
+            established=[conn(("10.1.1.27", 8081), ("10.1.0.9", 41000)),
+                         conn(("10.1.1.27", 8081), ("10.1.0.9", 41001))]))
+    payload = gateway.parse(snap, "gateway")
+    assert pipe_by_port(payload, 443)["connections"] == 0
+    assert payload["unmatched"]["inbound"] == 2, "the page's 'N other connections'"
+    assert payload["unmatched"]["other"] == 1, "8081 is a listener no frontend claims"
+
+
+def test_one_port_named_by_both_configs_is_one_pipe_carrying_both_domains() -> None:
+    """A host cannot bind a port twice, so the socket space is one pipe and the facts merge."""
+    snap = snapshot(
+        haproxy=haproxy_section([frontend("http", [80])]),
+        nginx=nginx_section([nginx_server([80], ["a.example.test"])]),
+        sockets=sockets_section(listens=[listen(80)],
+                                established=[conn(("10.1.1.27", 80), ("203.0.113.1", 1234))]))
+    payload = gateway.parse(snap, "gateway")
+    ports = [pipe["port"] for pipe in payload["inbound"]]
+    assert ports == [80], "no second pipe, and no socket counted down both"
+    assert pipe_by_port(payload, 80)["connections"] == 1
+    assert pipe_by_port(payload, 80)["domains"] == ["a.example.test"]
+
+
+def test_the_client_list_is_unique_sorted_and_bounded() -> None:
+    ports = [80]
+    many = [conn(("10.1.1.27", 80), (f"203.0.113.{i % 256}", 40000 + i))
+            for i in range(gateway.MAX_CLIENTS + 12)]
+    many.append(conn(("10.1.1.27", 80), ("203.0.113.1", 45678)))
+    snap = snapshot(nginx=nginx_section([nginx_server(ports, ["a.example.test"])]),
+                    sockets=sockets_section(listens=[listen(80)], established=many))
+    pipe = pipe_by_port(gateway.parse(snap, "gateway"), 80)
+    assert len(pipe["clients"]) == gateway.MAX_CLIENTS
+    assert pipe["clients"] == sorted(pipe["clients"])
+    assert pipe["connections"] == len(many), "the count stays the truth when the list is capped"
+
+
+# --------------------------------------------------------------------------- #
+# the outbound join
+# --------------------------------------------------------------------------- #
+
+def test_a_target_written_as_an_address_matches_exactly_and_says_by_port_false() -> None:
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([8443], ["app.example.test"],
+                                          [proxy_pass(("127.0.0.1", 8081))])]),
+        sockets=sockets_section(
+            listens=[listen(8443)],
+            established=[conn(("10.1.1.27", 39000), ("127.0.0.1", 8081)),
+                         conn(("10.1.1.27", 39002), ("10.9.9.9", 8081))]))
+    payload = gateway.parse(snap, "gateway")
+    exact = pipe_to(payload, "127.0.0.1", 8081)
+    assert exact["target"]["by_port"] is False
+    assert exact["connections"] == 1, "the one socket dialled to that address"
+    assert exact["metric"] == "gateway.out_conns@127.0.0.1:8081"
+    assert payload["unmatched"]["outbound"] == 1, (
+        "the same port on a different address is not this target's connection")
+
+
+def test_a_domain_target_is_matched_on_port_alone_and_labelled_by_port() -> None:
+    """The address half is the config's, the socket half is the kernel's, and `by_port` says so."""
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([4443], ["cloud.example.test"],
+                                          [proxy_pass(("nextcloud.example.test", 11000))])]),
+        sockets=sockets_section(
+            listens=[listen(4443)],
+            established=[conn(("10.1.1.27", 4443), ("10.1.0.4", 55555)),
+                         conn(("10.1.1.27", 54340), ("10.1.0.254", 11000))]))
+    payload = gateway.parse(snap, "gateway")
+    pipe = pipe_to(payload, "nextcloud.example.test", 11000)
+    assert pipe["target"]["by_port"] is True
+    assert pipe["connections"] == 1
+    assert pipe["source"] == "nginx"
+    assert payload["unmatched"]["groups"] == []
+
+
+def test_a_port_shared_by_several_names_resolves_none_of_them() -> None:
+    """The real `stream.conf` shape: sixteen names on 443, and no basis for picking one.
+
+    The sockets go to `unmatched.outbound` with a group naming the candidates; each candidate's
+    own count stays `null` — a repeated port total drawn on sixteen pipes reads as sixteen times
+    the traffic, and a split of it is arithmetic nobody measured (rule 2).
+    """
+    entries = [(f"site{i}.example.test", (f"backend{i}.example.test", 443)) for i in range(4)]
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
+                            maps=[mapping_block("$targetBackend", entries)]),
+        sockets=sockets_section(
+            listens=[listen(443)],
+            established=[conn(("10.1.1.27", 443), ("203.0.113.5", 4000)),
+                         conn(("10.1.1.27", 50000), ("10.2.2.2", 443)),
+                         conn(("10.1.1.27", 50002), ("10.2.2.3", 443))]))
+    payload = gateway.parse(snap, "gateway")
+    assert len(payload["nginx"]["map_targets"]) == 4
+    for _key, (host, port) in entries:
+        assert pipe_to(payload, host, port)["connections"] is None, "unknown, never 0 and never 2"
+    group = payload["unmatched"]["groups"][0]
+    assert group == {"port": 443, "connections": 2,
+                     "hosts": ["backend0.example.test", "backend1.example.test",
+                               "backend2.example.test", "backend3.example.test"],
+                     "host_count": 4}
+    assert payload["unmatched"]["outbound"] == 2
+    assert payload["unmatched"]["inbound"] == 0
+
+
+def test_the_group_host_list_is_bounded_and_reports_the_real_total() -> None:
+    entries = [(f"site{i}.example.test", (f"backend{i}.example.test", 443))
+               for i in range(gateway.MAX_GROUP_HOSTS + 6)]
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="be")])],
+                            maps=[mapping_block("$be", entries)]),
+        sockets=sockets_section(established=[conn(("10.1.1.27", 50000), ("10.2.2.2", 443))]))
+    group = gateway.parse(snap, "gateway")["unmatched"]["groups"][0]
+    assert len(group["hosts"]) == gateway.MAX_GROUP_HOSTS
+    assert group["host_count"] == len(entries)
+
+
+def test_no_join_ever_asks_a_resolver() -> None:
+    """The whole point of `by_port`: this module must not turn a name into an address.
+
+    Spied on the live `socket` module the stdlib resolvers use, and run over the captured
+    gateway as well as a synthetic one, so a "helpful" future `socket.getaddrinfo` call inside
+    the join fails here rather than making 33 DNS queries every 30 seconds from the poll loop.
+    """
+    calls: list[tuple[Any, ...]] = []
+
+    def spy(*args: Any, **kwargs: Any) -> Any:  # pragma: no cover - must never run
+        calls.append((args, kwargs))
+        raise AssertionError("the gateway join must never resolve a name")
+
+    original = socket_module.getaddrinfo
+    socket_module.getaddrinfo = spy
+    try:
+        gateway.parse(snapshot(
+            haproxy=haproxy_section([frontend("mc", [25565])],
+                                    [backend("mc-be", [server("mc1", "minecraft.example.test",
+                                                              25565)])]),
+            nginx=nginx_section([nginx_server([443], ["a.example.test"],
+                                              [proxy_pass(("app.example.test", 8080))])]),
+            sockets=sockets_section(listens=[listen(25565), listen(443)],
+                                    established=[conn(("10.1.1.27", 443), ("203.0.113.5", 900)),
+                                                 conn(("10.1.1.27", 1), ("10.2.2.2", 8080))])),
+            "gateway")
+    finally:
+        socket_module.getaddrinfo = original
+    assert calls == []
+
+
+def test_two_services_on_one_host_are_two_pipes_not_one_sum() -> None:
+    """The reason the outbound series name carries its port (`minecraft` 25565 and 25566)."""
+    snap = snapshot(
+        haproxy=haproxy_section([frontend("mc1", [25565]), frontend("mc2", [25566])],
+                                [backend("testcraft", [server("mc1", "mc.example.test", 25565)]),
+                                 backend("nimcraft", [server("mc2", "mc.example.test", 25566)])]),
+        sockets=sockets_section(
+            established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 25565)),
+                         conn(("10.1.1.27", 40002), ("10.5.5.5", 25566)),
+                         conn(("10.1.1.27", 40004), ("10.5.5.5", 25566))]))
+    payload = gateway.parse(snap, "gateway")
+    assert pipe_to(payload, "mc.example.test", 25565)["connections"] == 1
+    assert pipe_to(payload, "mc.example.test", 25566)["connections"] == 2
+    assert [pipe["metric"] for pipe in payload["outbound"]] == [
+        "gateway.out_conns@mc.example.test:25565", "gateway.out_conns@mc.example.test:25566"]
+    names = {pipe["target"]["name"] for pipe in payload["outbound"]}
+    assert names == {"mc1", "mc2"}, "the config's own server names survive the dedup"
+
+
+def test_a_target_with_no_port_is_a_pipe_with_an_unknown_count() -> None:
+    """`server x host` names no port, so there is nothing to join on and nothing to invent."""
+    snap = snapshot(
+        haproxy=haproxy_section([], [backend("be", [server("solo", "solo.example.test")])]),
+        sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080))]))
+    pipe = pipe_to(gateway.parse(snap, "gateway"), "solo.example.test", None)
+    assert pipe["target"]["by_port"] is False
+    assert pipe["connections"] is None
+    assert pipe["metric"] == "gateway.out_conns@solo.example.test"
+
+
+def test_the_same_target_from_two_configs_is_one_pipe() -> None:
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([4443], ["a.example.test"],
+                                          [proxy_pass(("keycloak.example.test", 8080))]),
+                             nginx_server([8443], ["admin.a.example.test"],
+                                          [proxy_pass(("keycloak.example.test", 8080))])]),
+        sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 8080))]))
+    payload = gateway.parse(snap, "gateway")
+    targets = [pipe for pipe in payload["outbound"]
+               if pipe["target"]["host"] == "keycloak.example.test"]
+    assert len(targets) == 1
+    assert targets[0]["connections"] == 1, "one pipe, so one count, so no double draw"
+
+
+# --------------------------------------------------------------------------- #
+# the map: which maps route, and which are just header words
+# --------------------------------------------------------------------------- #
+
+def test_a_map_no_proxy_pass_references_contributes_nothing() -> None:
+    """`$connection_upgrade`'s values are `upgrade`/`close` — header words, never a host."""
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="targetBackend")])],
+                            maps=[mapping_block("connection_upgrade",
+                                                [("default", ("upgrade", None)),
+                                                 ("", ("close", None))]),
+                                  mapping_block("targetBackend",
+                                                [("a.example.test", ("be.example.test", 443))])]),
+        sockets=sockets_section(listens=[listen(443)]))
+    payload = gateway.parse(snap, "gateway")
+    assert [row["key"] for row in payload["nginx"]["map_targets"]] == ["a.example.test"]
+    assert [pipe["target"]["host"] for pipe in payload["outbound"]] == ["be.example.test"]
+
+
+def test_a_map_entry_with_no_address_is_skipped() -> None:
+    snap = snapshot(
+        nginx=nginx_section([nginx_server([443], [], [proxy_pass(None, variable="be")])],
+                            maps=[mapping_block("be", [("a.example.test", None),
+                                                       ("b.example.test", ("be2.example.test", 443))])]))
+    payload = gateway.parse(snap, "gateway")
+    assert [row["key"] for row in payload["nginx"]["map_targets"]] == ["b.example.test"]
+
+
+def test_a_duplicate_endpoint_keeps_the_name_the_config_has() -> None:
+    """haproxy is claimed first, so nginx writing the same endpoint loses nothing."""
+    snap = snapshot(
+        haproxy=haproxy_section([frontend("app", [80])],
+                                [backend("app-be", [server("app1", "app.example.test", 9000)])]),
+        nginx=nginx_section([nginx_server([8443], ["app.example.test"],
+                                          [proxy_pass(("app.example.test", 9000))])]),
+        sockets=sockets_section(established=[conn(("10.1.1.27", 40000), ("10.5.5.5", 9000))]))
+    payload = gateway.parse(snap, "gateway")
+    app_pipes = [pipe for pipe in payload["outbound"]
+                 if pipe["target"]["host"] == "app.example.test"]
+    assert len(app_pipes) == 1
+    assert app_pipes[0]["target"]["name"] == "app1"
+    assert app_pipes[0]["source"] == "haproxy"
+    assert app_pipes[0]["connections"] == 1
+
+
+def test_a_proxy_pass_naming_no_host_is_no_pipe() -> None:
+    """`target: {host: ""}` is drift, and drift yields no pipe rather than a pipe to nowhere."""
+    snap = snapshot(nginx=nginx_section([nginx_server(
+        [443], [], [{"location": "/", "scheme": "http", "target": {"host": "  ", "port": 80},
+                     "variable": None}])]))
+    payload = gateway.parse(snap, "gateway")
+    assert payload["outbound"] == []
+    assert payload["nginx"]["servers"] == [{"port": 443, "server_names": []}]
+
+
+def test_the_readers_refuse_the_shapes_that_are_not_values() -> None:
+    """The three guards every section read goes through, on the inputs that are not data.
+
+    Pinned directly (the way `tests/test_config.py` pins `_flag`/`_num`) because each one is the
+    difference between a pipe and no pipe: an empty string is not an address, a bare string is
+    not a list of names, and whitespace is not a host.
+    """
+    assert gateway._is_ip_literal("") is False
+    assert gateway._is_ip_literal(None) is False
+    assert gateway._is_ip_literal("203.0.113.7") is True
+    assert gateway._is_ip_literal("[::1]") is False, "/proc spells addresses without brackets"
+    assert gateway._is_ip_literal("app.example.test") is False
+    assert gateway._names("not a list") == []
+    assert gateway._names(None) == []
+    assert gateway._names([None, "a-be", "a-be", "   ", "b-be"]) == ["a-be", "b-be"]
+    assert gateway._target_of({"host": "   ", "port": 80}) is None
+    assert gateway._target_of(["host", 80]) is None
+    assert gateway._target_of({"host": "be.example.test", "port": "nope"}) == (
+        "be.example.test", None)
+
+
+# --------------------------------------------------------------------------- #
+# the degradation arms: three ways to have nothing, none of them red
+# --------------------------------------------------------------------------- #
+
+def test_an_agent_without_the_sections_predates_gateway_support(gateway_snapshot: Any) -> None:
+    """The 1.1.0 / `--no-proxy` arm: same keys, empty and null, one sentence, no verdict."""
+    payload = gateway.parse(gateway_snapshot("predates"), "gateway")
+    assert payload["supported"] is False
+    assert payload["why"] == gateway.WHY_PREDATES
+    assert "predates gateway support" in payload["why"]
+    assert payload["haproxy"] == {"frontends": []}
+    assert payload["nginx"] == {"servers": [], "map_targets": []}
+    assert payload["inbound"] == [] and payload["outbound"] == []
+    assert payload["unmatched"] == {"inbound": None, "outbound": None, "other": None,
+                                    "groups": []}
+    assert payload["sockets"] == {"listens": None, "established": None}
+    assert payload["sections"] == {"haproxy": None, "nginx": None, "sockets": None}
+    assert gateway.samples(payload, 1.0) == [], "an unknown count publishes no gauge"
+
+
+def test_a_host_running_no_haproxy_says_so_and_keeps_the_nginx_pipes(gateway_snapshot: Any) -> None:
+    payload = gateway.parse(gateway_snapshot("no_haproxy"), "gateway")
+    assert payload["supported"] is True, "the agent knows the feature; the host has no haproxy"
+    assert payload["sections"]["haproxy"] is False
+    assert "no haproxy" in (payload["why"] or "")
+    assert "not found" not in (payload["why"] or ""), "the page's sentence, not the agent's log"
+    assert payload["section_errors"], "the agent's own line still travels, verbatim"
+    assert any("not found" in line for line in payload["section_errors"])
+    assert [row["port"] for row in payload["nginx"]["servers"]] == [80, 443, 4443, 8443]
+    assert [pipe["port"] for pipe in payload["inbound"] if pipe["source"] == "nginx"] == \
+        [80, 443, 4443, 8443]
+    assert payload["haproxy"]["frontends"] == []
+
+
+def test_an_unreadable_config_is_a_different_sentence_from_no_config() -> None:
+    snap = snapshot(haproxy=haproxy_section([], available=None),
+                    nginx=nginx_section([nginx_server([80], ["a.example.test"])]),
+                    sockets=sockets_section(listens=[listen(80)]))
+    payload = gateway.parse(snap, "gateway")
+    assert payload["sections"]["haproxy"] is None
+    assert "could not be read or parsed" in (payload["why"] or "")
+    assert [pipe["port"] for pipe in payload["inbound"]] == [80]
+
+
+def test_a_socket_table_that_did_not_answer_leaves_every_count_null(
+        gateway_snapshot: Any) -> None:
+    """`available: null` is "no answer", and the only honest counts are `null` (rule 2)."""
+    payload = gateway.parse(gateway_snapshot("sockets_silent"), "gateway")
+    assert payload["sections"]["sockets"] is None
+    assert "sockets:" in (payload["why"] or "")
+    assert payload["sockets"] == {"listens": None, "established": None}
+    assert payload["inbound"], "the configured pipes are still drawn — the config did answer"
+    assert all(pipe["connections"] is None for pipe in payload["inbound"])
+    assert all(pipe["connections"] is None for pipe in payload["outbound"])
+    assert payload["unmatched"] == {"inbound": None, "outbound": None, "other": None,
+                                    "groups": []}
+    assert gateway.samples(payload, 1.0) == []
+
+
+def test_a_partial_snapshot_names_the_missing_section_only() -> None:
+    """A hypothetical agent publishing `sockets` but no configs: supported, with a gap named."""
+    snap = snapshot(sockets=sockets_section(listens=[listen(9999)]))
+    payload = gateway.parse(snap, "gateway")
+    assert payload["supported"] is True
+    assert payload["sections"] == {"haproxy": None, "nginx": None, "sockets": True}
+    assert payload["unmatched"]["other"] == 1, "9999 is a listener nothing claims"
+    assert "haproxy" in (payload["why"] or "") and "nginx" in (payload["why"] or "")
+
+
+# --------------------------------------------------------------------------- #
+# what the payload must never contain, and the gauges it must
+# --------------------------------------------------------------------------- #
+
+def test_no_count_carries_a_verdict_or_a_window(gateway_snapshot: Any) -> None:
+    """`net.*` precedent for colour, rule 2 for the `Δ` label: a count is a count.
+
+    Both arms are walked, so a `severity` bolted onto a pipe or a `window_seconds` copied from
+    `sources/host.py` fails here — the first would be a judgement `severity.py` never made, the
+    second would claim a measurement this source does not take.
+    """
+    for mode in ("full", "predates", "sockets_silent"):
+        keys = walk_keys(gateway.parse(gateway_snapshot(mode), "gateway"))
+        assert "severity" not in keys, mode
+        assert not {key for key in keys if "window" in key or key.startswith("Δ")}, mode
+
+
+def test_the_payload_is_json_serialisable_end_to_end(gateway_snapshot: Any) -> None:
+    """It goes out over `/api/gateway/snapshot` verbatim, so nothing in it may be exotic."""
+    json.dumps(gateway.parse(gateway_snapshot("full"), "gateway"))
+
+
+def test_samples_are_one_gauge_per_pipe_under_the_gateway_family(gateway_snapshot: Any) -> None:
+    snap = gateway_snapshot("full")
+    payload = gateway.parse(snap, "gateway")
+    rows = gateway.samples(payload, 1_760_000_000.0)
+    assert rows, "a live gateway publishes gauges"
+    assert {row.source for row in rows} == {gateway.SOURCE}
+    assert {row.host for row in rows} == {"gateway"}
+    named = {row.metric: row.value for row in rows}
+    assert named["gateway.in_conns@443"] == float(pipe_by_port(payload, 443)["connections"])
+    counted = [pipe for pipe in payload["inbound"] + payload["outbound"]
+               if pipe["connections"] is not None]
+    assert len(named) == len(counted), "one gauge per pipe with a number, and none without"
+    for pipe in counted:
+        assert named[pipe["metric"]] == float(pipe["connections"]), (
+            "the stored gauge is the number the page prints, never a second reading of it")
+
+
+def test_parse_gateway_snapshot_stamps_its_samples_from_the_snapshot() -> None:
+    """The x-axis is when the agent took the snapshot, the way `sources/disks.py` does it."""
+    result = gateway.parse_gateway_snapshot(
+        "gateway", snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
+                            sockets=sockets_section(listens=[listen(22)])), now=999.0)
+    assert result.events == [], "a connection count has no transition worth an event line"
+    assert [sample.ts for sample in result.samples] == [1_760_000_000.0]
+    assert [sample.metric for sample in result.samples] == ["gateway.in_conns@22"]
+
+
+def test_a_snapshot_with_no_timestamp_falls_back_to_the_callers_clock() -> None:
+    doc = snapshot(haproxy=haproxy_section([frontend("ssh", [22])]),
+                   sockets=sockets_section(listens=[listen(22)]))
+    doc.pop("ts")
+    result = gateway.parse_gateway_snapshot("gateway", doc, now=42.0)
+    assert [sample.ts for sample in result.samples] == [42.0]
+
+
+def test_the_gauges_skip_a_pipe_with_no_number() -> None:
+    """No sample for an unknown count — a chart of zeros for a dead socket table is the
+    failure mode rule 2 names, and `sources/disks.py` already refuses it for unreadable drives."""
+    payload = gateway.parse(snapshot(haproxy=haproxy_section([frontend("ssh", [22])])), "gateway")
+    assert payload["inbound"][0]["connections"] is None
+    assert gateway.samples(payload, 1.0) == []
+
+
+def test_a_snapshot_whose_sections_are_empty_is_an_honest_empty_page() -> None:
+    """A 1.2.0 agent on a box with two empty configs: supported, and nothing drawn."""
+    snap = snapshot(haproxy=haproxy_section([]), nginx=nginx_section(),
+                    sockets=sockets_section())
+    payload = gateway.parse(snap, "gateway")
+    assert payload["supported"] is True and payload["why"] is None
+    assert payload["inbound"] == [] and payload["outbound"] == []
+    assert payload["sockets"] == {"listens": 0, "established": 0}
+    assert payload["unmatched"] == {"inbound": 0, "outbound": 0, "other": 0, "groups": []}
+
+
+def test_junk_in_the_sections_yields_no_pipes_and_no_exception() -> None:
+    """The source is a projection of a document it does not control — drift is `n/a`, not a crash."""
+    snap: dict[str, Any] = {"ts": 1.0, "host": "gateway",
+                           "haproxy": {"available": True, "frontends": [3, None],
+                                       "backends": {"nope": 1}},
+                           "nginx": {"available": True, "servers": "not a list", "maps": None},
+                           "sockets": {"available": True, "listens": [None],
+                                       "established": [{"local": 5, "remote": None}]}}
+    payload = gateway.parse(snap, "gateway")
+    assert payload["inbound"] == [] and payload["outbound"] == []
+    assert payload["unmatched"]["inbound"] == 0
+    assert payload["sockets"] == {"listens": 0, "established": 1}
+
+
+def test_a_port_of_zero_or_a_name_of_nothing_is_no_match() -> None:
+    """`0` is not a port and an empty target is not a host, so neither becomes a pipe (rule 2)."""
+    snap = snapshot(
+        haproxy=haproxy_section([frontend("u", [None, 0]), frontend("ok", [22])],
+                                [backend("b", [server("?", "", 22), server("z", "z.example.test",
+                                                                            0)])]),
+        sockets=sockets_section(listens=[listen(0), listen(22)]))
+    payload = gateway.parse(snap, "gateway")
+    assert [row["port"] for row in payload["haproxy"]["frontends"]] == [22]
+    assert [pipe["target"]["host"] for pipe in payload["outbound"]] == ["z.example.test"]
+    assert payload["outbound"][0]["target"]["port"] is None
+    assert payload["unmatched"]["other"] == 0, "a 0 is not a port, claimed or otherwise"
+
+
+def test_a_use_backend_map_is_reported_as_routes_not_one_chosen_backend() -> None:
+    """haproxy picks per request here; the payload names every candidate and chooses none."""
+    snap = snapshot(haproxy=haproxy_section(
+        [frontend("fe", [80], default="base-be", use=("blue-be", "green-be"))]))
+    row = gateway.parse(snap, "gateway")["haproxy"]["frontends"][0]
+    assert row["backend"] == "base-be"
+    assert row["routes"] == ["base-be", "blue-be", "green-be"]

tests/test_live_fleet.py

@@ -33,7 +33,7 @@ from dashboards import certs, collectors
 from dashboards.config import Settings, load
 from dashboards.rates import RateEstimator
 from dashboards.severity import classify
-from dashboards.sources import nut
+from dashboards.sources import gateway, nut
 from dashboards.sources.disks import SMART_SERIES, DisksCollector
 from dashboards.sources.host import HostCollector
 from slots_contract import assert_slot_shape
@@ -255,6 +255,46 @@ async def test_disks_source_charts_the_physical_drives_the_agent_reports(
         pytest.skip("no host agent answered")
 
 
+async def test_the_gateway_source_joins_the_live_ingress(settings: Settings,
+                                                         agent_tls: dict[str, collectors.AgentTls]) -> None:
+    """Phase 29: `sources/gateway.py` over the **real** gateway agent, read-only.
+
+    The production path — the same `fetch_snapshot` the poller will run, through the stored
+    mTLS authority — so what fails here is a drift between the agent's sections and the join,
+    not a typo in a fixture. The assertions are the ones that hold whatever the traffic is:
+    the agent answers the three sections, every configured frontend is a pipe, and the
+    inbound/outbound/unmatched tallies **partition** the kernel's ESTABLISHED table, which is
+    what lets the page's numbers reconcile against a direct `/proc/net/tcp` read.
+    """
+    target = next((item for item in settings.host_targets if item.name == settings.gateway_host),
+                  None)
+    if target is None:
+        pytest.skip(f"GATEWAY_HOST={settings.gateway_host!r} is not in HOST_TARGETS")
+    async with httpx.AsyncClient(timeout=settings.gateway_timeout) as client:
+        try:
+            snapshot = await collectors.fetch_snapshot(client, target,
+                                                       timeout=settings.gateway_timeout,
+                                                       tls=agent_tls.get(target.name))
+        except collectors.UpstreamError as exc:
+            pytest.skip(f"{target.name} did not answer: {exc}")
+    payload = gateway.parse(snapshot, target.name)
+    if not payload["supported"]:
+        pytest.fail(f"{target.name} runs agent {payload.get('agent_version')} — the gateway page "
+                    "needs 1.2.0's haproxy/nginx/sockets sections (AGENTS.md gap 11's rollout)")
+    assert payload["host"] == target.name
+    assert payload["sockets"]["established"] == len(snapshot["sockets"]["established"])
+    assert payload["inbound"], "no frontend on either config would leave the page empty"
+    for pipe in payload["inbound"]:
+        assert isinstance(pipe["connections"], int), "the socket table answered, so a count is known"
+    for group in payload["unmatched"]["groups"]:
+        assert group["connections"] > 0 and group["host_count"] > 1, (
+            "a group is only ever the unresolved fan-out of several names on one port")
+    attributed = sum(pipe["connections"] or 0 for pipe in payload["inbound"] + payload["outbound"])
+    assert attributed + payload["unmatched"]["inbound"] + payload["unmatched"]["outbound"] == \
+        payload["sockets"]["established"], "every socket counted exactly once"
+    assert all(sample.source == gateway.SOURCE for sample in gateway.samples(payload, 0.0))
+
+
 @pytest.mark.parametrize("engine", ["gufo", "vllm"])
 async def test_engine_still_publishes_the_series_we_parse(engine: str,
                                                           settings: Settings) -> None: