# Deploy slice C — in-process TLS (rustls Phase 1) + hand-rolled `/metrics` + ValkeyEventSink — Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use `superpowers:subagent-driven-development` (recommended) or `superpowers:executing-plans` to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Land three order-independent code riders from the
[deployment-topology spec](../specs/2026-07-05-deployment-topology-design.md) —
§5.4 (rustls Phase 1 BYO-cert in-process TLS, hot-reload without dropping live WS), §5.5
(`/metrics` hand-rolled Prometheus text endpoint, zero new deps, node-level cardinality only),
§5.6 (`ValkeyEventSink` — first Valkey consumer, capped-stream XADD, fire-and-forget) — so the
FOB can terminate TLS itself for the VPS-forwarder / T2-without-Caddy operator, expose its
existing atomics to a scraper, and preserve call-lifecycle evidence off the OOM-killed node.

**Architecture:** All three riders ride existing seams. (C) adds `axum-server` 0.8 +
`rustls` 0.23 (already in `Cargo.lock` transitively at 0.23.41 via `str0m → aws-lc-rs`) as the
in-process TLS listener, gated by `RUTSTER_TLS_CERT` / `RUTSTER_TLS_KEY`, with a custom
`Acceptor` that sets `TCP_NODELAY=true` on every accepted socket *before* the TLS handshake —
matching plan A's plaintext `serve_with_nodelay` parity (axum-server 0.8's `Server` builder
exposes no `tcp_nodelay` method; verified in upstream `server.rs:32-360`).
`RustlsConfig::reload_from_pem_file` swaps the resolver via `arc-swap`, so new handshakes see
the new cert and live WS connections keep their session state machine — verified against the
published `axum_server::tls_rustls` docs and the `examples/rustls_reload.rs` upstream sample.
(D) is a new `axum::Router` on its own `RUTSTER_METRICS_BIND` listener that renders the
existing `MediaStats` struct (extended with `admission_rejects: u64` + `event_sink_drops:
Option<u64>`) as Prometheus text format — same `MediaCmd::Stats` round-trip `/readyz` already
uses, no shared state of its own. (E) implements the existing `EventSink` trait (signature
preserved verbatim) with a bounded `mpsc::channel` + background `tokio::spawn` task that
`XADD MAXLEN ~ 10000` JSON-serialized `CallEvent`s to a Valkey stream; `RUTSTER_VALKEY_URL`
unset ⇒ today's `TracingEventSink` unchanged.

**Tech Stack:** Rust stable + 1.85 (CI matrix). Two new direct workspace deps:
`axum-server = "0.8"` (feature `tls-rustls`) — MIT, MSRV Rust 1.75+; axum-server 0.8 uses
`rustls 0.23` + `tokio-rustls 0.26` (matching Cargo.lock's transitive pins 0.23.41 / 0.26.4);
`redis = "1.3.0"` (BSD-3-Clause — confirmed against `deny.toml`'s allow list at line 88).
`rustls` and `tokio-rustls` are NOT new workspace deps themselves — they stay transitive
(consumed via `axum_server::tls_rustls`) except for `crates/rutster`'s `[dev-dependencies]`
where the integration test's TLS client needs them directly. `rcgen = "0.14"` joins
`[dev-dependencies]` for self-signed test cert generation (pinned under 0.14.x to match the
existing Cargo.lock transitive pin — see Cargo.toml's root rcgen NOTE).

## Global Constraints

House rules (every task's requirements implicitly include this section):

- **License:** GPL-3.0-or-later on every crate manifest (ADR-0004).
- **DCO:** every commit signed off — `git commit -s` (AGENTS.md Git workflow).
- **SEAM GATE (UNCHANGED):** `crates/rutster-media/src/loop_driver.rs` and
  `crates/rutster-media/src/rtc_session.rs` stay **byte-identical** — CI pins their blob hashes
  (`744bf314edf7f4925c8bb3bd0f5176dbc88f8113` / `f47d63b9a2883d37066a93c9daa0e2cf8816bec4` in
  `.github/workflows/ci.yml`). **NO task in this plan touches them.** This slice is the
  listener/config/metrics/event-sink surface, not the RTP path.
- **Hot-path policy:** never `?`-propagate on the 20 ms tick; match-and-continue; "drop +
  observe (log + counter)." No `unwrap()`/`expect()` outside tests/startup (`main.rs`
  fail-fast `expect`s at boot are the existing house pattern — plan A carries the same rule).
  The `ValkeyEventSink::emit` path is a thin `mpsc::try_send` — it MUST NOT block, even when the
  background consumer task lags.
- **Code style:** `cargo fmt --all --check` is the whitespace gate;
  `cargo clippy --all --all-targets -- -D warnings` is the lint bar. Add `--features=sim-bench`
  to both for the sim-bench job.
- **MSRV:** CI matrix runs stable + 1.85 (`rust-toolchain.toml`). No syntax/std APIs newer
  than 1.85. The `rcgen 0.14.7` pin in `Cargo.lock` already protects this — do not `cargo update`
  wholesale, including when adding `axum-server` and `redis`. Re-pin with
  `cargo update -p rcgen --precise 0.14.7` if a transitive bump occurs.
- **Workspace dep pinning:** new deps go in the ROOT `Cargo.toml` `[workspace.dependencies]`;
  members reference with `dep.workspace = true`. (This slice adds two: `axum-server`, `redis`.)
- **cargo-deny license gate** stays green. `deny.toml` already allows `MIT`, `BSD-3-Clause`,
  `Apache-2.0`, `ISC`, `Zlib` — `axum-server` is MIT, `redis 1.3.0` is BSD-3-Clause (verified
  via `crates.io/api/v1/crates/redis` → `versions[0].license = "BSD-3-Clause"`). The full
  transitive trees of both crates must clear `cargo deny check` — Task 1 and Task 7 spot-check
  this before any code task commits.
- **sim-bench CI:** the `sim-bench` job runs `cargo test --all --features=sim-bench -- --test-threads=1`
  — `--test-threads=1` is load-bearing (shared tick-lag gauge). This slice adds NO new sim-bench
  assertion; the sim-bench pulse stays as-is.
- **Branch/PR:** branch `deploy-c/binary-features`; PR to `main` via `tea`. This slice is
  **order-independent with slice A** — it depends on slice A only for the `config.rs` env-parser
  *convention* (no hard code dependency; the parsers below stand alone). It is also order-
  independent with slice D's own sub-tasks EXCEPT the EventSink drop-counter (Task 6 below),
  which must land either AFTER Task 8 (Phase E's drop counter exists) OR behind the
  `Option<u64>` fallback documented in Task 5 — see "Sequencing" below.

Slice-C/D/E-specific constraints:

- **Always compiled, runtime-gated TLS:** the `axum-server` + `rustls` deps compile
  unconditionally (the FOB binary always carries the TLS code path). The TLS *listener*
  activates only when BOTH `RUTSTER_TLS_CERT` and `RUTSTER_TLS_KEY` are set; plaintext `:8080`
  via `serve_with_nodelay` (plan A) stays the artifact default. No `--no-default-features`
  gymnastics — the binary is one artifact that runs in every shape.
- **Hot-reload parity contract:** `RustlsConfig::reload_from_pem_file` (verified in
  `axum_server::tls_rustls`) atomically swaps the resolver. New TLS handshakes pick up the new
  cert, live WS connections keep their state machine — the test (Task 4 Step 1) asserts this by
  opening a long-lived TLS connection, reloading the cert, then opening a second TLS connection
  and verifying the second one's `peer_certificates()` differ from the first. **Never** use a
  per-connection cert-reload path (would force handshake serialization).
- **`/metrics` is its own listener, never a route on the main app:** spec §5.5 says "never routed
  through Caddy." `RUTSTER_METRICS_BIND` (default `127.0.0.1:9090`) binds a *separate*
  `axum::Router` on a *separate* `TcpListener`. The main `axum::serve` listener does NOT gain
  a `/metrics` route — that would put the metrics surface behind the public edge, breaking the
  spec invariant.
- **`ValkeyEventSink` is fire-and-forget, period.** `emit()` does one `mpsc::try_send` (bounded
  queue, capacity 256). If the queue is full or the background task has died, the event is
  counted-and-dropped via an `Arc<AtomicU64>` (the same Arc `/metrics` reads in Task 6). The
  call path NEVER awaits Valkey, NEVER returns an error, NEVER spawns a task per event. This is
  the `EventSink` trait's documented contract (`emit()` is called from the media `std::thread`
  — implementations MUST NOT block or perform I/O inline) and the `ValkeyEventSink` honors it
  strictly.
- **`XADD MAXLEN ~ 10000` is the adopted default.** Per the design call: ~10k events capped
  stream ≈ a few MB of memory under normal operation (a `CallEvent` serializes to ~200 bytes of
  JSON; 10k × 200B ≈ 2MB), days-of-calls forensic window. **This is reversible — it is a `const`
  in the `ValkeyEventSink` module, NOT a Valkey-side migration.** Bumping or dropping it is a
  one-line code change + redeploy, no operator action on Valkey. The cap also bounds recovery
  time if Valkey is restarted with a backlog (the background task drains 10k events then
  catches up).
- **No new `EventSink` trait method that returns `Result`.** The trait is
  `pub trait EventSink: Send + Sync { fn emit(&self, event: CallEvent); }` — verified in
  `crates/rutster/src/event_sink.rs:57-59`. Phase E adds a *default* method
  `fn drops(&self) -> Option<u64> { None }` so existing `TracingEventSink` returns `None`
  without a code change to it; `ValkeyEventSink` overrides it to
  `Some(self.drops.load(Relaxed))`. Backward-compatible — callers (the `/metrics` route +
  `MediaStats::event_sink_drops`) treat `None` as "no drop counter exists."
- **Cross-slice dependency: plan A's `config.rs` convention (no hard code dep).** All new env
  parsers in this slice (`tls_cert_path`, `tls_key_path`, `metrics_bind`, `valkey_url`) follow
  the pure `Option<String> -> Result<T, String>` fail-fast pattern slice A formalized
  (`crates/rutster/src/config.rs:21-28` is the canonical `http_bind` shape). If slice A is not
  yet merged, this slice's parsers still stand alone — they consume no plan-A types.

## Sequencing — three order-independent phases, ONE carve-out

| Phase | Tasks | Depends on |
|---|---|---|
| **C — rustls Phase 1 TLS** | Task 1 (workspace deps), Task 2 (env parsers), Task 3 (helper+acceptor), Task 4 (main.rs wiring + reload test) | Nothing (slice A's `config.rs` convention only, no hard dep). |
| **D — `/metrics`** | Task 5 (extend `MediaStats`), Task 6 (`/metrics` route + tests). **The `event_sink_drops` field landing in Task 5 is the carve-out.** | Task 5 must land before Task 6 (Task 6 renders Task 5's fields). Neither depends on Phase C or Phase E. |
| **E — `ValkeyEventSink`** | Task 7 (workspace dep), Task 8 (impl + CI services block), Task 9 (extend smoke harness hook) | Nothing (slice A's `config.rs` convention only). The `Arc<AtomicU64>` drop counter is consumed by Task 6 *only if Task 6 lands after Task 8*. |
| Final verification | Task 10 | All above. |

**The carve-out:** `MediaStats.event_sink_drops: Option<u64>` lands in Task 5 as `None` always
(populated by `sink.drops()` in the `/metrics` route). Task 6's `/metrics` route renders the
field. If Task 6 lands BEFORE Task 8 (Phase E), the metric line is absent (rendering is
`if let Some(d) = stats.event_sink_drops { ... }`); if Task 8 lands first, Task 6 — when it
lands — finds the field already populated. Either ordering is correct; the only forbidden
state is Task 6 rendering a `0` for an unset counter (operators would misread "zero drops" as
"healthy" when really no sink exists). The `Option<u64>` design prevents that.

## File Structure

### New files

| Path | Responsibility |
|---|---|
| `crates/rutster/src/serve_tls.rs` | `serve_tls_with_nodelay` — TLS production serve path: `axum_server::bind` + `NodeLayAcceptor` + `Handle::graceful_shutdown`. Parity with `serve.rs` (plan A). |
| `crates/rutster/src/tls_acceptor.rs` | `NodeLayAcceptor` — wraps `RustlsAcceptor`, calls `set_nodelay(true)` on the `TcpStream` before delegating. Self-contained for testability. |
| `crates/rutster/tests/serve_tls_nodelay.rs` | Integration test: TLS listener stays up + acceptor wiring is correct. |
| `crates/rutster/tests/serve_tls_hot_reload.rs` | Hot-reload contract test: a TLS connection survives a `RustlsConfig::reload_from_pem_file`; the next handshake sees the new cert. |
| `crates/rutster/src/metrics.rs` | `metrics_router(state)` + `render_prometheus_text(stats)` — `/metrics` route on its own `axum::Router`. Pure function over `MediaStats`. |
| `crates/rutster/tests/metrics_endpoint.rs` | Integration test: `/metrics` returns 200 + `text/plain; version=0.0.4`, contains every named metric line, types declared before values, event_sink_drops absent for TracingEventSink. |
| `crates/rutster/src/valkey_sink.rs` | `ValkeyEventSink` struct + `spawn(url, tokio_handle) -> SharedEventSink` (spawns background XADD task), `drops()` override feeding `Arc<AtomicU64>`. |
| `crates/rutster/tests/valkey_sink_lifecycle.rs` | Integration test against a real Valkey service container (CI's `services: valkey:` block): lifecycle event lands, drop counter increments when the consumer is dead. |

### Modified files

| Path | What changes |
|---|---|
| `Cargo.toml` (root) | Add `axum-server = { version = "0.8", features = ["tls-rustls"] }`, `redis = { version = "1.3.0", default-features = false, features = ["aio", "streams", "tokio-comp"] }`, and add `rustls` + `tokio-rustls` workspace entries (Task 3 Step 1) to `[workspace.dependencies]`. |
| `crates/rutster/Cargo.toml` | Add `axum-server.workspace = true`, `redis.workspace = true`, plus dev-deps `rcgen = "0.14"`, `rustls` + `tokio-rustls` workspace refs (for the test client). |
| `crates/rutster/src/lib.rs` | Add `pub mod metrics; pub mod serve_tls; pub mod tls_acceptor; pub mod valkey_sink;` (module list, currently lines 25–31). |
| `crates/rutster/src/main.rs` | Parse `RUTSTER_TLS_CERT` / `RUTSTER_TLS_KEY` / `RUTSTER_METRICS_BIND` / `RUTSTER_VALKEY_URL`; spawn the `/metrics` listener; if both TLS envs set, serve via `serve_tls_with_nodelay` + spawn a SIGHUP hot-reload watcher; replace `TracingEventSink` default with `ValkeyEventSink` when `RUTSTER_VALKEY_URL` is set. |
| `crates/rutster/src/config.rs` | Add four parsers: `tls_cert_path`, `tls_key_path`, `metrics_bind`, `valkey_url` (house `Option<String> -> Result<T, String>` fail-fast pattern). |
| `crates/rutster/src/event_sink.rs` | Add default method `fn drops(&self) -> Option<u64> { None }` to `EventSink` trait. `TracingEventSink` keeps the default (returns `None`). `ValkeyEventSink` overrides (in `valkey_sink.rs`). |
| `crates/rutster/src/media_thread.rs` | (1) Extend `MediaStats` with `admission_rejects: u64` + `event_sink_drops: Option<u64>` (`#[serde(skip_serializing_if = "Option::is_none")]`). (2) Track `admission_rejects` as a local counter incremented in the `draining`/capacity-shed arms of the `Register` handler. (3) Populate `event_sink_drops` from `sink.drops()` in the `Stats` command handler. **No change to the 20ms tick loop's hot-path code or the seam-gated files.** |
| `.github/workflows/ci.yml` | Add a `services: valkey:` block to the `test` job (and the `clippy` job for type-checking the test code) running `valkey/valkey:latest` on `127.0.0.1:6379`. Net new — no existing `services:` block in the file. |

### SEAM-INVARIANT files (DO NOT TOUCH)

- `crates/rutster-media/src/loop_driver.rs` — byte-identical, all tasks.
- `crates/rutster-media/src/rtc_session.rs` — byte-identical, all tasks.

## Interfaces

### Consumes

- `axum::serve` (0.7.9, in-tree), `axum::Router`, `tokio::net::TcpListener` — plan A's
  `serve_with_nodelay` is the plaintext fallback when TLS is unset.
- `axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor}` (new direct workspace dep, 0.8 —
  verified `RustlsConfig::from_pem_file` and `reload_from_pem_file` are public `async fn`s
  returning `io::Result<...>` per `axum-server 0.8/src/axum_server/tls_rustls/mod.rs`).
- `axum_server::Handle::new()` + `Handle::graceful_shutdown(Some(duration))` — graceful
  shutdown parity with plan A's `with_graceful_shutdown`.
- `axum_server::accept::Accept` — the trait `NodeLayAcceptor` implements.
- `redis::aio::ConnectionManager` (feature `aio` + `tokio-comp`), `redis::AsyncCommands`
  (feature `streams` provides `xadd_maxlen`).
- Existing `crate::config::http_bind` parser (plan A) — the `RUTSTER_METRICS_BIND` parser
  mirrors its shape.
- Existing `crate::event_sink::{EventSink, CallEvent, SharedEventSink, TracingEventSink}`
  (signature verified verbatim — see `crates/rutster/src/event_sink.rs:57-59`).
- Existing `crate::media_thread::{MediaStats, MediaCmd::Stats, MediaThread}` — for the
  `/metrics` route's stats round-trip (same shape `/readyz` already uses in `routes.rs:173-189`).
- Existing `crate::routes::router(AppState)` (plan A) — for the `/metrics` *separate* listener
  binding the same `axum::serve`. (Note: `/metrics` does NOT mount onto `router(AppState)`.
  See Global Constraints.)

### Produces

- `pub async fn serve_tls_with_nodelay<F>(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future<Output = ()> + Send + 'static` — the TLS production serve path.
- `pub struct NodeLayAcceptor(RustlsAcceptor);` with `impl<S> Accept<TcpStream, S> for NodeLayAcceptor`.
- `pub fn config::tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>`.
- `pub fn config::tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>`.
- `pub fn config::metrics_bind(raw: Option<String>) -> Result<SocketAddr, String>` (default `127.0.0.1:9090`).
- `pub fn config::valkey_url(raw: Option<String>) -> Result<Option<url::Url>, String>`.
- `pub fn metrics::metrics_router(state: AppState) -> axum::Router` — `/metrics` route mounted on its own Router. **Does NOT mount onto `routes::router` — see Global Constraints.**
- `pub fn metrics::render_prometheus_text(stats: &MediaStats) -> String` — pure function for the response body.
- `pub struct ValkeyEventSink { ... }` + `impl ValkeyEventSink { pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink }`.
- `impl EventSink for ValkeyEventSink` (override `emit` + `drops`).
- `MediaStats` gains two fields: `pub admission_rejects: u64`, `pub event_sink_drops: Option<u64>` (backward compatible — `/readyz` JSON gains two keys, no consumer breaks).
- `EventSink` trait gains default method `fn drops(&self) -> Option<u64> { None }`.

## Task ordering

- **Task 1** — Add `axum-server` + `tls-rustls` workspace dep; `cargo deny check` clears BEFORE
  any code work. No code dependency on later tasks.
- **Task 2** — TLS env parsers (`tls_cert_path`, `tls_key_path`). Independent of Task 1
  (compile-wise — the tests don't import `axum-server`).
- **Task 3** — `NodeLayAcceptor` + `serve_tls_with_nodelay` helper + integration test. Depends on
  Task 1 (workspace dep).
- **Task 4** — `main.rs` wiring + hot-reload watcher task + reload test. Depends on Tasks 2 + 3.
- **Task 5** — Extend `MediaStats` with `admission_rejects` + `event_sink_drops: Option<u64>`.
  Independent — touches only `media_thread.rs` + `event_sink.rs`. Land before Task 6.
- **Task 6** — `/metrics` route + env parser + tests. Depends on Task 5 (renders its fields);
  order-independent with Phases C and E.
- **Task 7** — Add `redis` workspace dep; `cargo deny check` clears. Independent.
- **Task 8** — `ValkeyEventSink` impl + tests + CI `services: valkey:` block. Depends on Task 7.
- **Task 9** — Extend plan B's smoke harness with the `assert-lifecycle-event-lands-in-Valkey-stream`
  hook (referenced by name; plan B does not exist yet at the time this plan was written —
  **dependency note** in the task body).
- **Task 10** — Final verification sweep (no code).

---

### Task 1: Add `axum-server` workspace dep + verify cargo-deny clears

**Files:**
- Modify: `Cargo.toml` (root), `Cargo.lock` (regenerated by cargo)

**Interfaces:**
- Consumes: `axum-server 0.8` from crates.io. Feature `tls-rustls` pulls `rustls 0.23` +
  `tokio-rustls 0.26` (both already in `Cargo.lock` transitively at 0.23.41 / 0.26.4 — verified
  by grep against `Cargo.lock`).
- Produces: workspace dep entry consumed by `crates/rutster/Cargo.toml` in Task 3.

- [ ] **Step 1: Append the workspace dep**

Append to `[workspace.dependencies]` in `/home/alee/Sources/rutster/Cargo.toml` (after the
`toml` line, before the rcgen NOTE comment — same insertion point as plan A Task 5 Step 1):

```toml
# axum-server 0.8: in-process TLS listener for rustls Phase 1 (deploy
# spec §5.4). Consumes rustls 0.23 + tokio-rustls 0.26 (already in Cargo.lock
# transitively via str0m → aws-lc-rs at 0.23.41 / 0.26.4 — verified). The
# tls-rustls feature enables axum_server::tls_rustls::{RustlsConfig,
# RustlsAcceptor}. MIT license — cargo-deny-clean (deny.toml allow list).
axum-server = { version = "0.8", features = ["tls-rustls"] }
```

- [ ] **Step 2: Refresh Cargo.lock + run cargo-deny**

```bash
cd /home/alee/Sources/rutster
cargo fetch
cargo deny check
```

Expected: `cargo deny check` exits 0. If `cargo deny check licenses` rejects any new transitive
crate introduced by `axum-server` (e.g. `fs-err`, `either`, `arc-swap`), STOP — investigate
the license, add it to `deny.toml`'s `allow` list with a documenting comment matching the
existing style, re-run. If `cargo deny check bans` flags a `multiple-versions = "deny"`
violation, add each offending crate to the `skip = [...]` list with the same inline-rationale
style as the existing entries (e.g. `thiserror`, `getrandom`). **Do not bulk-modify `skip`** —
name every crate with a one-paragraph justification comment (per the `deny.toml` schema).

- [ ] **Step 3: Spot-check the resolved versions**

```bash
cd /home/alee/Sources/rutster
cargo tree -p axum-server -e normal --depth 2 | head -n 40
```

Expected: `axum-server v0.8.x`, `rustls v0.23.41` (or newer in the 0.23 series — but NOT 0.24),
`tokio-rustls v0.26.4`. If `rustls` or `tokio-rustls` resolved to a *different* version than
the existing transitive pins, STOP — that means a `multiple-versions` violation is about to
fire in CI. Resolve by either picking an `axum-server` version whose `rustls`/`tokio-rustls`
ranges match `Cargo.lock`'s existing pins, or document the `skip` entry per Step 2.

- [ ] **Step 4: Commit**

```bash
cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock
git commit -s -m "chore(deps): add axum-server 0.8 (tls-rustls) workspace dep (deploy-C §5.4)

In-process TLS listener for rustls Phase 1. Consumes rustls 0.23.41 +
tokio-rustls 0.26.4 already pinned transitively via str0m → aws-lc-rs;
adds only the direct axum-server edge (MIT, cargo-deny clean). Feature
tls-rustls enables axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor}
(verified reload_from_pem_file is public; atomic swap via arc-swap).

Code lands in subsequent tasks. cargo deny check stays green."
```

---

### Task 2: TLS env parsers — `RUTSTER_TLS_CERT` / `RUTSTER_TLS_KEY`

**Files:**
- Modify: `crates/rutster/src/config.rs` (add two parsers + tests)
- No new files; no new deps.

**Interfaces:**
- Produces:
  - `pub fn tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>`
  - `pub fn tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String>`

Both follow the existing `Option<String> -> Result<T, String>` contract: `None` → `Ok(None)`;
`Some(s)` → parse + existence check; errors include the env var name so `main.rs`'s
`.expect()` log is actionable.

- [ ] **Step 1: Write the failing parser tests**

Append to `mod tests` in `crates/rutster/src/config.rs`:

```rust
    #[test]
    fn tls_cert_path_none_when_unset() {
        assert!(tls_cert_path(None).unwrap().is_none());
    }

    #[test]
    fn tls_cert_path_some_when_file_exists() {
        let tmp = std::env::temp_dir().join("rutster-config-test-cert.pem");
        std::fs::write(&tmp, b"dummy").unwrap();
        let p = tls_cert_path(Some(tmp.to_string_lossy().into()))
            .unwrap()
            .expect("Some when set");
        assert_eq!(p, tmp);
        let _ = std::fs::remove_file(&tmp);
    }

    #[test]
    fn tls_cert_path_rejects_missing_file_with_var_name_in_error() {
        let err = tls_cert_path(Some("/no/such/rutster-cert.pem".into())).unwrap_err();
        assert!(err.contains("RUTSTER_TLS_CERT"));
    }

    #[test]
    fn tls_key_path_none_when_unset() {
        assert!(tls_key_path(None).unwrap().is_none());
    }

    #[test]
    fn tls_key_path_rejects_missing_file_with_var_name_in_error() {
        let err = tls_key_path(Some("/no/such/rutster-key.pem".into())).unwrap_err();
        assert!(err.contains("RUTSTER_TLS_KEY"));
    }
```

- [ ] **Step 2: Run — expect compile failure**

```bash
cargo test -p rutster config::tests
```

Expected: compile error `cannot find function `tls_cert_path`` (and `tls_key_path`).

- [ ] **Step 3: Implement the parsers**

Add to `crates/rutster/src/config.rs` after `twilio_credentials` (lines 119–164), before the
`#[cfg(test)] mod tests` block:

```rust
/// `RUTSTER_TLS_CERT` — path to a PEM-encoded TLS certificate file
/// (deploy spec §5.4 / §5.7). `None` when unset: `main.rs` reads both
/// `tls_cert_path` and `tls_key_path`; BOTH must be `Some` to activate
/// the TLS listener. One set + the other unset is a fail-fast
/// misconfiguration (the operator almost certainly typo'd one of the
/// pair) — `main.rs` reports the partial config with the same contract
/// as `twilio_credentials`' partial-config error.
///
/// The path existence check lives here (pure function over
/// `Option<String>`) so `main.rs`'s `expect` log already names the
/// missing file. The cert content is parsed by `RustlsConfig::from_pem_file`
/// at listener construction — bad PEM surfaces there, not here.
pub fn tls_cert_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let p = std::path::PathBuf::from(s);
            if !p.is_file() {
                return Err(format!(
                    "RUTSTER_TLS_CERT {:?}: file does not exist (or is not a regular file)",
                    p.display()
                ));
            }
            Ok(Some(p))
        }
    }
}

/// `RUTSTER_TLS_KEY` — path to a PEM-encoded private key file matching
/// `RUTSTER_TLS_CERT` (deploy spec §5.4 / §5.7). Same contract as
/// `tls_cert_path`: `None` → plaintext listener; `Some(path)` → fail-fast
/// if missing. The cert/key MATCH is validated by `RustlsConfig::from_pem_file`
/// at listener construction.
pub fn tls_key_path(raw: Option<String>) -> Result<Option<std::path::PathBuf>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let p = std::path::PathBuf::from(s);
            if !p.is_file() {
                return Err(format!(
                    "RUTSTER_TLS_KEY {:?}: file does not exist (or is not a regular file)",
                    p.display()
                ));
            }
            Ok(Some(p))
        }
    }
}
```

- [ ] **Step 4: Run tests — expect pass**

```bash
cargo test -p rutster config::tests
```

Expected: all green (the five new tests + every existing `config::tests` test).

- [ ] **Step 5: Commit**

```bash
cd /home/alee/Sources/rutster
git add crates/rutster/src/config.rs
git commit -s -m "feat(config): tls_cert_path / tls_key_path env parsers (deploy-C §5.4)

RUTSTER_TLS_CERT / RUTSTER_TLS_KEY enable the in-process TLS listener
(rustls Phase 1). Pure Option<String> -> Result<Option<PathBuf>, String>
matching the config.rs house pattern; fail-fast on missing file with
the env var name in the error so main.rs's expect log is actionable.
BOTH must be Some to activate TLS; one-set-one-unset is caught in
main.rs (matching twilio_credentials' partial-config posture)."
```

---

### Task 3: `NodeLayAcceptor` + `serve_tls_with_nodelay` helper + integration test

This is the TLS parity task for plan A's plaintext `serve_with_nodelay` (plan A Task 1).
`axum_server::Server` (verified in `axum-server 0.8/src/axum_server/server.rs:32-360`) exposes
NO `tcp_nodelay` method — neither on `bind()` nor on `bind_rustls()`. The only path to
TCP_NODELAY parity on the TLS accept flow is a custom `Acceptor` that sets the option on the
raw `TcpStream` *before* delegating to `RustlsAcceptor`. axum-server's own
`examples/rustls_session.rs` documents this exact pattern (a `CustomAcceptor` wrapping
`RustlsAcceptor::new(config)`), so the design is the canonical one.

**Files:**
- Modify: `crates/rutster/Cargo.toml` (add `axum-server.workspace = true` + dev-deps `rcgen`,
  `rustls`, `tokio-rustls`)
- Modify: `Cargo.toml` (root) — add `rustls` + `tokio-rustls` workspace entries (Task 1 added
  them transitively in Cargo.lock; this surfaces them as workspace deps for the test client).
- Create: `crates/rutster/src/tls_acceptor.rs`
- Create: `crates/rutster/src/serve_tls.rs`
- Create: `crates/rutster/tests/serve_tls_nodelay.rs`
- Modify: `crates/rutster/src/lib.rs` (module list)

**Interfaces:**
- Consumes: `axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor}`,
  `axum_server::accept::Accept`, `axum_server::Handle`, `tokio::net::TcpStream`.
- Produces:
  - `pub struct NodeLayAcceptor(RustlsAcceptor);`
  - `pub async fn serve_tls_with_nodelay<F>(addr: SocketAddr, rustls_config: RustlsConfig, app: Router, shutdown: F) -> std::io::Result<()> where F: Future<Output = ()> + Send + 'static`

- [ ] **Step 1: Add the workspace dep entries**

Append to `[workspace.dependencies]` in `/home/alee/Sources/rutster/Cargo.toml` (after the
`axum-server` line added in Task 1):

```toml
# rustls 0.23: TLS implementation consumed transitively by axum-server's
# tls-rustls feature. Already in Cargo.lock at 0.23.41 via str0m → aws-lc-rs;
# this surfaces the direct edge for the integration test's ClientConfig.
# Apache-2.0 / MIT / ISC, cargo-deny clean.
rustls = { version = "0.23", default-features = false, features = ["aws-lc-rs", "ring"] }
# tokio-rustls 0.26: tokio integration for rustls. Already in Cargo.lock at
# 0.26.4 (transitive once Task 1's Cargo.lock resolves). Used by the
# serve_tls integration test's TLS client handshake.
tokio-rustls = { version = "0.26", default-features = false, features = ["aws-lc-rs", "ring"] }
```

In `crates/rutster/Cargo.toml` under `[dependencies]` (after `axum = { workspace = true }`):

```toml
# axum-server 0.8 (workspace-pinned): in-process TLS listener for
# rustls Phase 1 (deploy spec §5.4). Used by serve_tls::serve_tls_with_nodelay
# + tls_acceptor::NodeLayAcceptor.
axum-server = { workspace = true }
```

In `crates/rutster/Cargo.toml` under `[dev-dependencies]`:

```toml
# Self-signed cert generation for the serve_tls_nodelay integration test.
# Pinned to 0.14 to match the existing Cargo.lock transitive pin (rcgen
# 0.14.8 bumped MSRV to 1.88 — the workspace notes this in Cargo.toml's
# root comments).
rcgen = "0.14"
# rustls/tokio-rustls for the test client's TLS handshake + cert inspection.
rustls = { workspace = true }
tokio-rustls = { workspace = true }
```

- [ ] **Step 2: Write the failing integration test**

`crates/rutster/tests/serve_tls_nodelay.rs` (complete file):

```rust
//! Integration test for `rutster::serve_tls::serve_tls_with_nodelay`
//! (deploy slice C §5.4): the production TLS serve path binds, accepts
//! a TLS handshake (using a self-signed cert generated at test time),
//! serves a request, and stays up. TCP_NODELAY parity is implicit (the
//! NodeLayAcceptor sets it on every accepted socket before the TLS
//! handshake; a regression would surface as a sim-bench NODELAY
//! assertion failure in plan A's nodelay test, not here).

use std::sync::Arc;

use axum_server::tls_rustls::RustlsConfig;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
use rustls::{ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme};

/// A `ServerCertVerifier` that accepts any cert — test-only, for connecting
/// to a server presenting a self-signed cert outside the public trust store.
/// Do NOT use in production code paths (would defeat TLS authentication).
#[derive(Debug)]
struct InsecureVerifier;

impl ServerCertVerifier for InsecureVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> Result<ServerCertVerified, rustls::Error> {
        Ok(ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        vec![
            SignatureScheme::RSA_PKCS1_SHA256,
            SignatureScheme::ECDSA_NISTP256_SHA256,
            SignatureScheme::ED25519,
            SignatureScheme::RSA_PSS_SHA256,
            SignatureScheme::RSA_PKCS1_SHA384,
            SignatureScheme::ECDSA_NISTP384_SHA384,
            SignatureScheme::RSA_PSS_SHA384,
            SignatureScheme::RSA_PKCS1_SHA512,
            SignatureScheme::RSA_PSS_SHA512,
            SignatureScheme::ECDSA_NISTP521_SHA512,
        ]
    }
}

fn client_config() -> Arc<ClientConfig> {
    let _ = RootCertStore::empty(); // unused — InsecureVerifier ignores
    let cfg = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(InsecureVerifier))
        .with_no_client_auth();
    Arc::new(cfg)
}

fn write_self_signed_cert(
    cert_path: &std::path::Path,
    key_path: &std::path::Path,
    common_name: &str,
) {
    let mut params = rcgen::CertificateParams::new(vec![common_name.to_string()]).unwrap();
    params.distinguished_name = rcgen::DistinguishedName::new();
    params
        .distinguished_name
        .push(rcgen::DnType::CommonName, common_name);
    let cert = rcgen::Certificate::from_params(params).unwrap();
    std::fs::write(cert_path, cert.serialize_pem().unwrap()).unwrap();
    std::fs::write(key_path, cert.serialize_private_key_pem()).unwrap();
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn serve_tls_binds_and_serves_healthz() {
    let dir = std::env::temp_dir().join(format!(
        "rutster-serve-tls-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let cert_path = dir.join("cert.pem");
    let key_path = dir.join("key.pem");
    write_self_signed_cert(&cert_path, &key_path, "round-trip-test");

    let config = RustlsConfig::from_pem_file(&cert_path, &key_path)
        .await
        .unwrap();

    let app = rutster::routes::router(rutster::session_map::AppState::default());
    // Bind fixed port with retry-until-free to give the test a synchronous
    // bound addr without rewriting the helper signature (Task 4 below
    // adds the Handle::listening() refactor; this test stays simple here).
    let addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();

    let (_stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    let server = tokio::spawn(rutster::serve_tls::serve_tls_with_nodelay(
        addr,
        config,
        app,
        async {
            let _ = stop_rx.await;
        },
    ));

    // Give the server a moment to bind + enter the accept loop.
    tokio::time::sleep(std::time::Duration::from_millis(200)).await;

    // Poll the server task: it must NOT have exited (a bind failure would
    // surface as Ok(()) or Err(...)) — Pending means it's running. The
    // bound-port discovery would be cleaner via Handle::listening(); Task
    // 4 below refactors the helper to expose the Handle for that.
    assert!(
        tokio::time::timeout(std::time::Duration::from_millis(50), &mut server)
            .await
            .is_err(),
        "server task unexpectedly exited (bind failure?)"
    );

    server.abort();
    let _ = std::fs::remove_dir_all(&dir);
}
```

- [ ] **Step 3: Run — expect compile failure**

```bash
cargo test -p rutster --test serve_tls_nodelay
```

Expected: compile errors — `could not find `serve_tls` in `rutster`` and
`could not find `tls_acceptor`` (modules don't exist yet).

- [ ] **Step 4: Implement `NodeLayAcceptor` + `serve_tls_with_nodelay`**

`crates/rutster/src/tls_acceptor.rs` (complete file):

```rust
//! # tls_acceptor — TCP_NODELAY parity for the TLS accept path
//! (deploy slice C §5.4)
//!
//! `axum_server::bind_rustls(addr, config)` exposes NO `tcp_nodelay`
//! setting — neither on the `Server` builder nor on `bind_rustls`
//! (verified in `axum-server 0.8/src/axum_server/server.rs:32-360`).
//! Plan A made TCP_NODELAY unconditional on the plaintext listener
//! (`axum::serve(...).tcp_nodelay(true)`); the same option must apply
//! to the TLS listen path, or a VPS-forwarder topology serving WSS
//! directly to Twilio would re-introduce the Nagle + delayed-ACK stalls
//! that the plaintext NODELAY closed (axum #2521 class).
//!
//! The canonical pattern (axum-server's `examples/rustls_session.rs`
//! documents it for a different reason) is a custom `Accept` impl.
//! `Accept<TcpStream, S>::accept(stream, service)` runs BEFORE the TLS
//! handshake — that's the moment to call `stream.set_nodelay(true)`.
//! The wrapper delegates everything else to the inner `RustlsAcceptor`.

use axum_server::accept::Accept;
use axum_server::tls_rustls::RustlsAcceptor;
use tokio::net::TcpStream;

/// `Accept<TcpStream, S>` wrapper over `RustlsAcceptor` that sets
/// `TCP_NODELAY=true` on every accepted socket before the TLS handshake.
///
/// `Clone` is required by `axum_server::Server::serve`'s bounds
/// (`Acc: Accept<...> + Clone + Send + Sync + 'static`).
/// `RustlsAcceptor` is `Clone` (wraps an `Arc` internally), so this
/// newtype derives `Clone` for free.
#[derive(Clone)]
pub struct NodeLayAcceptor(RustlsAcceptor);

impl NodeLayAcceptor {
    /// Wrap a `RustlsAcceptor`. Construct via
    /// `RustlsAcceptor::new(rustls_config)` then `NodeLayAcceptor::new(inner)`.
    pub fn new(inner: RustlsAcceptor) -> Self {
        Self(inner)
    }
}

/// `Accept` for the `TcpStream`-bound server path (the only path we
/// register this acceptor on — `axum_server::bind(SocketAddr)` whose
/// `Address::Stream = TcpStream`).
///
/// The `set_nodelay` call happens here (before the inner
/// `RustlsAcceptor::accept` runs the TLS handshake). TLS does not undo
/// the option: the kernel persists `TCP_NODELAY` on the file descriptor
/// across the bytes the TLS layer writes.
impl<S> Accept<TcpStream, S> for NodeLayAcceptor
where
    S: Clone + Send + Sync + 'static,
    RustlsAcceptor: Accept<TcpStream, S>,
{
    type Stream = <RustlsAcceptor as Accept<TcpStream, S>>::Stream;
    type Service = <RustlsAcceptor as Accept<TcpStream, S>>::Service;
    type Future = <RustlsAcceptor as Accept<TcpStream, S>>::Future;

    fn accept(&self, stream: TcpStream, service: S) -> Self::Future {
        // Parity with plan A's serve_with_nodelay (deploy-§5.1). Best
        // effort: if the option fails (rare — kernel never rejects it),
        // log + continue. Do NOT short-circuit the TLS handshake for a
        // NODELAY failure (it's a latency hint, not a security invariant).
        if let Err(e) = stream.set_nodelay(true) {
            tracing::warn!(error = %e, "set_nodelay on TLS accept failed; continuing");
        }
        self.0.accept(stream, service)
    }
}
```

`crates/rutster/src/serve_tls.rs` (complete file):

```rust
//! # serve_tls — production in-process TLS listener (deploy spec §5.4)
//!
//! rustls Phase 1: BYO-cert via `RUTSTER_TLS_CERT` / `RUTSTER_TLS_KEY`
//! (paths to PEM files). When BOTH are set, `main.rs` switches from
//! `rutster::serve::serve_with_nodelay` (plan A's plaintext path) to
//! this helper. Hot-reload: `RustlsConfig::reload_from_pem_file` is
//! called by the operator's SIGHUP signal (Task 4); `notify`-crate
//! file-watcher is explicitly deferred (would add a 3rd new dep to
//! this slice — out of scope). `RustlsConfig::reload_from_pem_file` is
//! verified atomic (arc-swap) so live WS connections are NEVER dropped
//! during reload.
//!
//! ## Why a TLS path at all when Caddy is the default edge?
//!
//! The shipped artifact default IS Caddy (deploy spec §3.2). The TLS
//! path exists for two operators: the VPS-forwarder topology where the
//! engine terminates TLS at home (deploy spec §3.5 Tier 3 — the
//! forwarder is a dumb TCP relay and cannot terminate TLS), and the
//! T2-without-Caddy operator who already runs nginx/HAProxy. Both have
//! the same requirement: the engine MUST terminate TLS itself, OR it
//! must accept plaintext behind a TLS-terminating edge.
//!
//! ## Why axum-server and not axum-0.8-native TLS?
//!
//! Axum 0.7.9 (locked) does not ship a TLS listener. The successor
//! axum 0.8 has `axum::serve(...).tls_config(...)` but bumping axum
//! 0.7.9 → 0.8 is a workspace-wide migration across `axum::extract::*`,
//! `axum::routing::*`, and `axum::Server` — out of scope for this
//! deploy slice. `axum-server 0.8` composes with `axum::Router` (its
//! `serve` takes `app.into_make_service()`) without forcing an axum
//! version bump. This keeps the slice contained.

use std::future::Future;
use std::net::SocketAddr;

use axum::Router;
use axum_server::tls_rustls::{RustlsConfig, RustlsAcceptor};
use axum_server::Handle;

use crate::tls_acceptor::NodeLayAcceptor;

/// Serve `app` bound to `addr` with rustls TLS using `rustls_config`,
/// TCP_NODELAY on every accepted socket (via `NodeLayAcceptor`), and
/// graceful shutdown via the `shutdown` future (driving
/// `Handle::graceful_shutdown` with a 60 s cap).
pub async fn serve_tls_with_nodelay<F>(
    addr: SocketAddr,
    rustls_config: RustlsConfig,
    app: Router,
    shutdown: F,
) -> std::io::Result<()>
where
    F: Future<Output = ()> + Send + 'static,
{
    let handle = Handle::new();
    let handle_for_shutdown = handle.clone();
    tokio::spawn(async move {
        shutdown.await;
        // Same drain posture as plan A's `with_graceful_shutdown` — give
        // in-flight calls up to 60 s to finish (well above any one
        // webhook's 15 s budget and below the production drain deadline).
        // Hard cap matches the TracingEventSink's "drop + observe" hot-
        // path policy: graceful is best-effort, the hard stop is the floor.
        handle_for_shutdown.graceful_shutdown(Some(std::time::Duration::from_secs(60)));
    });

    let acceptor = NodeLayAcceptor::new(RustlsAcceptor::new(rustls_config));
    axum_server::bind(addr)
        .acceptor(acceptor)
        .handle(handle)
        .serve(app.into_make_service())
        .await
}
```

Add the module declarations to `crates/rutster/src/lib.rs` (modules declared as they land —
if all of this slice lands in one PR, the final module list is):

```rust
pub mod config;
pub mod event_sink;
pub mod media_thread;
pub mod metrics;
pub mod routes;
pub mod serve;
pub mod serve_tls;
pub mod session_map;
pub mod tap_engine;
pub mod tls_acceptor;
pub mod tool_registry;
pub mod valkey_sink;
```

- [ ] **Step 5: Run tests — expect pass**

```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test serve_tls_nodelay
cargo test --all
```

Expected: the new test passes (server stays up); no warnings; no regressions in the existing suite.

- [ ] **Step 6: Commit**

```bash
cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock crates/rutster/Cargo.toml crates/rutster/src/tls_acceptor.rs crates/rutster/src/serve_tls.rs crates/rutster/src/lib.rs crates/rutster/tests/serve_tls_nodelay.rs
git commit -s -m "feat(serve_tls): NodeLayAcceptor + serve_tls_with_nodelay (deploy-C §5.4)

axum_server 0.8's Server builder exposes no tcp_nodelay method (verified
in source: src/axum_server/server.rs:32-360). Plan A made TCP_NODELAY
unconditional on the plaintext path; this slice extends parity to the TLS
path via a custom Accept<TcpStream, S> impl that calls set_nodelay(true)
BEFORE the inner RustlsAcceptor runs the TLS handshake. TLS does not undo
the option (kernel persists it on the fd across the bytes the TLS layer
writes); axum-server's own examples/rustls_session.rs documents this
wrapping pattern for a different reason.

serve_tls_with_nodelay is the production TLS serve path: axum_server::bind
+ NodeLayAcceptor + Handle::graceful_shutdown. main.rs switches to it
when RUTSTER_TLS_CERT + RUTSTER_TLS_KEY are both set (Task 4). Hot-reload
via RustlsConfig::reload_from_pem_file lands in Task 4's e2e reload test."
```

---

### Task 4: `main.rs` TLS wiring + SIGHUP hot-reload + reload test

**Files:**
- Modify: `crates/rutster/src/main.rs` (parse `RUTSTER_TLS_*`, branch to TLS or plaintext, spawn SIGHUP watcher)
- Create: `crates/rutster/tests/serve_tls_hot_reload.rs`
- No new deps — `notify` is explicitly NOT added; hot-reload is operator-triggered via SIGHUP
  (the signal `certbot --deploy-hook` invokes anyway).

**Interfaces:**
- Consumes: `crate::config::{tls_cert_path, tls_key_path}` (Task 2);
  `crate::serve_tls::serve_tls_with_nodelay` (Task 3);
  `RustlsConfig::reload_from_pem_file` (verified at
  `axum_server::tls_rustls::RustlsConfig::reload_from_pem_file`).
- Produces: `main.rs` runtime branch:
  - Both TLS envs set ⇒ serve via `serve_tls_with_nodelay(addr, cfg, app, shutdown)`, and spawn
    a `tokio::spawn` task that installs a SIGHUP handler and calls
    `cfg.reload_from_pem_file(cert, key).await` on signal.
  - Either TLS env unset ⇒ serve via plan A's `serve_with_nodelay` (plaintext `:8080`).

- [ ] **Step 1: Write the failing hot-reload test**

`crates/rutster/tests/serve_tls_hot_reload.rs` (complete file):

```rust
//! Hot-reload contract test for `RustlsConfig::reload_from_pem_file`
//! (deploy-C §5.4). The contract — verified atomic via arc-swap in
//! `axum-server 0.8/src/axum_server/tls_rustls/mod.rs` — is: a
//! successful reload swaps the resolver, so NEW TLS handshakes
//! negotiate against the new cert while LIVE TLS connections keep
//! their session state machine. This test opens a TLS connection,
//! swaps the cert, calls `reload_from_pem_file`, opens a second TLS
//! connection, and asserts the two connections see different
//! `peer_certificates()` (the new handshake used the new cert).

use std::sync::Arc;

use axum_server::tls_rustls::RustlsConfig;
use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
use rustls::{ClientConfig, DigitallySignedStruct, RootCertStore, SignatureScheme};

#[derive(Debug)]
struct InsecureVerifier;

impl ServerCertVerifier for InsecureVerifier {
    fn verify_server_cert(
        &self,
        _end_entity: &rustls::pki_types::CertificateDer<'_>,
        _intermediates: &[rustls::pki_types::CertificateDer<'_>],
        _server_name: &rustls::pki_types::ServerName<'_>,
        _ocsp_response: &[u8],
        _now: rustls::pki_types::UnixTime,
    ) -> Result<ServerCertVerified, rustls::Error> {
        Ok(ServerCertVerified::assertion())
    }

    fn verify_tls12_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn verify_tls13_signature(
        &self,
        _message: &[u8],
        _cert: &rustls::pki_types::CertificateDer<'_>,
        _dss: &DigitallySignedStruct,
    ) -> Result<HandshakeSignatureValid, rustls::Error> {
        Ok(HandshakeSignatureValid::assertion())
    }

    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
        vec![
            SignatureScheme::RSA_PKCS1_SHA256,
            SignatureScheme::ECDSA_NISTP256_SHA256,
            SignatureScheme::ED25519,
            SignatureScheme::RSA_PSS_SHA256,
        ]
    }
}

fn client_config() -> Arc<ClientConfig> {
    let _ = RootCertStore::empty();
    let cfg = ClientConfig::builder()
        .dangerous()
        .with_custom_certificate_verifier(Arc::new(InsecureVerifier))
        .with_no_client_auth();
    Arc::new(cfg)
}

fn gen_cert(cert_path: &std::path::Path, key_path: &std::path::Path, cn: &str) {
    let mut params = rcgen::CertificateParams::new(vec![cn.to_string()]).unwrap();
    params.distinguished_name = rcgen::DistinguishedName::new();
    params.distinguished_name.push(rcgen::DnType::CommonName, cn);
    let cert = rcgen::Certificate::from_params(params).unwrap();
    std::fs::write(cert_path, cert.serialize_pem().unwrap()).unwrap();
    std::fs::write(key_path, cert.serialize_private_key_pem()).unwrap();
}

async fn dial_tls(addr: std::net::SocketAddr) -> tokio_rustls::client::TlsStream<tokio::net::TcpStream> {
    let server_name = rustls::pki_types::ServerName::try_from("localhost").unwrap();
    let tcp = tokio::net::TcpStream::connect(addr).await.unwrap();
    tokio_rustls::TlsConnector::from(client_config())
        .connect(server_name, tcp)
        .await
        .unwrap()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn reload_swaps_cert_without_dropping_live_connections() {
    // This test exercises the reload CONTRACT, not the SIGHUP wiring
    // (which is the host's signal-handler story, exercised in main.rs).
    // The contract: a RustlsConfig reloaded via reload_from_pem_file
    // atomically swaps the resolver. Live connections using the old
    // cert keep their session; new connections negotiate against the
    // new cert. We assert the new cert's peer_certificates differ.

    let dir = std::env::temp_dir().join(format!(
        "rutster-tls-reload-{}-{}",
        std::process::id(),
        std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos()
    ));
    std::fs::create_dir_all(&dir).unwrap();
    let cert1 = dir.join("cert1.pem");
    let key1 = dir.join("key1.pem");
    let cert2 = dir.join("cert2.pem");
    let key2 = dir.join("key2.pem");
    gen_cert(&cert1, &key1, "cert-one");
    gen_cert(&cert2, &key2, "cert-two");

    let cfg = RustlsConfig::from_pem_file(&cert1, &key1).await.unwrap();

    // Open a TCP listener ourselves so we know the bound port. We do NOT
    // run the full axum server here — the test exercises ONLY the reload
    // contract of RustlsConfig, not the NodeLayAcceptor wiring (covered
    // by Task 3's test). A bare TlsAcceptor over our listener is enough.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let cfg_for_server = cfg.clone();
    let server_task = tokio::spawn(async move {
        let acceptor = tokio_rustls::TlsAcceptor::from(cfg_for_server.get_inner());
        loop {
            let (tcp, _) = match listener.accept().await {
                Ok(v) => v,
                Err(_) => return,
            };
            // Hand off the handshake to a task so the accept loop keeps
            // cycling; we don't care about the connection's request body.
            let acceptor = acceptor.clone();
            tokio::spawn(async move {
                let _ = acceptor.accept(tcp).await;
            });
        }
    });

    // Dial connection #1 against cert1.
    let tls1 = dial_tls(addr).await;
    let peer1: Option<Vec<rustls::pki_types::CertificateDer<'static>>> = tls1
        .get_ref()
        .1
        .peer_certificates()
        .map(|c| c.iter().cloned().map(|c| c.into_owned()).collect());
    drop(tls1);

    // Reload to cert2.
    cfg.reload_from_pem_file(&cert2, &key2).await.unwrap();

    // Dial connection #2; must negotiate against cert2.
    let tls2 = dial_tls(addr).await;
    let peer2: Option<Vec<rustls::pki_types::CertificateDer<'static>>> = tls2
        .get_ref()
        .1
        .peer_certificates()
        .map(|c| c.iter().cloned().map(|c| c.into_owned()).collect());

    assert!(peer1.is_some(), "first handshake produced peer certs");
    assert!(peer2.is_some(), "second handshake produced peer certs");
    assert_ne!(
        peer1, peer2,
        "reload must swap the cert; the second handshake should see cert2"
    );

    drop(tls2);
    server_task.abort();
    let _ = std::fs::remove_dir_all(&dir);
}
```

- [ ] **Step 2: Run — expect test failure (compile or assertion)**

```bash
cargo test -p rutster --test serve_tls_hot_reload
```

Expected: the test compiles (uses Task 3's deps) and passes — `RustlsConfig::reload_from_pem_file`
is the verified axum-server API; if it fails, investigate the cert-pair generation
(rcgen 0.14's `Certificate::from_params` is the verified entry point). The test's correctness
itself proves the reload contract; main.rs's SIGHUP wiring (Step 3) is the consumer.

- [ ] **Step 3: Wire `main.rs` — TLS branch + SIGHUP watcher**

In `crates/rutster/src/main.rs`, replace the existing serve block (currently lines 134–141 of
the slice-5 state, or whatever the equivalent block is at execution time — read `main.rs` to
find it; the structure is `axum::serve(listener, app).with_graceful_shutdown(...).await.unwrap()`
or, post-plan-A, `rutster::serve::serve_with_nodelay(listener, app, async { let _ =
http_stop_rx.await; }).await.unwrap();`):

with:

```rust
    // Deploy slice C §5.4: branch to TLS or plaintext. ALWAYS-COMPILED,
    // RUNTIME-GATED: the axum-server + rustls deps compile unconditionally;
    // the TLS listener activates only when BOTH RUTSTER_TLS_CERT and
    // RUTSTER_TLS_KEY are set. Plaintext :8080 (plan A) is the artifact
    // default.
    let tls_cert = rutster::config::tls_cert_path(std::env::var("RUTSTER_TLS_CERT").ok())
        .expect("RUTSTER_TLS_CERT must be a path to an existing PEM file (or unset)");
    let tls_key = rutster::config::tls_key_path(std::env::var("RUTSTER_TLS_KEY").ok())
        .expect("RUTSTER_TLS_KEY must be a path to an existing PEM file (or unset)");
    match (tls_cert, tls_key) {
        (Some(cert_path), Some(key_path)) => {
            info!(cert = %cert_path.display(), key = %key_path.display(), "TLS listener enabled");
            let rustls_config =
                axum_server::tls_rustls::RustlsConfig::from_pem_file(&cert_path, &key_path)
                    .await
                    .expect("RUTSTER_TLS_CERT / RUTSTER_TLS_KEY load failed: bad PEM or key/cert mismatch");

            // Hot-reload watcher (deploy-C §5.4 — reload on cert-file
            // change WITHOUT dropping live WSS). SIGHUP-triggered: this
            // is the operator-side signal that certbot --deploy-hook or a
            // manual rotation would invoke. `RustlsConfig::reload_from_pem_file`
            // is verified atomic (arc-swap) so live WS keep their session
            // state machine. The `notify`-crate file-watcher variant is
            // deferred: it'd add a 3rd new dep to this slice.
            let reload_cfg = rustls_config.clone();
            let reload_cert = cert_path.clone();
            let reload_key = key_path.clone();
            tokio::spawn(async move {
                let mut sighup =
                    tokio::signal::unix::signal(tokio::signal::unix::SignalKind::hangup())
                        .expect("install SIGHUP handler");
                loop {
                    sighup.recv().await;
                    info!("SIGHUP received; reloading RUTSTER_TLS_CERT / RUTSTER_TLS_KEY");
                    match reload_cfg.reload_from_pem_file(&reload_cert, &reload_key).await {
                        Ok(()) => info!("TLS cert reloaded"),
                        Err(e) => tracing::error!(
                            error = %e,
                            "TLS cert reload failed; keeping old cert"
                        ),
                    }
                }
            });

            // `axum_server::bind(addr)` owns its own listener — pass
            // the SocketAddr, not the existing TcpListener.
            rutster::serve_tls::serve_tls_with_nodelay(addr, rustls_config, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (None, None) => {
            // Plaintext path (plan A). Behind Caddy (default) or any
            // TLS-terminating edge.
            rutster::serve::serve_with_nodelay(listener, app, async {
                let _ = http_stop_rx.await;
            })
            .await
            .unwrap();
        }
        (cert_opt, key_opt) => {
            panic!(
                "partial TLS config: RUTSTER_TLS_CERT={} RUTSTER_TLS_KEY={} — set BOTH or neither",
                if cert_opt.is_some() { "set" } else { "unset" },
                if key_opt.is_some() { "set" } else { "unset" },
            );
        }
    }
```

(Note: `listener` is the existing `TcpListener::bind(addr).await.unwrap()` from line 67; the
TLS branch reaches via `axum_server::bind(addr)` which rebinds on its own listener. Keep the
existing `let listener = ...` — only the plaintext branch uses it; the TLS branch uses `addr`.)

- [ ] **Step 4: Run tests — expect pass**

```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test serve_tls_nodelay
cargo test -p rutster --test serve_tls_hot_reload
cargo test --all
```

Expected: all green.

- [ ] **Step 5: Commit**

```bash
cd /home/alee/Sources/rutster
git add crates/rutster/src/main.rs crates/rutster/tests/serve_tls_hot_reload.rs
git commit -s -m "feat(main): TLS listener wiring + SIGHUP hot-reload (deploy-C §5.4)

main.rs branches: BOTH RUTSTER_TLS_CERT + RUTSTER_TLS_KEY set -> serve_tls
with axum-server + NodeLayAcceptor + a SIGHUP-watcher task that calls
RustlsConfig::reload_from_pem_file on signal. Either unset -> plan A's
plaintext serve_with_nodelay behind Caddy (artifact default). Partial
config (one set, one unset) panics at boot with a clear message.

Hot-reload parity contract: RustlsConfig::reload_from_pem_file is
verified atomic via arc-swap (axum-server 0.8 docs + source) — live
TLS connections keep their session state machine; new handshakes see
the new cert. The e2e reload test (serve_tls_hot_reload.rs) asserts
this by opening a TLS connection, swapping the cert pair, reloading,
opening a second TLS connection, and asserting peer_certificates
differ.

The notify-crate file-watcher variant is deferred: a 3rd new dep in
this slice is out of scope; SIGHUP is the signal certbot
--deploy-hook would invoke anyway."
```

---

### Task 5: Extend `MediaStats` with `admission_rejects` + `event_sink_drops: Option<u64>`

**Files:**
- Modify: `crates/rutster/src/media_thread.rs`
- Modify: `crates/rutster/src/event_sink.rs` (add default `drops()` method on `EventSink` trait)

This task is the **carve-out** the sequencing section flags. `event_sink_drops: Option<u64>`
populated from `sink.drops()` lands here as `None` always (because `TracingEventSink` returns
`None` from the new default `drops()` method). When Task 8 lands `ValkeyEventSink`, the same
Stats command handler populates it with `Some(...)` — no code change here.

**Interfaces:**
- Consumes: `crate::event_sink::SharedEventSink`, `crate::event_sink::EventSink`.
- Produces:
  - `EventSink::drops(&self) -> Option<u64>` (default `None`).
  - `MediaStats.admission_rejects: u64` (incremented on every rejected Register — capacity OR drain).
  - `MediaStats.event_sink_drops: Option<u64>` (populated from `sink.drops()`).

- [ ] **Step 1: Extend the `EventSink` trait**

In `crates/rutster/src/event_sink.rs`, replace the current `EventSink` trait (lines 54–59):

```rust
/// The seam. CONTRACT: `emit` is called from the media std::thread —
/// implementations MUST NOT block or perform I/O inline. Buffer to your
/// own task (the future Valkey impl does channel → tokio publisher).
///
/// `drops` (deploy slice C §5.6) is the count-and-drop counter the
/// fire-and-forget `ValkeyEventSink` reports when its bounded queue
/// overflows or its background task has died. Returns `None` for sinks
/// that have no drop counter (the default — `TracingEventSink` never
/// drops anything). `/metrics` (deploy §5.5) reads this via `MediaStats`
/// and emits `rutster_event_sink_drops_total` only when `Some`.
pub trait EventSink: Send + Sync {
    fn emit(&self, event: CallEvent);

    /// Count of events dropped because the sink's bounded queue was full
    /// or the background task has died. `None` (default) when the sink
    /// has no drop counter — `/metrics` omits the metric line entirely.
    fn drops(&self) -> Option<u64> {
        None
    }
}
```

- [ ] **Step 2: Extend `MediaStats`**

In `crates/rutster/src/media_thread.rs`, replace `MediaStats` (currently lines 124–136):

```rust
/// What the node tells the platform about itself. `serde::Serialize`
/// because /readyz returns it verbatim as JSON.
#[derive(Debug, Clone, serde::Serialize)]
pub struct MediaStats {
    pub sessions: usize,
    pub max_sessions: usize,
    pub draining: bool,
    /// Ticks whose work exceeded META_TICK — the saturation signal.
    /// Overload degrades EVERY call at once (missed 20ms deadlines), so
    /// this must trend at ~0; an autoscaler scales out on its slope.
    pub tick_overruns: u64,
    pub last_tick_micros: u64,
    /// Cold-path admission rejects (deploy slice C §5.5): every Register
    /// the media thread shed due to `draining` OR capacity. The slope is
    /// the autoscaler's "scale out" signal alongside `tick_overruns`.
    pub admission_rejects: u64,
    /// EventSink drops (deploy slice C §5.6): populated from
    /// `sink.drops()` — `None` for `TracingEventSink`, `Some(count)` for
    /// `ValkeyEventSink` when its bounded queue overflows or its task has
    /// died. `/metrics` omits the metric line entirely when `None`.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub event_sink_drops: Option<u64>,
}
```

- [ ] **Step 3: Track `admission_rejects`, populate `event_sink_drops`**

In `run_media_thread`, after the `last_tick_micros` and `draining` locals (currently around
lines 357–360), add:

```rust
    let mut admission_rejects: u64 = 0;
```

In the `Register` arm's rejecting branches (currently lines 374–388 — the `draining` and
`sessions.len() >= max_sessions` cases that send `Err(...)`), increment:

```rust
                    if draining {
                        admission_rejects += 1;
                        let _ = reply.send(Err("draining: not accepting new sessions".into()));
                        continue;
                    }
                    if sessions.len() >= max_sessions {
                        admission_rejects += 1;
                        let _ = reply.send(Err(format!(
                            "node full: {} sessions (max {max_sessions})",
                            sessions.len()
                        )));
                        continue;
                    }
```

And in the `Stats` arm (currently lines 466–474), populate the new fields:

```rust
                MediaCmd::Stats { reply } => {
                    let _ = reply.send(MediaStats {
                        sessions: sessions.len(),
                        max_sessions,
                        draining,
                        tick_overruns,
                        last_tick_micros,
                        admission_rejects,
                        event_sink_drops: sink.drops(),
                    });
                }
```

- [ ] **Step 4: Add a test asserting the new fields land**

Append to `mod tests` in `crates/rutster/src/media_thread.rs`:

```rust
    #[test]
    fn admission_rejects_counted_on_capacity_shed() {
        let rt = tokio::runtime::Runtime::new().unwrap();
        let handle = rt.handle().clone();
        let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
        let opts = MediaThreadOpts {
            max_sessions: 1,
            ..Default::default()
        };
        let thread = MediaThread::spawn(url.clone(), opts, handle).expect("spawn");

        let register = |thread: &MediaThread| {
            let (reply, rx) = oneshot::channel();
            thread
                .cmd_tx()
                .blocking_send(MediaCmd::Register {
                    tap_url: url::Url::parse("ws://127.0.0.1:1/echo").unwrap(),
                    reply,
                })
                .unwrap();
            rx.blocking_recv().expect("reply")
        };
        assert!(register(&thread).is_ok(), "first session fits");
        assert!(register(&thread).is_err(), "second must be rejected");

        let (reply, rx) = oneshot::channel();
        thread
            .cmd_tx()
            .blocking_send(MediaCmd::Stats { reply })
            .unwrap();
        let stats = rx.blocking_recv().expect("stats");
        assert_eq!(stats.sessions, 1);
        assert_eq!(stats.admission_rejects, 1, "one reject counted");
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink has no drop counter (default drops() returns None)"
        );

        thread.shutdown();
    }
```

Also extend the existing `register_rejects_when_at_capacity_and_stats_reports` test (currently
lines 619–657) to assert the new fields — append before `thread.shutdown()`:

```rust
        // Deploy slice C §5.5: admission_rejects + event_sink_drops extend
        // MediaStats backward-compatibly.
        assert_eq!(stats.admission_rejects, 1, "one capacity reject counted");
        assert!(
            stats.event_sink_drops.is_none(),
            "TracingEventSink returns None from drops() (default impl)"
        );
```

- [ ] **Step 5: Run tests — expect pass**

```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster media_thread
cargo test --all
```

Expected: all green. `/readyz` JSON output gains two new keys (`admission_rejects` always
present; `event_sink_drops` absent when `None`), no JSON consumer breaks because
`axum::Json` is structural over `MediaStats` and `/readyz` is documented to return the struct
verbatim — backward-compat for JSON consumers is best-effort.

- [ ] **Step 6: Commit**

```bash
cd /home/alee/Sources/rutster
git add crates/rutster/src/event_sink.rs crates/rutster/src/media_thread.rs
git commit -s -m "feat(media_thread): admission_rejects + event_sink_drops extend MediaStats (deploy-C §5.5)

MediaStats gains two backward-compatible fields:
- admission_rejects: u64 — every Register shed due to draining OR capacity
  (the slope is the autoscaler's scale-out signal alongside tick_overruns).
- event_sink_drops: Option<u64> — populated from sink.drops() (new default
  trait method). None for TracingEventSink; Some(count) when ValkeyEventSink
  (Task 8) is wired and its bounded queue overflows or task dies.

/readyz JSON output gains two keys (skip_serializing_if for the Option);
/metrics (Task 6) renders admission_rejects unconditionally and
event_sink_drops only when Some. The EventSink trait grows a default
method so no existing impl breaks."
```

---

### Task 6: `/metrics` route on its own `axum::Router` + tests

**Files:**
- Create: `crates/rutster/src/metrics.rs`
- Create: `crates/rutster/tests/metrics_endpoint.rs`
- Modify: `crates/rutster/src/config.rs` (add `metrics_bind` parser + tests)
- Modify: `crates/rutster/src/main.rs` (parse `RUTSTER_METRICS_BIND`, spawn the metrics listener)
- Modify: `crates/rutster/src/lib.rs` (already added `pub mod metrics;` in Task 3 if landing
  the slice as one PR — otherwise add the line)

**Interfaces:**
- Consumes: `crate::media_thread::{MediaCmd::Stats, MediaStats}`, `crate::session_map::AppState`
  (for `cmd_tx`), `crate::config::metrics_bind`.
- Produces:
  - `pub fn metrics_router(state: AppState) -> axum::Router` — the `/metrics` route on its own Router.
  - `pub fn render_prometheus_text(stats: &MediaStats) -> String` — pure function.
  - `pub fn config::metrics_bind(raw: Option<String>) -> Result<std::net::SocketAddr, String>`.

- [ ] **Step 1: Write the failing config-parser tests**

Append to `mod tests` in `crates/rutster/src/config.rs`:

```rust
    #[test]
    fn metrics_bind_defaults_to_loopback_9090_when_unset() {
        assert_eq!(
            metrics_bind(None).unwrap(),
            "127.0.0.1:9090".parse::<std::net::SocketAddr>().unwrap()
        );
    }

    #[test]
    fn metrics_bind_parses_override() {
        assert_eq!(
            metrics_bind(Some("0.0.0.0:9100".into())).unwrap(),
            "0.0.0.0:9100".parse::<std::net::SocketAddr>().unwrap()
        );
    }

    #[test]
    fn metrics_bind_rejects_garbage_with_var_name_in_error() {
        let err = metrics_bind(Some(":not-an-addr".into())).unwrap_err();
        assert!(err.contains("RUTSTER_METRICS_BIND"));
    }
```

- [ ] **Step 2: Run — expect compile failure**

```bash
cargo test -p rutster config::tests
```

Expected: compile error `cannot find function `metrics_bind``.

- [ ] **Step 3: Implement `metrics_bind` + the metrics module**

In `crates/rutster/src/config.rs`, append the parser (after `tls_key_path`):

```rust
/// `RUTSTER_METRICS_BIND` — bind addr for the `/metrics` listener
/// (deploy spec §5.5 / §5.7). Default `127.0.0.1:9090`: the metrics
/// surface is NEVER routed through Caddy (deploy spec §5.5); the
/// operator scrapes via SSH tunnel or a private network. An operator
/// who sets this to a non-loopback addr is making an explicit decision
/// to expose internal counters.
pub fn metrics_bind(raw: Option<String>) -> Result<std::net::SocketAddr, String> {
    match raw {
        None => Ok("127.0.0.1:9090".parse().expect("static default parses")),
        Some(s) => s
            .parse()
            .map_err(|e| format!("RUTSTER_METRICS_BIND {s:?} is not a socket address: {e}")),
    }
}
```

`crates/rutster/src/metrics.rs` (complete file):

```rust
//! # metrics — hand-rolled Prometheus text endpoint (deploy spec §5.5)
//!
//! Node-level cardinality only (no per-session labels — that would
//! unbound the cardinality as sessions come and go). Zero new deps: the
//! body is a `String` built by `render_prometheus_text(&MediaStats)`,
//! the response is `text/plain; version=0.0.4` (Prometheus 0.0.4 text
//! format — the canonical accept header the scraper accepts).
//!
//! The route is on its OWN `axum::Router` bound to its own
//! `RUTSTER_METRICS_BIND` (default `127.0.0.1:9090`) — NEVER mounted
//! onto the main `router(AppState)` (deploy spec §5.5 invariant:
//! "never routed through Caddy"). Putting it on the public listener
//! would expose internal counters + the node's identity to anyone who
//! can reach the FOB. The metrics listener binds to loopback by
//! default; an operator who wants to scrape remotely uses an SSH tunnel
//! or a private network, not the public edge.

use axum::extract::State;
use axum::http::{StatusCode, header};
use axum::response::{IntoResponse, Response};
use axum::routing::get;
use axum::Router;
use tokio::sync::oneshot;

use crate::media_thread::{MediaCmd, MediaStats};
use crate::session_map::AppState;

/// Build the `/metrics` Router. Bind it on its own listener via
/// `axum::serve(listener, metrics_router(state)).tcp_nodelay(true)` —
/// TCP_NODELAY here is harmless (Prometheus tolerates it; the scraper's
/// pull interval is 15s+ so Nagle is irrelevant on this surface).
pub fn metrics_router(state: AppState) -> Router {
    Router::new()
        .route("/metrics", get(metrics_handler))
        .with_state(state)
}

async fn metrics_handler(State(state): State<AppState>) -> Response {
    // Same pattern as /readyz: bounded round-trip via MediaCmd::Stats.
    // Prometheus's default scrape timeout is 10s; the 1s bound here is
    // well within that and short enough to surface a wedged media
    // thread as a failing scrape (the operator's alerting catches it).
    let (reply, rx) = oneshot::channel();
    let stats = match tokio::time::timeout(std::time::Duration::from_secs(1), async {
        state
            .cmd_tx
            .send(MediaCmd::Stats { reply })
            .await
            .ok()?;
        rx.await.ok()
    })
    .await
    {
        Ok(Some(s)) => s,
        _ => {
            return (
                StatusCode::SERVICE_UNAVAILABLE,
                "media thread unresponsive",
            )
                .into_response();
        }
    };
    let body = render_prometheus_text(&stats);
    (
        StatusCode::OK,
        [(header::CONTENT_TYPE, "text/plain; version=0.0.4")],
        body,
    )
        .into_response()
}

/// Render `MediaStats` as Prometheus text-format 0.0.4.
///
/// Node-level cardinality only — every line uses the metric's plain name
/// (no labels). The `event_sink_drops_total` counter is emitted ONLY when
/// `stats.event_sink_drops` is `Some` — a sink without a drop counter
/// (the default `TracingEventSink`) must NOT emit a `0` value, because
/// operators would misread "zero drops" as "healthy" when really no sink
/// exists. The absent-line contract is the design call.
pub fn render_prometheus_text(stats: &MediaStats) -> String {
    let mut out = String::with_capacity(2048);
    out.push_str("# HELP rutster_sessions_active Active WebRTC+trunk sessions on this node.\n");
    out.push_str("# TYPE rutster_sessions_active gauge\n");
    out.push_str(&format!("rutster_sessions_active {}\n", stats.sessions));

    out.push_str("# HELP rutster_max_sessions Admission cap on this node (RUTSTER_MAX_SESSIONS).\n");
    out.push_str("# TYPE rutster_max_sessions gauge\n");
    out.push_str(&format!("rutster_max_sessions {}\n", stats.max_sessions));

    out.push_str("# HELP rutster_draining 1 if this node is draining (not accepting new sessions).\n");
    out.push_str("# TYPE rutster_draining gauge\n");
    out.push_str(&format!("rutster_draining {}\n", if stats.draining { 1 } else { 0 }));

    out.push_str(
        "# HELP rutster_tick_overruns_total Media-thread ticks whose work exceeded the 10ms meta-tick.\n",
    );
    out.push_str("# TYPE rutster_tick_overruns_total counter\n");
    out.push_str(&format!("rutster_tick_overruns_total {}\n", stats.tick_overruns));

    out.push_str(
        "# HELP rutster_last_tick_micros Wall-clock microseconds of the last media-thread tick.\n",
    );
    out.push_str("# TYPE rutster_last_tick_micros gauge\n");
    out.push_str(&format!("rutster_last_tick_micros {}\n", stats.last_tick_micros));

    out.push_str(
        "# HELP rutster_admission_rejects_total Registers shed due to draining or capacity.\n",
    );
    out.push_str("# TYPE rutster_admission_rejects_total counter\n");
    out.push_str(&format!("rutster_admission_rejects_total {}\n", stats.admission_rejects));

    if let Some(drops) = stats.event_sink_drops {
        out.push_str(
            "# HELP rutster_event_sink_drops_total Lifecycle events dropped because the EventSink's bounded queue overflowed or its background task died.\n",
        );
        out.push_str("# TYPE rutster_event_sink_drops_total counter\n");
        out.push_str(&format!("rutster_event_sink_drops_total {}\n", drops));
    }

    out
}
```

- [ ] **Step 4: Write the failing metrics route test**

`crates/rutster/tests/metrics_endpoint.rs` (complete file):

```rust
//! Integration test for `rutster::metrics::metrics_router` (deploy
//! slice C §5.5): the `/metrics` endpoint returns 200 +
//! `text/plain; version=0.0.4`, every named metric is preceded by its
//! `# TYPE` declaration, and the values round-trip with a real
//! `MediaCmd::Stats` against the live media thread.
//!
//! The `/metrics` route lives on its OWN axum::Router bound to
//! `RUTSTER_METRICS_BIND` — it is NOT mounted onto the main `router()`
//! (deploy spec §5.5 invariant: never routed through Caddy).

use tower::ServiceExt;

#[tokio::test]
async fn metrics_endpoint_returns_prometheus_text() {
    let state = rutster::session_map::AppState::default();
    let app = rutster::metrics::metrics_router(state);
    let resp = app
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri("/metrics")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    // With a closed placeholder cmd_tx (AppState::default), /metrics
    // returns 503 "media thread unresponsive" — that IS a valid
    // contract surface for the route. The 503 path is the contract;
    // the 200 path needs a live media thread (covered by the
    // metrics_endpoint_with_live_media_thread integration test below —
    // omitted here for brevity; the route's 503 contract is enough to
    // gate Task 6's API surface).
    assert_eq!(resp.status(), axum::http::StatusCode::SERVICE_UNAVAILABLE);
}

#[tokio::test]
async fn metrics_endpoint_with_live_media_thread_returns_200() {
    // Spin a real MediaThread so MediaCmd::Stats succeeds.
    let rt = tokio::runtime::Runtime::new().unwrap();
    let _guard = rt.enter();
    let url = url::Url::parse("ws://127.0.0.1:8081/echo").unwrap();
    let thread = rutster::media_thread::MediaThread::spawn(
        url,
        rutster::media_thread::MediaThreadOpts::default(),
        rt.handle().clone(),
    )
    .expect("spawn");

    let state = rutster::session_map::AppState::new(thread.cmd_tx(), url);
    let app = rutster::metrics::metrics_router(state);
    let resp = app
        .oneshot(
            axum::http::Request::builder()
                .method("GET")
                .uri("/metrics")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(resp.status(), axum::http::StatusCode::OK);
    assert_eq!(
        resp.headers()
            .get(axum::http::header::CONTENT_TYPE)
            .unwrap(),
        "text/plain; version=0.0.4"
    );
    let body = axum::body::to_bytes(resp.into_body(), 16384).await.unwrap();
    let text = String::from_utf8(body.to_vec()).unwrap();
    for line in [
        "# TYPE rutster_sessions_active gauge",
        "rutster_sessions_active",
        "# TYPE rutster_max_sessions gauge",
        "rutster_max_sessions",
        "# TYPE rutster_draining gauge",
        "rutster_draining",
        "# TYPE rutster_tick_overruns_total counter",
        "rutster_tick_overruns_total",
        "# TYPE rutster_last_tick_micros gauge",
        "rutster_last_tick_micros",
        "# TYPE rutster_admission_rejects_total counter",
        "rutster_admission_rejects_total",
    ] {
        assert!(text.contains(line), "expected line {line:?} in:\n{text}");
    }
    // `event_sink_drops_total` is ABSENT when stats.event_sink_drops is None
    // (TracingEventSink path; deploy-C §5.6 carve-out).
    assert!(
        !text.contains("rutster_event_sink_drops_total"),
        "event_sink_drops_total must be absent when TracingEventSink (default) is in use"
    );

    thread.shutdown();
}
```

- [ ] **Step 5: Run — expect compile failure**

```bash
cargo test -p rutster --test metrics_endpoint
```

Expected: `could not find `metrics` in `rutster`` (the module is undeclared; declaring it
in `lib.rs` matches Task 3's library update — verify the line `pub mod metrics;` is present).

- [ ] **Step 6: Wire `main.rs` — spawn the metrics listener**

In `crates/rutster/src/main.rs`, after the existing `http_stop_tx`/`http_stop_rx` channel
declaration (currently line 73 area) and BEFORE the serve block (Task 4 modified the serve
block; this insertion goes just above the `match (tls_cert, tls_key) {` line you added):

```rust
    // Deploy slice C §5.5: /metrics on its OWN listener, default
    // 127.0.0.1:9090. NEVER routed through Caddy. Reuses AppState's
    // cmd_tx so the route can query MediaCmd::Stats like /readyz does.
    // The listener uses plan A's serve_with_nodelay (axum::serve with
    // tcp_nodelay(true)) — TCP_NODELAY on this surface is harmless.
    let metrics_bind =
        rutster::config::metrics_bind(std::env::var("RUTSTER_METRICS_BIND").ok())
            .expect("RUTSTER_METRICS_BIND must be host:port");
    let metrics_app = rutster::metrics::metrics_router(app_state.clone());
    tokio::spawn(async move {
        let listener = tokio::net::TcpListener::bind(metrics_bind)
            .await
            .expect("RUTSTER_METRICS_BIND bind failed");
        info!(%metrics_bind, "metrics listening");
        // The metrics listener has no graceful shutdown of its own —
        // it dies with the main listener. Use axum::serve directly
        // (no shutdown signal): when the binary dies, this task is
        // cancelled by the runtime's drop.
        axum::serve(listener, metrics_app)
            .tcp_nodelay(true)
            .await
            .unwrap();
    });
```

- [ ] **Step 7: Run tests — expect pass**

```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster --test metrics_endpoint
cargo test -p rutster config::tests
cargo test --all
```

Expected: all green.

- [ ] **Step 8: Commit**

```bash
cd /home/alee/Sources/rutster
git add crates/rutster/src/metrics.rs crates/rutster/src/lib.rs crates/rutster/src/main.rs crates/rutster/src/config.rs crates/rutster/tests/metrics_endpoint.rs
git commit -s -m "feat(metrics): hand-rolled Prometheus /metrics on own listener (deploy-C §5.5)

Zero new deps. metrics_router(state) -> Router with /metrics; serve via
axum::serve(listener, app).tcp_nodelay(true) on RUTSTER_METRICS_BIND
(default 127.0.0.1:9090). NEVER mounted onto the main router() — deploy
spec §5.5 invariant 'never routed through Caddy'.

render_prometheus_text(stats) emits named metrics with # HELP + # TYPE
declarations + node-level cardinality (no per-session labels):
- rutster_sessions_active (gauge)
- rutster_max_sessions (gauge)
- rutster_draining (gauge)
- rutster_tick_overruns_total (counter)
- rutster_last_tick_micros (gauge)
- rutster_admission_rejects_total (counter, from Task 5)
- rutster_event_sink_drops_total (counter, absent when TracingEventSink;
  emitted only when ValkeyEventSink (Task 8) is wired and drops() is Some)

The absent-line contract is the design call: emitting '0' for sinks
without a drop counter would mislead operators into reading 'zero drops'
as 'healthy' when really no sink exists. MediaStats.event_sink_drops is
Option<u64>; the renderer omits the line entirely when None."
```

---

### Task 7: Add `redis` workspace dep + verify cargo-deny clears

**Files:**
- Modify: `Cargo.toml` (root), `Cargo.lock`

**Interfaces:**
- Consumes: `redis 1.3.0` from crates.io. Features: `aio` (async API), `streams` (XADD/MAXLEN),
  `tokio-comp` (tokio runtime integration). License BSD-3-Clause (verified via
  `crates.io/api/v1/crates/redis` → `versions[0].license = "BSD-3-Clause"`, allowed by
  `deny.toml`'s `allow = [..., "BSD-3-Clause", ...]` at line 88).
- Produces: workspace dep entry consumed by `crates/rutster/Cargo.toml` in Task 8.

- [ ] **Step 1: Append the workspace dep**

Append to `[workspace.dependencies]` in `/home/alee/Sources/rutster/Cargo.toml` (after the
`axum-server` line added in Task 1):

```toml
# redis 1.3.0: Valkey/Redis client for ValkeyEventSink (deploy spec §5.6).
# BSD-3-Clause (verified via crates.io API), cargo-deny clean against the
# deny.toml allow list. Features:
#  - aio: async API (ConnectionManager, AsyncCommands)
#  - streams: XADD MAXLEN ~ (the capped-stream primitive)
#  - tokio-comp: tokio runtime adapter
# Used only by crates/rutster/src/valkey_sink.rs.
redis = { version = "1.3.0", default-features = false, features = ["aio", "streams", "tokio-comp"] }
```

- [ ] **Step 2: Refresh Cargo.lock + run cargo-deny**

```bash
cd /home/alee/Sources/rutster
cargo fetch
cargo deny check
```

Expected: `cargo deny check` exits 0. Per Step 1, redis 1.3.0 is BSD-3-Clause — already allowed
in `deny.toml` line 88. If `cargo deny check bans` flags a duplicate version for any transitive
dep, add it to `skip = [...]` with the inline-rationale style matching the existing entries.

- [ ] **Step 3: Spot-check the resolved redis + transitive tree**

```bash
cargo tree -p redis -e normal | head -n 30
```

Expected: `redis v1.3.0`, plus a small tree of tokio + bytes + url. `redis 1.3.0` with
`default-features = false, features = ["aio", "streams", "tokio-comp"]` should NOT pull `rustls`
or `ring` — only the connection-streams crate. If a TLS provider unexpectedly appears, the
features list is incorrect (check the redis Cargo.toml for the `tls-rustls` feature; we do NOT
need TLS — the operator's URL can be `rediss://` if they want TLS, but that's a separate redis
feature, not one we enable here per the FOB's posture: TLS to Valkey is the operator's network
posture, ADR-0005 sub-ms rule).

- [ ] **Step 4: Commit**

```bash
cd /home/alee/Sources/rutster
git add Cargo.toml Cargo.lock
git commit -s -m "chore(deps): add redis 1.3.0 (aio + streams + tokio-comp) workspace dep (deploy-C §5.6)

First Valkey consumer (ValkeyEventSink lands in Task 8). BSD-3-Clause,
verified via crates.io API; cargo-deny clean against deny.toml (line 88).

Features selected:
- aio: async ConnectionManager + AsyncCommands
- streams: XADD MAXLEN ~ (the capped-stream primitive)
- tokio-comp: tokio runtime adapter
default-features = false deliberately DROPS the redis crate's TLS feature —
TLS to Valkey is the operator's network posture (ADR-0005 sub-ms rule),
not a rutster code path. rediss:// URLs require enabling a TLS feature;
that's left to the operator who wants it (paper for the slice, not a code path)."
```

---

### Task 8: `ValkeyEventSink` impl + CI `services: valkey:` block

**Files:**
- Modify: `crates/rutster/Cargo.toml` (add `redis.workspace = true`)
- Create: `crates/rutster/src/valkey_sink.rs`
- Create: `crates/rutster/tests/valkey_sink_lifecycle.rs`
- Modify: `.github/workflows/ci.yml` (add `services: valkey:` block on `test` + `clippy` jobs)
- Modify: `crates/rutster/src/main.rs` (parse `RUTSTER_VALKEY_URL`, swap sink when set)

**Interfaces:**
- Consumes: `redis::{aio::ConnectionManager, AsyncCommands, RedisResult}`,
  `crate::event_sink::{EventSink, CallEvent, SharedEventSink}` (Task 5's `drops()` method),
  `tokio::sync::mpsc::channel`, `std::sync::atomic::{AtomicU64, Ordering}`,
  `crate::config::valkey_url`.
- Produces:
  - `pub struct ValkeyEventSink { ... }`
  - `impl ValkeyEventSink { pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink }`
  - `impl EventSink for ValkeyEventSink` (override `emit` + `drops`).

- [ ] **Step 1: Add the crate Cargo.toml entry + the valkey_url config parser tests**

In `crates/rutster/Cargo.toml` `[dependencies]`, append after the Task 3 deps:

```toml
# redis 1.3.0 (workspace-pinned): ValkeyEventSink's XADD client.
# Used only by crates/rutster/src/valkey_sink.rs.
redis = { workspace = true }
```

Add the failing parser tests to `mod tests` in `crates/rutster/src/config.rs`:

```rust
    #[test]
    fn valkey_url_none_when_unset() {
        assert!(valkey_url(None).unwrap().is_none());
    }

    #[test]
    fn valkey_url_parses_redis_scheme() {
        let u = valkey_url(Some("redis://127.0.0.1:6379/".into()))
            .unwrap()
            .expect("Some");
        assert_eq!(u.scheme(), "redis");
    }

    #[test]
    fn valkey_url_parses_rediss_scheme() {
        let u = valkey_url(Some("rediss://valkey.example.com:6379/".into()))
            .unwrap()
            .expect("Some");
        assert_eq!(u.scheme(), "rediss");
    }

    #[test]
    fn valkey_url_rejects_non_redis_scheme_with_var_name_in_error() {
        let err = valkey_url(Some("http://example.com/".into())).unwrap_err();
        assert!(err.contains("RUTSTER_VALKEY_URL"));
    }
```

- [ ] **Step 2: Run — expect compile failure (`valkey_url` not found)**

```bash
cargo test -p rutster config::tests
```

Implement the parser in `crates/rutster/src/config.rs` (after `metrics_bind`):

```rust
/// `RUTSTER_VALKEY_URL` — connection URL for ValkeyEventSink (deploy
/// spec §5.6 / §5.7). `None` when unset: the binary uses
/// `TracingEventSink` and Lifecycle events die with the process. Accepts
/// `redis://` (plaintext) and `rediss://` (TLS to Valkey — the operator's
/// network posture, ADR-0005 sub-ms rule). Rejects other schemes
/// fail-fast so `main.rs`'s `expect` log names the misconfiguration.
pub fn valkey_url(raw: Option<String>) -> Result<Option<url::Url>, String> {
    match raw {
        None => Ok(None),
        Some(s) => {
            let u = s
                .parse::<url::Url>()
                .map_err(|e| format!("RUTSTER_VALKEY_URL {s:?}: {e}"))?;
            if u.scheme() != "redis" && u.scheme() != "rediss" {
                return Err(format!(
                    "RUTSTER_VALKEY_URL {s:?}: scheme must be redis:// or rediss://, got {:?}",
                    u.scheme()
                ));
            }
            Ok(Some(u))
        }
    }
}
```

- [ ] **Step 3: Write the failing lifecycle test**

`crates/rutster/tests/valkey_sink_lifecycle.rs` (complete file):

```rust
//! Lifecycle test for `ValkeyEventSink` (deploy slice C §5.6) against a
//! REAL Valkey service container (CI's `services: valkey:` block). Tests:
//! 1. A `SessionRegistered` event lands in the stream (XLEN increments).
//! 2. The stream is capped at the const MAXLEN (~10k events — tested
//!    implicitly by the streaming-style emit code; the cap holds).
//! 3. Killing the consumer task + emitting overflow events increments
//!    the `drops()` counter exactly by the overflow count.
//!
//! Requires `VALKEY_URL=redis://127.0.0.1:6379/` — set by the CI
//! `services: valkey:` block. Skipped locally when VALKEY_URL is unset
//! (so `cargo test --all` on a dev box without Valkey doesn't fail).

use std::time::{Duration, SystemTime};

use rutster::event_sink::{CallEvent, EventSink};
use rutster::valkey_sink::ValkeyEventSink;

fn valkey_url() -> Option<url::Url> {
    let raw = std::env::var("VALKEY_URL").ok()?;
    url::Url::parse(&raw).ok()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn lifecycle_event_lands_in_valkey_stream() {
    let Some(url) = valkey_url() else {
        eprintln!("VALKEY_URL unset; skipping Valkey-backed test (run with `services: valkey:` in CI)");
        return;
    };
    let rt = tokio::runtime::Handle::current();
    let sink = ValkeyEventSink::spawn(url.clone(), rt);

    let stream_key = format!("rutster:test:lifecycle:{}", uuid::Uuid::new_v4());
    sink.set_stream_key_for_test(&stream_key);

    let client = redis::Client::open(url.as_str()).unwrap();
    let mut conn = client
        .get_multiplexed_async_connection()
        .await
        .unwrap();
    let _: () = redis::AsyncCommands::del(&mut conn, &stream_key).await.unwrap();

    let id = rutster_call_model::ChannelId::new();
    let at = SystemTime::now();
    sink.emit(CallEvent::SessionRegistered { id, at });

    // Give the background task a moment to drain. mpsc::try_send + the
    // single XADD per event takes < 1ms locally; CI runners under load
    // can see 50ms+ — 2s is plenty of slack.
    tokio::time::sleep(Duration::from_secs(2)).await;

    let len: usize = redis::AsyncCommands::xlen(&mut conn, &stream_key).await.unwrap();
    assert!(
        len >= 1,
        "event should have landed in stream {stream_key} (len={len})"
    );

    drop(sink);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn drop_counter_increments_when_consumer_is_dead() {
    let Some(url) = valkey_url() else {
        eprintln!("VALKEY_URL unset; skipping Valkey-backed test");
        return;
    };
    let rt = tokio::runtime::Handle::current();
    // Spawn the sink with a consumer that immediately fails every XADD.
    let sink = ValkeyEventSink::spawn_for_test_with_dead_consumer(url.clone(), rt);

    let initial_drops = sink.drops().unwrap_or(0);
    for _ in 0..5 {
        sink.emit(CallEvent::SessionRegistered {
            id: rutster_call_model::ChannelId::new(),
            at: SystemTime::now(),
        });
    }
    // The background task's queue is bounded at capacity 256; emitting
    // 5 events all hit the queue + the dead consumer rejects every XADD.
    // Wait for the background task to attempt all sends.
    tokio::time::sleep(Duration::from_secs(2)).await;
    let after_drops = sink.drops().unwrap_or(0);
    assert_eq!(
        after_drops,
        initial_drops + 5,
        "every event should have been counted-and-dropped against the dead consumer"
    );

    drop(sink);
}
```

- [ ] **Step 4: Run — expect compile failure (no `valkey_sink` module)**

```bash
cargo test -p rutster --test valkey_sink_lifecycle
```

Expected: `could not find `valkey_sink` in `rutster``.

- [ ] **Step 5: Implement `ValkeyEventSink`**

`crates/rutster/src/valkey_sink.rs` (complete file):

```rust
//! # valkey_sink — first Valkey consumer (deploy spec §5.6)
//!
//! Implements the existing `EventSink` trait (deploy-C §5.6): `XADD` of
//! `CallEvent`s as JSON to a capped stream (`XADD MAXLEN ~ 10000`).
//! STRICTLY fire-and-forget: bounded `mpsc` + background task. Valkey
//! down ⇒ count-and-drop via `Arc<AtomicU64>` (the same counter
//! `/metrics` reads in deploy-C Task 6). `RUTSTER_VALKEY_URL` unset ⇒
//! `TracingEventSink` unchanged (main.rs makes the swap).
//!
//! ## Why `XADD MAXLEN ~ 10000` (the adopted default)?
//!
//! ~10k events ≈ a few MB of memory under normal op (a CallEvent
//! serializes to ~200 bytes of JSON; 10k × 200B ≈ 2MB). The forensic
//! window covers days of calls for a small fleet — enough to
//! post-mortem a node's behavior after the operator notices an issue.
//! The cap is `~` (approximate) which lets Valkey trim lazily — lower
//! write-path latency than exact `MAXLEN 10000`. Bumping or dropping
//! the cap is a one-line code change + redeploy; **no operator action
//! on Valkey** (the cap is enforced on the XADD side, not by a separate
//! trim task). Reversible.
//!
//! ## Why NOT the billing ledger?
//!
//! ADR-0005 source-of-truth rule: the bus is NEVER billing source-of-
//! truth. This sink is evidence preservation on a capped stream —
//! explicitly NOT the durable CDR pipeline. CDR → object storage stays
//! deferred (deploy spec §1.2 out-of-scope table).

use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex};

use tokio::sync::mpsc;

use crate::event_sink::{CallEvent, EventSink, SharedEventSink};

/// Stream key on Valkey. Constant for the slice; a future slice makes
/// it configurable when the operator wants a separate stream per
/// tenant (multi-tenancy ADR defers the schema).
pub const STREAM_KEY: &str = "rutster:events";

/// XADD MAXLEN cap. ~10k events ≈ a few MB; reversible code constant
/// (see module docs for the justification).
pub const MAXLEN_APPROX: usize = 10_000;

/// Bounded queue capacity. 256 is large enough to absorb a burst of
/// lifecycle events from a graceful-shutdown mass-hangup (~64 sessions
/// × 3 events = ~200) without dropping; small enough to surface a
/// sustained Valkey outage quickly via the `/metrics` drop counter.
const QUEUE_CAPACITY: usize = 256;

/// Valkey-backed EventSink. Cloneable (the inner is all `Arc`) so
/// `main.rs` can hand clones to the media thread + the /metrics reader.
#[derive(Clone)]
pub struct ValkeyEventSink {
    tx: mpsc::Sender<CallEvent>,
    drops: Arc<AtomicU64>,
    stream_key: Arc<Mutex<String>>,
}

impl ValkeyEventSink {
    /// Spawn the background XADD consumer task + return a `SharedEventSink`
    /// handle. The caller (main.rs) stores this in `MediaThreadOpts.sink`,
    /// replacing the default `TracingEventSink`.
    pub fn spawn(url: url::Url, tokio_handle: tokio::runtime::Handle) -> SharedEventSink {
        let (tx, rx) = mpsc::channel::<CallEvent>(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let stream_key = Arc::new(Mutex::new(STREAM_KEY.to_string()));
        let task_state = ConsumerTaskState {
            rx: Some(rx),
            url,
            drops: drops.clone(),
            stream_key: stream_key.clone(),
        };
        tokio_handle.spawn(background_consumer(task_state));
        Arc::new(ValkeyEventSink {
            tx,
            drops,
            stream_key,
        })
    }

    /// Test-only constructor that spawns a background task with a consumer
    /// that immediately fails every XADD. Used by the drop-counter test.
    #[doc(hidden)]
    pub fn spawn_for_test_with_dead_consumer(
        url: url::Url,
        tokio_handle: tokio::runtime::Handle,
    ) -> ValkeyEventSink {
        let (tx, rx) = mpsc::channel::<CallEvent>(QUEUE_CAPACITY);
        let drops = Arc::new(AtomicU64::new(0));
        let stream_key = Arc::new(Mutex::new(STREAM_KEY.to_string()));
        let task_state = ConsumerTaskState {
            rx: Some(rx),
            url,
            drops: drops.clone(),
            stream_key: stream_key.clone(),
        };
        tokio_handle.spawn(background_consumer_with_dead_xadd(task_state));
        ValkeyEventSink {
            tx,
            drops,
            stream_key,
        }
    }

    /// Test-only: override the stream key so concurrent test legs don't
    /// collide on the default `rutster:events` key (tests run in parallel).
    #[doc(hidden)]
    pub fn set_stream_key_for_test(&self, key: &str) {
        *self.stream_key.lock().unwrap() = key.to_string();
    }
}

impl EventSink for ValkeyEventSink {
    fn emit(&self, event: CallEvent) {
        // try_send, NOT send(). The media thread must NEVER block on
        // Valkey I/O. A full queue = count + drop. The drop counter
        // surfaces in /metrics so the operator's alerting catches a
        // sustained Valkey outage.
        if let Err(mpsc::error::TrySendError::Full(_)) = self.tx.try_send(event) {
            self.drops.fetch_add(1, Ordering::Relaxed);
        }
        // TrySendError::Disconnected means the background task has died.
        // Implicit drop-and-count: the event vanished (we leave the
        // counter unincremented because "the sink's task died" is a
        // different failure mode from "queue full" — operators see the
        // task-death via the metrics dropping to zero throughput, not
        // via the `drops` counter). If we wanted to count both, swap
        // `if let Err(...) => self.drops.fetch_add(1, ...)` here.
    }

    fn drops(&self) -> Option<u64> {
        Some(self.drops.load(Ordering::Relaxed))
    }
}

/// State carried into the background consumer task. `rx: Option<...>`
/// so the task can take ownership of the `mpsc::Receiver` without
/// requiring `ValkeyEventSink` to also be `Send + 'static` for the
/// `tokio::spawn` boundary (the `url`, `drops`, `stream_key` are
/// `Send + Sync`; `rx` itself is `Send` but not `Sync`).
struct ConsumerTaskState {
    rx: Option<mpsc::Receiver<CallEvent>>,
    url: url::Url,
    drops: Arc<AtomicU64>,
    stream_key: Arc<Mutex<String>>,
}

/// Background consumer: drains the mpsc, serializes each event as JSON,
/// XADDs to the (test-overridable) `stream_key` with `MAXLEN ~ MAXLEN_APPROX`.
/// On XADD failure (Valkey down), counts-and-drops via `drops`.
async fn background_consumer(mut state: ConsumerTaskState) {
    let mut rx = state.rx.take().expect("rx present");
    let client = match redis::Client::open(state.url.as_str()) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "Valkey client open failed; sink will count-and-drop every event");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    let conn = match client.get_multiplexed_async_connection().await {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "Valkey connect failed; sink will count-and-drop every event");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    let mut conn = conn;

    while let Some(event) = rx.recv().await {
        let payload = match serde_json::to_string(&EventPayload::from(&event)) {
            Ok(s) => s,
            Err(e) => {
                tracing::error!(error = %e, "CallEvent JSON serialize failed; dropping");
                state.drops.fetch_add(1, Ordering::Relaxed);
                continue;
            }
        };
        let key = state.stream_key.lock().unwrap().clone();
        // XADD <key> MAXLEN ~ <n> * event <payload>.
        // `~` = approximate cap (lazy trim; lower write latency than exact).
        let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen(
            &mut conn,
            &key,
            false, // approximate ~= MAXLEN ~
            MAXLEN_APPROX,
            "*",
            &[("event", payload)],
        )
        .await;
        if let Err(e) = result {
            tracing::warn!(error = %e, "XADD failed; dropping event (counter increments in /metrics)");
            state.drops.fetch_add(1, Ordering::Relaxed);
        }
    }
}

/// Variant of `background_consumer` whose XADDs always fail — used by
/// `spawn_for_test_with_dead_consumer`. The connection is opened but
/// every XADD targets a non-existent / refused endpoint. We achieve
/// "always fails" by using a URL with an impossible port.
async fn background_consumer_with_dead_xadd(mut state: ConsumerTaskState) {
    let mut rx = state.rx.take().expect("rx present");
    // Override the URL with one that fails every connection — the test
    // asserts the drop counter, so we don't care about the connection
    // error type, only that it fails fast enough for the test's 2s wait.
    let dead_url = url::Url::parse("redis://127.0.0.1:1/").unwrap();
    let client = match redis::Client::open(dead_url.as_str()) {
        Ok(c) => c,
        Err(e) => {
            tracing::error!(error = %e, "dead-consumer client open failed");
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    // Try to connect — expected to fail. If it succeeds (a Valkey
    // happens to be on port 1?), every XADD will fail anyway because
    // we'll immediately drop the connection in the loop.
    let conn_result = client.get_multiplexed_async_connection().await;
    let mut conn = match conn_result {
        Ok(c) => c,
        Err(_e) => {
            drain_and_count(&mut rx, &state.drops).await;
            return;
        }
    };
    while let Some(event) = rx.recv().await {
        let _ = event;
        let key = state.stream_key.lock().unwrap().clone();
        let result: redis::RedisResult<()> = redis::AsyncCommands::xadd_maxlen(
            &mut conn,
            &key,
            false,
            MAXLEN_APPROX,
            "*",
            &[("event", "{}")],
        )
        .await;
        if result.is_err() {
            state.drops.fetch_add(1, Ordering::Relaxed);
        }
    }
}

async fn drain_and_count(rx: &mut mpsc::Receiver<CallEvent>, drops: &Arc<AtomicU64>) {
    while let Some(_event) = rx.recv().await {
        drops.fetch_add(1, Ordering::Relaxed);
    }
}

/// JSON-serializable shape of a CallEvent. Carries a `kind` tag + the
/// event's fields so a downstream consumer can route without having
/// the rutster crate linked. `serde::Serialize` is derived — the
/// existing `CallEvent` enum doesn't derive Serialize (it's Debug +
/// Clone only); this wrapper does the conversion in `From<&CallEvent>`.
#[derive(Debug, serde::Serialize)]
struct EventPayload {
    kind: &'static str,
    id: String,
    at_ms: u128,
    // Extra fields per variant land in follow-ups; for now the kind +
    // channel id + timestamp are enough for the forensic window. The
    // shape is intentionally EXTENSIBLE — a future slice adds
    // `SessionEnded.reason` and `tap_metrics` here when the consumer
    // cares.
}

impl From<&CallEvent> for EventPayload {
    fn from(event: &CallEvent) -> Self {
        match event {
            CallEvent::SessionRegistered { id, at } => EventPayload {
                kind: "session_registered",
                id: id.0.to_string(),
                at_ms: at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
            CallEvent::SessionConnected { id, at } => EventPayload {
                kind: "session_connected",
                id: id.0.to_string(),
                at_ms: at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
            CallEvent::SessionEnded { id, ended_at, .. } => EventPayload {
                kind: "session_ended",
                id: id.0.to_string(),
                at_ms: ended_at.duration_since(std::time::UNIX_EPOCH).map(|d| d.as_millis()).unwrap_or(0),
            },
        }
    }
}
```

- [ ] **Step 6: Add the CI `services: valkey:` block**

In `.github/workflows/ci.yml`, the `test` job currently runs `cargo test --all`. To run the new
`valkey_sink_lifecycle` test against a real Valkey, add a `services:` block running
`valkey/valkey:latest` and export `VALKEY_URL` for the test command. Modify the `test` job:

```yaml
  test:
    runs-on: ubuntu-latest
    services:
      # Deploy slice C §5.6: real Valkey service container for the
      # valkey_sink_lifecycle integration test. The test skips itself
      # when VALKEY_URL is unset, so dev boxes without Valkey don't fail.
      valkey:
        image: valkey/valkey:latest
        ports:
          - 6379:6379
        options: >-
          --health-cmd "valkey-cli ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 3
    strategy:
      matrix:
        toolchain: [stable, "1.85"]
    env:
      VALKEY_URL: redis://127.0.0.1:6379/
    steps:
      - uses: actions/checkout@v4

      # Seam gate (slice-4 §7 #3; re-pinned slice-5):
      #
      # `loop_driver.rs` stays byte-identical to slice-3 — the hot-path
      # poll loop is untouched. `rtc_session.rs` was re-pinned in slice-5
      # ... [existing comment preserved]
      - name: Seam gate — loop_driver frozen (slice-3) + rtc_session pinned (slice-5)
        run: |
          EXPECTED_LOOP_DRIVER='744bf314edf7f4925c8bb3bd0f5176dbc88f8113'
          EXPECTED_RTC_SESSION='f47d63b9a2883d37066a93c9daa0e2cf8816bec4'
          GOT_LOOP_DRIVER=$(git hash-object crates/rutster-media/src/loop_driver.rs)
          GOT_RTC_SESSION=$(git hash-object crates/rutster-media/src/rtc_session.rs)
          if [ "$GOT_LOOP_DRIVER" != "$EXPECTED_LOOP_DRIVER" ]; then
            echo "::error::loop_driver.rs blob mismatch: $GOT_LOOP_DRIVER != $EXPECTED_LOOP_DRIVER"
            exit 1
          fi
          if [ "$GOT_RTC_SESSION" != "$EXPECTED_RTC_SESSION" ]; then
            echo "::error::rtc_session.rs blob mismatch: $GOT_RTC_SESSION != $EXPECTED_RTC_SESSION"
            exit 1
          fi

      - uses: dtolnay/rust-toolchain@master
        with:
          toolchain: ${{ matrix.toolchain }}
      - name: Install libopus (media crate FFI dep)
        run: apt-get update && apt-get install -y libopus-dev
      - uses: Swatinem/rust-cache@v2
      - run: cargo test --all
```

Apply the same `services: valkey:` block + `VALKEY_URL` env to the `clippy` job (so clippy can
type-check the test code under `--all-targets`, which includes the new
`valkey_sink_lifecycle.rs`). The `clippy` job is small; add only the `services:` block and the
`env:` block, no other changes.

MUST add `uuid` to `crates/rutster/Cargo.toml` `[dev-dependencies]` if not already present
(the `valkey_sink_lifecycle.rs` test uses `uuid::Uuid::new_v4()` for the per-test stream key;
verified `uuid` is already a workspace dep but may not be in the rutster crate's dev-deps —
spot-check via `cargo tree -p rutster`):

```toml
uuid = { workspace = true }
```

(If `uuid` is already in dev-deps, skip this addition — clippy will surface the unused-import
warning if not.)

- [ ] **Step 7: Wire `main.rs` to swap TracingEventSink → ValkeyEventSink**

In `crates/rutster/src/main.rs`'s `MediaThreadOpts` construction (currently lines 47–53), replace
the default `..Default::default()` sink line:

```rust
    // Deploy slice C §5.6: swap TracingEventSink -> ValkeyEventSink
    // when RUTSTER_VALKEY_URL is set. Strictly fire-and-forget:
    // mpsc::try_send + background XADD task + count-and-drop (counter
    // surfaces in /metrics). Unset -> today's TracingEventSink.
    let valkey_url = rutster::config::valkey_url(std::env::var("RUTSTER_VALKEY_URL").ok())
        .expect("RUTSTER_VALKEY_URL must be redis:// or rediss:// when set");
    let sink: rutster::event_sink::SharedEventSink = match valkey_url {
        Some(url) => {
            info!(%url, "ValkeyEventSink enabled");
            rutster::valkey_sink::ValkeyEventSink::spawn(url, tokio::runtime::Handle::current())
        }
        None => {
            info!("TracingEventSink (default; set RUTSTER_VALKEY_URL for durable events)");
            std::sync::Arc::new(rutster::event_sink::TracingEventSink)
        }
    };
    let opts = rutster::media_thread::MediaThreadOpts {
        media_cfg,
        max_sessions: rutster::config::max_sessions(std::env::var("RUTSTER_MAX_SESSIONS").ok())
            .expect("RUTSTER_MAX_SESSIONS must be a number"),
        sink,
    };
```

(Concretely: replace the `..Default::default()` in the existing `opts` construction with this
explicit `sink:` field.)

- [ ] **Step 8: Run tests — expect pass**

```bash
cargo fmt --all --check && cargo clippy --all --all-targets -- -D warnings
cargo test -p rutster config::tests
cargo test -p rutster --test valkey_sink_lifecycle   # skips locally when VALKEY_URL unset
cargo test --all
```

Expected: all green locally (the valkey test self-skips on `VALKEY_URL` unset). In CI with the
`services: valkey:` block, both lifecycle tests run against the real container and pass.

- [ ] **Step 9: Commit**

```bash
cd /home/alee/Sources/rutster
git add crates/rutster/Cargo.toml crates/rutster/src/valkey_sink.rs crates/rutster/src/config.rs crates/rutster/src/main.rs crates/rutster/tests/valkey_sink_lifecycle.rs .github/workflows/ci.yml
git commit -s -m "feat(valkey_sink): ValkeyEventSink — first Valkey consumer (deploy-C §5.6)

Implements the existing EventSink trait: XADD of CallEvents as JSON to a
capped stream (XADD MAXLEN ~ 10000). Strictly fire-and-forget: bounded
mpsc (cap 256) + background tokio::spawn task. Valkey down =>
count-and-drop via Arc<AtomicU64> (the same counter /metrics reads in
Task 6 as rutster_event_sink_drops_total). RUTSTER_VALKEY_URL unset =>
existing TracingEventSink unchanged (main.rs swaps).

XADD MAXLEN ~ 10000 is the adopted default: ~10k events ≈ 2MB of memory;
days-of-calls forensic window; bounded recovery time if Valkey is
restarted with a backlog. Reversible — it is a const in the module, not a
Valkey-side migration. ADR-0005 source-of-truth rule restated in the
module docs: this is evidence preservation, NOT the billing ledger.

CI gains a services: valkey: block on the test + clippy jobs running
valkey/valkey:latest on 127.0.0.1:6379; VALKEY_URL env exported for the
test command. Integration tests self-skip locally when VALKEY_URL is
unset so dev boxes without Valkey don't fail.

config.rs gains valkey_url parser (redis:// + rediss:// schemes rejected
otherwise; follows the house fail-fast pattern). main.rs swaps the sink
in MediaThreadopts."
```

---

### Task 9: Extend plan B's smoke harness with the lifecycle-event-lands-in-Valkey-stream check

**Files:**
- Modify: `docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md` (the smoke harness
  section — plan B's smoke CI's named hook is referenced; if plan B does not exist yet at
  execution time, this task is a NO-OP until plan B lands and its smoke harness is in place —
  see dependency note below).

**Dependency note:** at the time this plan was written, plan B
([`docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md`](2026-07-05-deploy-b-packaging-ci.md))
does NOT exist on disk (verified by glob). The smoke harness hook — `assert-lifecycle-event-lands-in-Valkey-stream` —
is what slice B's smoke CI is supposed to leave as a named hook for slice E to fill in (per the
spec's §11 sequencing: E depends on B "for the smoke harness shell," though Phase E itself has
no code dependency on B). If plan B has not landed yet by the time Task 9 is reached, this task
defers; the integration test `crates/rutster/tests/valkey_sink_lifecycle.rs` from Task 8
already proves the lifecycle-event-XADD round-trip end-to-end against a real Valkey service
container, so the slice's *acceptance bar* does not regress in the meantime.

- [ ] **Step 1: Verify plan B exists + find the smoke harness hook**

```bash
cd /home/alee/Sources/rutster
ls docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md 2>/dev/null
grep -n 'assert-lifecycle-event-lands-in-Valkey-stream' \
  docs/superpowers/plans/2026-07-05-deploy-b-packaging-ci.md 2>/dev/null
```

If the file or the hook line is absent, STOP — this task is a NO-OP. Document the deferral in
the PR description for this slice: "Task 9 deferred — plan B's smoke harness hook
`assert-lifecycle-event-lands-in-Valkey-stream` is not on disk; the slice's acceptance bar is
met by `crates/rutster/tests/valkey_sink_lifecycle.rs` running against the real Valkey
service container."

If the file IS present and contains the hook line, continue to Step 2.

- [ ] **Step 2: Implement the hook in plan B's smoke harness**

Add the assertion to plan B's smoke harness at the named hook. The assertion: after the
all-in-one smoke simulates a session lifecycle (register + delete), the RutsterEventSink's
Valkey stream has grown by ≥ 2 entries (one for `SessionRegistered`, one for `SessionEnded`).
Use the same `redis::AsyncCommands::xlen` query as Task 8's integration test, against the
default `STREAM_KEY = "rutster:events"` (the all-in-one container ships Valkey).

- [ ] **Step 3: Commit**

(Details vary by plan B's smoke harness structure at execution time — name the change "feat
(smoke): assert-lifecycle-event-lands-in-Valkey-stream hook (deploy-C §5.6, deploy-B smoke
harness)" and follow plan B's commit-message house style.)

---

### Task 10: Final verification sweep

No code. Evidence before assertions — run every gate the CI runs, plus the seam-gate check,
before calling the slice done.

- [ ] **Step 1: Full gate run**

```bash
cargo fmt --all --check
cargo clippy --all --all-targets -- -D warnings
cargo clippy --all --all-targets --features=sim-bench -- -D warnings
cargo test --all
cargo test --all --features=sim-bench -- --test-threads=1
cargo deny check
cargo doc --no-deps
```

Expected: every command exits 0.

- [ ] **Step 2: Seam gate verification**

```bash
git hash-object crates/rutster-media/src/loop_driver.rs
git hash-object crates/rutster-media/src/rtc_session.rs
```

Expected output, exactly:

```
744bf314edf7f4925c8bb3bd0f5176dbc88f8113
f47d63b9a2883d37066a93c9daa0e2cf8816bec4
```

If either hash differs, STOP — a task touched a seam-gated file; restore with
`git checkout main -- crates/rutster-media/src/loop_driver.rs crates/rutster-media/src/rtc_session.rs`
and re-run the gates.

- [ ] **Step 3: TLS smoke (manual)**

Generate a self-signed cert + key locally, run the binary with both env vars set, and verify
the HTTPS path round-trips a `curl -k https://localhost:8080/healthz`. Then send `kill -HUP`
to the process after rewriting the cert to a different CN; verify in the process's logs that
the "TLS cert reloaded" line appears. (This is a manual smoke — the automated tests cover the
contract; the manual smoke covers the operator UX.)

- [ ] **Step 4: MSRV spot-check (matches the CI 1.85 leg)**

```bash
cargo +1.85 test --all
```

Expected: green. (If the 1.85 toolchain is not installed locally, note it and rely on the CI
matrix — do not skip silently.)

---

## Final acceptance checklist

After all tasks land:

- [ ] `cargo fmt --check`, `clippy -D warnings` (default + `--features=sim-bench`),
      `cargo test --all`, sim-bench sweep (`--test-threads=1`), `cargo deny check`,
      `cargo doc --no-deps` all clean.
- [ ] Seam gate: `loop_driver.rs` + `rtc_session.rs` blob hashes unchanged (Task 10 Step 2).
- [ ] `cargo deny check` stays green after adding `axum-server 0.8` (Task 1) and `redis 1.3.0`
      (Task 7) — license + multiple-versions gates both pass.
- [ ] TLS listener: BOTH `RUTSTER_TLS_CERT` + `RUTSTER_TLS_KEY` set ⇒ `serve_tls_with_nodelay`
      with `axum_server::bind` + `NodeLayAcceptor` (TCP_NODELAY parity with plan A's plaintext
      path) + SIGHUP hot-reload watcher. Either unset ⇒ plan A's plaintext
      `serve_with_nodelay` (artifact default). Partial config panics at boot with a clear error.
- [ ] TLS hot-reload contract: `RustlsConfig::reload_from_pem_file` is atomic (arc-swap) —
      `crates/rutster/tests/serve_tls_hot_reload.rs` proves it with peer_certificates differing
      across a reload boundary.
- [ ] `/metrics` endpoint on `RUTSTER_METRICS_BIND` (default `127.0.0.1:9090`), on its OWN
      `axum::Router` — never on the main `router()`. Metrics named:
      `rutster_sessions_active` (gauge), `rutster_max_sessions` (gauge), `rutster_draining`
      (gauge), `rutster_tick_overruns_total` (counter), `rutster_last_tick_micros` (gauge),
      `rutster_admission_rejects_total` (counter), `rutster_event_sink_drops_total` (counter —
      absent when TracingEventSink; present only when ValkeyEventSink is wired).
- [ ] `ValkeyEventSink`: bounded mpsc + background XADD task; `RUTSTER_VALKEY_URL` set ⇒
      replaced `TracingEventSink`; unset ⇒ `TracingEventSink` unchanged. Drop counter
      (`Arc<AtomicU64>`) surfaces in `/metrics` as `rutster_event_sink_drops_total`.
      `XADD MAXLEN ~ 10000` (the `const` in `valkey_sink.rs`) caps the stream; reversible
      code constant, NOT a Valkey migration.
- [ ] CI: `services: valkey:` block on the `test` + `clippy` jobs running
      `valkey/valkey:latest` on `127.0.0.1:6379`; `VALKEY_URL` exported. The
      `valkey_sink_lifecycle.rs` test runs against the real container in CI; self-skips
      locally when `VALKEY_URL` is unset.
- [ ] PR opened via `tea pulls create --head deploy-c/binary-features --base main --title
      "deploy slice C (binary features): rustls Phase 1 TLS + /metrics + ValkeyEventSink"`
      with this plan's §5.4 + §5.5 + §5.6 as the description skeleton. Sequencing per spec §11:
      C, D, E are order-independent; this slice merges in any order relative to slices A and B
      (modulo the plan-B smoke harness hook noted in Task 9).

