fix(beacon): split session failures by status code (#2815)

## Description

The session beacon reports `failures` as a single count, incremented
whenever a turn ends `>= 500` (`headroom/telemetry/session.py`). Across
the current corpus that reads **3,969 failures on 595,445 turns
(0.67%)** — and the number cannot answer the only question anyone asks
of it: an Anthropic `529` is the provider shedding load and there is
nothing to fix; a `500` is usually ours. Today the two are
indistinguishable, so diagnosis falls back to inference from time-of-day
curves and per-install concentration.

This counts the status alongside the total.

```json
"failures": 3,
"failure_statuses": {"529": 2, "500": 1}
```

Motivating investigation on the live corpus (0.67% of turns, 6% of
sessions, 63% of all failures from 48 installs, a 2.5% plateau at 08–11
UTC decaying to 0.03% during the fleet's busiest hour) strongly suggests
provider-side 529 after retry exhaustion — but "strongly suggests" is
exactly the gap this field closes.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- **`headroom/telemetry/session.py`** — `_Session.failure_statuses`,
incremented next to `failures` in `record_outcome`. Keys are the bare
status string for the 5xx range, `"other"` beyond it. Emitted as a
sibling of `failures` in `payload()`.
- **`deploy/beacon/worker.js`** — `failure_statuses` added to
`ALLOWED_KEYS`. Without this the ingest allowlist silently drops it.
- **`deploy/beacon/sample-event.json`** — sample carries the new key in
OTLP `kvlistValue` form.

### Why no slug bounding

`skips` runs values through `_safe_slug` because they arrive as free
strings. A status code is an `int` the proxy itself produced; the `500
<= status < 600` check is what keeps a garbage value from inventing map
keys. Nothing here is user-derived, so the field stays content-free.

### Why `schema_version` stays 1

Additive, matching the precedent set by #2796, which added
`tokens.tool_saved` and the two `all_layers_*` rates without a bump.
Bumping signals a break to consumers when nothing about older rows
becomes invalid.

## Testing

- [x] Unit tests pass (`pytest`) — the module's own self-check, extended
- [x] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`) — see note
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ python -m headroom.telemetry.session
ok

$ ruff check headroom/telemetry/session.py
All checks passed!

$ ruff format --check headroom/telemetry/session.py
1 file already formatted

$ mypy --python-version 3.12 headroom/telemetry/session.py
Success: no issues found in 1 source file
# --python-version 3.12 only to skip a pre-existing numpy-stub syntax error the
# repo's python_version = "3.10" triggers locally; unrelated to this diff.

$ node --check deploy/beacon/worker.js     # ok
$ python -c "import json; json.load(open('deploy/beacon/sample-event.json'))"   # parses
```

The self-check in `headroom/telemetry/session.py` now records two 529s
and one 500 and asserts both the total and the split:

```python
assert emitted[-1]["failures"] == 3
assert emitted[-1]["failure_statuses"] == {"529": 2, "500": 1}
```

plus `assert event["failure_statuses"] == {}` on the clean-session path.

## Real Behavior Proof

- **Environment:** macOS 25.4.0, Python 3.12 venv, this branch.
- **Exact command / steps:** drive `SessionAggregator` with three
failing outcomes and encode the payload through the same `_any_value`
the wire uses.

```text
payload: 3 {'529': 2, '500': 1}
otlp   : {"kvlistValue": {"values": [{"key": "529", "value": {"intValue": "2"}},
                                     {"key": "500", "value": {"intValue": "1"}}]}}
```

The OTLP form matches `deploy/beacon/sample-event.json` byte-for-byte in
shape, and `unwrap()` in `worker.js` turns `kvlistValue` back into a
plain object, so it lands in R2 as `{"529": 2, "500": 1}` — the same
shape as `skips`, which DuckDB reads as `MAP(VARCHAR, BIGINT)`.

- **Observed result:** as above. Verified against the live corpus that
schema evolution here is already routine — 3,836 of 3,884 existing rows
have `rates.all_layers_saved_pct = NULL` from #2796 landing mid-corpus,
and every report still runs.
- **Not tested:** the deployed Worker (no staging R2 binding locally);
`node --check` covers syntax only. The allowlist addition is one array
entry consumed by the existing `pick()`.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`

## Screenshots (if applicable)

N/A — wire-format change, covered by the output above.

## Additional Notes

**Deploy order matters.** The Worker allowlist drops unknown keys, so
`deploy/beacon/worker.js` must be deployed *before* a client release
that emits the field — otherwise it is discarded at the door. No
corruption either way, just missing data until the Worker catches up.

**Old data is unaffected.** R2 objects are immutable NDJSON written per
request; nothing rewrites history. The corpus reader already passes
`union_by_name = true`, which fills the column with NULL for rows
written before this ships.

deploy/beacon/sample-event.json

@@ -284,6 +284,21 @@
                       "value": {
                         "intValue": "2"
                       }
+                    },
+                    {
+                      "key": "failure_statuses",
+                      "value": {
+                        "kvlistValue": {
+                          "values": [
+                            {
+                              "key": "529",
+                              "value": {
+                                "intValue": "2"
+                              }
+                            }
+                          ]
+                        }
+                      }
                     }
                   ]
                 }

deploy/beacon/worker.js

@@ -42,6 +42,7 @@ const ALLOWED_KEYS = [
   'providers',
   'models',
   'failures',
+  'failure_statuses',
 ];
 
 // Resource attributes we keep. Same rule: allowlist, not denylist.

headroom/telemetry/session.py

@@ -246,6 +246,7 @@ class _Session:
     cache_write_tokens: int = 0
     uncached_tokens: int = 0
     failures: int = 0
+    failure_statuses: dict[str, int] = field(default_factory=dict)
     passthrough_turns: int = 0
     response_cache_hits: int = 0
     overhead_ms: float = 0.0
@@ -392,6 +393,12 @@ class _Session:
             "providers": sorted(self.providers),
             "models": sorted(self.models),
             "failures": self.failures,
+            # The same failures split by status, because the count alone cannot
+            # answer the only question worth asking about it: a 529 is the
+            # provider shedding load (nothing to fix here) and a 500 is usually
+            # ours. Keys are the bare status string; the set is closed and tiny
+            # (500/502/503/504/529), so this needs no slug bounding.
+            "failure_statuses": dict(self.failure_statuses),
         }
         self.seq += 1
         return snapshot
@@ -513,8 +520,14 @@ def _fold(sess: _Session, outcome: Any, now: float, source: str = "proxy") -> No
     sess.uncached_tokens += int(get("uncached_input_tokens") or 0)
     sess.overhead_ms += float(get("overhead_ms", 0.0) or 0.0)
     sess.latency_ms += float(get("total_latency_ms", 0.0) or 0.0)
-    if int(get("status_code", 200) or 200) >= 500:
+    status = int(get("status_code", 200) or 200)
+    if status >= 500:
         sess.failures += 1
+        # ponytail: str(status) verbatim for the 5xx range, one bucket for
+        # anything outside it. Nothing here can be user data, and the range
+        # check is what keeps a garbage status_code from inventing map keys.
+        key = str(status) if status < 600 else "other"
+        sess.failure_statuses[key] = sess.failure_statuses.get(key, 0) + 1
     if get("from_response_cache", False):
         sess.response_cache_hits += 1
 
@@ -855,6 +868,7 @@ def demo() -> None:
     assert event["compression"]["transforms"] == {"crush": 2, "dedupe": 2}
     assert event["providers"] == ["anthropic"]
     assert event["failures"] == 0
+    assert event["failure_statuses"] == {}
 
     # The new burst is a distinct session, not a continuation.
     agg.flush_all()
@@ -976,10 +990,18 @@ def demo() -> None:
     class Failed(FakeOutcome):
         status_code = 529
 
+    class Broke(FakeOutcome):
+        status_code = 500
+
     agg2 = SessionAggregator(emitted.append)
     agg2.record(Failed(), now=2000.0)
+    agg2.record(Failed(), now=2001.0)
+    agg2.record(Broke(), now=2002.0)
     agg2.flush_all()
-    assert emitted[-1]["failures"] == 1
+    assert emitted[-1]["failures"] == 3
+    # Provider load-shedding and our own 500s have to be separable, or the
+    # count says "0.7% of turns failed" and nothing about whose fault it is.
+    assert emitted[-1]["failure_statuses"] == {"529": 2, "500": 1}
 
     # Flushing an empty aggregator is a no-op, not a null event.
     before = len(emitted)