diff --git a/bench/health_hol_repro.py b/bench/health_hol_repro.py
new file mode 100644
index 00000000..9dd043e0
--- /dev/null
+++ b/bench/health_hol_repro.py
@@ -0,0 +1,290 @@
+#!/usr/bin/env python3
+"""Does a slow upstream delay /_health, and at what concurrency?
+
+Two candidate mechanisms produce Dom's ECS symptom (three UNHEALTHY samples,
+flat CPU, self-recovery):
+
+  saturation      all `thread_pool_count` handler threads are busy, the worker
+                  hits max_conn and pauses accept. Needs ~128 concurrent slow
+                  requests before a health probe suffers.
+  head-of-line    httpz batches up to 16 ready requests and hands the whole
+                  batch to ONE pool thread (worker.zig flush), each pool thread
+                  owns a private queue, and stealing is one hop. A probe queued
+                  behind one slow request waits for it, however many threads are
+                  idle. Needs ~1 concurrent slow request.
+
+So the concurrency at which health latency first rises separates them. This
+runs a slow fake upstream, the real edge binary, and probes /_health on a fresh
+TCP connection (what an ECS health check does) while k requests are in flight.
+
+Usage: python3 bench/health_hol_repro.py [--edge zig-out/bin/edge] [--delay 3.0]
+"""
+
+import argparse
+import json
+import os
+import socket
+import statistics
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+
+BODY = json.dumps([{"message": "x" * 200, "ddsource": "repro"}]).encode()
+
+
+class SlowUpstream(BaseHTTPRequestHandler):
+    """Reads the whole body, waits, then answers 200. `delay` is set below."""
+
+    delay = 3.0
+    protocol_version = "HTTP/1.1"
+
+    def do_POST(self):  # noqa: N802 - BaseHTTPRequestHandler's name
+        length = int(self.headers.get("content-length") or 0)
+        remaining = length
+        while remaining > 0:
+            chunk = self.rfile.read(min(65536, remaining))
+            if not chunk:
+                break
+            remaining -= len(chunk)
+        time.sleep(self.delay)
+        self.send_response(200)
+        self.send_header("content-length", "2")
+        self.end_headers()
+        self.wfile.write(b"{}")
+
+    def log_message(self, *args):
+        pass
+
+
+def free_port():
+    s = socket.socket()
+    s.bind(("127.0.0.1", 0))
+    port = s.getsockname()[1]
+    s.close()
+    return port
+
+
+def probe_health(port, timeout=40.0):
+    """One health check on a fresh connection. Returns seconds, or None."""
+    started = time.monotonic()
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=timeout) as s:
+            s.settimeout(timeout)
+            s.sendall(b"GET /_health HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")
+            data = b""
+            while b"\r\n\r\n" not in data:
+                chunk = s.recv(4096)
+                if not chunk:
+                    break
+                data += chunk
+    except (socket.timeout, OSError):
+        return None
+    if not data.startswith(b"HTTP/1.1 200"):
+        return None
+    return time.monotonic() - started
+
+
+def post_logs(port, timeout=60.0):
+    """One POST /api/v2/logs on a fresh connection. Returns seconds, or None."""
+    started = time.monotonic()
+    head = (
+        "POST /api/v2/logs HTTP/1.1\r\nHost: x\r\n"
+        "Content-Type: application/json\r\n"
+        "Connection: close\r\n"
+        "Content-Length: %d\r\n\r\n" % len(BODY)
+    ).encode()
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=timeout) as s:
+            s.settimeout(timeout)
+            s.sendall(head + BODY)
+            data = b""
+            while b"\r\n\r\n" not in data:
+                chunk = s.recv(4096)
+                if not chunk:
+                    break
+                data += chunk
+    except (socket.timeout, OSError):
+        return None
+    return time.monotonic() - started
+
+
+def metric(port, name):
+    """One value from /_edge/metrics, or None."""
+    try:
+        with socket.create_connection(("127.0.0.1", port), timeout=10) as s:
+            s.sendall(b"GET /_edge/metrics HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")
+            data = b""
+            while True:
+                chunk = s.recv(65536)
+                if not chunk:
+                    break
+                data += chunk
+    except OSError:
+        return None
+    for line in data.decode("utf-8", "replace").splitlines():
+        if line.startswith(name + " ") or line.startswith(name + "{"):
+            return line
+    return None
+
+
+def stats(samples):
+    good = [s for s in samples if s is not None]
+    misses = len(samples) - len(good)
+    if not good:
+        return "no successful probe (%d misses)" % misses
+    good.sort()
+    p50 = statistics.median(good)
+    p90 = good[min(len(good) - 1, int(0.9 * len(good)))]
+    return "p50 %6.0f ms   p90 %6.0f ms   max %6.0f ms   misses %d/%d" % (
+        p50 * 1000, p90 * 1000, max(good) * 1000, misses, len(samples),
+    )
+
+
+def load_and_probe(port, concurrency, seconds, probe_every=0.25):
+    """Hold `concurrency` POSTs in flight; probe health on fresh connections."""
+    stop = threading.Event()
+
+    def loader():
+        while not stop.is_set():
+            post_logs(port)
+
+    threads = [threading.Thread(target=loader, daemon=True) for _ in range(concurrency)]
+    for t in threads:
+        t.start()
+    time.sleep(1.0)  # let the load reach steady state
+
+    # Read the saturation gauge while the load is on: it should track the
+    # requests actually held by handler threads.
+    gauge = metric(port, "edge_requests_in_flight")
+
+    samples = []
+    deadline = time.monotonic() + seconds
+    while time.monotonic() < deadline:
+        samples.append(probe_health(port))
+        time.sleep(probe_every)
+    stop.set()
+    for t in threads:
+        t.join(timeout=30)
+    return samples, gauge
+
+
+def burst_with_health(port, posts):
+    """Fire `posts` POSTs and one health probe together, so they land in one
+    event-loop batch. Returns the health latency."""
+    result = {}
+
+    def health():
+        result["health"] = probe_health(port)
+
+    start = threading.Barrier(posts + 1)
+
+    def post():
+        start.wait()
+        post_logs(port)
+
+    def health_racer():
+        start.wait()
+        health()
+
+    threads = [threading.Thread(target=post, daemon=True) for _ in range(posts)]
+    threads.append(threading.Thread(target=health_racer, daemon=True))
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join(timeout=60)
+    return result.get("health")
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--edge", default="zig-out/bin/edge")
+    ap.add_argument("--delay", type=float, default=3.0, help="upstream latency, seconds")
+    ap.add_argument("--seconds", type=float, default=10.0, help="probe window per step")
+    ap.add_argument("--threads", type=int, default=0, help="TERO_THREAD_POOL_COUNT override")
+    ap.add_argument("--mode", choices=("all", "burst", "load"), default="all")
+    ap.add_argument("--repeats", type=int, default=3, help="burst repeats")
+    ap.add_argument("--posts", type=int, default=15, help="POSTs per burst")
+    args = ap.parse_args()
+
+    SlowUpstream.delay = args.delay
+    upstream_port = free_port()
+    upstream = ThreadingHTTPServer(("127.0.0.1", upstream_port), SlowUpstream)
+    upstream.daemon_threads = True
+    threading.Thread(target=upstream.serve_forever, daemon=True).start()
+
+    edge_port = free_port()
+    config = {
+        "listen_address": "127.0.0.1",
+        "listen_port": edge_port,
+        "upstream_url": "http://127.0.0.1:%d" % upstream_port,
+        "log_level": "info",
+        "max_body_size": 1048576,
+    }
+    cfg = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False)
+    json.dump(config, cfg)
+    cfg.close()
+
+    env = dict(os.environ)
+    if args.threads:
+        env["TERO_THREAD_POOL_COUNT"] = str(args.threads)
+    log = open("/tmp/edge-repro.log", "wb")
+    edge = subprocess.Popen([args.edge, cfg.name], stdout=log, stderr=log, env=env)
+
+    try:
+        for _ in range(100):
+            if probe_health(edge_port, timeout=1.0) is not None:
+                break
+            if edge.poll() is not None:
+                sys.exit("edge exited early; see /tmp/edge-repro.log")
+            time.sleep(0.1)
+        else:
+            sys.exit("edge never answered /_health; see /tmp/edge-repro.log")
+
+        print("edge pid %d, port %d, upstream latency %.1fs" % (edge.pid, edge_port, args.delay))
+        print("startup line: %s" % (open("/tmp/edge-repro.log").readline().strip(),))
+        print()
+
+        idle = [probe_health(edge_port) for _ in range(20)]
+        print("idle                         %s" % stats(idle))
+        print()
+
+        if args.mode in ("all", "burst"):
+            print("one batch: N slow POSTs + 1 health probe fired together")
+            sizes = (1, 4, 8, args.posts) if args.mode == "all" else (args.posts,)
+            for posts in sizes:
+                got = [burst_with_health(edge_port, posts) for _ in range(args.repeats)]
+                print("  %3d POSTs in the batch     %s" % (posts, stats(got)))
+                over = sorted(int(g * 1000) for g in got if g is not None and g > 1.0)
+                if over:
+                    print("      probes over 1 s (ms): %s" % (over,))
+            print()
+
+        if args.mode in ("all", "load"):
+            print("sustained load, health probed every 250 ms on a fresh connection")
+            for k in (1, 4, 8, 16, 32, 64, 128):
+                samples, gauge = load_and_probe(edge_port, k, args.seconds)
+                print("  %3d in flight              %s   [%s]" % (k, stats(samples), gauge))
+            print()
+
+        for name in (
+            "edge_requests_in_flight",
+            "edge_upstream_timeouts_total",
+            "edge_upstream_retries_total",
+        ):
+            print("%s" % (metric(edge_port, name) or (name + " (absent)")))
+    finally:
+        edge.terminate()
+        try:
+            edge.wait(timeout=10)
+        except subprocess.TimeoutExpired:
+            edge.kill()
+        upstream.shutdown()
+        log.close()
+        os.unlink(cfg.name)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/bench/idle_conn_check.py b/bench/idle_conn_check.py
new file mode 100644
index 00000000..9157cb40
--- /dev/null
+++ b/bench/idle_conn_check.py
@@ -0,0 +1,87 @@
+#!/usr/bin/env python3
+"""Hold N idle connections open, then ask whether the frontend recovers.
+
+An ECS sidecar sees half-open sockets: a client that connects and sends
+nothing. httpz caps that with its request timeout (limits.REQUEST_TIMEOUT_
+SECONDS). The stdio frontend has no inbound timeout at all, so this measures
+whether an idle connection ever gives its slot back.
+
+usage: idle_conn_check.py <edge-bin> [conns] [wait-seconds]
+"""
+import json, os, socket, subprocess, sys, tempfile, threading, time
+sys.path.insert(0, "bench")
+from health_hol_repro import SlowUpstream, free_port, probe_health, metric
+from http.server import ThreadingHTTPServer
+
+edge_bin = sys.argv[1]
+conns = int(sys.argv[2]) if len(sys.argv) > 2 else 300
+wait = float(sys.argv[3]) if len(sys.argv) > 3 else 45.0
+
+SlowUpstream.delay = 0.01
+up = free_port()
+srv = ThreadingHTTPServer(("127.0.0.1", up), SlowUpstream)
+srv.daemon_threads = True
+threading.Thread(target=srv.serve_forever, daemon=True).start()
+
+port = free_port()
+cfg = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False)
+json.dump({"listen_address": "127.0.0.1", "listen_port": port,
+           "upstream_url": "http://127.0.0.1:%d" % up,
+           "log_level": "info", "max_body_size": 1048576}, cfg)
+cfg.close()
+log = open("/tmp/edge-idle.log", "wb")
+edge = subprocess.Popen([edge_bin, cfg.name], stdout=log, stderr=log)
+
+def report(tag):
+    h = probe_health(port, timeout=5.0)
+    series = [metric(port, n) for n in (
+        "edge_connections_active", "edge_connections_max", "edge_connections_total")]
+    shed = metric(port, "edge_connections_shed_total")
+    # httpz publishes its own counters through endpoints.zig; the stdio build
+    # has none of them, so this shows both sides of the parity gap.
+    for n in ("edge_inbound_timeouts_total", "httpz_connections",
+              "httpz_timeout_request", "httpz_timeout_keepalive"):
+        v = metric(port, n)
+        if v:
+            series.append(v)
+    print("  %-22s health %-10s %s" % (
+        tag, ("%.0f ms" % (h * 1000)) if h is not None else "NO 200",
+        "  ".join(x for x in series + [shed] if x) or "(no conn series)"))
+
+held = []
+try:
+    for _ in range(100):
+        if probe_health(port, timeout=1.0) is not None:
+            break
+        time.sleep(0.1)
+    report("idle")
+
+    for _ in range(conns):
+        try:
+            s = socket.create_connection(("127.0.0.1", port), timeout=5)
+            held.append(s)  # connected, never sends a byte
+        except OSError:
+            break
+    print("  held %d idle sockets" % len(held))
+    time.sleep(2)
+    report("with idle sockets")
+
+    # Does anything reclaim those slots while the sockets stay open?
+    waited = 0.0
+    while waited < wait:
+        time.sleep(15)
+        waited += 15
+        report("after %2.0f s" % waited)
+
+    for s in held:
+        s.close()
+    held = []
+    time.sleep(3)
+    report("after client close")
+finally:
+    for s in held:
+        s.close()
+    edge.terminate()
+    try: edge.wait(timeout=10)
+    except subprocess.TimeoutExpired: edge.kill()
+    srv.shutdown(); log.close(); os.unlink(cfg.name)
diff --git a/bench/partial_request_check.py b/bench/partial_request_check.py
new file mode 100644
index 00000000..a91afcb6
--- /dev/null
+++ b/bench/partial_request_check.py
@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Does a stalled partial request get an answer, or is it dropped in silence?
+
+A sender that delivers a head and then stalls has a request in flight. The
+frontend must not close in silence: it owes a status and a log line. This
+sends a `Content-Length: 100` head plus 10 bytes, then stalls.
+
+usage: partial_request_check.py <edge-bin>
+"""
+import json, os, socket, subprocess, sys, tempfile, threading, time
+sys.path.insert(0, "bench")
+from health_hol_repro import SlowUpstream, free_port, probe_health
+from http.server import ThreadingHTTPServer
+
+edge_bin = sys.argv[1]
+SlowUpstream.delay = 0.01
+up = free_port()
+srv = ThreadingHTTPServer(("127.0.0.1", up), SlowUpstream)
+srv.daemon_threads = True
+threading.Thread(target=srv.serve_forever, daemon=True).start()
+
+port = free_port()
+cfg = tempfile.NamedTemporaryFile("w", suffix=".json", delete=False)
+json.dump({"listen_address": "127.0.0.1", "listen_port": port,
+           "upstream_url": "http://127.0.0.1:%d" % up,
+           "log_level": "info", "max_body_size": 1048576}, cfg)
+cfg.close()
+log_path = "/tmp/edge-partial.log"
+log = open(log_path, "wb")
+edge = subprocess.Popen([edge_bin, cfg.name], stdout=log, stderr=log)
+
+try:
+    for _ in range(100):
+        if probe_health(port, timeout=1.0) is not None:
+            break
+        time.sleep(0.1)
+
+    s = socket.create_connection(("127.0.0.1", port), timeout=90)
+    s.sendall(b"POST /api/v2/logs HTTP/1.1\r\nHost: x\r\n"
+              b"Content-Type: application/json\r\nContent-Length: 100\r\n\r\n")
+    s.sendall(b'[{"m":"x"')  # 9 of 100 bytes, then stall
+    started = time.monotonic()
+    s.settimeout(90)
+    data = b""
+    while True:
+        chunk = s.recv(4096)
+        if not chunk:
+            break
+        data += chunk
+        if b"\r\n\r\n" in data:
+            break
+    waited = time.monotonic() - started
+    first = data.split(b"\r\n")[0].decode() if data else "connection closed with no response"
+    print("  after %4.1f s the sender got: %s" % (waited, first))
+    s.close()
+    time.sleep(0.5)
+    print("  log lines naming the drop:")
+    for line in open(log_path):
+        if "timeout" in line.lower() or "408" in line:
+            print("    " + line.rstrip())
+finally:
+    edge.terminate()
+    try: edge.wait(timeout=10)
+    except subprocess.TimeoutExpired: edge.kill()
+    srv.shutdown(); log.close(); os.unlink(cfg.name)
diff --git a/src/frontend/exchange.zig b/src/frontend/exchange.zig
index a7ebaccf..61942846 100644
--- a/src/frontend/exchange.zig
+++ b/src/frontend/exchange.zig
@@ -19,6 +19,19 @@ const UpstreamConnectionEvicted = struct { path: []const u8, err: []const u8 };
 /// The upstream answered before the body was fully sent and stopped reading;
 /// the send failed but a real status was already on the wire.
 const UpstreamEarlyResponse = struct { path: []const u8, status: u16, err: []const u8 };
+/// The watchdog cut this attempt off at its deadline. Without this line the
+/// request reports a generic transport failure, and nothing names the stalled
+/// intake as the cause.
+const UpstreamTimedOut = struct { path: []const u8, phase: []const u8 };
+/// A dial slow enough to matter. `std.http.Client` takes no connect timeout,
+/// so a stalled dial holds its handler thread and the watchdog has no socket
+/// to interrupt; this line is the only way to see one.
+const UpstreamDialSlow = struct { path: []const u8, ms: f64 };
+
+/// Dial warn threshold. A pooled connection dials in microseconds and a cold
+/// TCP plus TLS handshake to a public intake costs tens of milliseconds, so a
+/// whole second means the dial is stalling.
+const dial_slow_ns: i128 = std.time.ns_per_s;
 
 /// The parts of an inbound request the upstream leg needs, already lifted
 /// out of whatever HTTP server produced it.
@@ -102,6 +115,17 @@ fn dialUpstream(
     choice: service_mod.UpstreamChoice,
     client: *std.http.Client,
 ) !std.http.Client.Request {
+    const started_ns = std.Io.Timestamp.now(ctx.io, .awake).toNanoseconds();
+    defer {
+        const elapsed_ns = std.Io.Timestamp.now(ctx.io, .awake).toNanoseconds() - started_ns;
+        if (elapsed_ns >= dial_slow_ns) {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            ctx.bus.warn(UpstreamDialSlow{
+                .path = in.path,
+                .ms = @as(f64, @floatFromInt(elapsed_ns)) / std.time.ns_per_ms,
+            });
+        }
+    }
     return exec.openUpstreamWithClient(ctx, in.arena, in.method, in.target, in.headers, choice, client) catch |err| {
         // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
         ctx.bus.info(UpstreamRetried{ .path = in.path, .err = @errorName(err) });
@@ -109,6 +133,15 @@ fn dialUpstream(
     };
 }
 
+/// Record a watchdog timeout: the warn line names the phase and the path, the
+/// counter makes the rate alertable.
+fn timedOut(ctx: *exec.SharedCtx, path: []const u8, phase: []const u8) error{UpstreamTimeout} {
+    // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+    ctx.bus.warn(UpstreamTimedOut{ .path = path, .phase = phase });
+    if (ctx.metrics) |metrics| metrics.recordUpstreamTimeout();
+    return error.UpstreamTimeout;
+}
+
 /// Send `body` upstream and relay the response into `res`.
 ///
 /// Retry: a replayable buffered body (or a bodiless method) gets a second
@@ -153,7 +186,7 @@ pub fn exchange(
                 } else |_| {}
             }
             evictUpstream(ctx, &upstream_req, in.path, err);
-            if (bufs.timed_out.load(.acquire)) return error.UpstreamTimeout;
+            if (bufs.timed_out.load(.acquire)) return timedOut(ctx, in.path, "head");
             if (!retryableTransportError(err)) return err;
             if (attempt + 1 == attempts) return error.UpstreamTransportFailed;
             // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
@@ -165,7 +198,7 @@ pub fn exchange(
         relayResponse(sink, in.arena, &upstream_res, max_response, bufs) catch |err| {
             if (bufs.timed_out.load(.acquire)) {
                 evictUpstream(ctx, &upstream_req, in.path, err);
-                return error.UpstreamTimeout;
+                return timedOut(ctx, in.path, "relay");
             }
             switch (err) {
                 error.BodyTooLarge => {
diff --git a/src/frontend/httpz/server.zig b/src/frontend/httpz/server.zig
index 9a436167..59910e7f 100644
--- a/src/frontend/httpz/server.zig
+++ b/src/frontend/httpz/server.zig
@@ -130,6 +130,13 @@ const log = std.log.scoped(.httpz_server);
 const RequestFailed = struct { method: []const u8, path: []const u8, err: []const u8 };
 /// Per-request trace at debug level.
 const RequestCompleted = struct { method: []const u8, path: []const u8, status: u16, duration_ms: f64 };
+/// Same shape at warn level, for a request that held its handler thread.
+const RequestSlow = struct { method: []const u8, path: []const u8, status: u16, duration_ms: f64 };
+
+/// Warn past this. `RequestCompleted` is debug level, which production turns
+/// off, so without this line a handler that sat on a stalled upstream for
+/// seconds leaves no record at all.
+const slow_request_seconds: f64 = 5;
 
 pub fn configFromLimits(limits: limits_mod.Limits, address: [4]u8, port: u16) httpz.Config {
     const requested_workers = limits.worker_count orelse 1;
@@ -241,6 +248,8 @@ pub const Handler = struct {
         }
         const ctx = self.ctx;
         const start_ns = std.Io.Timestamp.now(ctx.io, .awake).toNanoseconds();
+        if (ctx.metrics) |metrics| metrics.recordInFlight(1);
+        defer if (ctx.metrics) |metrics| metrics.recordInFlight(-1);
         const method = serviceMethod(req.method);
         const known_path = exec.classifyKnownPath(req.url.path, method);
         if (ctx.metrics) |metrics| {
@@ -269,13 +278,23 @@ pub const Handler = struct {
             metrics.recordRequestDuration(known_path, elapsed_s);
             metrics.recordResponse(known_path, runtime_metrics.statusClass(res.status));
         }
-        // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
-        ctx.bus.debug(RequestCompleted{
-            .method = @tagName(req.method),
-            .path = req.url.path,
-            .status = res.status,
-            .duration_ms = elapsed_s * std.time.ms_per_s,
-        });
+        if (elapsed_s >= slow_request_seconds) {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            ctx.bus.warn(RequestSlow{
+                .method = @tagName(req.method),
+                .path = req.url.path,
+                .status = res.status,
+                .duration_ms = elapsed_s * std.time.ms_per_s,
+            });
+        } else {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            ctx.bus.debug(RequestCompleted{
+                .method = @tagName(req.method),
+                .path = req.url.path,
+                .status = res.status,
+                .duration_ms = elapsed_s * std.time.ms_per_s,
+            });
+        }
     }
 
     fn dispatch(self: *Handler, req: *httpz.Request, res: *httpz.Response) !void {
diff --git a/src/frontend/stdio/conn.zig b/src/frontend/stdio/conn.zig
index 0b1abd80..db2b5289 100644
--- a/src/frontend/stdio/conn.zig
+++ b/src/frontend/stdio/conn.zig
@@ -28,6 +28,7 @@ const runtime_metrics = @import("../../runtime/runtime_metrics.zig");
 const exchange = @import("../exchange.zig");
 const paths = @import("../paths.zig");
 const endpoints = @import("../endpoints.zig");
+const deadline_reader_mod = @import("deadline_reader.zig");
 
 const Inbound = exchange.Inbound;
 const InboundBody = paths.InboundBody;
@@ -39,6 +40,26 @@ const log = std.log.scoped(.conn);
 const RequestFailed = struct { method: []const u8, path: []const u8, err: []const u8 };
 /// Per-request trace at debug level.
 const RequestCompleted = struct { method: []const u8, path: []const u8, status: u16, duration_ms: f64 };
+/// Same shape at warn level, for a request that held its connection task.
+const RequestSlow = struct { method: []const u8, path: []const u8, status: u16, duration_ms: f64 };
+/// A connection refused before it carried a request, with the 503 sent.
+const ConnectionShed = struct { reason: []const u8, answered: u16 };
+/// An inbound read hit its deadline. `idle` is a keep-alive wait with no
+/// request in flight; `request` means a partial request stalled, which drops
+/// that request, so it answers 408 first.
+const InboundTimeout = struct { phase: []const u8, answered: u16 };
+/// A head that failed to parse. Answered 400, then closed.
+const RequestRejected = struct { reason: []const u8, answered: u16 };
+/// The response was already on the wire when the request failed. The body
+/// truncates and the connection closes, so the sender must retry: there is no
+/// status left to change.
+const ResponseTruncated = struct { path: []const u8, status: u16, err: []const u8 };
+/// Even the fixed error response failed to reach the client.
+const ResponseUndeliverable = struct { answered: u16, err: []const u8 };
+
+/// Warn past this, matching the httpz frontend: `RequestCompleted` is debug
+/// level, which production turns off.
+const slow_request_seconds: f64 = 5;
 
 /// Per-connection environment: the frontend-neutral shared context plus the
 /// stdio frontend's own state (slab slot buffers, arena pool).
@@ -46,6 +67,10 @@ const Env = struct {
     shared: *exec.SharedCtx,
     slab: *conn_slab_mod.ConnSlab,
     arenas: *arena_pool_mod.ArenaPool,
+    /// The inbound reader, so a body read that hit its deadline is reported
+    /// as the client stall it is (408) instead of a generic read failure
+    /// (502), which would point at the upstream.
+    inbound: *deadline_reader_mod.DeadlineReader,
 };
 
 /// The response side of the sink contract (exchange.zig) over a
@@ -86,38 +111,57 @@ pub fn serveConnection(
     const io = shared.io;
     defer stream.close(io);
 
-    var env: Env = .{ .shared = shared, .slab = slab, .arenas = arenas };
+    // `inbound` is filled once the slab slot provides its receive buffer.
+    var env: Env = undefined;
 
     const conn_id = slab.claim(io) orelse {
         // Load shed: no slab slot. One fixed write, then close.
+        if (shared.metrics) |metrics| metrics.recordConnectionShed(.slab_full);
+        // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+        shared.bus.warn(ConnectionShed{ .reason = "connection_slab_full", .answered = 503 });
         const shed = "HTTP/1.1 503 Service Unavailable\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
-        writeRawResponse(io, stream, shed);
+        writeRawResponse(shared, io, stream, shed, 503);
         return;
     };
     defer slab.release(io, conn_id);
+    // Tracks slab occupancy, so it pairs with the claim, not with the accept:
+    // a shed connection never counts as active.
+    if (shared.metrics) |metrics| metrics.recordConnectionsActive(1);
+    defer if (shared.metrics) |metrics| metrics.recordConnectionsActive(-1);
     const arena_slot = arenas.claim(io);
     defer arenas.release(io, arena_slot);
 
-    var net_reader = std.Io.net.Stream.Reader.init(stream, io, slab.recvBuf(conn_id));
+    var inbound: deadline_reader_mod.DeadlineReader = .init(io, stream, slab.recvBuf(conn_id));
+    env = .{ .shared = shared, .slab = slab, .arenas = arenas, .inbound = &inbound };
     var net_writer = std.Io.net.Stream.Writer.init(stream, io, slab.sendBuf(conn_id));
-    var server = std.http.Server.init(&net_reader.interface, &net_writer.interface);
+    var server = std.http.Server.init(&inbound.interface, &net_writer.interface);
 
     while (server.reader.state == .ready) {
         var request = server.receiveHead() catch |err| switch (err) {
-            // Cancellation surfaces as ReadFailed through the net reader.
-            error.HttpConnectionClosing, error.ReadFailed => return,
+            // Cancellation surfaces as ReadFailed through the reader, as does
+            // a deadline; `inbound` says which.
+            error.HttpConnectionClosing, error.ReadFailed => {
+                reportReadEnd(shared, io, stream, &inbound, err);
+                return;
+            },
             else => {
                 // Malformed head (incl. unsupported content-encoding, see
                 // wiring-notes): answer 400 on the raw writer and close.
+                if (shared.metrics) |metrics| metrics.recordInvalidRequest();
+                // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+                shared.bus.warn(RequestRejected{ .reason = @errorName(err), .answered = 400 });
                 const reject = "HTTP/1.1 400 Bad Request\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
-                writeRawResponse(io, stream, reject);
+                writeRawResponse(shared, io, stream, reject, 400);
                 return;
             },
         };
         handleRequest(&env, conn_id, arena_slot, &request) catch |err| {
+            // handleRequest already reported and answered what it could; this
+            // only decides the connection's fate.
             log.debug("request handling failed: {s}", .{@errorName(err)});
             return; // connection state unknown; close it
         };
+        inbound.endRequest();
         arenas.reset(arena_slot);
     }
 }
@@ -134,6 +178,8 @@ fn handleRequest(
     const ctx = env.shared;
     const arena = env.arenas.allocator(arena_slot);
     const start_ns = std.Io.Timestamp.now(ctx.io, .awake).toNanoseconds();
+    if (ctx.metrics) |metrics| metrics.recordInFlight(1);
+    defer if (ctx.metrics) |metrics| metrics.recordInFlight(-1);
 
     // Head strings die when the body reader is created; the target must
     // outlive that for the upstream leg, logs and metrics.
@@ -147,7 +193,15 @@ fn handleRequest(
 
     var sink: Sink = .{ .request = request, .buffer = env.slab.bodyBuf(conn_id) };
     var failed: ?anyerror = null;
-    dispatch(env, conn_id, arena, request, target, &sink) catch |err| {
+    dispatch(env, conn_id, arena, request, target, &sink) catch |raw_err| {
+        // The shared path reports a stalled read as a generic read failure.
+        // Only this frontend knows the deadline fired, so name it here: the
+        // sender stalled, the upstream did not.
+        const client_stalled = env.inbound.expired == .request;
+        const err = if (client_stalled) error.InboundBodyTimeout else raw_err;
+        if (client_stalled) {
+            if (ctx.metrics) |metrics| metrics.recordInboundTimeout(.request);
+        }
         // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
         ctx.bus.err(RequestFailed{
             .method = @tagName(request.head.method),
@@ -158,6 +212,14 @@ fn handleRequest(
             metrics.recordRequestError(known_path, .uncaught);
         }
         failed = err;
+        if (sink.status != 0) {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            ctx.bus.warn(ResponseTruncated{
+                .path = path,
+                .status = sink.status,
+                .err = @errorName(err),
+            });
+        }
         if (sink.status == 0) {
             sink.status = exchange.errorStatus(err);
             request.respond("", .{
@@ -175,13 +237,23 @@ fn handleRequest(
         metrics.recordRequestDuration(known_path, elapsed_s);
         metrics.recordResponse(known_path, runtime_metrics.statusClass(sink.status));
     }
-    // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
-    ctx.bus.debug(RequestCompleted{
-        .method = @tagName(request.head.method),
-        .path = path,
-        .status = sink.status,
-        .duration_ms = elapsed_s * std.time.ms_per_s,
-    });
+    if (elapsed_s >= slow_request_seconds) {
+        // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+        ctx.bus.warn(RequestSlow{
+            .method = @tagName(request.head.method),
+            .path = path,
+            .status = sink.status,
+            .duration_ms = elapsed_s * std.time.ms_per_s,
+        });
+    } else {
+        // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+        ctx.bus.debug(RequestCompleted{
+            .method = @tagName(request.head.method),
+            .path = path,
+            .status = sink.status,
+            .duration_ms = elapsed_s * std.time.ms_per_s,
+        });
+    }
     if (failed) |err| return err;
 }
 
@@ -320,14 +392,63 @@ fn collectRequestHeaders(
     return buffer[0..count];
 }
 
-/// Best-effort fixed response on the raw stream (pre-HTTP-state failures:
-/// load shed, malformed head). Errors are ignored — the connection is being
-/// closed either way.
-fn writeRawResponse(io: std.Io, stream: std.Io.net.Stream, response: []const u8) void {
+/// Reports why the read side ended, and answers when a request was in flight.
+///
+/// Three outcomes hide behind one error: the peer closed a keep-alive
+/// connection (routine), our idle deadline reclaimed the slot (routine, but it
+/// is the capacity signal), or a partial request stalled past the request
+/// deadline. The last one drops a request the sender believes is in progress,
+/// so it gets a 408 and a warn line.
+fn reportReadEnd(
+    shared: *exec.SharedCtx,
+    io: std.Io,
+    stream: std.Io.net.Stream,
+    inbound: *deadline_reader_mod.DeadlineReader,
+    err: anyerror,
+) void {
+    const phase = inbound.expired orelse {
+        // No deadline fired: the peer went away, or we were canceled.
+        log.debug("inbound read ended: {s}", .{@errorName(inbound.err orelse err)});
+        return;
+    };
+    if (shared.metrics) |metrics| metrics.recordInboundTimeout(switch (phase) {
+        .idle => .idle,
+        .request => .request,
+    });
+    switch (phase) {
+        .idle => {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            shared.bus.debug(InboundTimeout{ .phase = "idle", .answered = 0 });
+        },
+        .request => {
+            // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+            shared.bus.warn(InboundTimeout{ .phase = "request", .answered = 408 });
+            const timeout = "HTTP/1.1 408 Request Timeout\r\ncontent-length: 0\r\nconnection: close\r\n\r\n";
+            writeRawResponse(shared, io, stream, timeout, 408);
+        },
+    }
+}
+
+/// Fixed response on the raw stream, for failures before the HTTP state
+/// machine can answer (load shed, malformed head, read deadline). A failure
+/// here means the client never learned the status, which is the one drop we
+/// cannot back-propagate, so it is logged rather than ignored.
+fn writeRawResponse(
+    shared: *exec.SharedCtx,
+    io: std.Io,
+    stream: std.Io.net.Stream,
+    response: []const u8,
+    status: u16,
+) void {
     var buf: [256]u8 = undefined;
     var writer = std.Io.net.Stream.Writer.init(stream, io, &buf);
-    writer.interface.writeAll(response) catch return;
-    writer.interface.flush() catch return;
+    writer.interface.writeAll(response) catch |err| return undeliverable(shared, status, err);
+    writer.interface.flush() catch |err| return undeliverable(shared, status, err);
+}
+
+fn undeliverable(shared: *exec.SharedCtx, status: u16, err: anyerror) void {
+    // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+    shared.bus.warn(ResponseUndeliverable{ .answered = status, .err = @errorName(err) });
 }
 
 // ============================== Tests ==============================
diff --git a/src/frontend/stdio/deadline_reader.zig b/src/frontend/stdio/deadline_reader.zig
new file mode 100644
index 00000000..40201c86
--- /dev/null
+++ b/src/frontend/stdio/deadline_reader.zig
@@ -0,0 +1,205 @@
+//! Inbound socket reads with a deadline, in std.Io terms.
+//!
+//! `std.Io.net.Stream.Reader` reads through `netRead`, which never times out.
+//! A client that connects and sends nothing therefore holds its connection
+//! slab slot until the process exits. Measured against 300 idle sockets and
+//! `max_connections` 256: the stdio frontend never answered `/_health` again,
+//! while httpz reclaimed every slot within 15 s through its request timeout.
+//!
+//! Reads here go through the `net_receive` operation under
+//! `Io.operateTimeout`, which on POSIX is a non-blocking `recvmsg` followed by
+//! `poll` with the deadline. That costs no extra task and no unit of
+//! concurrency, and it keeps `std.posix` out of the frontend (only
+//! core/io_select.zig may name a backend).
+//!
+//! Two deadlines, so both stalls are bounded:
+//!
+//!   idle     caps one read while no request is in flight (keep-alive wait).
+//!   request  caps the whole request once its first byte arrives.
+//!
+//! The second one is what httpz lacks. Its `SO_RCVTIMEO` restarts on every
+//! read, so a client that dribbles one byte per 29 s holds a handler thread
+//! for as long as it likes (see the DeadlineReader note in ../httpz/server.zig).
+const std = @import("std");
+const limits_mod = @import("../../core/limits.zig");
+
+/// Which deadline a read hit.
+pub const Phase = enum { idle, request };
+
+pub const DeadlineReader = struct {
+    interface: std.Io.Reader,
+    io: std.Io,
+    stream: std.Io.net.Stream,
+    /// Cap for a single read while no request is in flight.
+    idle: std.Io.Clock.Duration,
+    /// Cap for a whole request, measured from its first byte.
+    request: std.Io.Clock.Duration,
+    /// Armed by the first byte of a request; cleared by `endRequest`.
+    request_deadline: ?std.Io.Clock.Timestamp = null,
+    /// The real error behind `error.ReadFailed`, which `Io.Reader` cannot
+    /// carry. The connection driver reads it to tell a timeout from a peer
+    /// that went away.
+    err: ?anyerror = null,
+    /// Which deadline expired, set with `err = error.Timeout`.
+    expired: ?Phase = null,
+
+    pub fn init(io: std.Io, stream: std.Io.net.Stream, buffer: []u8) DeadlineReader {
+        return .{
+            .interface = .{
+                .vtable = &.{ .stream = streamImpl },
+                .buffer = buffer,
+                .seek = 0,
+                .end = 0,
+            },
+            .io = io,
+            .stream = stream,
+            .idle = .{ .raw = .fromSeconds(limits_mod.KEEPALIVE_TIMEOUT_SECONDS), .clock = .awake },
+            .request = .{ .raw = .fromSeconds(limits_mod.REQUEST_TIMEOUT_SECONDS), .clock = .awake },
+        };
+    }
+
+    /// Call after every response. The next read waits for a new request, so
+    /// the idle cap applies again.
+    pub fn endRequest(self: *DeadlineReader) void {
+        self.request_deadline = null;
+        self.expired = null;
+        self.err = null;
+    }
+
+    /// True when the last failure was this reader's own deadline.
+    pub fn timedOut(self: *const DeadlineReader) bool {
+        return self.expired != null;
+    }
+
+    fn currentTimeout(self: *const DeadlineReader) std.Io.Timeout {
+        if (self.request_deadline) |deadline| return .{ .deadline = deadline };
+        return .{ .duration = self.idle };
+    }
+
+    fn streamImpl(
+        io_r: *std.Io.Reader,
+        w: *std.Io.Writer,
+        limit: std.Io.Limit,
+    ) std.Io.Reader.StreamError!usize {
+        const self: *DeadlineReader = @alignCast(@fieldParentPtr("interface", io_r));
+        const dest = limit.slice(try w.writableSliceGreedy(1));
+        const phase: Phase = if (self.request_deadline == null) .idle else .request;
+
+        var message: std.Io.net.IncomingMessage = .init;
+        const result = self.io.operateTimeout(.{ .net_receive = .{
+            .socket_handle = self.stream.socket.handle,
+            .message_buffer = (&message)[0..1],
+            .data_buffer = dest,
+            .flags = .{},
+        } }, self.currentTimeout()) catch |err| {
+            self.err = err;
+            if (err == error.Timeout) self.expired = phase;
+            return error.ReadFailed;
+        };
+
+        const maybe_err, const count = result.net_receive;
+        if (maybe_err) |err| {
+            self.err = err;
+            return error.ReadFailed;
+        }
+        // A stream socket reports a peer close as a zero-length message.
+        if (count == 0 or message.data.len == 0) return error.EndOfStream;
+
+        // The first byte starts the whole-request clock.
+        if (self.request_deadline == null) {
+            self.request_deadline = .fromNow(self.io, self.request);
+        }
+        w.advance(message.data.len);
+        return message.data.len;
+    }
+};
+
+// ============================== Tests ==============================
+
+const testing = std.testing;
+
+const Pair = struct {
+    server: std.Io.net.Server,
+    client: std.Io.net.Stream,
+    accepted: std.Io.net.Stream,
+};
+
+/// Binds a loopback port, connects to it, and returns both ends. The same TCP
+/// path the frontend serves; `socketpair` cannot make an AF_INET pair.
+fn connectedPair(io: std.Io) !Pair {
+    var port: u16 = 21080;
+    while (port < 21280) : (port += 1) {
+        const address = std.Io.net.IpAddress.parse("127.0.0.1", port) catch continue;
+        var server = address.listen(io, .{ .reuse_address = true }) catch continue;
+        errdefer server.deinit(io);
+        const client = try address.connect(io, .{ .mode = .stream });
+        errdefer client.close(io);
+        const accepted = try server.accept(io);
+        return .{ .server = server, .client = client, .accepted = accepted };
+    }
+    return error.NoFreePort;
+}
+
+test "an idle peer hits the idle deadline instead of blocking forever" {
+    const io = testing.io;
+    var pair = try connectedPair(io);
+    defer pair.server.deinit(io);
+    defer pair.client.close(io);
+    defer pair.accepted.close(io);
+
+    var buffer: [256]u8 = undefined;
+    var reader: DeadlineReader = .init(io, pair.accepted, &buffer);
+    // The production caps are 30 s; this test only needs to prove the read
+    // returns at all.
+    reader.idle = .{ .raw = .fromMilliseconds(50), .clock = .awake };
+
+    // The client never writes.
+    try testing.expectError(error.ReadFailed, reader.interface.takeByte());
+    try testing.expectEqual(@as(?anyerror, error.Timeout), reader.err);
+    try testing.expectEqual(@as(?Phase, .idle), reader.expired);
+    try testing.expect(reader.timedOut());
+}
+
+test "a request that never finishes hits the request deadline" {
+    const io = testing.io;
+    var pair = try connectedPair(io);
+    defer pair.server.deinit(io);
+    defer pair.client.close(io);
+    defer pair.accepted.close(io);
+
+    var write_buffer: [64]u8 = undefined;
+    var client_writer = std.Io.net.Stream.Writer.init(pair.client, io, &write_buffer);
+    try client_writer.interface.writeAll("GET / HTTP/1.1\r\n");
+    try client_writer.interface.flush();
+
+    var buffer: [256]u8 = undefined;
+    var reader: DeadlineReader = .init(io, pair.accepted, &buffer);
+    reader.idle = .{ .raw = .fromMilliseconds(50), .clock = .awake };
+    reader.request = .{ .raw = .fromMilliseconds(50), .clock = .awake };
+
+    // The head arrives, so the request clock arms; the client then stalls.
+    _ = try reader.interface.take(16);
+    try testing.expect(reader.request_deadline != null);
+    try testing.expectError(error.ReadFailed, reader.interface.takeByte());
+    try testing.expectEqual(@as(?Phase, .request), reader.expired);
+
+    // A finished response returns the reader to the idle cap.
+    reader.endRequest();
+    try testing.expect(reader.request_deadline == null);
+}
+
+test "a peer that closes reports end of stream, not a timeout" {
+    const io = testing.io;
+    var pair = try connectedPair(io);
+    defer pair.server.deinit(io);
+    defer pair.accepted.close(io);
+
+    pair.client.close(io);
+
+    var buffer: [256]u8 = undefined;
+    var reader: DeadlineReader = .init(io, pair.accepted, &buffer);
+    reader.idle = .{ .raw = .fromMilliseconds(500), .clock = .awake };
+
+    try testing.expectError(error.EndOfStream, reader.interface.takeByte());
+    try testing.expect(!reader.timedOut());
+}
diff --git a/src/frontend/stdio/server.zig b/src/frontend/stdio/server.zig
index 08ac618d..826b4ae0 100644
--- a/src/frontend/stdio/server.zig
+++ b/src/frontend/stdio/server.zig
@@ -18,6 +18,10 @@ const thread_bufs = @import("../thread_bufs.zig");
 
 const log = std.log.scoped(.http_server);
 
+// Named event payloads: the type name is the telemetry event name.
+/// A connection refused before it carried a request, with the 503 sent.
+const ConnectionShed = struct { reason: []const u8, answered: u16 };
+
 pub const HttpServer = struct {
     listener: std.Io.net.Server,
     ctx: *exec.SharedCtx,
@@ -91,12 +95,16 @@ pub const HttpServer = struct {
                     continue;
                 },
             };
+            if (self.ctx.metrics) |metrics| metrics.recordConnectionAccepted();
             self.lifecycle.spawn(io, conn_mod.serveConnection, .{
                 self.ctx, &self.slab, &self.arenas, stream,
             }) catch |err| switch (err) {
                 error.ConcurrencyUnavailable => {
                     // The Io implementation is at its task limit; the slab
                     // would also have shed. Tell the client to back off.
+                    if (self.ctx.metrics) |metrics| metrics.recordConnectionShed(.concurrency);
+                    // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
+                    self.ctx.bus.warn(ConnectionShed{ .reason = "io_concurrency_unavailable", .answered = 503 });
                     shedConnection(io, stream);
                 },
             };
diff --git a/src/frontend/thread_bufs.zig b/src/frontend/thread_bufs.zig
index dbf9b756..407fa30f 100644
--- a/src/frontend/thread_bufs.zig
+++ b/src/frontend/thread_bufs.zig
@@ -144,7 +144,9 @@ pub fn expireTrackedUpstreams(ctx: *exec.SharedCtx, force: bool) void {
         if (bufs.timed_out.swap(true, .acq_rel)) continue;
         connection.closing = true;
         connection.stream_reader.stream.shutdown(ctx.io, .both) catch |err| {
-            log.debug("failed to interrupt upstream: {s}", .{@errorName(err)});
+            // The handler stays blocked on a socket nothing can interrupt, so
+            // this is the last record of that thread.
+            log.warn("failed to interrupt upstream: {s}", .{@errorName(err)});
         };
     }
 }
diff --git a/src/root.zig b/src/root.zig
index 8f93d6e4..9d8653a0 100644
--- a/src/root.zig
+++ b/src/root.zig
@@ -31,6 +31,7 @@ pub const frontend_upstream = @import("frontend/upstream.zig");
 pub const frontend_select = @import("frontend/select.zig");
 pub const frontend_stdio_server = @import("frontend/stdio/server.zig");
 pub const frontend_stdio_conn = @import("frontend/stdio/conn.zig");
+pub const frontend_stdio_deadline_reader = @import("frontend/stdio/deadline_reader.zig");
 pub const frontend_httpz_server = @import("frontend/httpz/server.zig");
 pub const frontend_exchange = @import("frontend/exchange.zig");
 pub const frontend_paths = @import("frontend/paths.zig");
@@ -99,6 +100,7 @@ test {
     _ = @import("frontend/upstream.zig");
     _ = @import("frontend/exec.zig");
     _ = @import("frontend/stdio/conn.zig");
+    _ = @import("frontend/stdio/deadline_reader.zig");
     // Both frontends compile in every test build regardless of -Dfrontend,
     // so the unselected one can't rot (PLAN-FRONTEND-SWAP.md §6).
     _ = @import("frontend/stdio/server.zig");
diff --git a/src/runtime/app.zig b/src/runtime/app.zig
index 40047214..62579498 100644
--- a/src/runtime/app.zig
+++ b/src/runtime/app.zig
@@ -517,6 +517,7 @@ pub fn run(init: std.process.Init, distribution: mode.Distribution) !void {
         });
     }
 
+    runtime_metrics.setMaxConnections(engine.limits.max_connections);
     // ziglint-ignore: Z010 (named type sets EventBus telemetry name)
     bus.info(DataPlaneBudget{
         .frontend = @tagName(build_options.frontend),
diff --git a/src/runtime/runtime_metrics.zig b/src/runtime/runtime_metrics.zig
index 17f4e662..eb53cf78 100644
--- a/src/runtime/runtime_metrics.zig
+++ b/src/runtime/runtime_metrics.zig
@@ -1,9 +1,17 @@
 const std = @import("std");
 const m = @import("metrics_zig");
 const ext = @import("extensions");
+const build_options = @import("build_options");
 
 const log = std.log.scoped(.runtime_metrics);
 
+/// Connection-level series need a hook at accept and at slot release. The
+/// stdio frontend owns both; httpz owns neither (it exports its own
+/// `httpz_connections` and `httpz_invalid_request` through endpoints.zig
+/// instead). Registering them on an httpz build would publish a flat zero,
+/// which reads as "no connections" rather than "not measured".
+const conn_metrics_enabled = build_options.frontend == .stdio;
+
 pub const DistributionLabel = enum {
     edge,
     datadog,
@@ -66,6 +74,30 @@ pub const ErrorClassLabel = enum {
     module,
 };
 
+/// Why a connection was refused before it ever carried a request.
+pub const ShedReasonLabel = enum {
+    /// No free connection slab slot: the process is at `max_connections`.
+    slab_full,
+    /// The Io implementation refused another concurrent task.
+    concurrency,
+};
+
+const ShedLabels = struct {
+    reason: ShedReasonLabel,
+};
+
+/// Which inbound deadline expired. `idle` is a keep-alive wait with no
+/// request in flight; `request` means a partial request stalled and was
+/// answered with 408.
+pub const InboundPhaseLabel = enum {
+    idle,
+    request,
+};
+
+const InboundTimeoutLabels = struct {
+    phase: InboundPhaseLabel,
+};
+
 pub const PolicyTelemetryLabel = enum {
     datadog_logs,
     datadog_metrics,
@@ -121,6 +153,50 @@ const InternalMetrics = struct {
         .{ .help = "Fresh-connection transport retries." },
         .{},
     ),
+    edge_upstream_timeouts_total: m.Counter(u64) = .init(
+        "edge_upstream_timeouts_total",
+        .{ .help = "Upstream attempts the watchdog cut off at its deadline." },
+        .{},
+    ),
+    /// Connections accepted, and the count currently holding a slab slot.
+    /// stdio only: see `conn_metrics_enabled`.
+    edge_connections_total: m.Counter(u64) = if (conn_metrics_enabled) .init(
+        "edge_connections_total",
+        .{ .help = "Inbound connections accepted." },
+        .{},
+    ) else .{ .noop = {} },
+    edge_connections_active: m.Gauge(i64) = if (conn_metrics_enabled) .init(
+        "edge_connections_active",
+        .{ .help = "Inbound connections currently holding a connection slot." },
+        .{},
+    ) else .{ .noop = {} },
+    /// Connections refused before a request: the exhaustion signal. Read it
+    /// against `edge_connections_max`.
+    edge_connections_shed_total: ConnectionsShedTotal,
+    /// Inbound reads cut off by their deadline. The `request` phase counts
+    /// dropped requests; the `idle` phase counts reclaimed keep-alive slots.
+    edge_inbound_timeouts_total: InboundTimeoutsTotal,
+    /// Heads that failed to parse, answered with 400 and a close. stdio only.
+    edge_requests_invalid_total: m.Counter(u64) = if (conn_metrics_enabled) .init(
+        "edge_requests_invalid_total",
+        .{ .help = "Requests rejected before routing because the head failed to parse." },
+        .{},
+    ) else .{ .noop = {} },
+    /// The configured connection ceiling. Both frontends report it, so an
+    /// alert can compare use against capacity without knowing the frontend.
+    edge_connections_max: m.Gauge(i64) = .init(
+        "edge_connections_max",
+        .{ .help = "Configured maximum concurrent inbound connections." },
+        .{},
+    ),
+    /// Handler threads currently holding a request. Against the configured
+    /// thread pool count this is the saturation signal: at the ceiling, a new
+    /// request (a health probe included) waits for a thread to come free.
+    edge_requests_in_flight: m.Gauge(i64) = .init(
+        "edge_requests_in_flight",
+        .{ .help = "Requests currently held by a handler thread." },
+        .{},
+    ),
     edge_requests_total: RequestsTotal,
     edge_request_duration_seconds: RequestDurationSeconds,
     edge_responses_total: ResponsesTotal,
@@ -146,11 +222,13 @@ const InternalMetrics = struct {
     const RequestDurationSeconds = m.HistogramVec(
         f64,
         DurationLabels,
-        &.{ 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5 },
+        &.{ 0.0001, 0.00025, 0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60 },
     );
     const ResponsesTotal = m.CounterVec(u64, ResponseLabels);
     const PrefilterDecisionsTotal = m.CounterVec(u64, PrefilterLabels);
     const RequestErrorsTotal = m.CounterVec(u64, ErrorLabels);
+    const ConnectionsShedTotal = m.CounterVec(u64, ShedLabels);
+    const InboundTimeoutsTotal = m.CounterVec(u64, InboundTimeoutLabels);
     const PolicyRecordsEvaluatedTotal = m.CounterVec(u64, PolicyLabels);
     const PolicyRecordsKeptTotal = m.CounterVec(u64, PolicyLabels);
     const PolicyRecordsDroppedTotal = m.CounterVec(u64, PolicyLabels);
@@ -208,6 +286,20 @@ pub const RuntimeMetrics = struct {
                     .{ .help = "Total number of request-level errors." },
                     .{},
                 ),
+                .edge_connections_shed_total = try InternalMetrics.ConnectionsShedTotal.init(
+                    allocator,
+                    io,
+                    "edge_connections_shed_total",
+                    .{ .help = "Connections refused before carrying a request." },
+                    .{},
+                ),
+                .edge_inbound_timeouts_total = try InternalMetrics.InboundTimeoutsTotal.init(
+                    allocator,
+                    io,
+                    "edge_inbound_timeouts_total",
+                    .{ .help = "Inbound reads cut off by the idle or request deadline." },
+                    .{},
+                ),
                 .edge_policy_records_evaluated_total = try InternalMetrics.PolicyRecordsEvaluatedTotal.init(
                     allocator,
                     io,
@@ -383,6 +475,44 @@ pub const RuntimeMetrics = struct {
         if (retry) self.internal.edge_upstream_retries_total.incr();
     }
 
+    pub fn recordUpstreamTimeout(self: *RuntimeMetrics) void {
+        self.internal.edge_upstream_timeouts_total.incr();
+    }
+
+    /// `delta` is +1 when a handler takes a request and -1 when it returns it.
+    pub fn recordInFlight(self: *RuntimeMetrics, delta: i64) void {
+        self.internal.edge_requests_in_flight.incrBy(delta);
+    }
+
+    pub fn recordConnectionAccepted(self: *RuntimeMetrics) void {
+        self.internal.edge_connections_total.incr();
+    }
+
+    /// `delta` is +1 when a connection claims a slot and -1 when it frees it.
+    pub fn recordConnectionsActive(self: *RuntimeMetrics, delta: i64) void {
+        self.internal.edge_connections_active.incrBy(delta);
+    }
+
+    pub fn recordConnectionShed(self: *RuntimeMetrics, reason: ShedReasonLabel) void {
+        self.internal.edge_connections_shed_total.incr(.{
+            .reason = reason,
+        }) catch |err| log.debug("failed to record shed metric: {}", .{err});
+    }
+
+    pub fn recordInboundTimeout(self: *RuntimeMetrics, phase: InboundPhaseLabel) void {
+        self.internal.edge_inbound_timeouts_total.incr(.{
+            .phase = phase,
+        }) catch |err| log.debug("failed to record inbound timeout metric: {}", .{err});
+    }
+
+    pub fn recordInvalidRequest(self: *RuntimeMetrics) void {
+        self.internal.edge_requests_invalid_total.incr();
+    }
+
+    pub fn setMaxConnections(self: *RuntimeMetrics, max_connections: usize) void {
+        self.internal.edge_connections_max.set(@intCast(max_connections));
+    }
+
     pub fn recordResponse(
         self: *RuntimeMetrics,
         known_path: KnownPathLabel,
@@ -455,3 +585,46 @@ pub fn statusClass(status: u16) StatusClassLabel {
     if (status >= 500 and status < 600) return .s5xx;
     return .other;
 }
+
+// ============================== Tests ==============================
+
+const testing = std.testing;
+
+test "connection and saturation series reach the scrape" {
+    var metrics: RuntimeMetrics = try .init(testing.allocator, testing.io, .datadog);
+    defer metrics.deinit();
+
+    metrics.setMaxConnections(256);
+    metrics.recordConnectionAccepted();
+    metrics.recordConnectionsActive(1);
+    metrics.recordConnectionsActive(1);
+    metrics.recordConnectionsActive(-1);
+    metrics.recordConnectionShed(.slab_full);
+    metrics.recordConnectionShed(.concurrency);
+    metrics.recordInvalidRequest();
+    metrics.recordInFlight(3);
+    metrics.recordInFlight(-1);
+    metrics.recordUpstreamTimeout();
+    metrics.recordInboundTimeout(.idle);
+    metrics.recordInboundTimeout(.request);
+
+    var out: std.Io.Writer.Allocating = .init(testing.allocator);
+    defer out.deinit();
+    try metrics.writePrometheus(&out.writer);
+    const text = out.written();
+
+    // Frontend-neutral series: present on every build.
+    try testing.expect(std.mem.indexOf(u8, text, "edge_connections_max 256") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_requests_in_flight 2") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_upstream_timeouts_total 1") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_connections_shed_total{reason=\"slab_full\"} 1") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_connections_shed_total{reason=\"concurrency\"} 1") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_inbound_timeouts_total{phase=\"idle\"} 1") != null);
+    try testing.expect(std.mem.indexOf(u8, text, "edge_inbound_timeouts_total{phase=\"request\"} 1") != null);
+
+    // Accept-time series exist only where a frontend can feed them.
+    const has_conn_series = std.mem.indexOf(u8, text, "edge_connections_active 1") != null;
+    try testing.expectEqual(conn_metrics_enabled, has_conn_series);
+    const has_invalid = std.mem.indexOf(u8, text, "edge_requests_invalid_total 1") != null;
+    try testing.expectEqual(conn_metrics_enabled, has_invalid);
+}