fix(backend): terminalize oversized conversation doc rejections and cap occasion size (#23341)

- Add proactive byte-size (OCCASION_MAX_DOCUMENT_BYTES = 786_432) and segment-count (OCCASION_MAX_TRANSCRIPT_SEGMENTS = 1_500) rollover caps to occasion conversations before Firestore rejects writes.
- Terminalize permanent Firestore 1 MiB document size rejections across durable conversation-finalization jobs, enrichment consumer callbacks, summary delivery, and v2 sync jobs.
- Fall back to shrinking status-only write (status="completed") in fit_document_limit when dead_letter_conversation_updates itself exceeds the 1 MiB ceiling.

Failure-Class: FC-permanent-write-rejection-retried-forever
Fixes #23341

Author: ocakciomursel-pixel

backend/config/occasion_conversations.py

@@ -19,11 +19,18 @@
 
 from __future__ import annotations
 
+from collections.abc import Mapping, Sequence
 import os
+from typing import Any
 
 OCCASION_CONVERSATIONS_ENABLED_ENV = 'OCCASION_CONVERSATIONS_ENABLED'
 OCCASION_GAP_SECONDS = 15 * 60
 OCCASION_MAX_SPAN_SECONDS = 3 * 60 * 60
+# Keep 256 KiB of headroom below Firestore's 1 MiB (1,048,576-byte) ceiling so
+# finalization bindings, summaries, action items, app results, and outbox
+# metadata can still commit without hitting InvalidArgument.
+OCCASION_MAX_DOCUMENT_BYTES = 1_048_576 - 262_144
+OCCASION_MAX_TRANSCRIPT_SEGMENTS = 1_500
 
 _OFF = frozenset({'0', 'false', 'off', 'no'})
 # Sources whose clients do not show episodes; their conversations keep today's boundaries.
@@ -45,6 +52,23 @@ def occasion_timeout(timeout: int) -> int:
     return max(int(timeout), OCCASION_GAP_SECONDS)
 
 
+def occasion_size_exceeded(
+    conversation: Mapping[str, Any],
+    document_path: str | None = None,
+) -> bool:
+    """Whether a live conversation has reached the segment-count or byte-size rollover cap."""
+    segments = conversation.get('transcript_segments')
+    if (
+        isinstance(segments, Sequence)
+        and not isinstance(segments, (str, bytes, bytearray))
+        and len(segments) >= OCCASION_MAX_TRANSCRIPT_SEGMENTS
+    ):
+        return True
+    from database.conversation_terminal_title import estimate_firestore_document_bytes
+
+    return estimate_firestore_document_bytes(conversation, document_path) >= OCCASION_MAX_DOCUMENT_BYTES
+
+
 # Explicit rollout bindings for inventory and operators. The generic parser in
 # refactor_modes owns every read, including the two suffixes derived from a mode.
 EPISODE_BOUNDARIES_MODE_ENV = 'EPISODE_BOUNDARIES_MODE'

backend/database/conversation_quiet_checks.py

@@ -15,6 +15,11 @@
 
 from database._client import get_firestore_client
 from database import conversation_finalization_jobs as jobs_db
+from database.conversation_terminal_title import (
+    FIRESTORE_MAX_DOCUMENT_BYTES,
+    TERMINAL_SIZE_HEADROOM_BYTES,
+    estimate_firestore_document_bytes,
+)
 from models.conversation_continuity import continuation_timeout
 from config.refactor_modes import serving_for
 from utils.conversations.lifecycle_clocks import speech_clock
@@ -50,6 +55,17 @@ def quiet_deadline(conversation: dict[str, Any], timeout: Any = 120, *, uid: str
     return finish + timedelta(seconds=continuation_timeout(conversation, quiet_timeout(timeout)))
 
 
+def _update_row_if_fits(transaction: Any, row_ref: Any, row: dict[str, Any], updates: dict[str, Any]) -> None:
+    # Skip arming stamps when the conversation is at the 1 MiB ceiling; quiet_finalization.run() retires the check.
+    path = getattr(row_ref, 'path', None)
+    doc_path = path if isinstance(path, str) else None
+    if (
+        estimate_firestore_document_bytes({**row, **updates}, doc_path) + TERMINAL_SIZE_HEADROOM_BYTES
+        <= FIRESTORE_MAX_DOCUMENT_BYTES
+    ):
+        transaction.update(row_ref, updates)
+
+
 def arm_quiet_check(
     uid: str, conversation_id: str, timeout: float, *, requires_byok: bool = False, firestore_client: Any = None
 ) -> dict[str, Any] | None:
@@ -84,8 +100,10 @@ def arm(transaction: Any) -> dict[str, Any] | None:
                 job.get('quiet_timeout_seconds') != quiet_timeout(timeout)
                 or row.get('listen_finalization_requires_byok') != requires_keys
             ):
-                transaction.update(
+                _update_row_if_fits(
+                    transaction,
                     row_ref,
+                    row,
                     {
                         'listen_quiet_timeout_seconds': quiet_timeout(timeout),
                         'listen_finalization_requires_byok': requires_keys,
@@ -106,8 +124,10 @@ def arm(transaction: Any) -> dict[str, Any] | None:
             'created_at': job.get('created_at') or now,
             'updated_at': now,
         }
-        transaction.update(
+        _update_row_if_fits(
+            transaction,
             row_ref,
+            row,
             {
                 'listen_quiet_timeout_seconds': quiet_timeout(timeout),
                 'listen_finalization_requires_byok': requires_keys,

backend/database/conversation_terminal_title.py

@@ -169,17 +169,29 @@ def fit_document_limit(
     ``base_update`` is the terminal's own status write, which the caller has
     always committed. ``extras`` (title, retry marker) are optional growth: an
     oversized update aborts the whole transaction and would strand the row on
-    ``processing``, so they are dropped rather than risk that.
+    ``processing``, so they are dropped rather than risk that. When the row is
+    already so close to 1 MiB that optional fields in ``base_update`` would
+    also exceed the ceiling, fall back to the status-only write.
     """
-    combined = {**base_update, **extras}
-    if not extras:
-        return combined
-    after = {**conversation, **combined}
     path = getattr(conversation_ref, 'path', None)
-    estimated = estimate_firestore_document_bytes(after, path if isinstance(path, str) else None)
-    if estimated + TERMINAL_SIZE_HEADROOM_BYTES > FIRESTORE_MAX_DOCUMENT_BYTES:
-        return dict(base_update)
-    return combined
+    doc_path = path if isinstance(path, str) else None
+    combined = {**base_update, **extras}
+    if extras:
+        after = {**conversation, **combined}
+        if (
+            estimate_firestore_document_bytes(after, doc_path) + TERMINAL_SIZE_HEADROOM_BYTES
+            <= FIRESTORE_MAX_DOCUMENT_BYTES
+        ):
+            return combined
+    base = dict(base_update)
+    if len(base) > 1 and 'status' in base:
+        after_base = {**conversation, **base}
+        if (
+            estimate_firestore_document_bytes(after_base, doc_path) + TERMINAL_SIZE_HEADROOM_BYTES
+            > FIRESTORE_MAX_DOCUMENT_BYTES
+        ):
+            return {'status': base['status']}
+    return base
 
 
 def _has_described_photo(conversation: Mapping[str, Any], conversation_ref: Any, transaction: Any) -> bool:

backend/routers/conversation_finalization.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import asyncio
+from datetime import timedelta
 import logging
 import os
 from typing import Any
@@ -22,6 +23,7 @@
 from utils.account_cutover.access import should_skip_background_account_mutation
 from utils.conversations import lifecycle as lifecycle_service
 from utils.conversations import quiet_finalization
+from utils.conversations.finalization_failure import classify_finalization_failure
 from utils.conversations.processing_trigger import trigger_for_finalization_job
 from utils.conversations.recovery import RecoveryStructureUnavailableError, recovery_minimum_terminal_enabled
 from utils.conversations.finalizer import (
@@ -105,6 +107,57 @@ async def _retry_or_dead_letter(
     return False
 
 
+async def _terminalize_oversized_finalization_job(
+    job: dict[str, Any],
+    job_id: str,
+    dispatch_generation: int,
+    claimed_lease_epoch: int,
+    this_attempt: int,
+) -> JSONResponse:
+    """Close a durable finalization job whose conversation hit Firestore's 1 MiB ceiling."""
+    if job.get('fanout_status') == 'completed':
+        completed = await run_blocking(
+            db_executor,
+            jobs_db.mark_finalization_completed,
+            job_id,
+            dispatch_generation,
+            claimed_lease_epoch,
+        )
+        if not completed:
+            return JSONResponse(status_code=409, content={'status': 'completion_conflict'})
+        return JSONResponse(status_code=200, content={'status': 'done'})
+    try:
+        await run_blocking(
+            db_executor,
+            lifecycle_service.close_oversized_in_progress_conversation,
+            job['uid'],
+            job['conversation_id'],
+            quiet_for=timedelta(0),
+        )
+    except Exception:
+        logger.warning(
+            'listen finalization oversized in_progress close skipped job_hash=%s',
+            finalization_diagnostic_id(job_id),
+        )
+    terminal = await run_blocking(
+        db_executor,
+        final_attempt_failed,
+        job_id,
+        dispatch_generation,
+        claimed_lease_epoch,
+        this_attempt,
+        failure_code='document_size_limit',
+    )
+    if not terminal:
+        return JSONResponse(status_code=409, content={'status': 'completion_conflict'})
+    logger.warning(
+        'listen finalization document size limit terminal job_hash=%s dispatch_generation=%s',
+        finalization_diagnostic_id(job_id),
+        dispatch_generation,
+    )
+    return JSONResponse(status_code=200, content={'status': 'dead_letter'})
+
+
 @router.post('/v1/conversation-finalization-jobs/run', include_in_schema=False)
 async def run_listen_finalization_job(
     request: Request,
@@ -273,6 +326,17 @@ async def run_listen_finalization_job(
                 >= get_listen_finalization_tasks_max_attempts_for_worker() - 1,
             )
         except ConversationFinalizationError as error:
+            if (
+                error.__cause__ is not None
+                and classify_finalization_failure(error.__cause__, job['conversation_id']).conversation_at_size_limit
+            ):
+                return await _terminalize_oversized_finalization_job(
+                    job,
+                    job_id,
+                    dispatch_generation,
+                    claimed_lease_epoch,
+                    max(task_retry_count, failed_attempts) + 1,
+                )
             if recovery_minimum_terminal_enabled() and isinstance(error.__cause__, RecoveryStructureUnavailableError):
                 # A completed fanout wins even if its worker failed before
                 # acknowledging the job. The lease-fenced transaction also
@@ -352,6 +416,25 @@ async def run_listen_finalization_job(
         logger.warning('listen finalization handler cancelled job=%s; preserving run lock until TTL', job_id)
         raise
     except Exception as error:
+        if (
+            claimed_lease_epoch is not None
+            and job is not None
+            and isinstance(job.get('conversation_id'), str)
+            and classify_finalization_failure(error, job['conversation_id']).conversation_at_size_limit
+        ):
+            try:
+                return await _terminalize_oversized_finalization_job(
+                    job,
+                    job_id,
+                    dispatch_generation,
+                    claimed_lease_epoch,
+                    max(task_retry_count, failed_attempts) + 1,
+                )
+            except Exception:
+                logger.error(
+                    'listen finalization oversized terminal update failed job=%s',
+                    job_id,
+                )
         logger.warning(
             'listen finalization worker failed stage=handler exception_type=%s job_hash=%s dispatch_generation=%s',
             type(error).__name__,

backend/routers/listen/conversations.py

@@ -46,7 +46,7 @@
     should_attach_to_existing_in_progress,
 )
 from utils.transcribe_store import calendar_db, conversations_db, redis_db
-from config.occasion_conversations import OCCASION_MAX_SPAN_SECONDS
+from config.occasion_conversations import OCCASION_MAX_SPAN_SECONDS, occasion_size_exceeded
 from utils.conversations.lifecycle_clocks import speech_clock
 from utils.episodes import boundaries, boundary_runtime
 from routers.listen.occasion import occasion_attach_target
@@ -161,12 +161,16 @@ async def _resume_continuation(self, pointer: dict[str, str]) -> bool:
             )
         if existing:
             existing = await self._continuity_row(existing)
-        if not existing or not resumable_continuation(
-            existing,
-            source=normalize_listen_source(self.host.request.source),
-            device_id=self.host.client_device_context.client_device_id,
-            now=self.clock(),
-            timeout=self.host.conversation_creation_timeout,
+        if (
+            not existing
+            or self._conversation_size_exceeded(existing)
+            or not resumable_continuation(
+                existing,
+                source=normalize_listen_source(self.host.request.source),
+                device_id=self.host.client_device_context.client_device_id,
+                now=self.clock(),
+                timeout=self.host.conversation_creation_timeout,
+            )
         ):
             return False
         self.host.recording_session_id = pointer['recording_session_id']
@@ -533,14 +537,18 @@ async def create_new_in_progress_conversation(self, *, rollover: bool = False) -
                 in_progress_status=ConversationStatus.in_progress,
             )
             if action == RecordingSessionReconnectAction.resume_current and not existing.get('deleted'):
-                if not self.host.is_multi_channel and not resumable_continuation(
-                    existing,
-                    source=normalize_listen_source(request.source),
-                    device_id=self.host.client_device_context.client_device_id,
-                    now=self.clock(),
-                    timeout=self.host.conversation_creation_timeout,
+                size_exceeded = self._conversation_size_exceeded(existing)
+                if not self.host.is_multi_channel and (
+                    size_exceeded
+                    or not resumable_continuation(
+                        existing,
+                        source=normalize_listen_source(request.source),
+                        device_id=self.host.client_device_context.client_device_id,
+                        now=self.clock(),
+                        timeout=self.host.conversation_creation_timeout,
+                    )
                 ):
-                    if (
+                    if size_exceeded or (
                         existing.get('source') == normalize_listen_source(request.source)
                         and existing.get('client_device_id') == self.host.client_device_context.client_device_id
                         and not any(
@@ -760,16 +768,20 @@ async def prepare(self) -> Optional[str]:
         existing = await self._continuity_row(existing)
         finished_at = speech_clock(self.host.request.uid, existing) or self.clock()
         seconds = (self.clock() - finished_at).total_seconds()
+        size_exceeded = self._conversation_size_exceeded(existing)
         if (
-            decide_existing_conversation_action(
+            size_exceeded
+            or decide_existing_conversation_action(
                 seconds_since_last_segment=seconds,
                 conversation_creation_timeout=continuation_timeout(existing, self.host.conversation_creation_timeout),
             )
             == ConversationLifecycleAction.process_and_create_new
         ):
             # This runs before STT initialization: an outage cannot indefinitely
             # defer empty-generation cleanup. The transaction still lets content win.
-            if not (existing.get('transcript_segments') or existing.get('photos') or existing.get('has_content')):
+            if size_exceeded or not (
+                existing.get('transcript_segments') or existing.get('photos') or existing.get('has_content')
+            ):
                 await self._finalize_isolated(self.process_conversation, existing['id'], stage='prepare_rollover')
             await self.create_new_in_progress_conversation()
             return existing['id']
@@ -922,9 +934,17 @@ async def recover_stale_in_progress(self) -> None:
                 continue
             await self._finalize_isolated(self.process_conversation, conversation['id'], stage='stale_in_progress')
 
+    def _conversation_size_exceeded(self, conversation: dict[str, Any]) -> bool:
+        cid = conversation.get('id') or getattr(getattr(self.host, 'state', None), 'current_conversation_id', None)
+        uid = getattr(getattr(self.host, 'request', None), 'uid', None)
+        doc_path = f'users/{uid}/conversations/{cid}' if uid and cid else None
+        return occasion_size_exceeded(conversation, document_path=doc_path)
+
     def _occasion_span_exceeded(self, conversation: dict[str, Any]) -> bool:
         if not getattr(self.host, 'occasion_mode', False):
             return False
+        if self._conversation_size_exceeded(conversation):
+            return True
         started = persisted_started_seconds(conversation.get('started_at'))
         return started is not None and self.clock().timestamp() - started >= OCCASION_MAX_SPAN_SECONDS
 
@@ -973,7 +993,7 @@ async def lifecycle_loop(self) -> None:
             if (
                 boundary_action != 'hold'
                 and action == ConversationLifecycleAction.continue_current
-                and self._occasion_span_exceeded(conversation)
+                and (self._occasion_span_exceeded(conversation) or self._conversation_size_exceeded(conversation))
             ):
                 # One occasion is capped like a smart-merged chain: all-day capture still rolls over.
                 logger.info('event=occasion_span_rollover conversation=%s', conversation_id)