diff --git a/CLAUDE.md b/CLAUDE.md
index 7391f360..9516207c 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -31,11 +31,15 @@ FastAPI backend and React front-ends for the RETINA passive-radar network.
   accounts in anything committed here.
 - **Configuration lives in `backend/.env`**, which is gitignored. Add new keys to
   `backend/.env.example` so the list stays current.
-- **Verify a deploy from the environment's API.** Green tests do not cover the
-  compose/env/frontend seams, so confirm the change against the environment
-  itself before calling a deploy done: `aircraft_on_map` in
-  `/api/test/dashboard`, the node set in `/api/radar/analytics`, and
-  `/api/radar/data/aircraft.json` for the data path. `/api/test/*` reads answer
-  only an administrator or the radar key, which stays on the droplet: read them
-  there with the runbook's `tst`. Scripted requests need a browser User-Agent or
-  Cloudflare answers `403 1010`.
+- **Verify a deploy with `just verify-deploy <prod|staging|test>`.** Green tests
+  do not cover the compose/env/frontend seams, so ask the environment itself
+  before calling a deploy done. Run it before the deploy too, and compare the
+  real nodes the two runs list. For a front-end change add
+  `--expect-string <text the change adds>`, which the earlier run must not find.
+  Its verdict rests on public reads; it runs its keyed checks (RETINA_ENV, task
+  health) only when `VERIFY_DEPLOY_RADAR_KEY` holds the radar key, and reports
+  them skipped otherwise. `/api/test/*` reads answer only an administrator or
+  the radar key, which stays on the droplet: read them there with the
+  runbook's `tst`. Scripted requests need a browser User-Agent or Cloudflare
+  answers `403 1010`. `deploy/verify-deploy.py` says what it checks and what it
+  cannot see.
diff --git a/ONBOARDING.md b/ONBOARDING.md
index e71cd2bb..6009e7c0 100644
--- a/ONBOARDING.md
+++ b/ONBOARDING.md
@@ -361,7 +361,8 @@ CI runs on every PR, on push to `main`, and on demand through
    they fail after it; the runbook's Environments section has the shape.
 
 So merging to `main` deploys to production automatically. Work on a feature
-branch, open a PR, get it green, then merge.
+branch, open a PR, get it green, then merge. Once the deploy has run,
+`just verify-deploy <env>` asks that environment whether it worked.
 
 ## Things that will bite you
 
diff --git a/backend/tests/test_verify_deploy.py b/backend/tests/test_verify_deploy.py
new file mode 100644
index 00000000..90b542c1
--- /dev/null
+++ b/backend/tests/test_verify_deploy.py
@@ -0,0 +1,789 @@
+"""deploy/verify-deploy.py's verdicts, against a fake environment.
+
+Each case is a way a deploy has been misjudged here: a degraded health that was
+only the baseline, an empty map read before it had time to fill, an asset the SPA
+fallback answered with a 200, a change searched for in the wrong chunk. The
+checks must fail on the real faults and pass the baseline, so both directions are
+asserted.
+"""
+
+from __future__ import annotations
+
+import email.utils
+import http.client
+import importlib.util
+import io
+import json
+import re
+import sys
+import time
+import urllib.error
+import urllib.parse
+from pathlib import Path
+
+import pytest
+
+_SCRIPT = Path(__file__).resolve().parents[2] / "deploy" / "verify-deploy.py"
+_spec = importlib.util.spec_from_file_location("verify_deploy", _SCRIPT)
+vd = importlib.util.module_from_spec(_spec)
+sys.modules["verify_deploy"] = vd
+_spec.loader.exec_module(vd)
+
+API = "api.example.test"
+APP = "app.example.test"
+NOW = 1_800_000_000.0
+MARKER = ("x-retina-origin", "retina-server")
+KEY = "rk-test-9f3c2a7e5b1d4c60"
+HEALTH = f"{API}/api/health"
+FEED = f"{APP}/api/radar/data/aircraft-live.json"
+ALL = f"{APP}/api/radar/data/aircraft.json"
+NODES = f"{API}/api/radar/nodes"
+DASHBOARD = f"{API}/api/test/dashboard"
+REAL = ("ndeaaa", "ndebbb", "ndeccc")
+
+INDEX_HTML = (
+    '<html><head><script src="/theme-boot.js"></script>'
+    '<link rel="stylesheet" href="https://unpkg.example.test/leaflet.css" />'
+    '<script type="module" crossorigin src="/assets/index-abc.js"></script>'
+    '<link rel="stylesheet" crossorigin href="/assets/index-xyz.css"></head></html>'
+)
+INDEX_JS = 'const a=()=>import("./Lazy-def.js");__vite__mapDeps(["assets/Other-ghi.js","assets/Lazy-def.css"]);'
+
+
+def _resp(body, *, status=200, ctype="application/json", marked=True, date=NOW) -> vd.Response:
+    headers = {"content-type": ctype, "date": email.utils.formatdate(date, usegmt=True)}
+    if marked:
+        headers[MARKER[0]] = MARKER[1]
+    raw = body if isinstance(body, bytes) else (body if isinstance(body, str) else json.dumps(body)).encode()
+    return vd.Response(status, headers, raw)
+
+
+def _feed(aircraft=2, age=1, by="ndeaaa", also=()):
+    """A feed of `aircraft` aircraft, each seen by node `by` and contributed to by `also`."""
+    seen = [{"hex": f"a{i}", "node_ref": by, "contributing_node_refs": list(also)} for i in range(aircraft)]
+    return _resp({"now": NOW - age, "aircraft": seen})
+
+
+def _health(fleet=False, polling=False, status="ok"):
+    return _resp({"status": status, "synthetic_fleet": fleet, "polled_radar_polling": polling})
+
+
+def _nodes(real=REAL, synthetic=("synth-1",), gone=("ndegone",)):
+    listed = {}
+    for refs, status, is_synthetic in (
+        (real, "active", False),
+        (synthetic, "active", True),
+        (gone, "disconnected", False),
+    ):
+        for ref in refs:
+            listed[f"k-{ref}"] = {"node_ref": ref, "status": status, "is_synthetic": is_synthetic}
+    return _resp({"nodes": listed})
+
+
+def _dashboard(env="test", stale=(), subsystems=None):
+    return {
+        "environment": env,
+        "subsystem_health": subsystems or {"tcp_server": "ok"},
+        "task_health": {"stale_tasks": list(stale)},
+    }
+
+
+def _site(**overrides) -> dict[str, object]:
+    """URL -> response (or a list of responses, served in turn) for a healthy box without the fleet."""
+    site = {
+        f"https://{HEALTH}": _health(),
+        f"https://{FEED}": _feed(),
+        f"https://{ALL}": _feed(3, by="synth-1"),
+        f"https://{NODES}": _nodes(),
+        f"https://{DASHBOARD}": _resp(_dashboard()),
+        f"https://{APP}/": _resp(INDEX_HTML, ctype="text/html", marked=False),
+        f"https://{APP}/assets/index-abc.js": _resp(INDEX_JS, ctype="text/javascript", marked=False),
+        f"https://{APP}/assets/Lazy-def.js": _resp('export const t="Shipped today";', ctype="text/javascript"),
+        f"https://{APP}/assets/Other-ghi.js": _resp("export const u=1;", ctype="text/javascript"),
+        f"https://{APP}/assets/index-xyz.css": _resp(".retina-banner{color:teal}", ctype="text/css", marked=False),
+        f"https://{APP}/assets/Lazy-def.css": _resp(".map-surface{inset:0}", ctype="text/css", marked=False),
+        f"https://{APP}/theme-boot.js": _resp('document.documentElement.dataset.theme="dark"', ctype="text/javascript"),
+    }
+    site.update({(k if k.startswith("https://") else f"https://{k}"): v for k, v in overrides.items()})
+    return site
+
+
+def _fleet_site(**overrides) -> dict[str, object]:
+    """The fleet's box, healthy: the flag on and the synthetic fleet connected and seeing aircraft."""
+    return _site(**{HEALTH: _health(fleet=True), **overrides})
+
+
+class _Clock:
+    def __init__(self):
+        self.t = NOW
+        self.sleeps = 0
+        self.calls = 0
+
+    def __call__(self):
+        self.calls += 1
+        return self.t
+
+    def sleep(self, seconds):
+        self.sleeps += 1
+        self.t += seconds
+
+
+class _Asked(list):
+    """The URLs asked for, in order, and the headers each was last asked with."""
+
+    def __init__(self):
+        super().__init__()
+        self.headers: dict[str, dict | None] = {}
+
+
+def _serve(site, url):
+    answer = site[url]
+    if isinstance(answer, list):
+        answer = answer.pop(0) if len(answer) > 1 else answer[0]
+    if isinstance(answer, Exception):
+        raise answer
+    return answer
+
+
+def _verify(site, expect=(), *, absent=(), wait_s=60, retina_env="test", fleet=False, polling=False, key=""):
+    clock = _Clock()
+    asked = _Asked()
+
+    def get(url, headers=None):
+        parts = urllib.parse.urlsplit(url)
+        path = f"{parts.scheme}://{parts.netloc}{parts.path}"
+        asked.append(path)
+        asked.headers[path] = headers
+        return _serve(site, path)
+
+    env = vd.Environment(API, APP, retina_env, fleet, polling)
+    verifier = vd.Verifier(
+        env, get=get, clock=clock, sleep=clock.sleep, log=lambda _: None, wait_s=wait_s, marker=MARKER, key=key
+    )
+    report = verifier.run(list(expect), list(absent))
+    return report, clock, asked
+
+
+def _rows(report, level):
+    return {name: detail for lvl, name, detail in report.rows if lvl == level}
+
+
+def test_a_healthy_environment_passes():
+    report, clock, asked = _verify(_site())
+    assert _rows(report, "FAIL") == {}
+    assert set(_rows(report, "PASS")) == {"health", "aircraft feed", "map", "box", "origin", "bundle"}
+    assert clock.sleeps == 0
+    # The unfiltered feed is the fleet's box's business alone.
+    assert f"https://{ALL}" not in asked
+
+
+def test_degraded_health_is_the_baseline_and_only_warns():
+    report, _, _ = _verify(_site(**{HEALTH: _health(status="degraded")}))
+    assert "health" in _rows(report, "WARN")
+    assert _rows(report, "FAIL") == {}
+
+
+def test_a_cloudflare_1010_refusal_is_named():
+    refused = _resp("error code: 1010", status=403, ctype="text/plain", marked=False)
+    report, _, _ = _verify(_site(**{HEALTH: refused}))
+    assert "1010" in _rows(report, "FAIL")["health"]
+    assert "User-Agent" in _rows(report, "FAIL")["health"]
+
+
+def test_an_answer_without_the_origin_marker_fails():
+    other = _resp({"now": NOW, "aircraft": []}, marked=False)
+    report, _, _ = _verify(_site(**{FEED: other}))
+    assert f"https://{FEED}" in _rows(report, "FAIL")["origin"]
+
+
+def test_a_feed_the_flush_stopped_writing_fails():
+    report, _, _ = _verify(_site(**{FEED: _feed(1, age=300)}))
+    assert "stopped" in _rows(report, "FAIL")["aircraft feed"]
+
+
+def test_an_empty_map_is_polled_until_it_fills():
+    report, clock, _ = _verify(_site(**{FEED: [_feed(0), _feed(0), _feed(4)]}), wait_s=600)
+    assert "4 aircraft on the public map" in _rows(report, "PASS")["map"]
+    assert clock.sleeps == 2
+
+
+def test_the_map_is_judged_on_the_feed_it_draws_not_the_unfiltered_one():
+    report, _, asked = _verify(_site(**{FEED: _feed(0), ALL: _feed(9)}), wait_s=60)
+    assert "no aircraft on the public map" in _rows(report, "WARN")["map"]
+    assert f"https://{FEED}" in asked
+
+
+# Real-node boxes: an empty public map with real nodes connected warns, and the
+# control beside it passes.
+
+
+def test_an_empty_public_map_over_real_nodes_only_warns():
+    report, _, _ = _verify(_site(**{FEED: _feed(0)}), wait_s=60)
+    assert "quiet sky" in _rows(report, "WARN")["map"]
+    assert _rows(report, "FAIL") == {}
+
+
+def test_the_same_box_with_aircraft_on_the_public_map_passes():
+    report, _, _ = _verify(_site(**{FEED: _feed(5)}), wait_s=60)
+    assert _rows(report, "PASS")["map"] == "5 aircraft on the public map; 3 real node(s) connected"
+
+
+# A feed that errors, times out or is not a feed fails; the healthy control is
+# test_a_healthy_environment_passes.
+
+
+def test_a_feed_that_errors_fails():
+    broken = _resp("bad gateway", status=502, ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{FEED: broken}), wait_s=60)
+    assert "502" in _rows(report, "FAIL")["aircraft feed"]
+
+
+def test_a_feed_that_times_out_fails():
+    report, _, _ = _verify(_site(**{FEED: vd.Unreachable(f"https://{FEED}: timed out")}), wait_s=60)
+    assert "timed out" in _rows(report, "FAIL")["aircraft feed"]
+
+
+@pytest.mark.parametrize(
+    "body",
+    [{"now": NOW, "planes": []}, {"now": "soon", "aircraft": []}, {"now": NOW, "aircraft": {}}],
+    ids=["no aircraft list", "now not a number", "aircraft not a list"],
+)
+def test_a_feed_of_the_wrong_shape_fails(body):
+    report, _, _ = _verify(_site(**{FEED: _resp(body)}), wait_s=60)
+    assert "is not an aircraft feed" in _rows(report, "FAIL")["aircraft feed"]
+
+
+def test_a_node_list_of_the_wrong_shape_fails():
+    report, _, _ = _verify(_site(**{NODES: _resp({"nodes": []})}), wait_s=60)
+    assert "carries no node list" in _rows(report, "FAIL")["map"]
+
+
+# The fleet's box: the fleet must be connected and must see aircraft.
+
+
+def test_the_fleet_s_box_passes_with_the_fleet_seeing_aircraft():
+    report, _, _ = _verify(_fleet_site(), fleet=True)
+    assert _rows(report, "PASS")["fleet"] == "3 aircraft seen by the 1 connected synthetic node(s)"
+    assert _rows(report, "FAIL") == {}
+
+
+def test_the_fleet_connected_with_no_aircraft_after_the_window_fails():
+    # The same box as the test above, with the fleet's aircraft taken away.
+    report, clock, _ = _verify(_fleet_site(**{ALL: _feed(0)}), wait_s=60, fleet=True)
+    assert "was seen by the 1 connected synthetic" in _rows(report, "FAIL")["fleet"]
+    assert "post-restart window" in _rows(report, "FAIL")["fleet"]
+    assert clock.t - NOW >= 60
+
+
+def test_on_the_fleet_s_box_real_nodes_aircraft_do_not_stand_in_for_the_fleet():
+    report, clock, _ = _verify(_fleet_site(**{ALL: _feed(9, by="ndeaaa")}), wait_s=60, fleet=True)
+    assert "was seen by the 1 connected synthetic" in _rows(report, "FAIL")["fleet"]
+    assert clock.sleeps > 0
+
+
+def test_an_aircraft_the_fleet_contributed_to_counts_for_it():
+    report, _, _ = _verify(_fleet_site(**{ALL: _feed(2, by="ndeaaa", also=("synth-1",))}), fleet=True)
+    assert "2 aircraft seen by" in _rows(report, "PASS")["fleet"]
+
+
+def test_on_the_fleet_s_box_a_fleet_not_connected_fails():
+    report, clock, _ = _verify(_fleet_site(**{NODES: _nodes(synthetic=())}), wait_s=60, fleet=True)
+    assert "synthetic fleet is not connected" in _rows(report, "FAIL")["fleet"]
+    assert clock.sleeps > 0
+
+
+def test_an_unfiltered_feed_of_the_wrong_shape_fails_the_fleet():
+    report, _, _ = _verify(_fleet_site(**{ALL: _resp({"now": NOW})}), wait_s=60, fleet=True)
+    assert "is not an aircraft feed" in _rows(report, "FAIL")["fleet"]
+
+
+def test_the_fleet_s_box_with_no_real_node_says_the_public_map_is_empty():
+    report, _, _ = _verify(_fleet_site(**{NODES: _nodes(real=()), FEED: _feed(0)}), fleet=True)
+    assert "shows real nodes only" in _rows(report, "INFO")["map"]
+    assert _rows(report, "FAIL") == {}
+
+
+def test_a_stale_feed_fails_even_under_a_quiet_sky():
+    # Every frame rewrites the feed, aircraft or none, so a quiet sky leaves it fresh.
+    unpublished = _resp({"now": 0, "aircraft": []})
+    report, _, _ = _verify(_site(**{FEED: unpublished}), wait_s=60)
+    assert "stopped" in _rows(report, "FAIL")["aircraft feed"]
+
+
+def test_no_node_connected_fails_even_on_a_real_node_environment():
+    report, _, _ = _verify(_site(**{FEED: _feed(0), NODES: _nodes(real=(), synthetic=())}), wait_s=60)
+    assert "no real node on the public list is connected" in _rows(report, "FAIL")["map"]
+
+
+def test_synthetic_nodes_do_not_stand_in_for_real_ones_off_the_fleet_s_box():
+    report, _, _ = _verify(_site(**{NODES: _nodes(real=())}), wait_s=60)
+    assert "no real node on the public list is connected" in _rows(report, "FAIL")["map"]
+
+
+def test_the_real_nodes_are_listed_for_comparison():
+    report, _, _ = _verify(_site())
+    assert _rows(report, "INFO")["real nodes"] == "3 connected: ndeaaa ndebbb ndeccc"
+
+
+def test_an_empty_node_list_is_read_again():
+    report, clock, _ = _verify(_site(**{NODES: [_resp({"nodes": {}}), _nodes(real=("ndeaaa",))]}))
+    assert _rows(report, "INFO")["real nodes"].startswith("1 connected")
+    assert clock.sleeps == 1
+
+
+# Which box answered, from /api/health's flags.
+
+
+def test_a_hostname_reaching_a_box_with_the_other_fleet_setting_fails():
+    report, _, _ = _verify(_site(**{HEALTH: _health(fleet=True)}))
+    assert "synthetic_fleet=True" in _rows(report, "FAIL")["box"]
+
+
+def test_prod_answering_for_staging_fails():
+    report, _, _ = _verify(_site(**{HEALTH: _health(polling=True)}), retina_env="test", polling=False)
+    assert "polled_radar_polling=True where the overlay leaves off" in _rows(report, "FAIL")["box"]
+    report, _, _ = _verify(_site(**{HEALTH: _health(polling=True)}), retina_env="production", polling=True)
+    assert "polled_radar_polling=True" in _rows(report, "PASS")["box"]
+
+
+def test_a_health_that_does_not_say_which_box_warns():
+    report, _, _ = _verify(_site(**{HEALTH: _resp({"status": "ok"})}))
+    assert "unproven" in _rows(report, "WARN")["box"]
+
+
+# The keyed checks.
+
+
+def test_without_the_key_the_keyed_checks_are_skipped_and_not_asked_for():
+    report, _, asked = _verify(_site())
+    assert vd.KEY_VARIABLE in _rows(report, "SKIP")["keyed checks"]
+    assert f"https://{DASHBOARD}" not in asked
+    assert "environment" not in _rows(report, "PASS")
+
+
+def test_the_key_goes_to_the_dashboard_alone_and_in_a_header():
+    report, _, asked = _verify(_site(), key=KEY)
+    assert asked.headers[f"https://{DASHBOARD}"] == {"X-API-Key": KEY}
+    assert all(not headers for url, headers in asked.headers.items() if url != f"https://{DASHBOARD}")
+    assert not any(KEY in url for url in asked)
+    assert "RETINA_ENV=test" in _rows(report, "PASS")["environment"]
+    assert "keyed checks" not in _rows(report, "SKIP")
+
+
+def test_with_the_key_a_hostname_reaching_the_wrong_box_fails():
+    report, _, _ = _verify(_site(), retina_env="production", key=KEY)
+    assert "RETINA_ENV='test'" in _rows(report, "FAIL")["environment"]
+
+
+def test_with_the_key_stale_tasks_and_waiting_subsystems_warn():
+    dash = _dashboard(stale=["solver"], subsystems={"tcp_server": "ok", "mqtt": "starting"})
+    report, _, _ = _verify(_site(**{DASHBOARD: _resp(dash)}), key=KEY)
+    assert "solver" in _rows(report, "WARN")["tasks"]
+    assert _rows(report, "WARN")["subsystems"] == "mqtt=starting"
+    assert _rows(report, "FAIL") == {}
+
+
+def test_a_key_the_environment_refuses_fails_the_keyed_checks():
+    refused = _resp({"detail": "Not authenticated"}, status=401, marked=False)
+    report, _, _ = _verify(_site(**{DASHBOARD: refused}), key=KEY)
+    assert "401 or 403 means the key is not this environment's" in _rows(report, "FAIL")["keyed checks"]
+
+
+@pytest.mark.parametrize(
+    ("key", "echo"),
+    [
+        (KEY, "x" * 190 + KEY + "y" * 40),  # straddles the 200-character cut
+        ("rk'quoted\\key", "key rk'quoted\\key is not valid here"),  # repr would escape it
+    ],
+    ids=["cut", "escaped"],
+)
+def test_a_key_echoed_back_is_redacted_before_the_body_is_quoted(key, echo):
+    refused = _resp(echo, status=403, ctype="text/plain", marked=False)
+    report, _, _ = _verify(_site(**{DASHBOARD: refused}), key=key)
+    shown = report.render()
+    assert "keyed checks" in _rows(report, "FAIL")
+    assert "[radar key" in shown
+    assert not any(key[i : i + 6] in shown for i in range(len(key) - 5))
+
+
+def test_the_report_redacts_the_key_whatever_brings_it():
+    # A second line of defence, for a detail that reaches the report other than through an answer's body.
+    report = vd.Report(secret=KEY)
+    report.add("FAIL", "keyed checks", f"https://{DASHBOARD}: {KEY!r} refused")
+    assert KEY not in report.render()
+    assert "[radar key]" in report.render()
+
+
+def _opener(site, asked):
+    """verify-deploy's _open, answering from `site` by path, as fetch() would see it."""
+    by_path = {urllib.parse.urlsplit(url).path: answer for url, answer in site.items()}
+
+    class Answer:
+        def __init__(self, resp):
+            self.status, self.headers, self._body = resp.status, resp.headers, resp.body
+
+        def read(self):
+            return self._body
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            return False
+
+    def _open(request, *, follow):
+        asked.append((request.full_url, dict(request.header_items()), follow))
+        resp = _serve(by_path, urllib.parse.urlsplit(request.full_url).path)
+        if resp.status != 200:
+            raise urllib.error.HTTPError(request.full_url, resp.status, "", resp.headers, io.BytesIO(resp.body))
+        return Answer(resp)
+
+    return _open
+
+
+def _main(monkeypatch, capsys, site, key=None):
+    asked = []
+    monkeypatch.setattr(vd, "_open", _opener(site, asked))
+    if key is None:
+        monkeypatch.delenv(vd.KEY_VARIABLE, raising=False)
+    else:
+        monkeypatch.setenv(vd.KEY_VARIABLE, key)
+    status = vd.main(["test", "--wait", "0"])
+    out = capsys.readouterr()
+    return status, out.out, out.err, asked
+
+
+@pytest.mark.parametrize(
+    ("dashboard", "status"),
+    [
+        (_resp(_dashboard()), 0),
+        # A server that echoed the key back in an error.
+        (_resp(f"key {KEY} is not valid here", status=403, ctype="text/plain", marked=False), 1),
+    ],
+)
+def test_the_key_never_appears_in_any_output(monkeypatch, capsys, dashboard, status):
+    code, out, err, asked = _main(monkeypatch, capsys, _fleet_site(**{DASHBOARD: dashboard}), key=KEY)
+    assert code == status
+    assert KEY not in out + err
+    sent = [(headers, follow) for url, headers, follow in asked if url.endswith("/api/test/dashboard")]
+    assert sent and all(headers.get("X-api-key") == KEY and not follow for headers, follow in sent)
+    assert all(follow and "X-api-key" not in headers for url, headers, follow in asked if "/api/test/" not in url)
+    assert not any(KEY in url for url, _, _ in asked)
+
+
+def test_a_key_no_header_can_carry_is_refused_without_being_shown(monkeypatch, capsys):
+    with pytest.raises(SystemExit) as exited:
+        _main(monkeypatch, capsys, _fleet_site(), key="rkNEWLINEsecret\nrest")
+    assert exited.value.code == 2
+    err = capsys.readouterr().err
+    assert vd.KEY_VARIABLE in err
+    assert "rkNEWLINEsecret" not in err
+
+
+def test_a_request_with_the_key_is_not_redirected(monkeypatch):
+    built = []
+    monkeypatch.setattr(vd.urllib.request, "build_opener", lambda *handlers: built.append(handlers) or None)
+    with pytest.raises(AttributeError):
+        vd._open(vd.urllib.request.Request("https://api.example.test/"), follow=False)
+    assert any(isinstance(h, vd._Stay) for h in built[-1])
+    with pytest.raises(AttributeError):
+        vd._open(vd.urllib.request.Request("https://api.example.test/"), follow=True)
+    assert not any(isinstance(h, vd._Stay) for h in built[-1])
+    assert vd._Stay().redirect_request(None, None, 302, "Found", {}, "https://elsewhere.example.test/") is None
+
+
+def test_keyed_checks_skipped_are_said_so_and_not_counted_as_passed(monkeypatch, capsys):
+    code, out, _, asked = _main(monkeypatch, capsys, _fleet_site())
+    assert code == 0
+    assert re.search(r"^  SKIP +keyed checks", out, re.MULTILINE)
+    assert out.rstrip().splitlines()[-1].startswith("PASS with keyed checks skipped, not passed.")
+    assert not any(url.endswith("/api/test/dashboard") for url, _, _ in asked)
+
+
+def test_the_spa_fallback_answering_the_bundle_fails():
+    fallback = _resp(INDEX_HTML, ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{f"{APP}/assets/index-abc.js": fallback}))
+    assert "SPA" in _rows(report, "FAIL")["bundle"]
+
+
+def test_a_module_script_whose_src_comes_first_is_found():
+    page = INDEX_HTML.replace(
+        'type="module" crossorigin src="/assets/index-abc.js"', 'src="/assets/index-abc.js" type="module"'
+    )
+    report, _, _ = _verify(_site(**{f"{APP}/": _resp(page, ctype="text/html", marked=False)}))
+    assert "index-abc.js" in _rows(report, "PASS")["bundle"]
+
+
+def test_the_bundle_is_named_by_the_module_script_and_other_origins_are_not_read():
+    report, _, asked = _verify(_site())
+    assert "index-abc.js" in _rows(report, "PASS")["bundle"]
+    assert f"https://{APP}/theme-boot.js" in asked
+    assert not any("unpkg.example.test" in url for url in asked)
+    assert "https://unpkg.example.test/leaflet.css" in _rows(report, "INFO")["other hosts"]
+
+
+def test_an_expected_string_is_found_in_a_lazy_chunk():
+    report, _, _ = _verify(_site(), expect=["Shipped today"])
+    assert "Lazy-def.js" in _rows(report, "PASS")["expect"]
+
+
+def test_every_asset_the_page_reaches_is_checked_without_being_asked():
+    report, _, asked = _verify(_site())
+    assert {f"https://{APP}/assets/Lazy-def.js", f"https://{APP}/assets/index-xyz.css"} <= set(asked)
+    assert "5 other asset(s)" in _rows(report, "PASS")["bundle"]
+
+
+def test_a_missing_expected_string_fails_after_searching_every_chunk():
+    report, _, asked = _verify(_site(), expect=["never shipped"])
+    assert "7 file(s)" in _rows(report, "FAIL")["expect"]
+    assert {f"https://{APP}/assets/Lazy-def.js", f"https://{APP}/assets/Other-ghi.js"} <= set(asked)
+
+
+def test_an_unreachable_host_fails_rather_than_raising():
+    report, _, _ = _verify(_site(**{HEALTH: vd.Unreachable("api: timed out")}))
+    assert "timed out" in _rows(report, "FAIL")["health"]
+
+
+@pytest.mark.parametrize(
+    ("env", "expected"),
+    [
+        ("prod", ("api.retina.fm", "app.retina.fm", "production", False, True)),
+        ("staging", ("staging-api.retina.fm", "staging-app.retina.fm", "test", False, False)),
+        ("test", ("test-api.retina.fm", "test-app.retina.fm", "test", True, False)),
+    ],
+)
+def test_each_environment_is_read_from_its_overlay(env, expected):
+    assert vd.environment(env) == vd.Environment(*expected)
+
+
+def test_the_marker_is_the_one_the_smoke_suites_assert():
+    assert vd.origin_marker() == MARKER
+
+
+def test_an_unknown_environment_is_a_usage_error():
+    with pytest.raises(SystemExit) as exited:
+        vd.main(["nowhere"])
+    assert exited.value.code == 2
+
+
+def test_a_feed_not_yet_published_after_a_restart_is_waited_for():
+    unpublished = _resp({"now": 0, "aircraft": []})
+    report, clock, _ = _verify(_site(**{FEED: [unpublished, _feed(1)]}), wait_s=600)
+    assert _rows(report, "FAIL") == {}
+    assert clock.sleeps == 1
+
+
+def test_a_feed_left_over_from_before_a_restart_is_waited_for():
+    report, clock, _ = _verify(_site(**{FEED: [_feed(3, age=400), _feed(3)]}), wait_s=600)
+    assert _rows(report, "FAIL") == {}
+    assert clock.sleeps == 1
+
+
+def test_a_server_answering_5xx_while_it_restarts_is_waited_for():
+    warming = _resp("bad gateway", status=502, ctype="text/html", marked=False)
+    report, clock, _ = _verify(_site(**{NODES: [warming, _nodes()]}), wait_s=600)
+    assert _rows(report, "FAIL") == {}
+    assert clock.sleeps == 1
+
+
+def test_the_fleet_s_unfiltered_feed_is_waited_for():
+    report, clock, _ = _verify(_fleet_site(**{ALL: [_feed(0), _feed(2, by="synth-1")]}), wait_s=600, fleet=True)
+    assert _rows(report, "FAIL") == {}
+    assert clock.sleeps == 1
+
+
+def test_an_error_from_the_edge_is_not_blamed_on_the_origin_rule():
+    refused = _resp("error code: 1010", status=403, ctype="text/plain", marked=False)
+    report, _, _ = _verify(_site(**{HEALTH: refused}))
+    assert "origin" not in _rows(report, "FAIL")
+
+
+def test_a_lazy_chunk_answered_by_the_spa_fallback_is_not_searched():
+    fallback = _resp(INDEX_HTML + "<title>Shipped today</title>", ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{f"{APP}/assets/Lazy-def.js": fallback}), expect=["Shipped today"])
+    assert "expect" in _rows(report, "FAIL")
+
+
+def test_a_connection_dropped_mid_response_is_unreachable_not_a_crash(monkeypatch):
+    def drop(*args, **kwargs):
+        raise http.client.IncompleteRead(b"partial")
+
+    monkeypatch.setattr(vd, "_open", drop)
+    monkeypatch.setattr(vd.time, "sleep", lambda s: None)
+    with pytest.raises(vd.Unreachable):
+        vd.fetch("https://api.example.test/api/health")
+
+
+def test_a_lazy_chunk_the_deploy_lacks_fails_the_bundle():
+    missing = _resp("<html>404</html>", status=404, ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{f"{APP}/assets/Other-ghi.js": missing}))
+    assert "the deploy does not carry it" in _rows(report, "FAIL")["bundle"]
+
+
+def test_an_asset_answering_5xx_once_is_asked_again():
+    flaky = [_resp("bad gateway", status=502, ctype="text/html", marked=False), _resp("x", ctype="text/javascript")]
+    report, _, _ = _verify(_site(**{f"{APP}/assets/Other-ghi.js": flaky}))
+    assert "bundle" in _rows(report, "PASS")
+
+
+def test_a_string_only_a_stylesheet_carries_is_found():
+    report, _, _ = _verify(_site(), expect=["retina-banner"])
+    assert "index-xyz.css" in _rows(report, "PASS")["expect"]
+
+
+def test_a_string_the_change_removes_must_be_gone():
+    report, _, _ = _verify(_site(), absent=["Shipped today"])
+    assert "still in Lazy-def.js" in _rows(report, "FAIL")["expect"]
+    report, _, _ = _verify(_site(), absent=["retired wording"])
+    assert "expect" in _rows(report, "PASS")
+
+
+def test_a_4xx_ends_the_wait_at_once():
+    gone = _resp("not found", status=404, ctype="text/html", marked=False)
+    report, clock, _ = _verify(_site(**{NODES: gone}), wait_s=600)
+    assert "404" in _rows(report, "FAIL")["map"]
+    assert clock.sleeps == 0
+
+
+def test_a_4xx_on_the_unfiltered_feed_ends_the_wait_at_once():
+    gone = _resp("not found", status=404, ctype="text/html", marked=False)
+    report, clock, _ = _verify(_fleet_site(**{ALL: gone}), wait_s=600, fleet=True)
+    assert "404" in _rows(report, "FAIL")["fleet"]
+    assert clock.sleeps == 0
+
+
+def test_a_lazy_page_s_stylesheet_is_read():
+    report, _, _ = _verify(_site(), expect=["map-surface"])
+    assert "Lazy-def.css" in _rows(report, "PASS")["expect"]
+    missing = _resp("<html>404</html>", status=404, ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{f"{APP}/assets/Lazy-def.css": missing}))
+    assert "Lazy-def.css" in _rows(report, "FAIL")["bundle"]
+
+
+def test_a_script_the_page_names_outside_assets_is_read():
+    report, _, _ = _verify(_site(), expect=["dataset.theme"])
+    assert "theme-boot.js" in _rows(report, "PASS")["expect"]
+
+
+def test_one_failed_read_on_the_last_poll_after_good_ones_is_a_blip():
+    blip = vd.Unreachable("api.example.test: timed out")
+    health = [_health(), _health(), _health(), blip]
+    report, _, _ = _verify(_site(**{FEED: _feed(0), HEALTH: health}), wait_s=60)
+    assert "health" not in _rows(report, "FAIL")
+    assert "quiet sky" in _rows(report, "WARN")["map"]
+
+
+def test_a_blip_is_forgiven_only_when_the_poll_before_it_was_good():
+    down = vd.Unreachable("api.example.test: timed out")
+    health = [_health(), down, down, down]
+    report, _, _ = _verify(_site(**{FEED: _feed(0), HEALTH: health}), wait_s=60)
+    assert "timed out" in _rows(report, "FAIL")["health"]
+
+
+def test_a_node_list_blip_on_the_last_poll_keeps_the_earlier_list():
+    blip = vd.Unreachable("api.example.test: timed out")
+    nodes = [_nodes(), _nodes(), _nodes(), blip]
+    report, _, _ = _verify(_site(**{FEED: _feed(0), NODES: nodes}), wait_s=60)
+    assert "quiet sky" in _rows(report, "WARN")["map"]
+    assert _rows(report, "INFO")["real nodes"].startswith("3 connected")
+
+
+def test_an_unfiltered_feed_blip_on_the_last_poll_keeps_the_earlier_feed():
+    blip = vd.Unreachable("app.example.test: timed out")
+    fleet = [_feed(0), _feed(0), _feed(0), blip]
+    report, _, _ = _verify(_fleet_site(**{ALL: fleet}), wait_s=60, fleet=True)
+    assert "was seen by the 1 connected synthetic" in _rows(report, "FAIL")["fleet"]
+
+
+def test_a_rate_limit_is_waited_out():
+    limited = _resp("too many requests", status=429, ctype="text/plain", marked=False)
+    report, clock, _ = _verify(_site(**{NODES: [limited, _nodes()]}), wait_s=600)
+    assert _rows(report, "FAIL") == {}
+    assert clock.sleeps == 1
+
+
+def test_the_page_is_asked_twice_on_a_5xx():
+    page = [_resp("bad gateway", status=502, ctype="text/html", marked=False), _resp(INDEX_HTML, ctype="text/html")]
+    report, _, _ = _verify(_site(**{f"{APP}/": page}))
+    assert "bundle" in _rows(report, "PASS")
+
+
+def test_a_walk_at_its_bound_still_reads_what_fits(monkeypatch):
+    monkeypatch.setattr(vd, "MAX_ASSETS", 4)
+    report, _, _ = _verify(_site(), expect=["Shipped today"])
+    assert "Lazy-def.js" in _rows(report, "PASS")["expect"]
+    assert "2 more unread" in _rows(report, "WARN")["bundle"]
+
+
+def test_a_fault_waiting_cannot_clear_ends_the_wait_at_once():
+    bad_cert = vd.Unreachable("api.example.test: certificate verification failed", final=True)
+    report, clock, _ = _verify(_fleet_site(**{HEALTH: bad_cert, ALL: _feed(0)}), wait_s=600, fleet=True)
+    assert "certificate" in _rows(report, "FAIL")["health"]
+    assert clock.sleeps == 0
+    # One poll was taken, so no row may claim the fleet stayed idle through a wait.
+    assert "was seen by the 1 connected synthetic" in _rows(report, "FAIL")["fleet"]
+    assert "of polling" not in report.render()
+    assert "post-restart window" not in report.render()
+
+
+def test_a_walk_cut_short_says_so_and_cannot_prove_an_absence(monkeypatch):
+    monkeypatch.setattr(vd, "MAX_ASSETS", 3)
+    report, _, _ = _verify(_site(), absent=["retired wording"])
+    assert "unread" in _rows(report, "WARN")["bundle"]
+    assert "not read" in _rows(report, "FAIL")["expect"]
+
+
+def test_the_script_content_types_match_the_smoke_suite_s():
+    shell = (Path(__file__).resolve().parents[2] / "deploy" / "page-asset.sh").read_text()
+    accepted = set(re.findall(r"200:\*(\w+)\*", shell))
+    assert accepted == set(vd.JAVASCRIPT.pattern.split("|"))
+
+
+def test_the_page_itself_is_searched():
+    report, _, _ = _verify(_site(), expect=['<script src="/theme-boot.js">'])
+    assert "index.html" in _rows(report, "PASS")["expect"]
+    report, _, _ = _verify(_site(), absent=['crossorigin href="/assets/index-xyz.css"'])
+    assert "still in index.html" in _rows(report, "FAIL")["expect"]
+
+
+def test_a_short_wait_is_counted_in_seconds():
+    report, _, _ = _verify(_fleet_site(**{ALL: _feed(0)}), wait_s=30, fleet=True)
+    assert "after 40s of polling" in _rows(report, "FAIL")["fleet"]
+
+
+def test_a_carried_feed_ages_from_when_it_was_read():
+    earlier = vd.Reading(feed={"aircraft": []}, feed_age=1.0, at=100.0, feed_at=165.0)
+    later = vd.Reading(feed="timed out", at=185.0, feed_at=190.0)
+    assert later.over(earlier).feed_age == 21.0
+
+
+def test_a_stylesheet_whose_href_comes_first_is_read():
+    page = INDEX_HTML.replace("</head>", '<link href="/extra.css" rel="stylesheet"></head>')
+    missing = _resp("<html>404</html>", status=404, ctype="text/html", marked=False)
+    report, _, _ = _verify(_site(**{f"{APP}/": _resp(page, ctype="text/html"), f"{APP}/extra.css": missing}))
+    assert "extra.css" in _rows(report, "FAIL")["bundle"]
+
+
+def test_a_stale_earlier_feed_is_not_carried_over_a_failed_read():
+    earlier = vd.Reading(feed={"aircraft": []}, feed_age=90.0, at=100.0, feed_at=101.0)
+    later = vd.Reading(feed="timed out", at=121.0, feed_at=122.0)
+    assert later.over(earlier).feed == "timed out"
+
+
+def test_the_feed_is_timed_by_its_own_read_not_the_slowest_one():
+    clock = _Clock()
+    site = _site()
+
+    def get(url, headers=None):
+        parts = urllib.parse.urlsplit(url)
+        if parts.path == "/api/radar/nodes":
+            # Once the feed's read is timed (the poll's start is the first call), a slow read's worth passes.
+            deadline = time.monotonic() + 5
+            while clock.calls < 2 and time.monotonic() < deadline:
+                time.sleep(0.005)
+            clock.t += 65
+        return site[f"{parts.scheme}://{parts.netloc}{parts.path}"]
+
+    env = vd.Environment(API, APP, "test", False)
+    verifier = vd.Verifier(env, get=get, clock=clock, sleep=clock.sleep, log=lambda _: None, marker=MARKER)
+    assert verifier._read().feed_at == NOW
diff --git a/deploy/origin-marker.sh b/deploy/origin-marker.sh
index 611e3df3..19508f58 100644
--- a/deploy/origin-marker.sh
+++ b/deploy/origin-marker.sh
@@ -1,6 +1,7 @@
 #!/bin/bash
-# Which origin answered, and the single implementation of that assertion.
-# Sourced by staging-smoke-test.sh and by CI's production smoke tests.
+# Which origin answered, and the smoke suites' implementation of that assertion.
+# Sourced by staging-smoke-test.sh and by CI's production smoke tests;
+# deploy/verify-deploy.py reads the two values below and checks them in Python.
 #
 # A probe proves a hostname responded, not that THIS service responded. Both
 # origins behind retina.fm answer /api/health with 200 and the same CSP, HSTS,
diff --git a/deploy/page-asset.sh b/deploy/page-asset.sh
index 962157fa..6d723021 100644
--- a/deploy/page-asset.sh
+++ b/deploy/page-asset.sh
@@ -1,6 +1,9 @@
 #!/bin/bash
-# Does the page render, or merely answer? The single implementation of that
-# assertion. Sourced by staging-smoke-test.sh and by CI's production smoke tests.
+# Does the page render, or merely answer? The smoke suites' implementation of
+# that assertion, sourced by staging-smoke-test.sh and by CI's production smoke
+# tests. deploy/verify-deploy.py makes the same one in Python for every asset the
+# page reaches; backend/tests/test_verify_deploy.py pins the script content types
+# the two accept together.
 #
 # A status check cannot see the failure this exists for. A bundle built for one
 # mount point and served at another returns its index.html intact, with a 200 and
diff --git a/deploy/verify-deploy.py b/deploy/verify-deploy.py
new file mode 100644
index 00000000..a8ef5f4b
--- /dev/null
+++ b/deploy/verify-deploy.py
@@ -0,0 +1,816 @@
+#!/usr/bin/env python3
+"""Ask a deployed environment whether its deploy worked.
+
+    just verify-deploy <prod|staging|test> [--expect-string S ...] [--expect-absent S ...] [--wait MINUTES]
+
+Green CI does not cover the compose, env and front-end seams, so this asks the
+environment itself. It reads only, and only through the public hostnames, which
+come from the environment's compose overlay. Its verdict rests on public reads;
+/api/test/dashboard answers only an administrator or the radar key, so the
+checks that need it (RETINA_ENV, task health) run only when the key is in
+VERIFY_DEPLOY_RADAR_KEY, and are reported as skipped otherwise. The key is sent
+in one header, to that one route, and never printed.
+
+* A request from a script needs a browser User-Agent, or Cloudflare answers
+  403 with error 1010. A curl on the droplet answers 400, because nginx demands
+  Cloudflare's client certificate (authenticated origin pulls).
+* `x-retina-origin` on every /api/ answer proves this repo's nginx answered,
+  rather than whatever service the hostname currently points at. The
+  synthetic-fleet and radar-polling flags in /api/health, read back against the
+  overlay, tell prod, staging and test apart; RETINA_ENV, when keyed, agrees.
+* `/api/health` reads `degraded` for 5-11 minutes roughly hourly, and after
+  every restart, with aircraft on the map throughout. So it warns and never
+  fails; judge a suspected regression against the alert channel's history.
+* After a restart the server answers 5xx for a while, publishes an empty feed,
+  and takes minutes to refill the map while nodes reconnect and tracks form. So
+  health, the public map's feed, the public node list and, on the fleet's box,
+  the unfiltered feed are polled together for up to --wait minutes, and only
+  what is still wrong at the end fails; a failed read on the last poll alone,
+  after a good one, is a blip. A 4xx other than a rate limit, or a certificate
+  the interpreter cannot check, will not clear with waiting and ends the wait.
+* The public map on every deployed app host draws aircraft-live.json, which
+  carries real nodes only, so its aircraft are counted there rather than in the
+  map's toolbar, which a scripted browser can leave at 0; nodes are counted in
+  the public node list, which leaves out private nodes and polled radars in
+  probation. Real nodes can solve nothing for a while (quiet hours, or frames
+  that solve nothing), so an empty public map with real nodes connected warns,
+  at the cost of the whole --wait. No real node connected, or a feed that has
+  stopped, fails: every frame a vetted node sends rewrites the feed, aircraft or
+  none. Only the synthetic fleet guarantees aircraft, so on its box the
+  unfiltered aircraft.json must carry aircraft a synthetic node saw, and a fleet
+  not connected fails.
+* A status code on the page proves nothing about the front end. Every script
+  and stylesheet the page names from its own host, and every chunk and
+  stylesheet those reach, has to come back as what it is; ones on other hosts
+  are listed, not read. --expect-string proves a change shipped by
+  finding a string it adds, and --expect-absent by the absence of one it
+  removes; run both before the deploy as well, or a string that was always
+  there proves nothing. The page itself is searched too. A file the page never
+  names (one under public/ that only a link reaches) is not read.
+
+No public endpoint reports the deployed commit (ClickUp 123zgec4n48), so a
+front-end change is attributed with --expect-string. The connected real nodes
+on the public list are listed but not judged, having nothing to compare with;
+set the list beside one from a run before the deploy. Straight after a restart
+the list is empty until the first analytics refresh, and can then still hold
+nodes that have not reconnected (they are swept offline within about three
+minutes), so repeat a run made that early before comparing. Looking at the map
+in a browser stays a human check.
+
+Exit status: 0 when every check passed (warnings allowed), 1 when one failed,
+2 on a usage error.
+"""
+
+from __future__ import annotations
+
+import argparse
+import concurrent.futures
+import email.utils
+import functools
+import http.client
+import json
+import os
+import re
+import secrets
+import ssl
+import sys
+import threading
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+
+OVERLAYS = {
+    "prod": "docker-compose.prod.yml",
+    "staging": "docker-compose.staging.yml",
+    "test": "docker-compose.test.yml",
+}
+ALIASES = {"production": "prod"}
+
+# Any current desktop browser's string passes Cloudflare's browser integrity check.
+USER_AGENT = (
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36"
+)
+
+# What the public map draws (dashboard/src/pages/map/hooks.ts), and the unfiltered feed.
+MAP_FEED = "/api/radar/data/aircraft-live.json"
+FLEET_FEED = "/api/radar/data/aircraft.json"
+# /api/health's flags that say which box answered, and the overlay variable each
+# follows. Both are on at exactly "1" (routes/sim_ingest.py, services/blah2_poller.py).
+IDENTITY = (("synthetic_fleet", "SYNTHETIC_FLEET_ENABLED"), ("polled_radar_polling", "POLLED_RADAR_POLLING_ENABLED"))
+# The aircraft flush rewrites the feed about once a second.
+FEED_STALE_S = 60
+POLL_S = 20
+DEFAULT_WAIT_MIN = 10
+# Where the radar key comes from, when the keyed checks are wanted.
+KEY_VARIABLE = "VERIFY_DEPLOY_RADAR_KEY"
+# Timeouts and rate limits: 4xx answers that waiting can clear.
+TRANSIENT_4XX = {408, 425, 429}
+# A bound on the asset walk; the console has about fifty.
+MAX_ASSETS = 400
+# The content types deploy/page-asset.sh accepts as a script, pinned together by
+# backend/tests/test_verify_deploy.py.
+JAVASCRIPT = re.compile(r"javascript|ecmascript")
+STYLESHEET = re.compile(r"text/css")
+
+
+@dataclass
+class Response:
+    status: int
+    headers: dict[str, str]
+    body: bytes
+
+    @property
+    def text(self) -> str:
+        return self.body.decode("utf-8", "replace")
+
+
+class Unreachable(Exception):
+    def __init__(self, message: str, *, final: bool = False):
+        super().__init__(message)
+        # A certificate this interpreter cannot check. A name that does not resolve is
+        # not final: macOS reports a dropped network the same way.
+        self.final = final
+
+
+class Final(str):
+    """Why a reading failed, when waiting will not change it (a 4xx)."""
+
+
+@functools.lru_cache(maxsize=1)
+def _ssl_context() -> ssl.SSLContext:
+    # python.org's interpreter ships without a CA bundle; the backend venv has certifi.
+    try:
+        import certifi
+    except ImportError:
+        return ssl.create_default_context()
+    return ssl.create_default_context(cafile=certifi.where())
+
+
+class _Stay(urllib.request.HTTPRedirectHandler):
+    """Answers a redirect with the 3xx itself, so a credential never follows one to another host."""
+
+    def redirect_request(self, *args, **kwargs):
+        return None
+
+
+def _open(request: urllib.request.Request, *, follow: bool):
+    handlers = [urllib.request.HTTPSHandler(context=_ssl_context())] + ([] if follow else [_Stay()])
+    return urllib.request.build_opener(*handlers).open(request, timeout=30)
+
+
+def fetch(url: str, headers: dict[str, str] | None = None) -> Response:
+    """GET `url`. Extra headers carry a credential, so a request with them is not redirected."""
+    request = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*", **(headers or {})})
+    for attempt in (1, 2):
+        try:
+            try:
+                with _open(request, follow=not headers) as resp:
+                    return Response(resp.status, {k.lower(): v for k, v in resp.headers.items()}, resp.read())
+            except urllib.error.HTTPError as err:
+                return Response(err.code, {k.lower(): v for k, v in err.headers.items()}, err.read())
+        except (urllib.error.URLError, OSError, http.client.HTTPException) as err:
+            reason = getattr(err, "reason", err)
+            if isinstance(reason, ssl.SSLCertVerificationError):
+                raise Unreachable(
+                    f"{url}: certificate verification failed. This interpreter has no CA bundle; "
+                    "run it through `just verify-deploy`, whose venv carries certifi.",
+                    final=True,
+                ) from err
+            if attempt == 2:
+                raise Unreachable(f"{url}: {reason!r}") from err
+            time.sleep(5)
+    raise AssertionError("unreachable")
+
+
+def _overlay_setting(env: str, key: str) -> str | None:
+    text = (REPO / OVERLAYS[env]).read_text()
+    found = re.search(rf"^\s*-\s*{key}=(\S+)\s*$", text, re.MULTILINE)
+    return found.group(1) if found else None
+
+
+@dataclass(frozen=True)
+class Environment:
+    api: str
+    app: str
+    retina_env: str
+    synthetic_fleet: bool
+    polled_radar_polling: bool = False
+
+
+def environment(env: str) -> Environment:
+    """What `env`'s overlay deploys: its hostnames, and what the server should say it is."""
+    settings = {key: _overlay_setting(env, key) for key in ("HOST_API", "HOST_APP", "RETINA_ENV")}
+    missing = sorted(key for key, value in settings.items() if not value)
+    if missing:
+        raise SystemExit(f"{OVERLAYS[env]} sets no {', '.join(missing)}")
+    flags = {flag: _overlay_setting(env, variable) == "1" for flag, variable in IDENTITY}
+    return Environment(settings["HOST_API"], settings["HOST_APP"], settings["RETINA_ENV"], **flags)
+
+
+def origin_marker() -> tuple[str, str]:
+    """The header and value deploy/origin-marker.sh asserts, so the two cannot drift."""
+    text = (REPO / "deploy" / "origin-marker.sh").read_text()
+    header = re.search(r'^RETINA_ORIGIN_HEADER="([^"]+)"', text, re.MULTILINE)
+    value = re.search(r'^RETINA_ORIGIN_VALUE="([^"]+)"', text, re.MULTILINE)
+    if not (header and value):
+        raise SystemExit("deploy/origin-marker.sh no longer defines RETINA_ORIGIN_HEADER and RETINA_ORIGIN_VALUE")
+    return header.group(1), value.group(1)
+
+
+@dataclass
+class Report:
+    rows: list[tuple[str, str, str]] = field(default_factory=list)
+    # Never printed: every detail passes through here on its way in.
+    secret: str = ""
+
+    def add(self, level: str, name: str, detail: str) -> None:
+        if self.secret:
+            detail = detail.replace(self.secret, "[radar key]")
+        self.rows.append((level, name, detail))
+
+    def levels(self, level: str) -> list[str]:
+        return [name for lvl, name, _ in self.rows if lvl == level]
+
+    def render(self) -> str:
+        lines = []
+        for level, name, detail in self.rows:
+            first, *rest = detail.splitlines() or [""]
+            lines.append(f"  {level:<5} {name:<14} {first}")
+            lines.extend(f"{'':22}{line}" for line in rest)
+        return "\n".join(lines)
+
+
+@dataclass
+class Reading:
+    """One poll of the endpoints that recover after a restart."""
+
+    health: dict | str = ""  # the body, or why it could not be read
+    feed: dict | str = ""  # MAP_FEED
+    feed_age: float = 0.0
+    nodes: dict | str = ""  # /api/radar/nodes
+    fleet: dict | str | None = None  # FLEET_FEED, read on the fleet's box only
+    at: float = 0.0  # when the poll began
+    feed_at: float = 0.0  # when its feed was read
+
+    def over(self, earlier: Reading) -> Reading:
+        """This reading, with an endpoint it could not read taken from `earlier` if that one could."""
+        merged = Reading(self.health, self.feed, self.feed_age, self.nodes, self.fleet, self.at, self.feed_at)
+        if isinstance(self.health, str) and not isinstance(self.health, Final) and isinstance(earlier.health, dict):
+            merged.health = earlier.health
+        carry_feed = isinstance(earlier.feed, dict) and earlier.feed_age <= FEED_STALE_S
+        if isinstance(self.feed, str) and not isinstance(self.feed, Final) and carry_feed:
+            merged.feed, merged.feed_age = earlier.feed, earlier.feed_age + (self.at - earlier.feed_at)
+        if isinstance(self.nodes, str) and not isinstance(self.nodes, Final) and isinstance(earlier.nodes, dict):
+            merged.nodes = earlier.nodes
+        if isinstance(self.fleet, str) and not isinstance(self.fleet, Final) and isinstance(earlier.fleet, dict):
+            merged.fleet = earlier.fleet
+        return merged
+
+    @property
+    def final(self) -> bool:
+        return any(isinstance(x, Final) for x in (self.health, self.feed, self.nodes, self.fleet))
+
+    @property
+    def shown(self) -> int:
+        """Aircraft on the public map."""
+        return len(self.feed["aircraft"]) if isinstance(self.feed, dict) else 0
+
+    def connected(self, synthetic: bool) -> list[str]:
+        """The refs of the connected nodes on the public list, synthetic or real."""
+        listed = self.nodes["nodes"] if isinstance(self.nodes, dict) else {}
+        return sorted(
+            n.get("node_ref") or key
+            for key, n in listed.items()
+            if bool(n.get("is_synthetic")) == synthetic and n.get("status") != "disconnected"
+        )
+
+    def seen_by(self, refs: list[str]) -> int:
+        """How many aircraft in the unfiltered feed a node in `refs` saw."""
+        if not isinstance(self.fleet, dict):
+            return 0
+        wanted = set(refs)
+        return sum(
+            1 for a in self.fleet["aircraft"] if wanted & {a.get("node_ref"), *(a.get("contributing_node_refs") or [])}
+        )
+
+
+class Verifier:
+    def __init__(
+        self,
+        env: Environment,
+        *,
+        get: Callable[[str], Response] = fetch,
+        clock: Callable[[], float] = time.time,
+        sleep: Callable[[float], None] = time.sleep,
+        log: Callable[[str], None] = lambda line: print(line, file=sys.stderr),
+        wait_s: float = DEFAULT_WAIT_MIN * 60,
+        marker: tuple[str, str] | None = None,
+        key: str = "",
+    ):
+        self.env = env
+        self.key = key
+        self.api = f"https://{env.api}"
+        self.app = f"https://{env.app}"
+        self.get = get
+        self.clock = clock
+        self.sleep = sleep
+        self.log = log
+        self.wait_s = wait_s
+        self.marker = marker or origin_marker()
+        self.report = Report(secret=key)
+        self.waited = 0.0
+        self.foreign: list[str] = []
+        # Ordered sets: the settle loop reads the same URLs repeatedly, several at a
+        # time, so writes to them hold the lock.
+        self._unmarked: dict[str, None] = {}
+        self._marked: dict[str, None] = {}
+        self._marks = threading.Lock()
+
+    def run(self, expect: list[str], absent: list[str] = ()) -> Report:
+        reading = self._settle()
+        self._judge(reading)
+        self._keyed()
+        self._origin()
+        self._bundle(expect, list(absent))
+        return self.report
+
+    def _json(self, url: str, headers: dict[str, str] | None = None) -> tuple[Response, dict] | str:
+        """The response and its JSON object, or why there is none."""
+        try:
+            resp = self.get(url, headers) if headers else self.get(url)
+        except Unreachable as err:
+            return Final(err) if err.final else str(err)
+        if self.key:
+            # Before anything quotes the body: a quote cut short or escaped would slip past Report.
+            resp = Response(resp.status, resp.headers, resp.body.replace(self.key.encode(), b"[radar key]"))
+        if resp.status != 200:
+            why = f"{url} answered {resp.status}.{_explain(resp)}"
+            return Final(why) if 400 <= resp.status < 500 and resp.status not in TRANSIENT_4XX else why
+        # Only a 200 is judged for the marker: an error page from the edge never reached nginx.
+        header, value = self.marker
+        with self._marks:
+            if value in resp.headers.get(header, ""):
+                self._marked[urllib.parse.urlsplit(url).netloc] = None
+                self._unmarked.pop(url, None)
+            else:
+                self._unmarked[url] = None
+        try:
+            body = json.loads(resp.body)
+        except ValueError:
+            return f"{url} answered 200 with no JSON: {resp.text[:200]!r}"
+        return (resp, body) if isinstance(body, dict) else f"{url} answered JSON that is not an object"
+
+    def _feed(self, url: str) -> tuple[Response, dict] | str:
+        """An aircraft feed, or why it is not one."""
+        got = self._json(url)
+        if isinstance(got, str):
+            return got
+        if not isinstance(got[1].get("aircraft"), list) or not isinstance(got[1].get("now", 0), (int, float)):
+            return f"{url} is not an aircraft feed: it lacks an aircraft list or a numeric now"
+        return got
+
+    def _read(self) -> Reading:
+        reading = Reading(at=self.clock())
+
+        def read_feed() -> tuple[tuple[Response, dict] | str, float]:
+            return self._feed(f"{self.app}{MAP_FEED}"), self.clock()
+
+        def read_nodes() -> tuple[Response, dict] | str:
+            got = self._json(f"{self.api}/api/radar/nodes")
+            if isinstance(got, tuple) and not isinstance(got[1].get("nodes"), dict):
+                return f"{self.api}/api/radar/nodes carries no node list"
+            return got
+
+        with concurrent.futures.ThreadPoolExecutor(max_workers=4) as pool:
+            health = pool.submit(self._json, f"{self.api}/api/health")
+            feed = pool.submit(read_feed)
+            nodes = pool.submit(read_nodes)
+            fleet = pool.submit(self._feed, f"{self.app}{FLEET_FEED}") if self.env.synthetic_fleet else None
+            health, (got, reading.feed_at), nodes = health.result(), feed.result(), nodes.result()
+            if fleet is not None:
+                fleet = fleet.result()
+                reading.fleet = fleet[1] if isinstance(fleet, tuple) else fleet
+        reading.health = health[1] if isinstance(health, tuple) else health
+        reading.nodes = nodes[1] if isinstance(nodes, tuple) else nodes
+        if isinstance(got, str):
+            reading.feed = got
+        else:
+            reading.feed = got[1]
+            reading.feed_age = _served_at(got[0], self.clock) - float(got[1].get("now") or 0)
+        return reading
+
+    def _settle(self) -> Reading:
+        start = self.clock()
+        deadline = start + self.wait_s
+        announced = False
+        previous = Reading()
+        while True:
+            reading = self._read()
+            if self._settled(reading) or reading.final or self.clock() >= deadline:
+                self.waited = self.clock() - start
+                return reading.over(previous)
+            previous = reading
+            if not announced:
+                self.log(
+                    f"  waiting for the server to settle (a restart takes minutes to refill the map); "
+                    f"polling every {POLL_S}s for up to {self.wait_s / 60:.0f} min"
+                )
+                announced = True
+            self.sleep(POLL_S)
+
+    def _settled(self, reading: Reading) -> bool:
+        """Whether every reading the verdict rests on is as a healthy box leaves it."""
+        if not all(isinstance(x, dict) for x in (reading.health, reading.feed, reading.nodes)):
+            return False
+        if reading.feed_age > FEED_STALE_S:
+            return False
+        if self.env.synthetic_fleet and not reading.seen_by(reading.connected(synthetic=True)):
+            return False
+        if reading.connected(synthetic=False):
+            return reading.shown > 0
+        # With no real node the public map has nothing to show; only the fleet's box can do without one.
+        return self.env.synthetic_fleet
+
+    def _judge(self, reading: Reading) -> None:
+        after = f" after {_span(self.waited)} of polling" if self.waited >= POLL_S else ""
+        health = reading.health
+        if isinstance(health, str):
+            self.report.add("FAIL", "health", health)
+        elif health.get("status") == "ok":
+            self.report.add("PASS", "health", "ok")
+        elif health.get("status") == "degraded":
+            self.report.add(
+                "WARN",
+                "health",
+                "degraded, which is the baseline: roughly hourly for 5-11 minutes, and after every\n"
+                "restart. Compare with the environment's alert channel before calling it a regression.",
+            )
+        else:
+            self.report.add("FAIL", "health", f"status is {health.get('status')!r}, neither ok nor degraded")
+
+        feed = reading.feed
+        if isinstance(feed, str):
+            self.report.add("FAIL", "aircraft feed", feed)
+        elif reading.feed_age > FEED_STALE_S:
+            self.report.add(
+                "FAIL",
+                "aircraft feed",
+                f"last written {reading.feed_age:.0f}s ago{after}: the aircraft flush has stopped, or no\n"
+                "vetted node is sending (a polled radar in probation does not refresh it)",
+            )
+        else:
+            self.report.add(
+                "PASS",
+                "aircraft feed",
+                f"the public map's feed carries {reading.shown} aircraft, written {max(reading.feed_age, 0):.0f}s ago",
+            )
+
+        if isinstance(reading.nodes, str):
+            self.report.add("FAIL", "map", reading.nodes)
+        else:
+            self._map(reading, after)
+            self._fleet(reading, after)
+        if isinstance(health, dict):
+            self._box(health)
+
+    def _map(self, reading: Reading, after: str) -> None:
+        real = reading.connected(synthetic=False)
+        if not real and not self.env.synthetic_fleet:
+            self.report.add("FAIL", "map", f"no real node on the public list is connected{after}")
+        elif not real:
+            self.report.add("INFO", "map", "no real node is connected, and the public map shows real nodes only")
+        elif isinstance(reading.feed, str):
+            pass  # the aircraft feed row says why there is no count
+        elif reading.shown:
+            self.report.add(
+                "PASS", "map", f"{reading.shown} aircraft on the public map; {len(real)} real node(s) connected"
+            )
+        else:
+            self.report.add(
+                "WARN",
+                "map",
+                f"no aircraft on the public map with {len(real)} real node(s) connected{after}: a quiet sky,\n"
+                "or frames that solve nothing. Compare with the alert channel.",
+            )
+        self.report.add("INFO", "real nodes", f"{len(real)} connected: {' '.join(real) or 'none'}")
+
+    def _fleet(self, reading: Reading, after: str) -> None:
+        if not self.env.synthetic_fleet:
+            return
+        real, synthetic = reading.connected(synthetic=False), reading.connected(synthetic=True)
+        if not synthetic:
+            self.report.add(
+                "FAIL", "fleet", f"the synthetic fleet is not connected{after}; {len(real)} real node(s) are"
+            )
+        elif isinstance(reading.fleet, str):
+            self.report.add("FAIL", "fleet", reading.fleet)
+        elif seen := reading.seen_by(synthetic):
+            self.report.add(
+                "PASS", "fleet", f"{seen} aircraft seen by the {len(synthetic)} connected synthetic node(s)"
+            )
+        else:
+            window = " Past the post-restart window, that is a fault." if after else ""
+            self.report.add(
+                "FAIL",
+                "fleet",
+                f"no aircraft in {FLEET_FEED} was seen by the {len(synthetic)} connected synthetic\n"
+                f"node(s){after}.{window}",
+            )
+
+    def _box(self, health: dict) -> None:
+        """Which box answered, from the flags /api/health reports and the overlay sets."""
+        agreed, wrong, missing = [], [], []
+        for flag, variable in IDENTITY:
+            want = getattr(self.env, flag)
+            if flag not in health:
+                missing.append(flag)
+            elif bool(health[flag]) != want:
+                wrong.append(f"{flag}={health[flag]} where the overlay {'sets' if want else 'leaves off'} {variable}")
+            else:
+                agreed.append(f"{flag}={bool(health[flag])}")
+        if wrong:
+            self.report.add(
+                "FAIL",
+                "box",
+                f"the server says {'; '.join(wrong)}:\n"
+                "the hostname reaches another box, or the droplet's backend/.env sets the flag itself",
+            )
+        elif missing:
+            self.report.add(
+                "WARN", "box", f"/api/health does not report {', '.join(missing)}, so which box answered is unproven"
+            )
+        else:
+            self.report.add("PASS", "box", f"{' and '.join(agreed)}, as the overlay sets")
+
+    def _keyed(self) -> None:
+        """The checks /api/test/dashboard allows, run only with the radar key."""
+        if not self.key:
+            self.report.add(
+                "SKIP",
+                "keyed checks",
+                f"RETINA_ENV and task health were not read: set {KEY_VARIABLE} to run them.\n"
+                "The key stays on the droplet; the runbook's tst reads the same route there.",
+            )
+            return
+        got = self._json(f"{self.api}/api/test/dashboard", {"X-API-Key": self.key})
+        if isinstance(got, str):
+            self.report.add("FAIL", "keyed checks", f"{got} A 401 or 403 means the key is not this environment's.")
+            return
+        dash = got[1]
+        served_env = dash.get("environment")
+        if served_env == self.env.retina_env:
+            self.report.add("PASS", "environment", f"RETINA_ENV={served_env}, as the overlay sets")
+        else:
+            self.report.add(
+                "FAIL",
+                "environment",
+                f"the server says RETINA_ENV={served_env!r} where the overlay sets {self.env.retina_env!r}: "
+                "the hostname reaches another box",
+            )
+        stale = dash.get("task_health", {}).get("stale_tasks") or []
+        if stale:
+            self.report.add("WARN", "tasks", f"stale: {', '.join(map(str, stale))}")
+        waiting = {k: v for k, v in dash.get("subsystem_health", {}).items() if v != "ok"}
+        if waiting:
+            self.report.add("WARN", "subsystems", ", ".join(f"{k}={v}" for k, v in sorted(waiting.items())))
+
+    def _origin(self) -> None:
+        header, value = self.marker
+        if self._unmarked:
+            self.report.add(
+                "FAIL",
+                "origin",
+                f"no {header}: {value} on {', '.join(self._unmarked)}.\n"
+                "This repo's nginx did not answer: an Origin Rule points the hostname elsewhere,\n"
+                "or the vhost lost its security-headers include.",
+            )
+        elif self._marked:
+            self.report.add("PASS", "origin", f"{header}: {value} from {', '.join(self._marked)}")
+
+    def _bundle(self, expect: list[str], absent: list[str]) -> None:
+        page_url = f"{self.app}/?cb={secrets.token_hex(6)}"
+        try:
+            page = self.get(page_url)
+            if page.status >= 500:
+                page = self.get(page_url)
+        except Unreachable as err:
+            self.report.add("FAIL", "bundle", str(err))
+            return
+        if page.status != 200 or "text/html" not in page.headers.get("content-type", ""):
+            self.report.add(
+                "FAIL",
+                "bundle",
+                f"{page_url} answered {page.status} {page.headers.get('content-type')}.{_explain(page)}",
+            )
+            return
+        src = next(
+            (
+                ref.group(1)
+                for attrs in re.findall(r"<script\b([^>]*)>", page.text, re.IGNORECASE)
+                if re.search(r'\btype=["\']?module\b', attrs, re.IGNORECASE)
+                and (ref := re.search(r'\bsrc=["\']([^"\']+)["\']', attrs, re.IGNORECASE))
+            ),
+            None,
+        )
+        if not src:
+            self.report.add(
+                "FAIL", "bundle", f"{self.app}/ names no module script, so it is not the console's index.html"
+            )
+            return
+        index_url = urllib.parse.urljoin(page_url, src)
+        sources, broken, unread = self._walk(page_url, page.text, index_url)
+        searched = {**sources, f"{self.app}/index.html": page.text}
+        if index_url in broken:
+            self.report.add("FAIL", "bundle", f"{_name(index_url)} {broken[index_url]}")
+            return
+        if broken:
+            listed = "; ".join(f"{_name(url)} {why}" for url, why in list(broken.items())[:5])
+            self.report.add("FAIL", "bundle", f"{len(broken)} of the page's assets did not resolve: {listed}")
+        else:
+            self.report.add(
+                "PASS", "bundle", f"{_name(index_url)} and the {len(sources) - 1} other asset(s) it needs resolve"
+            )
+        if unread:
+            self.report.add("WARN", "bundle", f"stopped at {MAX_ASSETS} assets with {unread} more unread")
+        if self.foreign:
+            self.report.add("INFO", "other hosts", f"the page also names, not read: {' '.join(self.foreign)}")
+        for needle in expect:
+            hit = next((url for url, text in searched.items() if needle in text), None)
+            if hit:
+                self.report.add("PASS", "expect", f"{needle!r} is in {_name(hit)}")
+            else:
+                self.report.add("FAIL", "expect", f"{needle!r} is in none of the {len(searched)} file(s) read")
+        for needle in absent:
+            hit = next((url for url, text in searched.items() if needle in text), None)
+            if hit:
+                self.report.add("FAIL", "expect", f"{needle!r} is still in {_name(hit)}")
+            elif unread:
+                self.report.add("FAIL", "expect", f"{needle!r} may be in one of the {unread} asset(s) not read")
+            else:
+                self.report.add("PASS", "expect", f"{needle!r} is in none of the {len(searched)} file(s) read")
+
+    def _walk(self, page_url: str, page: str, index_url: str) -> tuple[dict[str, str], dict[str, str], int]:
+        """Every script and stylesheet the page names and every chunk and stylesheet those reach.
+
+        Returns their sources, why the rest did not resolve, and how many were left
+        unread at the MAX_ASSETS bound.
+        """
+        origin = urllib.parse.urlsplit(page_url).netloc
+        named = [index_url]
+        self.foreign = []
+        for tag, attrs in re.findall(r"<(script|link)\b([^>]*)>", page, re.IGNORECASE):
+            if tag.lower() == "link" and not re.search(r'\brel=["\']?stylesheet\b', attrs, re.IGNORECASE):
+                continue
+            ref = re.search(r'\b(?:src|href)=["\']([^"\']+)["\']', attrs, re.IGNORECASE)
+            if ref:
+                url = urllib.parse.urljoin(page_url, ref.group(1))
+                if urllib.parse.urlsplit(url).netloc == origin:
+                    named.append(url)
+                else:
+                    self.foreign.append(url)
+        sources: dict[str, str] = {}
+        broken: dict[str, str] = {}
+        seen = set(named)
+        level = list(dict.fromkeys(named))
+        with concurrent.futures.ThreadPoolExecutor(max_workers=8) as pool:
+            while level:
+                room = MAX_ASSETS - len(sources) - len(broken)
+                batch, rest = level[:room], level[room:]
+                following = []
+                for url, (text, why) in zip(batch, pool.map(self._asset, batch), strict=True):
+                    if text is None:
+                        broken[url] = why
+                        continue
+                    sources[url] = text
+                    if _kind(url) is JAVASCRIPT:
+                        for ref in _chunk_refs(text):
+                            ref_url = urllib.parse.urljoin(url, ref)
+                            if ref_url not in seen:
+                                seen.add(ref_url)
+                                following.append(ref_url)
+                if rest:
+                    return sources, broken, len(rest) + len(following)
+                level = following
+        return sources, broken, 0
+
+    def _asset(self, url: str) -> tuple[str, None] | tuple[None, str]:
+        """An asset's text, or why it is not what the page expects. A 5xx is asked twice."""
+        kind = _kind(url)
+        for attempt in (1, 2):
+            try:
+                resp = self.get(url)
+            except Unreachable as err:
+                # fetch has already asked twice.
+                return None, f"unreachable: {err}"
+            ctype = resp.headers.get("content-type", "")
+            if resp.status == 200 and kind.search(ctype):
+                return resp.text, None
+            why = f"answered {resp.status} {ctype}"
+            if resp.status == 200:
+                return None, f"{why}: the SPA fallback's index.html, so the page renders nothing"
+            if resp.status == 404:
+                return None, f"{why}: the deploy does not carry it"
+            if resp.status < 500 or attempt == 2:
+                return None, why
+        raise AssertionError("unreachable")
+
+
+def _kind(url: str) -> re.Pattern:
+    return STYLESHEET if urllib.parse.urlsplit(url).path.endswith(".css") else JAVASCRIPT
+
+
+def _span(seconds: float) -> str:
+    return f"{seconds:.0f}s" if seconds < 120 else f"{seconds / 60:.0f} min"
+
+
+def _name(url: str) -> str:
+    return url.rsplit("/", 1)[-1]
+
+
+def _chunk_refs(text: str) -> list[str]:
+    # Vite names lazy chunks relative to the importer ("./Page-hash.js") and lists
+    # preload dependencies, a lazy page's stylesheet among them, from the root
+    # ("assets/Page-hash.css").
+    refs = []
+    for ref in re.findall(r"""["'`]((?:\./|/?assets/)[\w.-]+\.(?:js|css))["'`]""", text):
+        refs.append("/" + ref.lstrip("/") if ref.startswith(("assets/", "/assets/")) else ref)
+    return list(dict.fromkeys(refs))
+
+
+def _served_at(resp: Response, clock: Callable[[], float]) -> float:
+    # The edge's Date header, so a skewed local clock cannot fake a stale feed.
+    try:
+        return email.utils.parsedate_to_datetime(resp.headers["date"]).timestamp()
+    except (KeyError, TypeError, ValueError):
+        return clock()
+
+
+def _explain(resp: Response) -> str:
+    body = resp.text
+    if resp.status == 403 and "1010" in body:
+        return " That is Cloudflare's browser integrity check (error 1010) refusing the User-Agent."
+    if resp.status == 400 and "SSL certificate" in body:
+        return " nginx wants Cloudflare's client certificate: this request bypassed the edge."
+    return f" {body[:200]!r}" if body else ""
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
+    parser.add_argument("env", help="prod, staging or test")
+    parser.add_argument(
+        "--expect-string",
+        action="append",
+        default=[],
+        metavar="S",
+        help="a string the deployed front end must contain; repeatable",
+    )
+    parser.add_argument(
+        "--expect-absent",
+        action="append",
+        default=[],
+        metavar="S",
+        help="a string the deployed front end must no longer contain; repeatable",
+    )
+    parser.add_argument(
+        "--wait",
+        type=float,
+        default=DEFAULT_WAIT_MIN,
+        metavar="MINUTES",
+        help=f"how long a restarting server may take to settle (default {DEFAULT_WAIT_MIN})",
+    )
+    args = parser.parse_args(argv)
+    name = ALIASES.get(args.env, args.env)
+    if name not in OVERLAYS:
+        parser.error(f"unknown environment {args.env!r}; use one of {', '.join(OVERLAYS)}")
+
+    env = environment(name)
+    print(f"verify-deploy {name}: api {env.api}, app {env.app}")
+    key = os.environ.get(KEY_VARIABLE, "").strip()
+    if key and not re.fullmatch(r"[!-~]+", key):
+        # Named, never shown: http.client would put the key itself in its error.
+        parser.error(f"{KEY_VARIABLE} holds a character an HTTP header cannot carry")
+    report = Verifier(env, wait_s=args.wait * 60, key=key).run(args.expect_string, args.expect_absent)
+    report.add(
+        "INFO", "commit", "not reported by any public endpoint; attribute a front-end change with --expect-string"
+    )
+    print(report.render())
+
+    failed, warned, skipped = report.levels("FAIL"), report.levels("WARN"), report.levels("SKIP")
+    if failed:
+        print(f"FAIL: {', '.join(dict.fromkeys(failed))}")
+        return 1
+    notes = [f"{len(warned)} warning(s)"] if warned else []
+    notes += [f"{', '.join(skipped)} skipped, not passed"] if skipped else []
+    summary = f" with {'; '.join(notes)}" if notes else ""
+    print(f"PASS{summary}. The in-browser look at the map is still yours.")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/justfile b/justfile
index 312fd670..1a1591d4 100644
--- a/justfile
+++ b/justfile
@@ -501,3 +501,11 @@ deploy-test-status:
 # Tail retina-test's container logs (Ctrl-C to stop; the stack keeps running)
 deploy-test-logs service="":
     ssh "{{host_test}}" "cd {{app_test}} && docker compose logs -f --tail 100 {{service}}"
+
+# ── deployed environments ────────────────────────────────────────────────────
+
+# Did the deploy work? Asks prod | staging | test over its public hostnames (--help for options)
+[positional-arguments]
+verify-deploy env *args:
+    @[ -x "{{py}}" ] || { echo "no backend venv, whose certifi verifies TLS: run just setup"; exit 1; }
+    @"{{py}}" "{{root}}/deploy/verify-deploy.py" "$@"