Enforce the transcription cap where cloud STT is dispatched (flag off)

.github/failure-classes/FC-stt-allowance-not-enforced-at-dispatch.json

@@ -0,0 +1,10 @@
+{
+  "schema_version": 1,
+  "id": "FC-stt-allowance-not-enforced-at-dispatch",
+  "violated_contract": "The transcription cap is computed but not enforced where Omi-funded provider audio is dispatched; replay, fallback and delayed execution can bypass request-time allowance checks.",
+  "canonical_prevention": "Authorize actual provider funding at creation and every send, share live admission across channels and successors, and recheck queued execution before dispatch. Persist denied sync audio before acknowledging it. Fence provider-bound recovery receipts before side effects and atomically prove ingestion, clear known non-calls, and reconcile ambiguous attempts with one bounded re-dispatch while retaining terminal failures and finalizing siblings. Exercise both flag states at provider and queue boundaries, including oversized empty retries without spending failure attempts when held.",
+  "canonical_prevention_artifact": ["backend/tests/unit/test_stt_cap_dispatch.py", "backend/tests/unit/test_sync_cap_hold.py", "backend/tests/unit/test_batch_pipeline_admission.py", "backend/tests/unit/test_batch_pipeline_worker.py", "backend/tests/unit/test_sync_cap_coordinator.py", "backend/tests/unit/test_sync_cap_hold_recovery.py", "backend/tests/unit/test_sync_cap_recovery_checkpoint.py", "backend/tests/unit/test_sync_cap_recovery_dispatch_signal.py", "backend/tests/unit/test_sync_cap_recovery_plan.py", "backend/tests/unit/test_sync_cap_reconciliation_metrics.py", "backend/tests/unit/test_prerecorded_dispatch_selection.py", "backend/tests/unit/test_sync_cap_content.py"],
+  "evidence_prs": [],
+  "scope_hints": ["backend/routers/listen/", "backend/utils/stt/", "backend/utils/sync/", "backend/utils/batch_pipeline/"],
+  "status": "open"
+}

app/changelog/unreleased/20261016-stt-cap-response-dto.json

@@ -0,0 +1,3 @@
+{
+  "kind": "none"
+}

app/lib/backend/schema/gen/conversation_wire.g.dart

@@ -1673,13 +1673,15 @@ class GeneratedSearchConversationsResponse {
 class GeneratedSyncLocalFilesResultResponse {
   final List<String>? errors;
   final int failedSegments;
+  final bool? heldAudio;
   final List<String>? newMemories;
   final int totalSegments;
   final List<String>? updatedMemories;
 
   const GeneratedSyncLocalFilesResultResponse({
     this.errors,
     this.failedSegments = 0,
+    this.heldAudio,
     this.newMemories,
     this.totalSegments = 0,
     this.updatedMemories,
@@ -1689,6 +1691,7 @@ class GeneratedSyncLocalFilesResultResponse {
     return GeneratedSyncLocalFilesResultResponse(
       errors: _readFieldValue<List<String>>(_readField(json, const ["errors"]), "errors", _readStringList, requiredField: false, nullable: true),
       failedSegments: _required(_readFieldValue<int>(_readField(json, const ["failed_segments"]), "failed_segments", _readInt, requiredField: false, nullable: false, defaultValue: 0), "failed_segments"),
+      heldAudio: _readFieldValue<bool>(_readField(json, const ["held_audio"]), "held_audio", _readBool, requiredField: false, nullable: true),
       newMemories: _readFieldValue<List<String>>(_readField(json, const ["new_memories"]), "new_memories", _readStringList, requiredField: false, nullable: true),
       totalSegments: _required(_readFieldValue<int>(_readField(json, const ["total_segments"]), "total_segments", _readInt, requiredField: false, nullable: false, defaultValue: 0), "total_segments"),
       updatedMemories: _readFieldValue<List<String>>(_readField(json, const ["updated_memories"]), "updated_memories", _readStringList, requiredField: false, nullable: true),
@@ -1699,6 +1702,7 @@ class GeneratedSyncLocalFilesResultResponse {
     return {
       'errors': errors,
       'failed_segments': failedSegments,
+      'held_audio': heldAudio,
       'new_memories': newMemories,
       'total_segments': totalSegments,
       'updated_memories': updatedMemories,

backend/AGENTS.md

@@ -148,7 +148,7 @@ STT policy: `config/stt_provider_policy.py`. Windowed live rollout: `docs/operat
 - **monitoring** (`backend/charts/monitoring/`) — Prometheus/Grafana, Loki/Alloy, alerts and HPA adapters. Sync export flags, sizing and CPU audit: [runbook](docs/runbooks/sync-cloudrun-rightsizing.md).
 - **backend-secrets** (`backend/charts/backend-secrets/`) — ExternalSecret and SecretStore resources that sync backend runtime secrets into GKE namespaces.
 
-[Reset recovery](docs/locked-content-reset-recovery.md).
+[Reset recovery](docs/locked-content-reset-recovery.md). Dark funded-STT admission and held-audio recovery: [dispatch contract](docs/stt-cap-dispatch.md).
 
 Runtime env contract: keep `backend/deploy/runtime_env.yaml` aligned with GKE Helm values and Cloud Run runtime env; run `backend/scripts/pre-deploy-check.sh` after backend runtime env or deploy workflow changes. The backend deploy must validate the exact JSON state emitted by `render_backend_runtime_env.py --state-output`, and any Cloud Run export/replace editor must preserve those literal values before apply. The `llm_gateway` manifest section owns the release, ingress, and static-address identity; a reserved address alone is never an endpoint contract. Gateway-mode promotion requires the control-plane gate plus `probe-llm-gateway-from-cloud-run.sh` before Cloud Run revisions are created.
 

backend/config/stt_cap_dispatch.py

@@ -0,0 +1,7 @@
+"""Dark dispatch policy. No environment/deployment changes enable this implicitly."""
+
+import os
+
+
+def enabled() -> bool:
+    return os.getenv('STT_CAP_DISPATCH_ENFORCEMENT', 'false').lower() == 'true'

backend/database/sync_cap_holds.py

@@ -0,0 +1,106 @@
+"""Transfer fenced sync-ledger ownership to durable, recoverable audio holds.
+
+A held upload is acknowledged, but is not proof of completed transcription.
+Only settlement of every referenced hold can publish that proof.
+"""
+
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+from google.cloud import firestore
+
+from database._client import get_firestore_client
+from database.sync_ledger import (
+    sync_content_reference,
+    sync_content_owner_matches,
+    is_valid_completed_sync_content_result,
+    is_valid_held_sync_content_result,
+    LEDGER_RETENTION_DAYS,
+)
+
+PROTOCOL = 'stt_cap_hold_v1'
+
+
+def mark_content_held(
+    uid: str,
+    content_id: str,
+    job_id: str,
+    result: dict,
+    hold_ids: list[str],
+    *,
+    run_token: str | None = None,
+    run_epoch: int | None = None,
+) -> bool:
+    if not hold_ids or not is_valid_held_sync_content_result(result):
+        raise ValueError('durable held audio receipts required')
+    client = get_firestore_client()
+    ref = sync_content_reference(client, uid, content_id)
+
+    @firestore.transactional
+    def commit(tx: Any) -> bool:
+        row = ref.get(transaction=tx).to_dict() or {}
+        if not sync_content_owner_matches(row, job_id, run_token, run_epoch):
+            return False
+        tx.update(
+            ref,
+            {
+                'status': 'held',
+                'dispatch_protocol': PROTOCOL,
+                'result': result,
+                'cap_hold_ids': sorted(set(hold_ids)),
+                'pending_cap_hold_ids': sorted(set(hold_ids)),
+                'updated_at': datetime.now(timezone.utc),
+                # An unprocessed capture must survive inactive accounts.
+                'expires_at': firestore.DELETE_FIELD,
+            },
+        )
+        return True
+
+    return commit(client.transaction())
+
+
+def recovery_status(uid: str, content_id: str, hold_id: str) -> str:
+    row = sync_content_reference(get_firestore_client(), uid, content_id).get().to_dict() or {}
+    if row.get('status') == 'completed' and is_valid_completed_sync_content_result(row.get('result')):
+        return 'completed'
+    if row.get('dispatch_protocol') == PROTOCOL and row.get('status') == 'held':
+        if hold_id not in row.get('cap_hold_ids', []):
+            return 'waiting'
+        return 'ready' if hold_id in row.get('pending_cap_hold_ids', []) else 'settled'
+    return 'waiting'  # The original worker still owns preparation/retry.
+
+
+def settle_hold(uid: str, content_id: str, hold_id: str, response: dict, pieces: int) -> bool:
+    """Idempotently join one recovered hold; never overwrite another owner."""
+    client = get_firestore_client()
+    ref = sync_content_reference(client, uid, content_id)
+
+    @firestore.transactional
+    def commit(tx: Any) -> bool:
+        row = ref.get(transaction=tx).to_dict() or {}
+        if row.get('status') == 'completed':
+            return is_valid_completed_sync_content_result(row.get('result'))
+        if row.get('dispatch_protocol') != PROTOCOL or row.get('status') != 'held':
+            return False
+        if hold_id not in row.get('cap_hold_ids', []):
+            return False
+        pending = set(row.get('pending_cap_hold_ids') or [])
+        if hold_id not in pending:
+            return True
+        pending.remove(hold_id)
+        result = dict(row['result'])
+        for key in ('new_memories', 'updated_memories'):
+            result[key] = sorted(set(result.get(key) or []) | set(response.get(key) or []))
+        result['recovered_segments'] = int(result.get('recovered_segments', 0)) + pieces
+        result['held_audio'] = bool(pending)
+        now = datetime.now(timezone.utc)
+        updates: dict[str, Any] = {'result': result, 'pending_cap_hold_ids': sorted(pending), 'updated_at': now}
+        if not pending:
+            # All source audio has reached content/usage settlement; only now
+            # can ordinary duplicate admission converge a completed result.
+            result.update(failed_segments=0, errors=[], outcome='success')
+            updates.update(status='completed', expires_at=now + timedelta(days=LEDGER_RETENTION_DAYS))
+        tx.update(ref, updates)
+        return True
+
+    return commit(client.transaction())

backend/database/sync_cap_recovery.py

@@ -0,0 +1,73 @@
+"""Piece-lineage proof committed in the exact conversation-ingest transaction."""
+
+import time
+from typing import Any
+
+from database._client import get_firestore_client
+
+
+class RecoveryIngestFenced(RuntimeError):
+    """An expired/replaced recovery attempt cannot append conversation content."""
+
+
+def ingest_hooks(uid: str, context: dict, incoming: dict):
+    ref = (
+        get_firestore_client()
+        .collection('users')
+        .document(uid)
+        .collection('stt_cap_holds')
+        .document(context['hold_id'])
+    )
+    current = {}
+    piece = context['piece']
+
+    def before(transaction):
+        row = ref.get(transaction=transaction).to_dict() or {}
+        receipt = (row.get('piece_receipts') or {}).get(piece) or {}
+        if (
+            row.get('lease_token') != context['lease_token']
+            or row.get('lease_until', 0) <= time.time()
+            or receipt.get('state') != 'dispatched'
+            or receipt.get('attempt', 1) != context['attempt']
+        ):
+            raise RecoveryIngestFenced('held piece ingest ownership changed')
+        current.clear()
+        current.update(row.get('piece_ingests') or {})
+
+    def after(transaction, assigned):
+        conversation, created, _ = assigned
+        cid = conversation['id']
+        response: dict[str, Any] = {'new_memories': [], 'updated_memories': []}
+        repeat = conversation.get('_sync_lineage_repeat_only')
+        if not repeat:
+            response['new_memories' if created else 'updated_memories'] = [cid]
+        if conversation.get('sync_relevance') == 'keep' and (
+            not repeat or conversation.get('_sync_lineage_completion_pending')
+        ):
+            response['_merged'] = {cid: incoming.get('language')}
+        current[piece] = {
+            **context['receipt'],
+            'state': 'transcribed',
+            'has_content': True,
+            'response': response,
+            'lineage_anchor': incoming['id'],
+            'conversation_id': cid,
+        }
+        # No reads here: assignment has already staged its conversation writes.
+        transaction.update(ref, {'piece_ingests': dict(current)})
+
+    return before, after
+
+
+def ingested_piece(uid: str, hold_id: str, piece: str) -> dict | None:
+    row = (
+        get_firestore_client()
+        .collection('users')
+        .document(uid)
+        .collection('stt_cap_holds')
+        .document(hold_id)
+        .get()
+        .to_dict()
+        or {}
+    )
+    return (row.get('piece_ingests') or {}).get(piece)

backend/database/sync_ledger.py

@@ -1,4 +1,4 @@
 """Durable Firestore ledger for once-only sync processing and metering."""
 
 from __future__ import annotations
 
@@ -73,6 +73,7 @@
 
     BOUND = 'bound'
     COMPLETED = 'completed'
+    HELD = 'held'
     LOST = 'lost'
 
 
@@ -91,6 +92,10 @@
     def completed(self) -> bool:
         return self.outcome is SyncContentRunBindingOutcome.COMPLETED
 
+    @property
+    def held(self) -> bool:
+        return self.outcome is SyncContentRunBindingOutcome.HELD
+
 
 def _ledger_ref(client: Any, uid: str, content_id: str) -> Any:
     collection = sync_stage.collection_name('sync_content_ledger')
@@ -122,10 +127,24 @@
     return bound_token == run_token and bound_epoch == run_epoch
 
 
+def sync_content_reference(client: Any, uid: str, content_id: str) -> Any:
+    """Share the canonical owner-scoped reference with a custody transfer."""
+    return _ledger_ref(client, uid, content_id)
+
+
+def sync_content_owner_matches(
+    existing: Dict[str, Any], job_id: str, run_token: str | None, run_epoch: int | None = None
+) -> bool:
+    """Custody transfer uses the same token/epoch fence as normal completion."""
+    return _ledger_owner_matches(existing, job_id, run_token, run_epoch)
+
+
 def is_valid_completed_sync_content_result(result: Any) -> bool:
     """A ledger completion may converge Redis only for an all-success result."""
     if not isinstance(result, dict):
         return False
+    if result.get('held_audio') is True:
+        return False
     failed_segments = result.get('failed_segments')
     total_segments = result.get('total_segments')
     errors = result.get('errors')
@@ -152,6 +171,12 @@
     return outcome in {'success', 'expected_silence'}
 
 
+def is_valid_held_sync_content_result(result: Any) -> bool:
+    if not isinstance(result, dict) or result.get('held_audio') is not True:
+        return False
+    return is_valid_completed_sync_content_result({**result, 'held_audio': False})
+
+
 def _processing_claim_updates(
     job_id: str,
     lane: str,
@@ -193,6 +218,10 @@
 ) -> Dict[str, Any]:
     snapshot = ref.get(transaction=transaction)
     existing = cast(Dict[str, Any], snapshot.to_dict() or {}) if getattr(snapshot, 'exists', False) else {}
+    if existing.get('dispatch_protocol') == 'stt_cap_hold_v1' and existing.get('status') == 'held':
+        if is_valid_held_sync_content_result(existing.get('result')):
+            return {'outcome': 'held', 'result': existing['result'], 'job_id': existing.get('job_id')}
+        return {'outcome': 'busy', 'job_id': existing.get('job_id')}
     if existing.get('dispatch_protocol') == 'batch_v1':
         result = existing.get('result')
         if existing.get('status') == 'completed' and is_valid_completed_sync_content_result(result):
@@ -293,6 +322,10 @@
     """
     snapshot = ref.get(transaction=transaction)
     existing = cast(Dict[str, Any], snapshot.to_dict() or {}) if getattr(snapshot, 'exists', False) else {}
+    if existing.get('dispatch_protocol') == 'stt_cap_hold_v1' and existing.get('status') == 'held':
+        if is_valid_held_sync_content_result(existing.get('result')):
+            return SyncContentRunBinding(SyncContentRunBindingOutcome.HELD, existing['result'])
+        return SyncContentRunBinding(SyncContentRunBindingOutcome.LOST)
     if existing.get('dispatch_protocol') == 'batch_v1':
         result = existing.get('result')
         if existing.get('status') == 'completed' and is_valid_completed_sync_content_result(result):