harden(pusher): persist playback projection rebuild debt

backend/charts/batch-pipeline-worker/templates/record-seal-repair.yaml

@@ -53,6 +53,7 @@ spec:
               image: "{{ required "image.repository required" .Values.image.repository }}@{{ required "image.digest required" .Values.image.digest }}"
               command: [python, scripts/record_seal_repair_job.py]
               env:
+                - {name: PUSHER_PROJECTION_DEBT_DURABLE_ENABLED, value: {{ .Values.recordSealRepair.projectionDebtDurableEnabled | quote }}}
                 - {name: RECORD_SEAL_REPAIR_JOBS, value: {{ .Values.recordSealRepair.jobsDiscovery | quote }}}
                 - {name: RECORD_SEAL_REPAIR_ENABLED, value: {{ .Values.recordSealRepair.runtimeEnabled | quote }}}
                 - {name: RECORD_AUDIO_PARTIAL_ENABLED, value: {{ .Values.recordSealRules.partialEnabled | quote }}}

backend/charts/batch-pipeline-worker/values.yaml

@@ -220,6 +220,7 @@ recordSealRepair:
   suspend: true
   runtimeEnabled: true
   jobsDiscovery: "on"
+  projectionDebtDurableEnabled: true
   mode: apply
   workers: 8
   rowSeconds: 270

backend/config/batch_pipeline_projection_rebuild.py

@@ -0,0 +1,7 @@
+"""Runtime kill switch for optional projection-debt handoff and maintenance."""
+
+import os
+
+
+def enabled() -> bool:
+    return os.getenv('PUSHER_PROJECTION_DEBT_DURABLE_ENABLED', '1').strip().lower() not in {'0', 'false', 'off'}

backend/config/flag_budget.json

@@ -24,8 +24,9 @@
   },
   "seed_review_by": "2026-11-15",
   "budget": 169,
-  "count": 259,
+  "count": 260,
   "names": {
+    "PUSHER_PROJECTION_DEBT_DURABLE_ENABLED": {"owner": "uicleanup / omi-lifecycle", "review_by": "2026-12-06"},
     "BATCH_PIPELINE_SERVING_PROJECTION_SWEEP": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
     "BATCH_PIPELINE_PROJECTION_PUMP_ENABLED": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05", "registry": "config/feature-flags.yaml"},
     "BATCH_PIPELINE_PUMP_SWEEP_PAGES": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-05"},

backend/database/record_projection_rebuild.py

@@ -0,0 +1,79 @@
+"""Content-free playback repair debt; no seal, producer-token or object authority."""
+
+import hashlib
+import json
+from datetime import datetime, timezone
+
+from google.cloud import firestore
+
+from database._client import get_firestore_client, run_transactional
+from database.batch_pipeline_privacy import BatchPrivacyBlocked, BatchPrivacyReason, assert_batch_write_allowed
+
+COLLECTION = 'record_projection_rebuild_jobs'
+KIND = 'record_projection_rebuild_due'
+MAX_MARKERS_PER_BATCH = 256
+
+
+def marker_id(uid, cid):
+    for value in (uid, cid):
+        if not isinstance(value, str) or not value or '/' in value or len(value.encode()) > 256:
+            raise ValueError('bounded projection owner required')
+    return hashlib.sha256(json.dumps([uid, cid], separators=(',', ':')).encode()).hexdigest()
+
+
+def mark_due(uid, cid, *, firestore_client=None):
+    client = firestore_client if firestore_client is not None else get_firestore_client()
+    client.collection(COLLECTION).document(marker_id(uid, cid)).set(
+        {'kind': KIND, 'owner_id': uid, 'conversation_id': cid, 'requested_at': datetime.now(timezone.utc)},
+        timeout=5,
+        retry=None,
+    )
+
+
+def mark_many(uid, conversation_ids, *, firestore_client=None):
+    """One bounded cleanup RPC, rather than one timeout per socket obligation."""
+    owners = set(conversation_ids)
+    if not 0 < len(owners) <= MAX_MARKERS_PER_BATCH:
+        raise ValueError('bounded projection marker batch required')
+    client = firestore_client if firestore_client is not None else get_firestore_client()
+    batch = client.batch()
+    requested_at = datetime.now(timezone.utc)
+    for cid in owners:
+        ref = client.collection(COLLECTION).document(marker_id(uid, cid))
+        batch.set(ref, {'kind': KIND, 'owner_id': uid, 'conversation_id': cid, 'requested_at': requested_at})
+    batch.commit(timeout=1, retry=None)
+
+
+def apply_projection(uid, cid, payload, *, expected_audio_files, firestore_client=None):
+    """Cold repair only: deletion intent and projection CAS share the commit."""
+    from database.conversations import AudioFilesChanged, audio_files_revision, retain_aliases
+
+    client = firestore_client if firestore_client is not None else get_firestore_client()
+    ref = client.collection('users').document(uid).collection('conversations').document(cid)
+
+    @firestore.transactional
+    def commit(tx):
+        snapshot = ref.get(transaction=tx)
+        if not snapshot.exists:
+            return False
+        row = snapshot.to_dict() or {}
+        if row.get('deleted') or row.get('discarded') or row.get('private_cloud_sync_enabled') is False:
+            return False
+        try:
+            assert_batch_write_allowed(client, tx, uid, conversation_ids=(cid,))
+        except BatchPrivacyBlocked as blocked:
+            if blocked.reason in {
+                BatchPrivacyReason.ACCOUNT_DELETED,
+                BatchPrivacyReason.ACCOUNT_MISSING,
+                BatchPrivacyReason.CONVERSATION_DELETED,
+            }:
+                return False
+            return None  # Migration, generation holds and unknown refusals retain debt without a write.
+        if audio_files_revision(row.get('audio_files') or []) != audio_files_revision(expected_audio_files):
+            raise AudioFilesChanged('audio projection changed')
+        tx.update(ref, {'audio_files': retain_aliases(payload['audio_files'], row.get('audio_files') or [])})
+        # A retained socket worker may still publish a later tail. Its existing
+        # producer fence is read-only here: repair never clears or seals it.
+        return None if row.get('record_upload_sessions') else True
+
+    return run_transactional(client, commit, transient=True)

backend/docs/feature-flag-registry.md

@@ -714,6 +714,7 @@ and an explicit empty literal renders as `''`.
 | `PUBLIC_SHARED_CONVERSATION_CHAT_MODE` | Enable chat on public shared conversations | backend | env | closed | declared (gke/backend-listen); gateway (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill) | gateway (backend-listen (chart), cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | gateway (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill); off (backend-listen (chart), gke/backend-listen) | — | keep | keep | — | dazheng |
 | `PUSHER_ABANDON_EVICTED_TOKENS` | On unless off; capacity-evicted pusher drains attest token abandonment only after worker and executor settlement | backend | env | open | — | — | on (pusher (chart)); {value: 'on', category: ops_kill} (gke/pusher) | — | keep | keep | — | two-lane lead |
 | `PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED` | Default on; refresh the live pusher audio_files projection from exact uploaded-object metadata between full prefix rebuilds; 0 restores per-flush listing | backend | env | open | — | — | — | — | keep | keep | — | planscost |
+| `PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` | Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob | backend | env | open | — | — | — | — | keep | keep | — | uicleanup / omi-lifecycle |
 | `PUSHER_PROJECTION_QUEUE_WAIT_BOUND_ENABLED` | Default on; bound only committed pusher projection queue capacity waits to the existing 300-second receive-liveness budget | backend | env | open | — | — | — | — | keep | keep | — | two-lane lead |
 | `RATE_LIMIT_SHADOW_MODE` | Shadow backend rate limits | backend | env | closed | — | — | — | — | keep | keep | — | unowned |
 | `RECORDING_SPEAKER_RENAME_ENABLED` | Default on; bounded scoped recording speaker naming. Off reports disabled and clients use explicit single-conversation assignments. | backend, mobile | env | open | — | — | — | — | keep | keep | — | speakerrename |
@@ -1378,6 +1379,7 @@ their code default (`fail` tells you which way a missing value resolves).
 - `PLAYBACK_MANIFEST_RECOVERY_ENABLED` — Default on; next-access recovery for merge_failed playback markers after 15 minutes, deduped by marker generation (fail: closed)
 - `PROJECTION_DRAIN_CLAIM_AWARE` — Default on; read-only fleet claim admission and jittered backoff for redundant claim-busy owner-drain requests (fail: open)
 - `PUSHER_AUDIO_FILE_INCREMENTAL_ENABLED` — Default on; refresh the live pusher audio_files projection from exact uploaded-object metadata between full prefix rebuilds; 0 restores per-flush listing (fail: open)
+- `PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` — Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob (fail: open)
 - `PUSHER_PROJECTION_QUEUE_WAIT_BOUND_ENABLED` — Default on; bound only committed pusher projection queue capacity waits to the existing 300-second receive-liveness budget (fail: open)
 - `RATE_LIMIT_SHADOW_MODE` — Shadow backend rate limits (fail: closed)
 - `RECORDING_SESSION_MODE` — Select recording session migration mode (fail: closed)

backend/docs/listen_pusher_pipeline.mdx

@@ -9351,3 +9351,59 @@ owner-drain admissions use a read-only Redis claim GET and fail open on errors.
 Claim-busy re-requests through `again` back off 5–15 seconds; productive
 `more=True` continuations retain immediate FIFO progress without the peek.
 The advisory claim and transactional ordering fences remain unchanged.
+
+
+### Durable playback projection rebuild debt
+
+`PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` defaults **on**; `0`, `false`, or `off`
+disables pusher handoff and repair consumption. Pusher receives the flag through
+its existing `runtimeConfigMapName` `envFrom` binding; no literal env entry
+shadows that ConfigMap key. The repair CronJob declares it independently through
+`recordSealRepair.projectionDebtDurableEnabled` (default `true`). Rollback sets
+the consumer option `false` first, then the Pusher runtime ConfigMap flag off.
+Re-enabling resumes retained markers. Consumption also requires the existing
+repair CronJob to be enabled, unsuspended and runtime-enabled; this contract
+does not activate it.
+
+Exhausted close retries, empty close discovery, socket cleanup, and terminal
+transient/InvalidArgument batch failures hand off one
+`record_projection_rebuild_due` marker per user/conversation into
+`record_projection_rebuild_jobs/{sha256(JSON([uid,cid]))}`. Fields are `kind`,
+`owner_id`, `conversation_id`, and UTC `requested_at`; no content or conversation
+status is stored. Successful handoffs coalesce. Terminal handoffs enqueue into
+one metadata-only background task and never await Firestore or change recording
+completion. A separate bounded retry inventory holds up to 128 overflow owners;
+cleanup can add the 128 cached owners (256 total). Each task attempts at most two
+batch writes with a one-second SDK timeout and 1.25-second async attempt budget.
+Cleanup shares one three-second total budget across every marker and retry;
+marker failure never gates socket close or recording completion. Failure emits
+`marker_write_failed`; inventory saturation emits `marker_retry_overflow`, with
+sanitized logs and `omi_pusher_projection_debt_total` outcomes. Persistent storage
+failure, saturation, or process death before commit can still prevent handoff;
+the marker is not a WAL.
+
+The existing repair CronJob runs a separate 60-second metadata-only pass over
+at most 128 markers ordered by request time. A bounded collection scan reads at
+most 10,001 markers; the maintenance ceiling is 10,000. A confirmed empty
+inventory retires as `retired_no_objects` after seven days from Firestore document
+creation (request-time rotation and coalesced writes do not reset that age), or
+immediately under collection pressure. Retiring deletes only the observed marker,
+never audio. Nonempty or privacy-held debt remains due even above the ceiling;
+capacity is enforced by retiring confirmed empty work, not discarding useful debt.
+The aggregate report and Prometheus gauges expose marker count (saturated at
+10,001), oldest age in that bounded scan, and capacity state. `capacity_exceeded`
+is alertable; these gauges are bounded observations, not an exact count/oldest
+claim when the collection is above the ceiling. Apply rotates discovered markers
+before validation/discovery, so refusal, interruption or malformed ordered
+markers cannot monopolize the page. Preview never rebuilds or writes. Full
+rebuild refuses a tombstone or absent conversation before discovery, then
+transactionally rechecks privacy/deletion and the prior `audio_files` revision.
+Only `audio_files` may change. Temporary privacy/generation holds retain debt
+without metadata writes. Nonempty existing producer fences retain the marker
+even after metadata refresh, until their existing owner settles them. No seal
+or token authority is added. Raw-index failure retains debt; confirmed empty
+discovery uses the bounded retirement policy above.
+Successful final rebuild or permanent owner revocation clears only the observed
+marker version; a replacement marker always wins. No PCM or audio-object
+mutation occurs. The [phase 5C change note](./listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md)
+records scope and verification limits.

backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md

@@ -0,0 +1,60 @@
+# Durable playback projection debt (phase 5C)
+
+After the bounded close rebuild retries fail, pusher hands playback-only debt
+into `record_projection_rebuild_jobs/{sha256([uid,cid])}` before discarding its
+socket state. Socket cleanup also hands off outstanding debt. Terminal transient/InvalidArgument batch
+projection failures persist by exact owner even if the bounded close cache is full. The marker kind
+is `record_projection_rebuild_due`; its only other fields are owner ID,
+conversation ID and request time. Repeated handoffs coalesce by owner.
+Ordinary flushes add no reads or writes. Terminal marker handoffs enqueue into
+one bounded background task. A separate metadata-only retry inventory preserves
+overflow outside the close cache; marker failure cannot change recording-close
+success. Socket cleanup has one three-second total deadline for batched writes
+and retries, independent of debt count. `marker_write_failed` and bounded-list
+saturation emit sanitized logs and counter outcomes. Temporary privacy holds
+retain markers without a metadata update.
+
+The [owning contract](../../listen_pusher_pipeline.mdx#durable-playback-projection-rebuild-debt)
+defines payloads, ordering, limits, gauges, flag bindings and rollback. Process
+kill, persistent Firestore failure, or saturation can still prevent handoff;
+this marker is not a WAL.
+
+The existing record-seal-repair CronJob runs a separate 60-second pass before
+seal work, selecting at most 128 markers by request time. It uses a fresh,
+socket-independent full metadata rebuild even if the incremental optimization
+is disabled. Successful rebuild or revoked owner clears only the observed
+marker version. Existing nonempty producer fences retain the marker after a
+metadata update: a retained worker may still upload a later tail. This reads
+the existing map and never clears a token or grants seal authority. Incomplete discovery or failure retains and rotates the
+marker; malformed markers also rotate. Concurrent replacement wins over cleanup. Preview mode never rebuilds
+or writes. Reports contain aggregate dispositions, never owner IDs or paths.
+The existing CronJob suspension/runtime settings still govern execution; no
+CronJob is unsuspended by this code change.
+
+`projection_owner_gone` refuses a tombstone or absent row before discovery.
+The cold projection transaction rechecks account/conversation privacy,
+tombstones, soft deletion, sync opt-out and the prior audio-files revision
+before updating only `audio_files`. A deletion or another projection arriving
+during discovery cannot recreate the owner or overwrite a newer projection.
+Raw-index failure cannot count as successful durable repair, including forced
+socket close. Empty close discovery hands off its obligation too. Empty live
+inventory retains the marker until seven days from Firestore creation, then
+retires as `retired_no_objects`. A 10,000-marker maintenance ceiling also retires
+confirmed empty work under pressure, never useful or privacy-held debt. Per-pass
+work stays capped at 128; the collection scan caps at 10,001. Aggregate logs and
+Prometheus gauges expose bounded count, observed oldest age and capacity state.
+No retirement deletes audio.
+
+`PUSHER_PROJECTION_DEBT_DURABLE_ENABLED` defaults **on**; `0`, `false`, or `off`
+disables both handoff and consumption. Rollback disables the consumer runtime
+first (`recordSealRepair.projectionDebtDurableEnabled=false`), then pusher; existing markers remain inert. Re-enabling the flag resumes retained markers.
+A normal socket close does not probe or clear a preexisting marker: the cold
+consumer clears it after its own full rebuild, avoiding hot-path writes.
+
+This adds no seal authority, producer token changes, entry clock, heartbeat,
+dead-letter, audio-object mutation or PCM reads. Object discovery reads metadata
+only, including the existing raw-pack index. No production measurements, IAM
+checks, cloud access, live CronJob activation, end-user playback acceptance or
+cost/latency claims are included. Hermetic unit tests and lifecycle emulator
+scenarios exercise exhaustion, eviction, deletion during discovery and marker
+replacement. Phase 5 A/B simulator changes remain outside this lane.

backend/routers/pusher.py

@@ -1,4 +1,4 @@
 """The /v1/trigger/listen WebSocket: binary device-audio ingestion, live STT and
 speaker dispatch, and record-barrier custody for realtime capture.
 
 Each long-lived connection supervises several concurrent tasks under a strict
@@ -21,7 +21,8 @@
 from utils.pusher_audio_window import append_window_run, v2_overlap_conflict, AUDIO_TIMELINE_CONTINUITY_TOLERANCE
 from config.audio_custody import framed_pack
 from config.audio_outbox import enabled as audio_outbox_enabled
-from utils.pusher_audio_projection import PusherAudioProjection, projection_owner_gone
+from utils.pusher_audio_projection import PusherAudioProjection
+from utils.batch_pipeline.record_projection_rebuild import projection_owner_gone
 from utils.pusher_record_barrier import PusherRecordBarrier
 from utils.pusher_receive_metrics import SocketObservation, observe_work
 from utils.pusher_record_finish import (
@@ -351,7 +352,7 @@
         observation.close()  # No await: even repeated cancellation cannot skip close accounting.
 
 
 async def _run_websocket_util_trigger(
     websocket: WebSocket,
     uid: str,
     sample_rate: int,
@@ -461,7 +462,7 @@
     audio_projection = PusherAudioProjection(uid)
     deleted_conversations: set[str] = set()
 
     async def process_durable_private_cloud_queue() -> None:
         # Pending conversations hold only receipts. Hydrate at most one bounded
         # contiguous batch for the legacy playback projection, preserving the
         # old 60-second write cadence without retaining 60 seconds of PCM.
@@ -545,6 +546,7 @@
                 logger.warning('Durable audio projection conflict; archive receipt retained')
             except InvalidArgument as exc:
                 audio_projection.inventory.pop(conv_id, None)
+                await audio_projection.defer_rebuild(conv_id)
                 # Custody and its generation-pinned manifest remain durable.
                 # Retrying a deterministic projection rejection cannot repair
                 # the payload, and a sticky custody failure would strand it.
@@ -629,6 +631,9 @@
             # counting toward the bounded retry map.
             projection_retries.pop(conv_id, None)
             audio_projection.inventory.pop(conv_id, None)
+            # The close-debt cache may be full. Persist terminal projection debt
+            # by its exact owner rather than relying on that bounded inventory.
+            await audio_projection.defer_rebuild(conv_id)
             await record_barrier.fail_recording(conv_id)
             record_failure('projection', reason='transient')
             record_fallback(
@@ -753,7 +758,7 @@
         record_close_requests.discard(conv_id)
         record_barrier.retire_close(conv_id)
 
     async def process_private_cloud_queue() -> None:
         """Background task that batches private cloud sync uploads by conversation_id.
 
         Chunks are accumulated per conversation and flushed when:
@@ -1175,7 +1180,7 @@
     # The conversation this session last streamed, for the episode moment flush at teardown.
     episode_session: Dict[str, Optional[str]] = {'conversation_id': None}
 
     async def receive_tasks() -> None:
         nonlocal websocket_active
         nonlocal inline_finish_conversation
         nonlocal websocket_close_code
@@ -1390,7 +1395,7 @@
                 websocket_active = False
                 audio_bytes_event.set()
 
         async def process_frame(data: bytes) -> None:
             nonlocal conversation_generation, custody_sequence
             nonlocal audiobuffer, trigger_audiobuffer, private_cloud_chunk_start_time, private_cloud_window_opened_at
             nonlocal current_conversation_id, audio_timeline_last_end
@@ -1822,6 +1827,7 @@
 
         all_to_cancel = [t for t in bg_main_tasks if not t.done()]
         await drain_tasks(all_to_cancel, timeout=5.0, label="pusher_cleanup", cancel=True)
+        await audio_projection.persist_pending()
 
         if application_failed and observation.reason == 'peer':
             observation.reason = 'application_error'