fix(pipeline): age-gate ancient provider and fresh-enrichment sweep wakes

backend/config/flag_budget.json

@@ -24,8 +24,11 @@
   },
   "seed_review_by": "2026-11-15",
   "budget": 169,
-  "count": 248,
+  "count": 251,
   "names": {
+    "BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS": {"owner": "omi-pipeline", "review_by": "2026-11-30"},
+    "BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS": {"owner": "omi-pipeline", "review_by": "2026-11-30"},
+    "BATCH_PIPELINE_SWEEP_AGE_GATE": {"owner": "omi-pipeline", "review_by": "2026-11-30"},
     "PARAKEET_SHAPE_CANARY_FILE": {"owner": "two-lane lead", "review_by": "2026-12-04"},
     "PARAKEET_LID_MODEL": {"owner": "planscost", "review_by": "2026-11-30"},
     "PARAKEET_LID_ONLY": {"owner": "planscost", "review_by": "2026-11-30"},

backend/config/policy/sweep_age.py

@@ -0,0 +1,29 @@
+"""Live sweep admission limits; never alter durable obligations or initial enqueue."""
+
+import math
+import os
+
+DEFAULT_MAX_AGE_DAYS = 14.0
+
+
+def enabled() -> bool:
+    """Default on; only explicit off restores the previous sweep behavior."""
+    return os.getenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on').strip().lower() != 'off'
+
+
+def _days(raw: str) -> float:
+    try:
+        days = float(raw)
+    except ValueError:
+        return DEFAULT_MAX_AGE_DAYS
+    return days if math.isfinite(days) and 0 < days <= 3650 else DEFAULT_MAX_AGE_DAYS
+
+
+def max_age_seconds(queue: str) -> float:
+    if queue == 'provider-stt':
+        days = _days(os.getenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', '14'))
+    elif queue == 'enrichment-fresh':
+        days = _days(os.getenv('BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS', '14'))
+    else:
+        raise ValueError('unknown sweep age queue')
+    return days * 86400

backend/database/outbox_provider_stt.py

@@ -19,6 +19,7 @@
 from typing import Any, Callable, Literal, Mapping
 
 from database.outbox import Claim, EnqueueResult, OutboxKind, Outcome, SweepCounts
+from utils.observability.sweep_age import skip_ancient
 
 Row = dict[str, Any]
 _ROW_ID = re.compile('[0-9a-f]{64}')
@@ -320,6 +321,8 @@ def sweep(self, *, now: float, limit: int = 100) -> SweepCounts:
                     or now - float(row.get('woken_at') or 0) < REWAKE_AFTER_SECONDS
                 ):
                     continue
+                if skip_ancient(row, 'provider-stt', now=now):
+                    continue
                 if row.get('lease_token'):
                     expired += 1
 

backend/docs/feature-flag-registry.md

@@ -553,6 +553,7 @@ and an explicit empty literal renders as `''`.
 | `BATCH_PIPELINE_SECOND_MULTILINGUAL_ENABLED` | Ship-on Soniox second chance after valid-empty Modulate held recovery; false restores Modulate-only ordinary recovery; independent language-guard recovery retains its separate switch and cap | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SERVING_METRICS_MIN_REFRESH_SECONDS` | Default 30; each batch serving instance refreshes its queue gauges at most once per interval | backend | env | closed | — | — | 60 | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SHARED_EMBEDDING_TRUNK_ENABLED` | Default on unless false; the serial batch GPU chain runs the WeSpeaker embedding trunk once per 10 s chunk of a unit and pools it per local speaker, instead of once per (chunk, speaker) pair | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_SWEEP_AGE_GATE` | Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline |
 | `BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED` | Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_VAD_SEGMENTATION_REUSE_ENABLED` | Default on unless false; on the serial batch GPU chain a unit diarized at full length takes the raw segmentation scores its own VAD pass computed instead of a second segmentation forward | backend | env | open | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_WORKER_ENABLED` | Default-off dedicated GPU process switch; drains admitted work without enabling upload admission | backend | env | closed | false | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | false (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | — | keep | keep | — | dazheng |
@@ -785,11 +786,13 @@ and an explicit empty literal renders as `''`.
 | `BATCH_PIPELINE_ASR_LEFT_CONTEXT_UNITS` | Process-local predecessor tails kept for ASR left context (default 64, range [0, 1024]; each <=8 s int16, 10 min TTL, keyed by owner, privacy generation and capture stream: source, device and lock from the staged intake context, or the record conversation) | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_CONTEXT_SECONDS` | GPU chain internal ASR boundary context seconds; default 2; requires AMI qualification | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_DIARIZER` | GPU chain diarizer registry selection; default pyannote; unavailable candidates fail closed | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline |
 | `BATCH_PIPELINE_FLIGHT_BATCH_SECONDS` | Audio seconds packed per pyannote flight group; default 1800, range (0, 7200]; longer units take the serial path | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_INDEPENDENT_RECOVERY_MAX_SECONDS` | Default 60 seconds; lower the independent-language recovery per-unit duration cap, never exceed the hard 60-second ceiling | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_MODEL_BATCH_SIZE` | GPU chain maximum model window batch size; default 8; hard ceiling 32 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_PADDED_SECONDS` | GPU chain padded waveform seconds per model batch; default 272 (8 windows with context); shared-live hard cap 120 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_POOL_KIND` | Dedicated batch GPU by default; shared_live requires bounded units and the existing live GPU arbiter | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
+| `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` | Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 | backend | env | closed | — | — | — | — | keep | keep | — | omi-pipeline |
 | `BATCH_PIPELINE_PUBLISH_CONCURRENCY` | GPU worker post-flight publication concurrency: bounded fenced transitions and one ordered drain per owner per flight; default 4, 1 = legacy serial kill switch, max 16 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_REUSE_LIVE_ENABLED` | Default-off GPU-chain pre-STT reuse; requires existing lineage gates and full-span words-of-record quality proof | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
 | `BATCH_PIPELINE_SHADOW_LANGUAGE_PROBES_PER_HOUR` | Default and maximum 500 fleet-wide pair attempts per UTC hour; zero or malformed denies; minute pacing and two 120-second slots also apply | backend | env | closed | — | — | — | — | keep | keep | — | dazheng |
@@ -1074,6 +1077,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_DIARIZER` — GPU chain diarizer registry selection; default pyannote; unavailable candidates fail closed (fail: closed)
 - `BATCH_PIPELINE_DIARIZE_SPEECH_ONLY_ENABLED` — Default on unless false; batch GPU worker diarizes only padded VAD and word evidence and maps turns back to capture clocks (fail: open)
 - `BATCH_PIPELINE_EMPTY_ASR_RECOVERY_ENABLED` — Default off; hand speech-bearing empty ASR after padded retry to fenced per-segment recovery (fail: closed)
+- `BATCH_PIPELINE_ENRICHMENT_FRESH_SWEEP_MAX_AGE_DAYS` — Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 (fail: closed)
 - `BATCH_PIPELINE_FLIGHT_BATCHING_ENABLED` — Default off; pack pyannote segmentation (shared by VAD and diarization) and per-chunk embedding trunks across one GPU flight (fail: closed)
 - `BATCH_PIPELINE_FLIGHT_BATCH_SECONDS` — Audio seconds packed per pyannote flight group; default 1800, range (0, 7200]; longer units take the serial path (fail: closed)
 - `BATCH_PIPELINE_FRESH_GPU_PRIORITY_ENABLED` — Default on unless false; batch uploads in the 6 h fresh sync lane whose oldest capture is over 10 min old lease GPU at RECENT priority, after fresh uploads (fail: open)
@@ -1102,6 +1106,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_PROJECTION_PREDECESSOR_REPLAY_ENABLED` — Default on; a blocked projection successor schedules its owner's missing predecessor through the existing ordered walk (fail: open)
 - `BATCH_PIPELINE_PROJECTION_PREFETCH` — Default 4; an owner-chain walk reads the next N rows' successor links, job bindings, result payloads and audio manifests ahead of the chain, and runs the candidate lookup beside context and build (fail: open)
 - `BATCH_PIPELINE_PROJECTION_SOURCE_DELETED_ENABLED` — Default on; settle deleted original sources without a tombstone or redirect as source_deleted, without content publication (fail: open)
+- `BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS` — Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14 (fail: closed)
 - `BATCH_PIPELINE_PUBLISH_CONCURRENCY` — GPU worker post-flight publication concurrency: bounded fenced transitions and one ordered drain per owner per flight; default 4, 1 = legacy serial kill switch, max 16 (fail: closed)
 - `BATCH_PIPELINE_RAMP_EXCLUDE_HEAVY_OWNERS` — Ship-on partial-ramp admission isolation; only off disables; owners over 30 open jobs or six hours pending audio use legacy (fail: closed)
 - `BATCH_PIPELINE_RECORD_PROTOCOL_READY` — One-time native record protocol cutover; default closed retains record_pending obligations without expiry (fail: closed)
@@ -1120,6 +1125,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `BATCH_PIPELINE_SHORT_SPEECH_SINGLE_ENABLED` — Default off; a batch unit whose speech and words form one region of at most 2 s (pauses up to 0.5 s inclusive) takes the existing skipped_single result without diarization (fail: closed)
 - `BATCH_PIPELINE_STAGE_PROFILING_ENABLED` — Dedicated worker omi_batch_pipeline_stage_seconds wall-clock histograms; default on only for OMI_ENV_STAGE=dev, prod/unset off, explicit true/false overrides (local/offline opt in) (fail: open)
 - `BATCH_PIPELINE_STRICT_VAD_SHADOW_PERCENT` — Share of units that run the CPU strict-VAD shadow, sampled deterministically by segment hash; default 10, 0 disables the shadow, 100 restores full sampling; invalid values keep 10 (fail: closed)
+- `BATCH_PIPELINE_SWEEP_AGE_GATE` — Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes (fail: closed)
 - `BATCH_PIPELINE_SWEEP_OWNER_CHAIN_ENABLED` — Default on; when the periodic projection sweep advances an owner's chain head it keeps walking that owner's result log in chain order instead of waiting for its cursor to wrap (fail: open)
 - `BATCH_PIPELINE_VAD_SEGMENTATION_REUSE_ENABLED` — Default on unless false; on the serial batch GPU chain a unit diarized at full length takes the raw segmentation scores its own VAD pass computed instead of a second segmentation forward (fail: open)
 - `BATCH_PIPELINE_WINDOW_SECONDS` — GPU chain internal ASR core window seconds; default 30; queue units remain conversations (fail: closed)

backend/tests/unit/test_provider_stt_outbox.py

@@ -220,3 +220,43 @@ def test_wake_task_shape(monkeypatch):
     monkeypatch.delenv('PROVIDER_STT_TASKS_QUEUE')
     with pytest.raises(RuntimeError):
         provider_stt_tasks.create_wake(ROW, 'a' * 32, due_at=0)
+
+
+@pytest.mark.parametrize(
+    'age,setting,skipped', [(15, 'on', True), (1, 'on', False), (14, 'on', False), (15, 'off', False)]
+)
+def test_sweep_age_gate_preserves_old_rows_and_counts(box, monkeypatch, age, setting, skipped):
+    from copy import deepcopy
+    from utils.observability.sweep_age import SKIPPED
+
+    monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', setting)
+    monkeypatch.delenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', raising=False)
+    enqueue(box, now=100)
+    box.woken.clear()
+    before = deepcopy(row(box))
+    versions = dict(box.client.versions)
+    counter = SKIPPED.labels(queue='provider-stt', reason='max_age')
+    count = counter._value.get()
+    result = box.outbox.sweep(now=100 + age * 86400)
+    assert result.rewoken == (0 if skipped else 1)
+    assert result.errors == 0
+    assert len(box.woken) == (0 if skipped else 1)
+    assert counter._value.get() - count == int(skipped)
+    if skipped:
+        assert row(box) == before
+        assert box.client.versions == versions
+
+
+def test_ancient_sweep_keeps_existing_privacy_discard(box, monkeypatch):
+    from utils.observability.sweep_age import SKIPPED
+
+    monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on')
+    enqueue(box, now=100)
+    box.blocked.add(ROW)
+    box.woken.clear()
+    counter = SKIPPED.labels(queue='provider-stt', reason='max_age')
+    before = counter._value.get()
+    counts = box.outbox.sweep(now=100 + 15 * 86400)
+    assert counts.discarded == 1 and counts.rewoken == 0
+    assert row(box)['closed_reason'] == 'privacy_blocked'
+    assert not box.woken and counter._value.get() == before