feat(batch): 2-hour timeliness alarm for unpublished batch audio (#24364)

## Why

David, 2026-11-02 06:09Z: "We should be processing most audio within 30
mins of finalization, and alarm on 2 hours." Failed-held, quarantined
and repair waits must count, because that is how 165 segments were lost
on 11-01. The existing projection gauge reads 14.9 days of held backlog,
so any plain age alarm would fire forever on day one.

## What

The batch queue sample (`QueueMetrics.refresh`, FirestoreStore only)
exports, at most once per 5 minutes per process:

| series | meaning |
|---|---|
| `omi_batch_timeliness_new_oldest_age_seconds{leg}` | oldest
unpublished unit admitted at or after the epoch, all clocked from
admission: `uncommitted` (policy and vendor holds skipped),
`quarantined` (every recovery state), `assignment_held` (committed,
projection receipt quarantined), `projection`; `quarantine_held_rows`
visible only |
| `omi_batch_timeliness_new_over_sla_units{leg}` | new-cohort units past
2 h; also `terminal_empty` (visible only) |
| `omi_batch_timeliness_backlog_units{leg}` | units from before the
epoch: visible counts, never alarmed |
| `omi_batch_timeliness_scan_truncated{leg}` /
`omi_batch_timeliness_samples_total{outcome}` | blind-spot flag and
sample counter |

- **Kill switch:** `BATCH_TIMELINESS_ALARM_ENABLED`, default on,
registered as `ops_kill`.
- **Epoch:** `BATCH_TIMELINESS_EPOCH`, default `2026-11-02T06:09:00Z`,
when the target was set.
- **Cadence:** `BATCH_TIMELINESS_INTERVAL_SECONDS`, default 300.
- **Terminal-empty switch:** `TERMINAL_EMPTY_IS_PROCESSED` in
`config/batch_timeliness.py`, **default `True`** (see below).
- **Alert-only:** no scheduler, ramp, gate or hard-3 rollback trigger
reads these series.
- **Cost:** under 300 reads per sample per process, at most about
$0.4–0.8/day across 16 workers.
- **Indexes:** two new jobs composites, declared in
`database/batch_pipeline_indexes.py` with `firestore.indexes.json`
regenerated: `(state, recovery_state, enqueued_at)` for
`assignment_held`, and `(state, recovery_state, recovery.failure.reason,
measured_held_vad_speech_seconds, enqueued_at)` for `terminal_empty`.
Until both are READY the sample counts `error` and the stale rule fires.
- **Rules:** three agent-tier Grafana rules sit in
`charts/monitoring/pending/batch-timeliness.json`, because the
monitoring contracts reject rules on series that don't exist in prod
yet:
- `omi-batch-unit-unpublished-2h`: max new age over the uncommitted,
quarantined, assignment_held and projection legs > 7200 s, with a fresh
sample.
  - `omi-batch-timeliness-sample-stale`
- `omi-batch-timeliness-scan-blind`: any leg's walk full for 30 minutes.
- **Apply step:** in `charts/monitoring/batch-projection/README.md`.

## Terminal empty_asr counts as processed (David, 2026-11-02 10:45Z)

David: "finished recording counts as processed". The lead ruled the
one-line switch `TERMINAL_EMPTY_IS_PROCESSED` defaults that way, alarm
only, with no speech threshold.
- **Default `True`:** units that are `failed_held` with recovery failure
`empty_asr` and measured held VAD speech exactly 0 leave the
`quarantined` leg. Its count subtracts them exactly, so a page of them
can hide a younger unit but never a breach.
- **`False`:** puts them back in the 2h alarm.
- Units with measured speech always stay in the alarm. `terminal_empty`
stays a visible count either way.
- "No recoverable speech" reuses the existing silence contract
(`held_silence_allowed`: census exactly 0).

Measured (ro-prod, 10:36Z, counts only): no quarantined `empty_asr`
failed-held unit had zero measured speech. All 13 admitted since the
epoch carry some: 10 under 2 s, 2 at 2–10 s, 1 over 10 s. Of 530 backlog
units, 137 have no VAD evidence at all. So the default changes nothing
in today's alarm.

## Measured (ro-prod, read-only, 2026-11-02)

- Day-one values at the 06:09Z epoch: new breaches 0. Backlog:
quarantined 609, sibling_held 16,921, uncommitted including holds 523,
projection 7.
- Fire test, the PR's real code against prod (`fire_test.py`): epoch
03:00Z FIRES ($A = 18,155 s); epoch 06:09Z is clear.
- Failed-held arrives at about 130/day (115 `empty_asr`), so once
deployed the alarm keeps firing until those are resolved within 2 h.
That is the intended signal.

## Round 1 fixes (Luna FIX-FIRST, lead ruling 2cef45)

- The projection leg clocks from the job's `enqueued_at` (oldest 50
pending results, then job reads), not from commit.
- Live vendor holds are walked apart from active work. Truncation is a
per-leg gauge.
- New `assignment_held` leg, the real holes per siblingrelease.
- Read ceiling corrected: under 300 reads per sample.
- The sample comes from `store.client` instead of a new store method,
which fixes `test_jobstore_protocol_parity`.

## Round 2 delta (lead accepted 5e44c47 without a round 3, 11:42Z)

- Terminal-empty switch and `terminal_empty` count, above.
- A walk ordered by a clock other than admission (ready by
`available_at`, retry by `retry_started_at`) reports truncation on a
full page even when it found admitted rows, since an older admission can
sit behind it. The blind rule now reads every leg.
- Switch default `True` per David's ruling.
- Merged origin/main, including the main-red fix #24378 (shard 4/6
goldens).

## Tests

- Red first: the new tests fail without the switch (ImportError of
`TERMINAL_EMPTY_IS_PROCESSED`).
- Green through `backend/test.sh`: `test_batch_pipeline_timeliness.py`
(15), `test_batch_pipeline_metrics.py`, `test_entitlement_versions.py`,
`test_storefront_presentation.py`, and the Firestore index/query-shape
guard (`test_firestore_query_shapes`, `test_firestore_indexes`,
`test_firestore_index_config`, `test_firestore_index_selector_contract`,
`test_firestore_query_contract`).

## Deploy

Batch worker image (lead's train), after the index reconcile. Rollback:
set `BATCH_TIMELINESS_ALARM_ENABLED=false`, or revert. Nothing else
reads the series.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Author: Git-on-my-level

backend/charts/monitoring/batch-projection/README.md

@@ -0,0 +1,72 @@
+# Batch projection and timeliness alarms
+
+`alerts.yaml` holds the projection-obligation PrometheusRule. Prod runs no Alertmanager, so that file is a record,
+not a live route. Live alerts are Grafana-managed rules in `../alerts/*.json`.
+
+## 2-hour timeliness alarm (David, 2026-11-02)
+
+Target: process most batch audio within 30 minutes of admission, and alarm when a unit is still unpublished after
+2 hours. Failed-held, quarantined and repair waits count, because that is how 165 segments were lost on 2026-11-01.
+
+The batch worker queue sample exports these series (`utils/batch_pipeline/metrics.py`, at most once per 5 minutes
+per process):
+
+| series | meaning |
+|---|---|
+| `omi_batch_timeliness_new_oldest_age_seconds{leg}` | oldest unpublished unit admitted at or after the epoch. `uncommitted`, `quarantined` and `assignment_held` clock from `enqueued_at`; `projection` and `quarantine_held_rows` clock from `committed_at`, a lower bound |
+| `omi_batch_timeliness_new_over_sla_units{leg}` | units admitted at or after the epoch and unpublished past 2 h, per leg |
+| `omi_batch_timeliness_backlog_units{leg}` | unpublished units from before the epoch; visible, never alarmed |
+| `omi_batch_timeliness_scan_truncated{leg}` | 1 when a walk page is full and may hide an older admission (`uncommitted`, `projection`); the blind rule fires after 30 minutes, which is expected during large catch-up uploads |
+| `omi_batch_timeliness_samples_total{outcome}` | `ok`, `error`, `disabled` |
+
+Legs:
+- `uncommitted`: `ready`, `leased`, `result_ready`, `retry`, `vendor_held`, skipping deliberate admission holds
+  (allowance, budget, fair use, daily ceiling) and live vendor holds.
+- `quarantined`: every recovery state, including `failed_held` and `prepared`.
+- `assignment_held`: committed, but the projection receipt is quarantined (job `recovery_state=held`), so the text
+  never reached the conversation.
+- `projection`: committed, projection `pending`, `persisted` or `ready`.
+- `quarantine_held_rows`: result rows at `quarantine_held`. Counted, never alarmed: most are ready receipts whose text
+  is already in the conversation and only the content ledger waits; the rest are the assignment holds above.
+- `terminal_empty` (counts only): the quarantined units that are `failed_held` with recovery failure `empty_asr`
+  and measured held VAD speech exactly 0, the existing silence contract for "no recoverable speech".
+
+David, 2026-11-02 10:45Z: "finished recording counts as processed". So `TERMINAL_EMPTY_IS_PROCESSED` in
+`config/batch_timeliness.py` defaults to `True`: terminal empty units are subtracted exactly from the `quarantined`
+leg, and `terminal_empty` stays visible. `False` puts them back in the alarm. On prod at 2026-11-02 10:36Z, no
+quarantined `empty_asr` unit had zero measured speech: 13 of 13 admitted since the epoch carried some (10 under
+2 s), so today the default changes nothing. Units with measured speech always stay in the alarm.
+
+Config: `BATCH_TIMELINESS_ALARM_ENABLED` (default on; `false` stops the Firestore reads),
+`BATCH_TIMELINESS_EPOCH` (default `2026-11-02T06:09:00Z`, when the target was set) and
+`BATCH_TIMELINESS_INTERVAL_SECONDS` (default 300, clamped to 60–3600).
+
+Reads per sample, worst case: 8 bounded walks of 20 documents (160; 9 with the terminal-empty switch on), the projection walk and its job reads (100), 3
+single-document reads, and 17 count aggregations (one read per 1,000 index entries each, about 25 at today's backlog).
+That is under 300 reads per process per sample. Sixteen workers at the default cadence read at most about 1.4M per day
+(roughly $0.4–0.8/day); a healthy queue reads far less, because walks stop at the first unheld row. The
+`assignment_held` leg needs the new jobs `(state, recovery_state, enqueued_at)` composite index, and `terminal_empty`
+needs jobs `(state, recovery_state, recovery.failure.reason, measured_held_vad_speech_seconds, enqueued_at)`. Until
+both are READY, the whole sample fails and counts as `error`, and the stale rule fires.
+
+The metric is alert-only. No scheduler, ramp, hard-3 rollback trigger or gate reads it.
+
+### Apply (lead)
+
+The rules live in `../pending/batch-timeliness.json` until the series exist in prod, because the monitoring
+contracts reject rules that read metrics missing from `metric-inventory.prod.json`.
+
+1. Deploy a batch worker image that contains this change.
+2. Confirm in Prometheus that `max(omi_batch_timeliness_observed_at_seconds) > 0` and that
+   `omi_batch_timeliness_samples_total{outcome="ok"}` increases.
+3. Move the three rule objects from `../pending/batch-timeliness.json` into `../alerts/batch-timeliness.json` and append
+   them to `../alert-rules.json`. Refresh `metric-inventory.prod.json` with its documented generator, then run the
+   monitoring contract tests.
+4. Import with the documented Grafana rule provisioning path (`--apply --backup-file <new-file>`).
+
+All three rules are agent tier: Grafana keeps their state, and Telegram stays muted (`omi-tier-silent`). Change `tier`
+to `page` only if David wants the alarm on his phone.
+
+Rollback: set `BATCH_TIMELINESS_ALARM_ENABLED=false` on the worker (gauges freeze and the stale rule fires), or pause
+the three Grafana rules. The index step: merge triggers the Firestore index reconcile; approve its prod job and wait
+for READY before shipping the worker (merge_pr.sh waits out these gates).