diff --git a/backend/charts/backend-listen/dev_omi_backend_listen_values.yaml b/backend/charts/backend-listen/dev_omi_backend_listen_values.yaml index 82fa3d4338b..ce198c825fd 100644 --- a/backend/charts/backend-listen/dev_omi_backend_listen_values.yaml +++ b/backend/charts/backend-listen/dev_omi_backend_listen_values.yaml @@ -138,6 +138,8 @@ env: value: "10" - name: SERVER_BATCH_UNSUPPORTED_LANGUAGES value: "" + - name: BATCH_HELD_DEAD_LETTER_ENABLED + value: "true" - name: BATCH_PIPELINE_ENABLED value: "false" - name: BATCH_PIPELINE_UID_ALLOWLIST diff --git a/backend/charts/backend-listen/prod_omi_backend_listen_values.yaml b/backend/charts/backend-listen/prod_omi_backend_listen_values.yaml index 5c4f497761c..d235d33aeeb 100644 --- a/backend/charts/backend-listen/prod_omi_backend_listen_values.yaml +++ b/backend/charts/backend-listen/prod_omi_backend_listen_values.yaml @@ -137,6 +137,8 @@ env: value: "af,ar,az,be,bn,bs,ca,cy,eu,fa,gl,gu,he,hi,id,kk,kn,ko,mk,ml,mr,ms,no,pa,sq,sr,sw,ta,te,th,tl,tr,ur,vi,zh" - name: SERVER_BATCH_ONLY_AUTO_LANGUAGE_PERCENT value: "100" + - name: BATCH_HELD_DEAD_LETTER_ENABLED + value: "true" - name: BATCH_PIPELINE_ENABLED value: "true" - name: FIRST_SUMMARY_DEADLINE_ENABLED diff --git a/backend/charts/batch-pipeline-worker/dev_omi_batch-pipeline-worker_values.yaml b/backend/charts/batch-pipeline-worker/dev_omi_batch-pipeline-worker_values.yaml index 5fe080f2d04..b02c09a0404 100644 --- a/backend/charts/batch-pipeline-worker/dev_omi_batch-pipeline-worker_values.yaml +++ b/backend/charts/batch-pipeline-worker/dev_omi_batch-pipeline-worker_values.yaml @@ -26,6 +26,7 @@ tolerations: value: present effect: NoSchedule worker: + heldDeadLetterEnabled: true servingMetricsMinRefreshSeconds: 30 shadowRecoveryApiUrl: "https://backend-integration-dt5lrfkkoa-uc.a.run.app" headPriorityEnabled: false diff --git a/backend/charts/batch-pipeline-worker/prod_omi_batch-pipeline-worker_values.yaml b/backend/charts/batch-pipeline-worker/prod_omi_batch-pipeline-worker_values.yaml index b646f0dd044..6b0b6843ea7 100644 --- a/backend/charts/batch-pipeline-worker/prod_omi_batch-pipeline-worker_values.yaml +++ b/backend/charts/batch-pipeline-worker/prod_omi_batch-pipeline-worker_values.yaml @@ -55,6 +55,7 @@ borrowedNodePools: enabled: false pools: [parakeet-pool, parakeet-pool-b, parakeet-pool-v2] worker: + heldDeadLetterEnabled: true # Two pump processes share each L4 (cl2-gpueff 2026-11-04, data/run-2026-11-04-gpueff # in omi-workspace): prod is claim-path bound (GPU util 16%, ~1.6 units per # 3.2s pull), so a second independent pump doubles concurrent claims per GPU. diff --git a/backend/charts/batch-pipeline-worker/staging_omi_batch-pipeline-worker_values.yaml b/backend/charts/batch-pipeline-worker/staging_omi_batch-pipeline-worker_values.yaml index 454c6e077ca..fb9fec1c60d 100644 --- a/backend/charts/batch-pipeline-worker/staging_omi_batch-pipeline-worker_values.yaml +++ b/backend/charts/batch-pipeline-worker/staging_omi_batch-pipeline-worker_values.yaml @@ -31,6 +31,7 @@ tolerations: value: present effect: NoSchedule worker: + heldDeadLetterEnabled: true apiUrl: https://backend-sync-staging-1031333818730.us-central1.run.app audience: https://backend-sync-staging-1031333818730.us-central1.run.app # Staging composes the prod manifest, so it inherits the shadow 25 pin; the diff --git a/backend/charts/batch-pipeline-worker/templates/deployment.yaml b/backend/charts/batch-pipeline-worker/templates/deployment.yaml index 83f7189794d..a1cfbbfa571 100644 --- a/backend/charts/batch-pipeline-worker/templates/deployment.yaml +++ b/backend/charts/batch-pipeline-worker/templates/deployment.yaml @@ -128,6 +128,7 @@ spec: - {name: BATCH_PIPELINE_WORKER_VALIDATION_DISPOSITION_ENABLED, value: {{ .Values.worker.workerValidationDispositionEnabled | quote }}} - {name: BATCH_PIPELINE_DIARIZE_SPEECH_ONLY_ENABLED, value: {{ .Values.worker.diarizeSpeechOnlyEnabled | quote }}} - {name: BATCH_PIPELINE_SHORT_SPEECH_SINGLE_ENABLED, value: {{ .Values.worker.shortSpeechSingleEnabled | quote }}} + - {name: BATCH_HELD_DEAD_LETTER_ENABLED, value: {{ .Values.worker.heldDeadLetterEnabled | quote }}} - {name: BATCH_PIPELINE_EMPTY_ASR_DISPOSITION_ENABLED, value: {{ .Values.worker.emptyAsrDispositionEnabled | quote }}} - {name: BATCH_PIPELINE_EMPTY_ASR_SAFETY_NET_ENABLED, value: {{ .Values.worker.emptyAsrSafetyNetEnabled | quote }}} - {name: BATCH_PIPELINE_EMPTY_ASR_RECOVERY_ENABLED, value: {{ .Values.worker.emptyAsrRecoveryEnabled | quote }}} diff --git a/backend/charts/batch-pipeline-worker/values.yaml b/backend/charts/batch-pipeline-worker/values.yaml index aa6acd440f9..9ef1997acfd 100644 --- a/backend/charts/batch-pipeline-worker/values.yaml +++ b/backend/charts/batch-pipeline-worker/values.yaml @@ -49,6 +49,7 @@ borrowedNodePools: # autoscaler's expendable cutoff (-10), so a Pending borrower is not ignored. priority: -1 worker: + heldDeadLetterEnabled: true # Match the serving hosts; prod overrides this below the 120s guard blind window. servingMetricsMinRefreshSeconds: 30 # Control-loop defaults; rollback independently with 64 / false / 2 / {}. diff --git a/backend/charts/monitoring/alert-rules.json b/backend/charts/monitoring/alert-rules.json index 251f32cdfae..b5709348023 100644 --- a/backend/charts/monitoring/alert-rules.json +++ b/backend/charts/monitoring/alert-rules.json @@ -28676,5 +28676,178 @@ }, "record": null, "missingSeriesEvalsToResolve": 2 + }, + { + "uid": "omi-batch-held-expired-volume", + "orgID": 1, + "folderUID": "betdycdziadc0e", + "ruleGroup": "Batch held dead letters", + "title": "Retained held batch dead-letter volume", + "condition": "E", + "data": [ + { + "refId": "A", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "max(omi_batch_held_expired_units and (time() - omi_batch_held_expired_observed_at_seconds < 900)) or vector(0)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "A" + } + }, + { + "refId": "B", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "sum(increase(omi_batch_held_expired_total[30m])) or vector(0)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "B" + } + }, + { + "refId": "C", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "(time() - max(omi_batch_held_expired_observed_at_seconds)) or vector(1e9)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "C" + } + }, + { + "refId": "D", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "($A > 100 && $C < 900) || $B > 10", + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "D", + "type": "math" + } + }, + { + "refId": "E", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "conditions": [ + { + "evaluator": { + "params": [], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": [ + "E" + ] + }, + "reducer": { + "params": [], + "type": "last" + }, + "type": "query" + } + ], + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "D", + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "last", + "refId": "E", + "type": "reduce" + } + } + ], + "updated": "2026-11-06T00:00:00Z", + "noDataState": "OK", + "execErrState": "KeepLast", + "for": "5m", + "keep_firing_for": "0s", + "annotations": { + "dashboard_url": "https://monitor.omi.me/d/omi-core-features", + "summary": "Held batch dead letters exceed the retained-stock or new-expiry volume threshold.", + "user_impact": "Recorded audio remains retained but these units have no published transcript.", + "scope": "Stock is only state=dead with reason held_expired, a new explicit cohort; legacy unrelated dead rows never count. Rate counts unique CAS expirations over 30 minutes. Stock requires a sample younger than 15 minutes. Agent tier only. Each replica stock sample is filtered by its own timestamp before max aggregation, so another replica cannot refresh stale stock.", + "verification": "Read omi_batch_held_expired_units and omi_batch_held_expired_total, then inspect reasoned rows using an authorized operator read.", + "safe_next_action": "Check nightly re-drive and worker failure reasons. Use the private OIDC requeue-quarantined path after resolving the cause. Never delete audio or clear privacy fences.", + "runbook": "backend/docs/runbooks/batch-held-dead-letter.md", + "threshold": "Fresh retained stock >100 or >10 expirations in 30m for 5m; initial agent thresholds, not calibrated from production.", + "evaluated_bad": "Synthetic: stock 101 sampled 300s ago, or 11 new expirations in 30m.", + "evaluated_good": "Synthetic: stock 100 and 10 new expirations, or stale stock without a rate breach." + }, + "labels": { + "alert_identity": "omi-batch-held-expired-volume", + "component": "batch-pipeline", + "impact": "product", + "severity": "warning", + "instatus_component": "speech-processing", + "tier": "agent", + "audience": "agent", + "owner": "uicleanup" + }, + "isPaused": false, + "notification_settings": { + "receiver": "Omi - Services Alerting (Telegram)", + "group_by": [ + "alertname" + ], + "group_wait": "5m", + "group_interval": "1h", + "repeat_interval": "24h", + "mute_time_intervals": [ + "omi-tier-silent" + ] + }, + "record": null } ] diff --git a/backend/charts/monitoring/alerts/batch-held-dead-letter.json b/backend/charts/monitoring/alerts/batch-held-dead-letter.json new file mode 100644 index 00000000000..5079f2266a4 --- /dev/null +++ b/backend/charts/monitoring/alerts/batch-held-dead-letter.json @@ -0,0 +1,175 @@ +[ + { + "uid": "omi-batch-held-expired-volume", + "orgID": 1, + "folderUID": "betdycdziadc0e", + "ruleGroup": "Batch held dead letters", + "title": "Retained held batch dead-letter volume", + "condition": "E", + "data": [ + { + "refId": "A", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "max(omi_batch_held_expired_units and (time() - omi_batch_held_expired_observed_at_seconds < 900)) or vector(0)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "A" + } + }, + { + "refId": "B", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "sum(increase(omi_batch_held_expired_total[30m])) or vector(0)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "B" + } + }, + { + "refId": "C", + "queryType": "", + "relativeTimeRange": { + "from": 900, + "to": 0 + }, + "datasourceUid": "prometheus", + "model": { + "datasource": { + "type": "prometheus", + "uid": "prometheus" + }, + "expr": "(time() - max(omi_batch_held_expired_observed_at_seconds)) or vector(1e9)", + "instant": true, + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "C" + } + }, + { + "refId": "D", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "($A > 100 && $C < 900) || $B > 10", + "intervalMs": 1000, + "maxDataPoints": 43200, + "refId": "D", + "type": "math" + } + }, + { + "refId": "E", + "queryType": "", + "relativeTimeRange": { + "from": 0, + "to": 0 + }, + "datasourceUid": "__expr__", + "model": { + "conditions": [ + { + "evaluator": { + "params": [], + "type": "gt" + }, + "operator": { + "type": "and" + }, + "query": { + "params": [ + "E" + ] + }, + "reducer": { + "params": [], + "type": "last" + }, + "type": "query" + } + ], + "datasource": { + "type": "__expr__", + "uid": "__expr__" + }, + "expression": "D", + "intervalMs": 1000, + "maxDataPoints": 43200, + "reducer": "last", + "refId": "E", + "type": "reduce" + } + } + ], + "updated": "2026-11-06T00:00:00Z", + "noDataState": "OK", + "execErrState": "KeepLast", + "for": "5m", + "keep_firing_for": "0s", + "annotations": { + "dashboard_url": "https://monitor.omi.me/d/omi-core-features", + "summary": "Held batch dead letters exceed the retained-stock or new-expiry volume threshold.", + "user_impact": "Recorded audio remains retained but these units have no published transcript.", + "scope": "Stock is only state=dead with reason held_expired, a new explicit cohort; legacy unrelated dead rows never count. Rate counts unique CAS expirations over 30 minutes. Stock requires a sample younger than 15 minutes. Agent tier only. Each replica stock sample is filtered by its own timestamp before max aggregation, so another replica cannot refresh stale stock.", + "verification": "Read omi_batch_held_expired_units and omi_batch_held_expired_total, then inspect reasoned rows using an authorized operator read.", + "safe_next_action": "Check nightly re-drive and worker failure reasons. Use the private OIDC requeue-quarantined path after resolving the cause. Never delete audio or clear privacy fences.", + "runbook": "backend/docs/runbooks/batch-held-dead-letter.md", + "threshold": "Fresh retained stock >100 or >10 expirations in 30m for 5m; initial agent thresholds, not calibrated from production.", + "evaluated_bad": "Synthetic: stock 101 sampled 300s ago, or 11 new expirations in 30m.", + "evaluated_good": "Synthetic: stock 100 and 10 new expirations, or stale stock without a rate breach." + }, + "labels": { + "alert_identity": "omi-batch-held-expired-volume", + "component": "batch-pipeline", + "impact": "product", + "severity": "warning", + "instatus_component": "speech-processing", + "tier": "agent", + "audience": "agent", + "owner": "uicleanup" + }, + "isPaused": false, + "notification_settings": { + "receiver": "Omi - Services Alerting (Telegram)", + "group_by": [ + "alertname" + ], + "group_wait": "5m", + "group_interval": "1h", + "repeat_interval": "24h", + "mute_time_intervals": [ + "omi-tier-silent" + ] + }, + "record": null + } +] diff --git a/backend/charts/monitoring/gmp-query-migration-inventory.json b/backend/charts/monitoring/gmp-query-migration-inventory.json index 91aa3ce66d9..011b7a8d777 100644 --- a/backend/charts/monitoring/gmp-query-migration-inventory.json +++ b/backend/charts/monitoring/gmp-query-migration-inventory.json @@ -118,6 +118,46 @@ "pointer": "/data/0/model/expr", "scope": "normalized-omi-consumer-source-review" }, + { + "expression": "max(omi_batch_held_expired_units and (time() - omi_batch_held_expired_observed_at_seconds < 900)) or vector(0)", + "file": "alerts/batch-held-dead-letter.json", + "issues": [], + "metrics": [ + "omi_batch_held_expired_observed_at_seconds", + "omi_batch_held_expired_units" + ], + "owner_title": "Retained held batch dead-letter volume", + "owner_uid": "omi-batch-held-expired-volume", + "panel": null, + "pointer": "/data/0/model/expr", + "scope": "normalized-omi-consumer-source-review" + }, + { + "expression": "sum(increase(omi_batch_held_expired_total[30m])) or vector(0)", + "file": "alerts/batch-held-dead-letter.json", + "issues": [], + "metrics": [ + "omi_batch_held_expired_total" + ], + "owner_title": "Retained held batch dead-letter volume", + "owner_uid": "omi-batch-held-expired-volume", + "panel": null, + "pointer": "/data/1/model/expr", + "scope": "normalized-omi-consumer-source-review" + }, + { + "expression": "(time() - max(omi_batch_held_expired_observed_at_seconds)) or vector(1e9)", + "file": "alerts/batch-held-dead-letter.json", + "issues": [], + "metrics": [ + "omi_batch_held_expired_observed_at_seconds" + ], + "owner_title": "Retained held batch dead-letter volume", + "owner_uid": "omi-batch-held-expired-volume", + "panel": null, + "pointer": "/data/2/model/expr", + "scope": "normalized-omi-consumer-source-review" + }, { "expression": "((max(time() - omi_batch_pipeline_metrics_refreshed_at_seconds{namespace=\"prod-omi-backend\"}) >= bool 120) or vector(1)) + (count(count by (priority) (omi_batch_pipeline_oldest_age_seconds{namespace=\"prod-omi-backend\"} != omi_batch_pipeline_oldest_age_seconds{namespace=\"prod-omi-backend\"})) or vector(0)) + (count(count by (priority) (omi_batch_pipeline_retry_age_seconds{namespace=\"prod-omi-backend\"} != omi_batch_pipeline_retry_age_seconds{namespace=\"prod-omi-backend\"})) or vector(0))", "file": "alerts/batch-queue-safety.json", @@ -3738,7 +3778,7 @@ ], "counts": { "explicit_bridge": 25, - "query_consumers": 248 + "query_consumers": 251 }, "schema_version": 1, "source": "repository, not live Grafana export" diff --git a/backend/charts/monitoring/metric-inventory.prod.json b/backend/charts/monitoring/metric-inventory.prod.json index 8d19af290d2..0e829348509 100644 --- a/backend/charts/monitoring/metric-inventory.prod.json +++ b/backend/charts/monitoring/metric-inventory.prod.json @@ -2304,6 +2304,9 @@ "workqueue_work_duration_seconds_sum" ], "declared_metrics": [ + "omi_batch_held_expired_total", + "omi_batch_held_expired_units", + "omi_batch_held_expired_observed_at_seconds", "listen_finalization_byok_session_abandoned_last_24h", "listen_finalization_byok_session_abandoned_total", "listen_finalization_oldest_dead_letter_age_seconds", diff --git a/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_omi_cloud_run_metrics_exporter.yaml b/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_omi_cloud_run_metrics_exporter.yaml index 291c1566d30..d23ee2402fd 100644 --- a/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_omi_cloud_run_metrics_exporter.yaml +++ b/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_omi_cloud_run_metrics_exporter.yaml @@ -17,6 +17,10 @@ stackdriver: - 'prometheus.googleapis.com/omi_enrichment_summary_repair_' # WP-C lifecycle dwell and observer health (closed labels). - 'prometheus.googleapis.com/omi_batch_record_pending_' + # Retained held dead-letter stock/rate alert; serving maintenance emits these. + - 'prometheus.googleapis.com/omi_batch_held_expired_total/' + - 'prometheus.googleapis.com/omi_batch_held_expired_units/' + - 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/' - 'prometheus.googleapis.com/omi_record_seal_nojob_' - 'prometheus.googleapis.com/omi_enrichment_obligation_' - 'prometheus.googleapis.com/omi_finalization_byok_' diff --git a/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_sync_omi_cloud_run_metrics_exporter.yaml b/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_sync_omi_cloud_run_metrics_exporter.yaml index 25460e004dd..7344ca6162e 100644 --- a/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_sync_omi_cloud_run_metrics_exporter.yaml +++ b/backend/charts/monitoring/prometheus-stackdriver-exporter/dev_sync_omi_cloud_run_metrics_exporter.yaml @@ -36,6 +36,10 @@ stackdriver: - 'prometheus.googleapis.com/omi_enrichment_summary_repair_' # WP-C lifecycle dwell and observer health (closed labels). - 'prometheus.googleapis.com/omi_batch_record_pending_' + # Retained held dead-letter stock/rate alert; serving maintenance emits these. + - 'prometheus.googleapis.com/omi_batch_held_expired_total/' + - 'prometheus.googleapis.com/omi_batch_held_expired_units/' + - 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/' - 'prometheus.googleapis.com/omi_record_seal_nojob_' - 'prometheus.googleapis.com/omi_enrichment_obligation_' - 'prometheus.googleapis.com/omi_finalization_byok_' diff --git a/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_omi_cloud_run_metrics_exporter.yaml b/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_omi_cloud_run_metrics_exporter.yaml index 7f1013f3ab8..5b8092afbe3 100644 --- a/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_omi_cloud_run_metrics_exporter.yaml +++ b/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_omi_cloud_run_metrics_exporter.yaml @@ -29,6 +29,10 @@ stackdriver: - 'prometheus.googleapis.com/omi_enrichment_summary_repair_' # WP-C lifecycle dwell and observer health (closed labels). - 'prometheus.googleapis.com/omi_batch_record_pending_' + # Retained held dead-letter stock/rate alert; serving maintenance emits these. + - 'prometheus.googleapis.com/omi_batch_held_expired_total/' + - 'prometheus.googleapis.com/omi_batch_held_expired_units/' + - 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/' - 'prometheus.googleapis.com/omi_record_seal_nojob_' - 'prometheus.googleapis.com/omi_enrichment_obligation_' - 'prometheus.googleapis.com/omi_finalization_byok_' diff --git a/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_sync_omi_cloud_run_metrics_exporter.yaml b/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_sync_omi_cloud_run_metrics_exporter.yaml index d2884a2f80a..278ad00ffee 100644 --- a/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_sync_omi_cloud_run_metrics_exporter.yaml +++ b/backend/charts/monitoring/prometheus-stackdriver-exporter/prod_sync_omi_cloud_run_metrics_exporter.yaml @@ -36,6 +36,10 @@ stackdriver: - 'prometheus.googleapis.com/omi_enrichment_summary_repair_' # WP-C lifecycle dwell and observer health (closed labels). - 'prometheus.googleapis.com/omi_batch_record_pending_' + # Retained held dead-letter stock/rate alert; serving maintenance emits these. + - 'prometheus.googleapis.com/omi_batch_held_expired_total/' + - 'prometheus.googleapis.com/omi_batch_held_expired_units/' + - 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/' - 'prometheus.googleapis.com/omi_record_seal_nojob_' - 'prometheus.googleapis.com/omi_enrichment_obligation_' - 'prometheus.googleapis.com/omi_finalization_byok_' diff --git a/backend/config/batch_pipeline.py b/backend/config/batch_pipeline.py index de36b5c814e..4ead00932b2 100644 --- a/backend/config/batch_pipeline.py +++ b/backend/config/batch_pipeline.py @@ -30,6 +30,16 @@ class Priority(IntEnum): # David's fresh bar (2026-10-20): capture lag at most 10 min at upload. FRESH_CAPTURE_LAG_SECONDS = 600.0 +# A full grace window is also given to pre-field rows on first observation. +HELD_DEAD_LETTER_SECONDS = 7 * 86400 + + +def batch_held_dead_letter_enabled() -> bool: + """Ship-on held deadlines/expiry; manual requeue remains available when off.""" + from config.policy.kill_switches import kill_switch + + return kill_switch('BATCH_HELD_DEAD_LETTER_ENABLED') + def projection_source_deleted_enabled() -> bool: """Ship-on no-text settlement for a deleted source without a tombstone or redirect.""" diff --git a/backend/config/flag_budget.json b/backend/config/flag_budget.json index 3d530a625c1..84dc8265e5a 100644 --- a/backend/config/flag_budget.json +++ b/backend/config/flag_budget.json @@ -24,8 +24,9 @@ }, "seed_review_by": "2026-11-15", "budget": 169, - "count": 262, + "count": 263, "names": { + "BATCH_HELD_DEAD_LETTER_ENABLED": {"owner": "uicleanup / omi-lifecycle", "review_by": "2026-12-06", "registry": "config/feature-flags.yaml"}, "BATCH_PIPELINE_SERVING_SWEEP_PAGE_ROWS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-06"}, "BATCH_PIPELINE_SERVING_TICK_SECONDS": {"owner": "omi-pipeline / lead2", "review_by": "2026-12-06"}, "PUSHER_PROJECTION_DEBT_DURABLE_ENABLED": {"owner": "uicleanup / omi-lifecycle", "review_by": "2026-12-06"}, diff --git a/backend/config/policy/kill_switches.py b/backend/config/policy/kill_switches.py index e0d29905ab7..4a216b19928 100644 --- a/backend/config/policy/kill_switches.py +++ b/backend/config/policy/kill_switches.py @@ -8,6 +8,7 @@ import os KILL_SWITCHES = { + 'BATCH_HELD_DEAD_LETTER_ENABLED': True, 'BATCH_PIPELINE_LEADER_LEASE_ENABLED': True, 'BATCH_PIPELINE_RECOVERY_TASKS_ENABLED': True, 'BATCH_PIPELINE_PROJECTION_PAGINATION_ENABLED': True, diff --git a/backend/database/batch_pipeline_projection.py b/backend/database/batch_pipeline_projection.py index f8346505b1a..1c2602b5ebf 100644 --- a/backend/database/batch_pipeline_projection.py +++ b/backend/database/batch_pipeline_projection.py @@ -882,14 +882,15 @@ def held_without_recovery(job: dict) -> bool: """ from config.policy import RecoveryPolicy - if job.get('state') != 'quarantined' or job.get('result'): + retained_dead = job.get('state') == 'dead' and job.get('dead_letter_reason') == 'held_expired' + if (job.get('state') != 'quarantined' and not retained_dead) or job.get('result'): return False if any( isinstance(job.get(key), dict) and job[key].get('state') not in {'completed', 'failed_held'} for key in OPERATOR_INTENTS ): return False - if job.get('recovery_state') == 'failed_held': + if retained_dead or job.get('recovery_state') == 'failed_held': return True raw = (job.get('policy_parts') or {}).get('recovery') return ( diff --git a/backend/database/batch_pipeline_quarantine_recovery.py b/backend/database/batch_pipeline_quarantine_recovery.py index 241323738a6..4029dc8551a 100644 --- a/backend/database/batch_pipeline_quarantine_recovery.py +++ b/backend/database/batch_pipeline_quarantine_recovery.py @@ -16,7 +16,8 @@ from config.batch_paid_attempt import attempted, renewed, retryable from utils.batch_pipeline.correctness_metrics import record as record_correctness from utils.batch_pipeline.contracts import WorkResult, held_vad_census, held_silence_allowed, held_result_allowed -from utils.batch_pipeline.scheduler import owner_key +from utils.batch_pipeline.scheduler import Scheduler, owner_key +from config.batch_pipeline import batch_held_dead_letter_enabled, HELD_DEAD_LETTER_SECONDS # A receipt proves words existed, but deliberately contains no transcript. # Permit one replacement recognition only after the original intent's dwell. @@ -115,6 +116,36 @@ def checked( raise ProjectionBlocked('recovery content owner lost') return job, binding + def expire_held(self, segment_id, now): + if not batch_held_dead_letter_enabled(): + return 'disabled' + + def commit(tx): + job, _ = self.checked(tx, segment_id) + return expire_held_in_transaction(tx, job, now) + + return self.store.run(commit) + + def requeue_quarantined(self, segment_id, now, *, dead_before=None): + """Operator and nightly entry: retain the content/privacy ownership CAS.""" + + def commit(tx): + job, _ = self.checked(tx, segment_id) + if dead_before is not None: + expired_at = job.get('held_expired_at') + if ( + job.get('state') != 'dead' + or job.get('dead_letter_reason') != 'held_expired' + or isinstance(expired_at, bool) + or not isinstance(expired_at, (int, float)) + or not math.isfinite(expired_at) + or expired_at >= dead_before + ): + return False + return Scheduler.requeue_quarantined_in_transaction(tx, job, now) + + return self.store.run(commit) + def prepare(self, segment_id, owner_id, now): load_stage_policy(self.store, segment_id, 'recovery', server_flags=self.server_flags) # Generated outside the retryable transaction; a lost reply is reconciled @@ -324,6 +355,8 @@ def commit(tx): # This is a local result failure, not an abandoned recognizer. marker.update(settled_at=now, failure={'stage': 'payload_validation', 'reason': 'operation_failed'}) job['recovery_state'] = 'failed_held' + if batch_held_dead_letter_enabled(): + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS tx.put('jobs', segment_id, job) return 'exhausted' marker['missing_result_history'] = [ @@ -364,6 +397,8 @@ def commit(tx): ): return False job['recovery_state'] = 'failed_held' + if batch_held_dead_letter_enabled(): + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS job['measured_held_vad_speech_seconds'] = held_vad_census(job, 0) job['recovery'].update( settled_at=now, @@ -402,6 +437,8 @@ def commit(tx): ): return False job['recovery_state'] = 'failed_held' + if batch_held_dead_letter_enabled(): + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS job['recovery'].update(settled_at=now, refusal={'reason': 'invalid_retained_manifest'}) tx.put('jobs', segment_id, job) return True @@ -548,6 +585,8 @@ def commit(tx): job['measured_held_vad_speech_seconds'] = census if result_uri is None: job.update(recovery_state='failed_held') + if batch_held_dead_letter_enabled(): + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS if failure: from utils.batch_pipeline.recovery_disposition import STAGES, REASONS @@ -577,6 +616,47 @@ def commit(tx): return changed +def expire_held_in_transaction(tx, job, now): + """CAS a dormant compute hold into a retained, reasoned dead letter. + + No artifact IO, TTL or conversation write. Callers retain RecoveryStore.checked + and the store's commit-boundary privacy fence. Old rows get a full window. + """ + if not batch_held_dead_letter_enabled(): + return 'disabled' + if not Scheduler.held_requeue_eligible(tx, job, now, allow_dead=False): + return 'held' + deadline = job.get('held_deadline_at') + if deadline is None: + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS + tx.put('jobs', job['segment_id'], job) + return 'stamped' + if isinstance(deadline, bool) or not isinstance(deadline, (int, float)) or not math.isfinite(deadline): + return 'held' # Malformed metadata is not permission to expire custody. + if deadline > now: + return 'held' + # The pre-existing redrive_dead route requires merge_pending. Preserve that + # recovery contract even if this feature is disabled or its code reverted. + job.update(state='dead', merge_pending=True, dead_letter_reason='held_expired', held_expired_at=now) + tx.put('jobs', job['segment_id'], job) + return 'expired' + + +def dead_held_candidates(client, namespace, limit=32, *, after_id=None): + """Only this reason is nightly redriven; historical dead stock is untouched.""" + collection = client.collection(f'{namespace}_jobs') + query = collection.where('state', '==', 'dead').where('dead_letter_reason', '==', 'held_expired') + query = query.order_by('__name__') + if after_id is not None: + query = query.start_after({'__name__': collection.document(after_id)}) + return [(doc.id, doc.to_dict() or {}) for doc in query.limit(max(1, min(limit, 64))).stream()] + + +def held_expired_stock(client, namespace): + query = client.collection(f'{namespace}_jobs').where('state', '==', 'dead') + return query.where('dead_letter_reason', '==', 'held_expired').count().get()[0][0].value + + def recovery_candidates(client, namespace: str, owner_id: str | None, limit: int = 10, *, after_id: str | None = None): """Only unattempted or reconcilable units; failed STT never hot-loops.""" collection = client.collection(f'{namespace}_jobs') diff --git a/backend/deploy/cloud_run_gmp_sync.yaml b/backend/deploy/cloud_run_gmp_sync.yaml index d5d9f474360..0a4824b9998 100644 --- a/backend/deploy/cloud_run_gmp_sync.yaml +++ b/backend/deploy/cloud_run_gmp_sync.yaml @@ -43,7 +43,7 @@ spec: - action: keep sourceLabels: - __name__ - regex: 'omi_(finalization_pair_.*|enrichment_summary_repair_.*|batch_record_pending_.*|record_seal_nojob_.*|enrichment_obligation_.*|enrichment_early_abort_total|enrichment_supersede_streak(_bucket|_count|_sum)?|finalization_byok_.*|listen_server_batch_suppression_total|listen_preview_batch_custody_total|delivery_outcome_total|people_scan_dedupe_total|pinecone_.*|playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_projection_drains_skipped_total|batch_projection_claim_busy_backoffs_total|batch_projection_page_collapsed_rows_total|batch_projection_drain_head_requeues_total|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*|transcription_final_pass_total|transcription_final_pass_(total|unit_skips_total)|transcription_final_pass_(total|unit_skips_total|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count))|transcription_final_pass_(total|unit_skips_total|retry_terminal_(total|observed_at_seconds)|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count)))|conversation_llm_input_dedupe_.*|privacy_(checks|check_calls|check_reads|decision_mismatch|shadow_checks|memo_requests|memo_ineligible|ttl_cache)_total' + regex: 'omi_(finalization_pair_.*|enrichment_summary_repair_.*|batch_record_pending_.*|record_seal_nojob_.*|enrichment_obligation_.*|enrichment_early_abort_total|enrichment_supersede_streak(_bucket|_count|_sum)?|finalization_byok_.*|listen_server_batch_suppression_total|listen_preview_batch_custody_total|delivery_outcome_total|people_scan_dedupe_total|pinecone_.*|playback_writes_total|sync_.*|journey_.*|client_journey_.*|owner_recognition_conversations_total|conversation_duration_seconds(_.*)?|conversation_relevance_decision_total|conversation_summary_vector_upserts_total|conversation_relevance_decision_total|firestore_documents_per_operation(_.*)?|capture_finalization_.*|queue_oldest_ready_.*|batch_pipeline_.*|batch_held_expired_(total|units|observed_at_seconds)|batch_projection_drains_skipped_total|batch_projection_claim_busy_backoffs_total|batch_projection_page_collapsed_rows_total|batch_projection_drain_head_requeues_total|batch_intake_outcome_total|batch_admission_sample_window_total|batch_ramp_heavy_owner_excluded_total|batch_shadow_.*|accept_to_summary_seconds(_.*)?|summary_terminal_total|summary_debt_.*|live_session_transcript_outcome_total|fallback_total|parakeet_request_seconds(_.*)?|firestore_billed_reads_by_(caller|origin)_total|firestore_caller_label_overflow_total|firestore_billed_writes(_by_caller)?_total|firestore_write_caller_label_overflow_total|firestore_ambiguous_commit_writes_total|firestore_noop_writes_by_caller_total|memory_released_projection_total|fsr_leader_lease_total|held_content_.*|held_recovery_pacing_.*|stt_cap_recovery_reconciliation_total|stt_cap_recovery_cohort_narrowed|stt_cap_cohort_.*|stt_cap_launch_config_errors.*|stt_cap_dispatch_decisions_total|stt_cap_provider_audio_seconds_total|legacy_sync_.*|legacy_dark_.*|jobstore_.*|capture_coverage_.*|capture_dedupe_shadow_total|capture_retry_cap_total|speaker_evidence_.*|speaker_evidence_confident_owner_total|lifecycle_refactor_shadow_total|executor_task_(active|queued)|clients_refactor_shadow_total|notes_refactor_shadow_total|notes_refactor_results_total|stt_language_gate_.*|prerecorded_vendor_.*|episode_compose_.*|transcription_final_pass_total|transcription_final_pass_(total|unit_skips_total)|transcription_final_pass_(total|unit_skips_total|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count))|transcription_final_pass_(total|unit_skips_total|retry_terminal_(total|observed_at_seconds)|(agreement|latency_seconds|owner_delta_ratio)_(bucket|sum|count)))|conversation_llm_input_dedupe_.*|privacy_(checks|check_calls|check_reads|decision_mismatch|shadow_checks|memo_requests|memo_ineligible|ttl_cache)_total' # These eagerly initialized live-STT combinations dominate empty series # on sync workers. Keep fallbacks from sync/summary/storage components. - action: drop diff --git a/backend/deploy/runtime_env.yaml b/backend/deploy/runtime_env.yaml index d96d0d00600..138a38bfa5b 100644 --- a/backend/deploy/runtime_env.yaml +++ b/backend/deploy/runtime_env.yaml @@ -45,6 +45,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '0', category: rollout} @@ -805,6 +808,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '0', category: rollout} @@ -1080,6 +1086,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '0', category: rollout} @@ -1385,6 +1394,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '0', category: rollout} @@ -1671,6 +1683,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '0', category: rollout} @@ -2199,6 +2214,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '100', category: rollout} @@ -3091,6 +3109,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'false', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '100', category: rollout} @@ -3497,6 +3518,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '100', category: rollout} @@ -3935,6 +3959,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '100', category: rollout} @@ -4334,6 +4361,9 @@ environments: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: + value: 'true' + category: ops_kill BATCH_PIPELINE_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_PRESERVE_ADMITTED_ENABLED: {value: 'true', category: rollout} FIRST_SUMMARY_DEADLINE_PERCENT: {value: '100', category: rollout} diff --git a/backend/deploy/runtime_env/_base.yaml b/backend/deploy/runtime_env/_base.yaml index c3a5006e88a..91cc6d54b50 100644 --- a/backend/deploy/runtime_env/_base.yaml +++ b/backend/deploy/runtime_env/_base.yaml @@ -67,6 +67,7 @@ environment_shared: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: {value: 'true', category: ops_kill} BATCH_PIPELINE_ENABLED: value: 'false' category: rollout @@ -927,6 +928,7 @@ environment_shared: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: {value: 'true', category: ops_kill} BATCH_PIPELINE_ENABLED: value: 'false' category: rollout @@ -1305,6 +1307,7 @@ environment_shared: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: {value: 'true', category: ops_kill} BATCH_PIPELINE_ENABLED: value: 'false' category: rollout @@ -1668,6 +1671,7 @@ environment_shared: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: {value: 'true', category: ops_kill} BATCH_PIPELINE_ENABLED: value: 'false' category: rollout @@ -2022,6 +2026,7 @@ environment_shared: BATCH_CONTROL_CANDIDATE_FANOUT: {value: '16', category: rollout} BATCH_ADMISSION_PARALLEL_CALLS: {value: 'true', category: rollout} BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS: {value: '10', category: rollout} + BATCH_HELD_DEAD_LETTER_ENABLED: {value: 'true', category: ops_kill} BATCH_PIPELINE_ENABLED: value: 'false' category: rollout diff --git a/backend/docs/feature-flag-registry.md b/backend/docs/feature-flag-registry.md index 619f6935e71..2bd9c0812b3 100644 --- a/backend/docs/feature-flag-registry.md +++ b/backend/docs/feature-flag-registry.md @@ -513,6 +513,7 @@ and an explicit empty literal renders as `''`. | `BATCH_CONVERSATION_SPEAKERS_ENABLED` | Default on within persisted authoritative batch records; link unit speaker groups acoustically and name conversation groups | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_FINAL_PASS_ENABLED` | Default on: whole-conversation diarization before summaries for batch authority with available staged artifacts | backend | env | open | — | — | config_map (gke/backend-listen, gke/pusher); valueFrom (backend-listen (chart), pusher (chart)); {value: 'true', category: ops_kill} (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill) | — | keep | keep | — | dazheng | | `BATCH_FINAL_PASS_PERCENT` | Whole-conversation batch final-pass percentage; default 100 | backend | env | closed | — | — | — | — | keep | keep | — | dazheng | +| `BATCH_HELD_DEAD_LETTER_ENABLED` | Default on; seven-day held deadlines, retained dead letters and bounded nightly re-drive | backend | env | open | {value: 'true', category: ops_kill} | true (backend-listen (chart)); {value: 'true', category: ops_kill} (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | true (backend-listen (chart)); {value: 'true', category: ops_kill} (cloud_run/backend, cloud_run/backend-integration, cloud_run/backend-sync, cloud_run/backend-sync-backfill, gke/backend-listen) | — | keep | keep | — | uicleanup | | `BATCH_LEGACY_CAPACITY_FALLBACK_ENABLED` | Default on; while the GPU ring reads saturated (see BATCH_VENDOR_OVERFLOW_MIN_READY_UNITS), a new fresh upload from an owner whose legacy sync is still on stays on legacy sync instead of queueing behind the ring. false queues it on batch as before | backend | env | open | — | — | — | — | keep | keep | — | two-lane lead | | `BATCH_PIPELINE_ASR_CUDA_GRAPHS_ENABLED` | Default on unless false; batch worker Parakeet TDT greedy_batch label-looping decoding runs NeMo CUDA graphs instead of eager kernels (same algorithm, identical words); the chosen mode is asserted before and after every transcription and a mismatch quarantines the process | backend | env | open | — | — | — | — | keep | keep | — | dazheng | | `BATCH_PIPELINE_CONTENTION_RETRY_ENABLED` | Default on; batch Firestore transactions retry explicit contention on fresh transaction IDs with bounded jitter; false restores legacy SDK transaction behavior | backend | env | open | — | — | — | — | keep | keep | — | dazheng | diff --git a/backend/docs/listen_pusher_pipeline/changes/2026-11-06-batch-held-dead-letter.md b/backend/docs/listen_pusher_pipeline/changes/2026-11-06-batch-held-dead-letter.md new file mode 100644 index 00000000000..410234fc643 --- /dev/null +++ b/backend/docs/listen_pusher_pipeline/changes/2026-11-06-batch-held-dead-letter.md @@ -0,0 +1,16 @@ +# Retained batch held expiry and rollback recovery + +GPU worker holds gain a seven-day deadline, retained `dead(held_expired)` state, +a bounded nightly re-drive and an agent-tier stock/rate alert. The default-on +`BATCH_HELD_DEAD_LETTER_ENABLED` kill switch stops new stamps and automatic +maintenance; manual requeue remains available. See the +[runbook](../../runbooks/batch-held-dead-letter.md) and +[pipeline contract](../../listen_pusher_pipeline.mdx). + +Expiry atomically sets `merge_pending=true` so the existing `redrive_dead` path +can recover rows after a code revert, even before the first nightly pass. +The reason, deadline, encrypted audio references and attempt history remain. +Nothing is deleted and privacy/content fences are unchanged. Rollback recovery +is covered by unit tests and a loopback Firestore race using only the legacy +re-drive with the feature disabled; live execution and alert delivery were not +measured. This PR does not deploy. diff --git a/backend/docs/runbooks/batch-held-dead-letter.md b/backend/docs/runbooks/batch-held-dead-letter.md new file mode 100644 index 00000000000..cde66022778 --- /dev/null +++ b/backend/docs/runbooks/batch-held-dead-letter.md @@ -0,0 +1,72 @@ +# Retained batch held dead letters + +GPU batch holds (`quarantined`, `recovery_state=held|failed_held`, +`quarantine_reason=worker_failed`) gain a seven-day `held_deadline_at`. +The initial quarantine and each failed recovery settlement stamp a full window. +A legacy row without the field receives a full window at its first eligible +sweep; capture/enqueue age is never used to expire old stock immediately. + +`BATCH_HELD_DEAD_LETTER_ENABLED` is default **on**. False stops new stamps +and automatic held maintenance without changing legacy quarantine writes. +The private requeue endpoint remains available. Rollback: set this flag false +on the serving and worker roles; revert the PR to remove the code. Expiry sets +`merge_pending=true` in the same transaction as `state=dead`, so the existing +`POST /v2/batch-pipeline/redrive-dead` route (`Scheduler.redrive_dead`) accepts +these rows even after the new route and nightly helper are reverted. It retains +`dead_letter_reason=held_expired`, `held_deadline_at` and `held_expired_at`. +No backward data rewrite or pre-revert drain is needed for these rows. The +rolled-back service requires an authorized operator to invoke its existing +re-drive; it does not run the new nightly pass. Deployment declarations are +source only; this change does not deploy. + +The existing serving maintenance leader scans 32 held rows every five minutes, +with a durable document-key cursor. The transaction rechecks content ownership, +privacy generation, dormant held state, no result (including the result receipt), +no worker token, and an expired lease. Active recovery/operator intents block it. +Eligible expired rows become `state=dead`, `dead_letter_reason=held_expired`, +with `held_expired_at` and `merge_pending=true` for legacy re-drive. They retain +their encrypted source references, intent and attempt history and all other fields. No storage call, TTL, audio deletion, +conversation-status write or privacy-fence change occurs. + +A nightly UTC pass re-attempts only `dead(held_expired)` rows expired before +that midnight. It shares the maintenance leader, uses the same 32-row page and +five-minute cadence, and resumes its durable cursor across restarts until the +pass ends. Each row goes through `RecoveryStore.checked` and the sanctioned +`Scheduler.requeue_quarantined_in_transaction` CAS into the GPU queue. Crash +before checkpoint replays safely; concurrent callers cannot append twice. +Normal scheduler admission, policy, capacity and attempt accounting still apply. +Provider-STT holds never enter the GPU DRR ring: their vendor-paid attempt +protocol needs a separate sanctioned re-drive, so expiry/requeue refuse them. +The state-table deadline xfail remains for those unstamped provider entry paths +and the deliberate flag-off entry. No projection state is added in phase 4a. + +Authorized operators can call the existing private worker-OIDC surface: +`POST /v2/batch-pipeline/requeue-quarantined` with +`{"stage":"prod|dev","segment_id":""}` using the configured +worker identity. It accepts quarantined holds and held-expired dead letters; +`accepted=false` is a fenced or already advanced row. It performs no provider +or artifact IO. Repair missing inputs through the existing artifact restoration +path first. Do not clear account-deletion, privacy or content ownership fences. + +Status readers keep dead-held units in the held custody answer and partial +accounting. They never acknowledge full completion merely because a hold aged. +Healthy siblings remain visible. `privacy_blocked`, projection holds, BYOK and +finalization dead-letter tools remain outside this change. + +`omi_batch_held_expired_total` counts committed expiry transitions; +`omi_batch_held_expired_units` is all retained dead stock for this explicit new +reason, and `omi_batch_held_expired_observed_at_seconds` timestamps successful +stock samples. The three exact metric families are pinned in both dev/prod +Cloud Run exporter releases (main and sync/integration), because serving +maintenance emits them on CPU hosts. Ship those metrics-egress declarations +with the alert through the existing release path; no live egress or alert +delivery was qualified here. The agent-tier `omi-batch-held-expired-volume` rule alerts on +fresh stock >100 or >10 expirations in 30 minutes for five minutes. Unrelated +legacy dead rows do not count, and this rule never pages. These are initial +operator thresholds, not production-calibrated measurements. Inspect worker +failure causes and nightly progress before requeueing. A stock sample is not +proof of user-visible recovery; verify real transcript publication separately. + +Not measured here: production stock, dwell distribution, vendor spend, live +nightly execution, alert delivery or deployed behavior. Tests use synthetic +identities and offline/emulator paths only. diff --git a/backend/route_policy_manifest.yaml b/backend/route_policy_manifest.yaml index 88513623e8f..fe5fcd79d64 100644 --- a/backend/route_policy_manifest.yaml +++ b/backend/route_policy_manifest.yaml @@ -4711,6 +4711,7 @@ routes: - {route_type: http, method: POST, path: /v2/batch-pipeline/complete, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v2/batch-pipeline/fail, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v2/batch-pipeline/redrive-dead, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} + - {route_type: http, method: POST, path: /v2/batch-pipeline/requeue-quarantined, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v2/batch-pipeline/gpu-idle, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: POST, path: /v2/batch-pipeline/reconcile, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} - {route_type: http, method: GET, path: /v2/batch-pipeline/metrics, policy: {review_status: reviewed, auth: {mechanisms: [service_oidc], placement: dependency, scopes: []}, byok: not_applicable, rate_limit: {policy_name: none, key_subject: none, enforcement: none, placement: none}, timeout_class: sync_job, surface: internal_task, visibility: internal, data_domain: sync_audio, deprecation: {state: active}, owner: backend}} diff --git a/backend/routers/batch_pipeline.py b/backend/routers/batch_pipeline.py index a7d8b5730c9..d02d94b1d1a 100644 --- a/backend/routers/batch_pipeline.py +++ b/backend/routers/batch_pipeline.py @@ -225,6 +225,21 @@ async def redrive_dead(body: RedriveRequest): return {'accepted': accepted} +@router.post('/requeue-quarantined', include_in_schema=False) +async def requeue_quarantined(body: RedriveRequest): + """Same private worker OIDC boundary; usable even with expiry disabled.""" + from database.batch_pipeline_quarantine_recovery import RecoveryStore + from database.batch_pipeline_projection_protocol import ProjectionBlocked + + try: + accepted = await run_blocking( + db_executor, RecoveryStore(body.service().store).requeue_quarantined, body.segment_id, time.time() + ) + except (BatchPrivacyBlocked, ProjectionBlocked): + accepted = False + return {'accepted': accepted} + + @router.post('/shadow-recover', include_in_schema=False) async def shadow_recover(body: RedriveRequest): from database.batch_pipeline_shadow import ShadowStore diff --git a/backend/scripts/verify_pusher_cohost_env_diff.py b/backend/scripts/verify_pusher_cohost_env_diff.py index c569a7ac686..c8934e689e7 100644 --- a/backend/scripts/verify_pusher_cohost_env_diff.py +++ b/backend/scripts/verify_pusher_cohost_env_diff.py @@ -100,6 +100,9 @@ # none of those surfaces and never starts the dedicated GPU process. _BATCH_PIPELINE_LISTEN_ONLY = frozenset( { + # Held expiry/requeue is owned by main.py maintenance/RPC and GPU workers; + # pusher does not run the held maintenance loop or worker failure writer. + 'BATCH_HELD_DEAD_LETTER_ENABLED', 'BATCH_CONTROL_CANDIDATE_FANOUT', 'BATCH_ADMISSION_PARALLEL_CALLS', 'BATCH_PIPELINE_IDLE_POLL_MAX_SECONDS', diff --git a/backend/testing/lifecycle_sim/clock.py b/backend/testing/lifecycle_sim/clock.py index 4db0fe0fe50..358ebb1284e 100644 --- a/backend/testing/lifecycle_sim/clock.py +++ b/backend/testing/lifecycle_sim/clock.py @@ -11,6 +11,8 @@ 'updated_at', 'next_attempt_at', 'lease_expires_at', + 'held_deadline_at', + 'held_expired_at', ) diff --git a/backend/testing/lifecycle_sim/test_scenarios.py b/backend/testing/lifecycle_sim/test_scenarios.py index d0ae4f35d0f..4365263b928 100644 --- a/backend/testing/lifecycle_sim/test_scenarios.py +++ b/backend/testing/lifecycle_sim/test_scenarios.py @@ -16,6 +16,14 @@ from database import conversation_finalization_jobs as finalization_jobs from testing.lifecycle_sim.faults import BoundaryFault, InjectedLifecycleFault from testing.lifecycle_sim.harness import LifecycleSim + +# Shared real-Firestore regressions also run in the hermetic lifecycle harness. +from tests.integration.test_batch_held_dead_letter_firestore import ( + lane, # noqa: F401 + test_worker_fail_expire_and_nightly_redrive_keep_audio_and_cold_queue_integrity, # noqa: F401 + test_two_forced_requeues_append_once_in_real_firestore, # noqa: F401 + test_expired_row_survives_rollback_using_only_legacy_dead_redrive, # noqa: F401 +) from tests.support.lifecycle_state_table import durable_titling_rearm from testing.listen_pusher_stack.run import Stack, _wait_for_port from utils.batch_pipeline import live diff --git a/backend/tests/integration/test_batch_held_dead_letter_firestore.py b/backend/tests/integration/test_batch_held_dead_letter_firestore.py new file mode 100644 index 00000000000..3b96ec11d18 --- /dev/null +++ b/backend/tests/integration/test_batch_held_dead_letter_firestore.py @@ -0,0 +1,87 @@ +"""Loopback-only worker-fail -> held -> expire -> nightly CAS requeue proof.""" + +from concurrent.futures import ThreadPoolExecutor +from copy import deepcopy + +import pytest +from prometheus_client import CollectorRegistry + +from database.batch_pipeline_quarantine_recovery import RecoveryStore, dead_held_candidates, held_expired_stock +from tests.integration.test_batch_pipeline_projection_firestore import lane # noqa: F401 +from testing.lifecycle_sim.clock import age_job_row +from utils.batch_pipeline.contracts import WorkResult +from utils.batch_pipeline.held_dead_letter import HeldDeadLetter +from utils.batch_pipeline.metrics import QueueMetrics + +pytestmark = pytest.mark.integration + + +def test_worker_fail_expire_and_nightly_redrive_keep_audio_and_cold_queue_integrity(lane): + batch = lane.seed() + sid = batch.ids[0] + ref = lane.client.collection(f'{lane.store.namespace}_jobs').document(sid) + assert lane.queue.fail(sid, batch.leases[sid].token, 10, quarantine=True) + held = ref.get().to_dict() + assert held['held_deadline_at'] > 10 + assert held['state'] == 'quarantined' + age_job_row(ref, 10 * 86400) + recovery = RecoveryStore(lane.store) + # Both racers read the same row; exactly one terminal transition commits. + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda _: recovery.expire_held(sid, 86400), range(2))) + assert sorted(results) == ['expired', 'held'] + expired = ref.get().to_dict() + assert expired['audio_uri'] == held['audio_uri'] + assert expired['attempt'] == held['attempt'] + assert 'expires_at' not in expired + assert dead_held_candidates(lane.client, lane.store.namespace) == [(sid, expired)] + assert held_expired_stock(lane.client, lane.store.namespace) == 1 + metrics = QueueMetrics(CollectorRegistry()) + assert HeldDeadLetter(lane.store, metrics=metrics).tick(172800) == {'expired': 0, 'requeued': 1} + before = deepcopy(ref.get().to_dict()) + assert not recovery.requeue_quarantined(sid, 172800) + assert ref.get().to_dict() == before + leases = lane.queue.pull(172800) + assert len(leases) == 1 and leases[0].item.segment_id == sid + # The new lease fences the old token, then normal ordered publication works. + result = WorkResult(sid, 'gs://synthetic/requeued-result') + assert not lane.queue.complete(result, batch.leases[sid].token, 172801) + assert lane.queue.complete(result, leases[0].token, 172801) + assert lane.merge.commit_next(batch.binding.owner_id, 172802)['segment_id'] == sid + assert held_expired_stock(lane.client, lane.store.namespace) == 0 + + +def test_two_forced_requeues_append_once_in_real_firestore(lane): + batch = lane.seed() + sid = batch.ids[0] + assert lane.queue.fail(sid, batch.leases[sid].token, 10, quarantine=True) + recovery = RecoveryStore(lane.store) + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda _: recovery.requeue_quarantined(sid, 11), range(2))) + assert sorted(results) == [False, True] + assert [v.item.segment_id for v in lane.queue.pull(12, 2)] == [sid] + + +def test_expired_row_survives_rollback_using_only_legacy_dead_redrive(lane, monkeypatch): + batch = lane.seed() + sid = batch.ids[0] + ref = lane.client.collection(f'{lane.store.namespace}_jobs').document(sid) + assert lane.queue.fail(sid, batch.leases[sid].token, 10, quarantine=True) + age_job_row(ref, 10 * 86400) + assert RecoveryStore(lane.store).expire_held(sid, 86400) == 'expired' + expired = ref.get().to_dict() + assert expired['merge_pending'] is True + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + # Simulate the old service: no held requeue route or nightly helper is called. + with ThreadPoolExecutor(max_workers=2) as workers: + results = list(workers.map(lambda _: lane.queue.redrive_dead(sid, 86401), range(2))) + assert sorted(results) == [False, True] + restored = ref.get().to_dict() + for key in ('audio_uri', 'attempt', 'held_deadline_at', 'dead_letter_reason', 'held_expired_at'): + assert restored[key] == expired[key] + leases = lane.queue.pull(86402, 2) + assert [v.item.segment_id for v in leases] == [sid] + result = WorkResult(sid, 'gs://synthetic/rollback-result') + assert not lane.queue.complete(result, batch.leases[sid].token, 86403) + assert lane.queue.complete(result, leases[0].token, 86403) + assert lane.merge.commit_next(batch.binding.owner_id, 86404)['segment_id'] == sid diff --git a/backend/tests/support/firestore_batch_pipeline_drivers.py b/backend/tests/support/firestore_batch_pipeline_drivers.py index d54c7b27e2c..7ebd2e1b4e3 100644 --- a/backend/tests/support/firestore_batch_pipeline_drivers.py +++ b/backend/tests/support/firestore_batch_pipeline_drivers.py @@ -192,6 +192,15 @@ def entries(): }, neutrals={'limit': (10, 'bounded page size')}, ), + DriverEntry( + 'database.batch_pipeline_quarantine_recovery.dead_held_candidates', + domains={'namespace': namespaces, 'after_id': [None, 'held-09']}, + neutrals={'limit': (32, 'bounded page size')}, + ), + DriverEntry( + 'database.batch_pipeline_quarantine_recovery.held_expired_stock', + domains={'namespace': namespaces}, + ), DriverEntry( 'database.batch_pipeline_quarantine_recovery.held_candidates', domains={'namespace': namespaces, 'after_id': [None, 'held-09']}, diff --git a/backend/tests/support/lifecycle_state_table.py b/backend/tests/support/lifecycle_state_table.py index 76d074669e6..8c037c02bdb 100644 --- a/backend/tests/support/lifecycle_state_table.py +++ b/backend/tests/support/lifecycle_state_table.py @@ -134,6 +134,11 @@ def entry(module, function, field, state=None, *, inplace=False, patches=(), **b # Automatic entry paths are distinct from conservative payload builders. DEADLINE_ENTRIES = { + 'batch:quarantined': ( + entry('utils/batch_pipeline/retry_policy.py', 'quarantine_job', 'state', 'quarantined', inplace=True), + entry('utils/batch_pipeline/provider_stt.py', 'ProviderSttStore.settle.commit', 'state', 'quarantined'), + entry('utils/batch_pipeline/provider_stt.py', 'ProviderSttStore.terminal.commit', 'state', 'quarantined'), + ), 'finalization:queued': ( entry( FINALIZATION_MODULE, @@ -559,9 +564,21 @@ def row( 'batch', name, ('state',), - exit=LEASE if name == 'leased' else BATCH, - deadline='lease_until' if name == 'leased' else None, - stamped=name == 'leased', + exit=( + LEASE + if name == 'leased' + else ( + Exit( + 'sweeper', + 'utils/batch_pipeline/held_dead_letter.py:HeldDeadLetter.tick', + (Writer('database/batch_pipeline_quarantine_recovery.py', 'expire_held_in_transaction'),), + ) + if name == 'quarantined' + else BATCH + ) + ), + deadline='lease_until' if name == 'leased' else 'held_deadline_at' if name == 'quarantined' else None, + stamped=name in {'leased', 'quarantined'}, metric=( 'omi_batch_record_pending_new_oldest_age_seconds' if name == 'record_pending' @@ -591,7 +608,11 @@ def row( '#24587 adds a per-attempt admission lease; total pending-dwell deadline is still unstamped. ' '#24595 supplies recent-cohort age/page' if name == 'record_pending' - else '' + else ( + 'GPU holds gain seven-day expiry and nightly requeue; provider-STT holds remain fenced from the GPU ring.' + if name == 'quarantined' + else '' + ) ), ) for name in ( @@ -607,6 +628,21 @@ def row( 'committed', ) ) + + ( + row( + 'batch', + 'dead', + ('state',), + terminal=True, + owner='database/batch_pipeline_quarantine_recovery.py', + deadline='held_deadline_at', + exit=Exit('sweeper', 'utils/batch_pipeline/held_dead_letter.py:HeldDeadLetter.tick'), + metric='omi_batch_held_expired_units', + alert=Alert('omi-batch-held-expired-volume', 'agent'), + alert_required=True, + note='Reason held_expired is retained stock, never TTL/deletion; nightly UTC and private OIDC requeue.', + ), + ) + tuple( row( 'batch_projection', diff --git a/backend/tests/support/lifecycle_writer_inventory.py b/backend/tests/support/lifecycle_writer_inventory.py index 84ebe489a88..d017a35daaa 100644 --- a/backend/tests/support/lifecycle_writer_inventory.py +++ b/backend/tests/support/lifecycle_writer_inventory.py @@ -215,6 +215,12 @@ "'prepared'", ('plan_digest', 'prepared_at', 'privacy_generation', 'state', 'token'), ), + ( + 'expire_held_in_transaction', + 'state', + "'dead'", + ('dead_letter_reason', 'held_expired_at', 'merge_pending', 'state'), + ), ], 'database/batch_pipeline_recovery.py': [ ( @@ -2503,6 +2509,12 @@ "'ready'", ('available_at', 'failure', 'lease_until', 'retry_started_at', 'state', 'token'), ), + ( + 'Scheduler.requeue_quarantined_in_transaction', + 'state', + "'ready'", + ('available_at', 'held_requeued_at', 'lease_until', 'merge_pending', 'state', 'token'), + ), ( 'Scheduler.restore_audio_in_transaction', 'state', diff --git a/backend/tests/unit/test_backend_runtime_env_validator.py b/backend/tests/unit/test_backend_runtime_env_validator.py index f32b277d83b..2e92fa88108 100644 --- a/backend/tests/unit/test_backend_runtime_env_validator.py +++ b/backend/tests/unit/test_backend_runtime_env_validator.py @@ -451,7 +451,8 @@ def with_cloud_run_oauth_secrets(payload: str) -> str: ) payload = re.sub( r'("backend(?:-sync|-sync-backfill|-integration)?":\s*\{.*?"env":\s*\[)', - r'\1\n {"name": "ACTION_ITEM_REFRESH_PRESERVE_ENABLED", "value": "true"},', + r'\1\n {"name": "ACTION_ITEM_REFRESH_PRESERVE_ENABLED", "value": "true"},' + r'\n {"name": "BATCH_HELD_DEAD_LETTER_ENABLED", "value": "true"},', payload, flags=re.DOTALL, ) @@ -1809,6 +1810,22 @@ def test_cloud_run_state_reports_missing_gateway_url(tmp_path): assert [error.message for error in errors] == ['missing env OMI_LLM_GATEWAY_URL'] assert errors[0].scope == 'cloud_run/backend' + # A complete synthetic revision carries the new kill switch. Removing it + # must still fail validation; completing the fixture must not weaken the guard. + state_path.write_text( + with_cloud_run_oauth_secrets(_MISSING_GATEWAY_CLOUD_RUN_STATE).replace( + '{"name": "BATCH_HELD_DEAD_LETTER_ENABLED", "value": "true"},\n ', '' + ), + encoding='utf-8', + ) + errors = validator.validate_runtime_env(env='dev', cloud_run_state_path=state_path) + assert {(error.scope, error.message) for error in errors} == { + ('cloud_run/backend', 'missing env BATCH_HELD_DEAD_LETTER_ENABLED'), + ('cloud_run/backend-sync', 'missing env BATCH_HELD_DEAD_LETTER_ENABLED'), + ('cloud_run/backend-integration', 'missing env BATCH_HELD_DEAD_LETTER_ENABLED'), + ('cloud_run/backend', 'missing env OMI_LLM_GATEWAY_URL'), + } + state_path.write_text( with_cloud_run_oauth_secrets(_MISSING_GATEWAY_CLOUD_RUN_STATE).replace( '{"name": "SYNC_LINEAGE_LIVE_DEDUPE_ENABLED", "value": "true"},\n ', '' diff --git a/backend/tests/unit/test_batch_held_dead_letter.py b/backend/tests/unit/test_batch_held_dead_letter.py new file mode 100644 index 00000000000..3771d06cbca --- /dev/null +++ b/backend/tests/unit/test_batch_held_dead_letter.py @@ -0,0 +1,351 @@ +"""Synthetic custody/CAS regressions for held expiry and nightly re-drive.""" + +from copy import deepcopy +from dataclasses import replace +from unittest.mock import Mock + +import pytest +from prometheus_client import CollectorRegistry + +from config.batch_pipeline import HELD_DEAD_LETTER_SECONDS, batch_held_dead_letter_enabled, scheduler_control_key +from database.batch_pipeline_privacy import BatchPrivacyBlocked, privacy_owner_key +from database.batch_pipeline_projection import held_without_recovery +from database.batch_pipeline_projection_protocol import ProjectionBlocked +from database.batch_pipeline_quarantine_recovery import RecoveryStore, expire_held_in_transaction +from tests.unit.test_batch_pipeline_quarantine_recovery import setup # noqa: F401 +from tests.unit.test_batch_pipeline_scheduler import MemoryStore, item +from utils.batch_pipeline import held_dead_letter as sweep +from utils.batch_pipeline.intake import _held_status +from utils.batch_pipeline.metrics import QueueMetrics +from utils.batch_pipeline.scheduler import Scheduler + + +def row(client): + return client.rows['batch_pipeline_prod_jobs', 'bad'] + + +def test_expiry_backfills_full_window_then_cas_expires_without_audio_or_ttl_mutation(setup): + client, store, _, artifacts, _, _ = setup + recovery = RecoveryStore(store) + original = deepcopy(row(client)) + assert recovery.expire_held('bad', 100) == 'stamped' + assert row(client)['held_deadline_at'] == 100 + HELD_DEAD_LETTER_SECONDS + assert recovery.expire_held('bad', 100 + HELD_DEAD_LETTER_SECONDS - 1) == 'held' + assert recovery.expire_held('bad', 100 + HELD_DEAD_LETTER_SECONDS) == 'expired' + assert recovery.expire_held('bad', 101 + HELD_DEAD_LETTER_SECONDS) == 'held' + expected = { + **original, + 'held_deadline_at': 100 + HELD_DEAD_LETTER_SECONDS, + 'state': 'dead', + 'merge_pending': True, + 'dead_letter_reason': 'held_expired', + 'held_expired_at': 100 + HELD_DEAD_LETTER_SECONDS, + } + assert row(client) == expected + artifacts.assert_not_called() + assert not artifacts.mock_calls + assert held_without_recovery(row(client)) + assert _held_status([row(client)]) == ('processing', 'batch_unit_held') + + +@pytest.mark.parametrize('dead', [False, 'legacy', 'rollback_compatible']) +def test_requeue_idempotent_and_usable_with_flag_off_retains_attempt_and_audio_history(setup, monkeypatch, dead): + client, store, _, _, _, _ = setup + job = row(client) + job.update( + attempt=5, + recovery_state='failed_held', + recovery={'token': 'spent', 'settled_at': 10, 'privacy_generation': 0}, + held_deadline_at=20, + ) + if dead: + job.update(state='dead', dead_letter_reason='held_expired', held_expired_at=20, merge_pending=dead != 'legacy') + original = deepcopy(job) + # Keep the strict document fake off query paths; the emulator covers a cold DRR ring. + client.rows['batch_pipeline_prod_control', scheduler_control_key(job['owner_key'])] = {'p0_active': True} + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + recovery = RecoveryStore(store) + assert recovery.requeue_quarantined('bad', 30) + after = deepcopy(client.rows) + assert not recovery.requeue_quarantined('bad', 31) + assert client.rows == after + for key in ('attempt', 'audio_uri', 'projection_binding', 'privacy_generation', 'recovery', 'held_deadline_at'): + assert row(client)[key] == original[key] + assert row(client)['state'] == 'ready' and row(client)['merge_pending'] is True + assert row(client)['token'] == '' and row(client)['lease_until'] == 0 + + +def test_expired_row_requeues_through_unchanged_legacy_redrive_with_flag_off(setup, monkeypatch): + client, store, _, artifacts, _, _ = setup + job = row(client) + job.update(held_deadline_at=20, attempt=5) + original = deepcopy(job) + assert RecoveryStore(store).expire_held('bad', 30) == 'expired' + assert row(client)['merge_pending'] is True + client.rows['batch_pipeline_prod_control', scheduler_control_key(job['owner_key'])] = {'p0_active': True} + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + # Only the unchanged, pre-feature path is available after a code rollback. + queue = Scheduler(store) + assert queue.redrive_dead('bad', 31) + after = deepcopy(client.rows) + assert not queue.redrive_dead('bad', 32) + assert client.rows == after + restored = row(client) + assert restored['state'] == 'ready' and restored['merge_pending'] is True + assert restored['dead_letter_reason'] == 'held_expired' + assert restored['held_deadline_at'] == 20 and restored['held_expired_at'] == 30 + for key in ('attempt', 'audio_uri', 'projection_binding', 'privacy_generation'): + assert restored[key] == original[key] + assert not artifacts.mock_calls + + +@pytest.mark.parametrize( + 'patch', + [ + {'state': 'ready'}, + {'state': 'privacy_blocked'}, + {'quarantine_reason': 'privacy'}, + {'state': 'dead', 'dead_letter_reason': 'other'}, + {'stt_route': {'stt_route': 'provider'}}, + {'recovery_state': 'running'}, + {'recovery_state': 'prepared'}, + {'token': 'live'}, + {'lease_until': 31}, + {'result': {'result_uri': 'gs://synthetic/result'}}, + {'merge_pending': True}, + {'recovery': {'token': 'active', 'privacy_generation': 0}}, + {'speech_redrive': {'state': 'running'}}, + {'silence_disposition': {'state': 'prepared'}}, + {'terminal_disposition': {'state': 'prepared'}}, + ], +) +def test_expiry_and_requeue_fence_matrix_is_read_only(setup, patch): + client, store, _, _, _, _ = setup + row(client).update(held_deadline_at=20, **patch) + before = deepcopy(client.rows) + recovery = RecoveryStore(store) + assert recovery.expire_held('bad', 30) == 'held' + assert not recovery.requeue_quarantined('bad', 30) + assert client.rows == before + + +@pytest.mark.parametrize('patch', [{'token': 'live'}, {'lease_until': 31}, {'result': {'result_uri': 'synthetic'}}]) +def test_pending_dead_letter_still_requires_dormant_requeue_fences(setup, patch): + client, store, _, _, _, _ = setup + row(client).update(state='dead', dead_letter_reason='held_expired', merge_pending=True, **patch) + before = deepcopy(client.rows) + assert not RecoveryStore(store).requeue_quarantined('bad', 30) + assert client.rows == before + + +def test_result_receipt_fences_expiry_and_requeue_even_if_job_result_missing(setup): + client, store, _, _, _, _ = setup + row(client)['held_deadline_at'] = 20 + client.rows['batch_pipeline_prod_results', 'bad'] = {'result': {'result_uri': 'gs://synthetic/result'}} + before = deepcopy(client.rows) + recovery = RecoveryStore(store) + assert recovery.expire_held('bad', 30) == 'held' + assert not recovery.requeue_quarantined('bad', 30) + assert client.rows == before + + +@pytest.mark.parametrize('fence', ['privacy', 'ledger']) +def test_content_and_privacy_fences_remain_authoritative(setup, fence): + client, store, binding, _, _, _ = setup + row(client)['held_deadline_at'] = 20 + if fence == 'privacy': + client.rows['batch_pipeline_privacy', privacy_owner_key(binding.owner_id)] = {'generation': 1} + error = BatchPrivacyBlocked + else: + client.rows['users', binding.owner_id, 'sync_content_ledger', binding.content_id]['job_id'] = 'other' + error = ProjectionBlocked + before = deepcopy(client.rows) + recovery = RecoveryStore(store) + with pytest.raises(error): + recovery.expire_held('bad', 30) + with pytest.raises(error): + recovery.requeue_quarantined('bad', 30) + assert client.rows == before + + +@pytest.mark.parametrize('bad', [False, 'old', float('nan'), float('inf')]) +def test_bad_deadline_never_expires(setup, bad): + client, store, _, _, _, _ = setup + row(client)['held_deadline_at'] = bad + assert RecoveryStore(store).expire_held('bad', 30) == 'held' + + +def test_flag_off_preserves_legacy_quarantine_and_sweep_is_zero_io(monkeypatch): + monkeypatch.setenv('BATCH_PIPELINE_MAX_ATTEMPTS', '1') + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + assert not batch_held_dead_letter_enabled() + store = MemoryStore() + queue = Scheduler(store) + queue.enqueue(item('bad'), 0) + lease = queue.pull(0)[0] + assert queue.fail('bad', lease.token, 1) + assert 'held_deadline_at' not in store.rows['jobs', 'bad'] + before = deepcopy(store.rows) + assert store.run(lambda tx: expire_held_in_transaction(tx, tx.get('jobs', 'bad'), 10**9)) == 'disabled' + assert store.rows == before + untouched = Mock() + assert sweep.HeldDeadLetter(untouched).tick(10**9) == {'expired': 0, 'requeued': 0} + assert not untouched.mock_calls + + +def model_store(monkeypatch, n=35): + store = MemoryStore() + store.client, store.namespace = None, 'synthetic' + queue = Scheduler(store) + for i in range(n): + sid = f'held-{i:03d}' + queue.enqueue(replace(item(sid), capture_at=i), 0) + lease = queue.pull(0)[0] + queue.fail(sid, lease.token, 1, quarantine=True) + store.rows['jobs', sid]['held_deadline_at'] = 2 + + def page(state, after_id, limit): + return [ + (sid, deepcopy(job)) + for (table, sid), job in sorted(store.rows.items()) + if table == 'jobs' and job['state'] == state and (after_id is None or sid > after_id) + ][:limit] + + monkeypatch.setattr( + sweep, 'held_candidates', lambda c, ns, limit, after_id=None: page('quarantined', after_id, limit) + ) + monkeypatch.setattr( + sweep, 'dead_held_candidates', lambda c, ns, limit, after_id=None: page('dead', after_id, limit) + ) + # Model synthetic authorized content; strict Firestore tests above exercise the real fence. + monkeypatch.setattr(RecoveryStore, 'checked', lambda self, tx, sid: (tx.get('jobs', sid), None)) + monkeypatch.setattr(sweep, 'held_expired_stock', lambda c, ns: len(page('dead', None, n))) + return store + + +def test_nightly_bounded_cursor_survives_restart_and_double_delivery(monkeypatch): + store = model_store(monkeypatch) + metrics = QueueMetrics(CollectorRegistry()) + runner = sweep.HeldDeadLetter(store, metrics=metrics) + assert runner.tick(86400) == {'expired': 32, 'requeued': 0} + assert metrics.held_expired._value.get() == 32 + before = deepcopy(store.rows) + assert runner.tick(86400) == {'expired': 0, 'requeued': 0} + assert store.rows == before + # A fresh process continues the durable held cursor, then waits for midnight. + runner = sweep.HeldDeadLetter(store, metrics=metrics) + assert runner.tick(86700) == {'expired': 3, 'requeued': 0} + assert runner.tick(172800) == {'expired': 0, 'requeued': 32} + assert metrics.held_expired_stock._value.get() == 3 + assert sweep.HeldDeadLetter(store).tick(173100) == {'expired': 0, 'requeued': 3} + assert not any(job['state'] == 'dead' for (table, _), job in store.rows.items() if table == 'jobs') + assert metrics.held_expired._value.get() == 35 + + +def test_crash_before_checkpoint_replays_without_duplicate_expiry(monkeypatch): + store = model_store(monkeypatch, 1) + runner = sweep.HeldDeadLetter(store) + run = store.run + crashed = False + + def crash(fn): + nonlocal crashed + if fn.__name__ == 'checkpoint' and not crashed: + crashed = True + raise RuntimeError('synthetic crash after row commit') + return run(fn) + + monkeypatch.setattr(store, 'run', crash) + with pytest.raises(RuntimeError, match='synthetic crash'): + runner.tick(86400) + assert sweep.HeldDeadLetter(store).tick(86400) == {'expired': 0, 'requeued': 0} + assert sweep.HeldDeadLetter(store).tick(172800) == {'expired': 0, 'requeued': 1} + + +def test_private_requeue_rpc_is_stage_bound_authenticated_and_idempotent(setup, monkeypatch): + from fastapi import FastAPI + from fastapi.testclient import TestClient + from routers import batch_pipeline as router + from types import SimpleNamespace + + client, store, _, _, _, _ = setup + job = row(client) + client.rows['batch_pipeline_prod_control', scheduler_control_key(job['owner_key'])] = {'p0_active': True} + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + monkeypatch.setattr(router, 'serving_for_stage', lambda stage: SimpleNamespace(store=store)) + app = FastAPI() + app.include_router(router.router) + app.dependency_overrides[router.verify_worker] = lambda: None + path = '/v2/batch-pipeline/requeue-quarantined' + with TestClient(app) as api: + assert api.post(path, json={'stage': 'dev', 'segment_id': 'bad'}).status_code == 409 + assert api.post(path, json={'stage': 'prod', 'segment_id': 'bad'}).json() == {'accepted': True} + assert api.post(path, json={'stage': 'prod', 'segment_id': 'bad'}).json() == {'accepted': False} + app.dependency_overrides.clear() + monkeypatch.delenv('BATCH_PIPELINE_WORKER_AUDIENCE', raising=False) + monkeypatch.delenv('BATCH_PIPELINE_WORKER_SERVICE_ACCOUNT', raising=False) + with TestClient(app) as api: + assert api.post(path, json={'stage': 'prod', 'segment_id': 'bad'}).status_code == 403 + + +def test_failed_recovery_writers_stamp_new_full_window_and_off_restores_exact_payload(setup, monkeypatch): + client, store, binding, _, _, _ = setup + recovery = RecoveryStore(store) + prepared = recovery.prepare('bad', binding.owner_id, 100) + token = prepared['recovery']['token'] + before = deepcopy(client.rows) + monkeypatch.setenv('BATCH_HELD_DEAD_LETTER_ENABLED', 'false') + assert recovery.hold_invalid_manifest('bad', token, 200, audio_uri=row(client)['audio_uri']) + legacy = deepcopy(row(client)) + assert 'held_deadline_at' not in legacy + client.rows = deepcopy(before) + monkeypatch.delenv('BATCH_HELD_DEAD_LETTER_ENABLED') + assert recovery.hold_invalid_manifest('bad', token, 200, audio_uri=row(client)['audio_uri']) + assert row(client) == {**legacy, 'held_deadline_at': 200 + HELD_DEAD_LETTER_SECONDS} + # An actual failed paid recovery settlement uses the same fresh window. + client.rows = deepcopy(before) + assert recovery.start('bad', token, 201) + assert recovery.settle('bad', token, 202, failure={'stage': 'provider', 'reason': 'operation_failed'}) + assert row(client)['held_deadline_at'] == 202 + HELD_DEAD_LETTER_SECONDS + + +def test_nightly_rechecks_dead_age_inside_transaction_instead_of_trusting_scan_snapshot(setup): + client, store, _, _, _, _ = setup + recovery = RecoveryStore(store) + row(client).update(state='dead', dead_letter_reason='held_expired', held_expired_at=31) + before = deepcopy(client.rows) + assert not recovery.requeue_quarantined('bad', 32, dead_before=30) + assert client.rows == before + row(client).update(state='quarantined', held_expired_at=29) + assert not recovery.requeue_quarantined('bad', 32, dead_before=30) + + +@pytest.mark.parametrize('lease', [None, False, 'expired', float('nan'), float('inf')]) +def test_unknown_lease_metadata_is_not_evidence_of_expiry(setup, lease): + client, store, _, _, _, _ = setup + row(client).update(lease_until=lease, held_deadline_at=20) + recovery = RecoveryStore(store) + assert recovery.expire_held('bad', 30) == 'held' + assert not recovery.requeue_quarantined('bad', 30) + assert row(client)['state'] == 'quarantined' + + +def test_one_unreadable_row_does_not_pin_expiry_or_nightly_pages(monkeypatch, caplog): + store = model_store(monkeypatch, 2) + original = RecoveryStore.checked + + def unreadable(self, tx, sid): + if sid == 'held-000': + raise RuntimeError('synthetic row read failure') + return original(self, tx, sid) + + monkeypatch.setattr(RecoveryStore, 'checked', unreadable) + runner = sweep.HeldDeadLetter(store) + assert runner.tick(86400) == {'expired': 1, 'requeued': 0} + assert store.rows['jobs', 'held-000']['state'] == 'quarantined' + # A corrupt old dead row is also skipped without blocking the next key. + store.rows['jobs', 'held-000'].update(state='dead', dead_letter_reason='held_expired', held_expired_at=86400) + assert runner.tick(172800) == {'expired': 0, 'requeued': 1} + assert store.rows['jobs', 'held-001']['state'] == 'ready' + assert 'action=expire outcome=retry exception_type=RuntimeError' in caplog.text + assert 'action=requeue outcome=retry exception_type=RuntimeError' in caplog.text diff --git a/backend/tests/unit/test_batch_pipeline_chart.py b/backend/tests/unit/test_batch_pipeline_chart.py index 9bbadc07485..32ca19b6c7c 100644 --- a/backend/tests/unit/test_batch_pipeline_chart.py +++ b/backend/tests/unit/test_batch_pipeline_chart.py @@ -970,3 +970,14 @@ def test_control_loop_defaults_and_value_only_rollback_render(tmp_path, rollback ) == expected ) + + +@pytest.mark.parametrize('enabled', ['true', 'false']) +def test_held_dead_letter_kill_switch_renders_literal_worker_setting(enabled): + result = render(*enabled_arguments(), '--set', f'worker.heldDeadLetterEnabled={enabled}') + assert result.returncode == 0, result.stderr + deployments = [v for v in yaml.safe_load_all(result.stdout) if v and v.get('kind') == 'Deployment'] + assert deployments + for deployment in deployments: + env = deployment['spec']['template']['spec']['containers'][0]['env'] + assert next(v['value'] for v in env if v['name'] == 'BATCH_HELD_DEAD_LETTER_ENABLED') == enabled diff --git a/backend/tests/unit/test_flag_budget.py b/backend/tests/unit/test_flag_budget.py index e2638969288..da5745c0ddb 100644 --- a/backend/tests/unit/test_flag_budget.py +++ b/backend/tests/unit/test_flag_budget.py @@ -191,6 +191,7 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): # outside the pipeline scanner's scope. Keep helper kill switches separate. rollout_name = 'SERVER_BATCH_ONLY_WEB_PERCENT' assert helper_names == HELPER_KILL_SWITCHES | { + 'BATCH_HELD_DEAD_LETTER_ENABLED', 'SERVER_BATCH_UNSUPPORTED_LANGUAGES', rollout_name, 'BATCH_PIPELINE_SERVING_PROJECTION_SWEEP', @@ -229,6 +230,10 @@ def test_helper_kill_switch_budget_entries_match_the_default_on_registry(): assert rollout['kind'] == 'env' assert rollout['lifecycle'] == 'rollout' assert rollout_name in env_reads((BACKEND_DIR / 'config/server_batch.py').read_text(encoding='utf-8')) + held = names['BATCH_HELD_DEAD_LETTER_ENABLED'] + assert held['owner'] == 'uicleanup / omi-lifecycle' + assert held['review_by'] == '2026-12-06' + assert flags['BATCH_HELD_DEAD_LETTER_ENABLED']['lifecycle'] == 'ops_kill' for name in sorted(HELPER_KILL_SWITCHES): entry = names[name] flag = flags[name] diff --git a/backend/tests/unit/test_lifecycle_invariants.py b/backend/tests/unit/test_lifecycle_invariants.py index 33abb24d301..d08b5cfa78f 100644 --- a/backend/tests/unit/test_lifecycle_invariants.py +++ b/backend/tests/unit/test_lifecycle_invariants.py @@ -314,8 +314,11 @@ def assert_clause(row, clause): ('batch:privacy_blocked', 'dwell_metric'): 'batch:privacy_blocked: missing authoritative dwell/stock gauge; ' 'phase contract follow-up; no fix PR yet; ANX ' 'lcinv-lifecycle-invariants', - ('batch:quarantined', 'entry_deadline'): 'batch:quarantined: missing deadline stamped at entry; phase contract ' - 'follow-up; no fix PR yet; ANX lcinv-lifecycle-invariants', + ( + 'batch:quarantined', + 'entry_deadline', + ): 'GPU holds stamp a flag-gated held_deadline_at; provider-STT settle/terminal entries ' + 'remain unstamped and cannot join the GPU requeue path. ANX omi-lifecycle phase 4a', ('batch:ready', 'entry_deadline'): 'batch:ready: missing deadline stamped at entry; phase contract follow-up; no ' 'fix PR yet; ANX lcinv-lifecycle-invariants', ('batch:record_pending', 'entry_deadline'): 'batch:record_pending: total pending-dwell deadline is still ' diff --git a/backend/tests/unit/test_lifecycle_sim_unit.py b/backend/tests/unit/test_lifecycle_sim_unit.py index ad8cf37ece8..18babbaae6e 100644 --- a/backend/tests/unit/test_lifecycle_sim_unit.py +++ b/backend/tests/unit/test_lifecycle_sim_unit.py @@ -189,7 +189,14 @@ def test_shell_routes_retained_diagnostics_to_its_own_subdirectory(tmp_path, onl def test_age_job_preserves_evidence_and_shifts_numeric_and_datetime_deadlines(): stamp = datetime.fromtimestamp(1000, timezone.utc) - row = {'enqueued_at': 1000.0, 'available_at': 1060.0, 'committed_at': stamp, 'state': 'record_pending'} + row = { + 'enqueued_at': 1000.0, + 'available_at': 1060.0, + 'committed_at': stamp, + 'state': 'record_pending', + 'held_deadline_at': 2000.0, + 'held_expired_at': 1000.0, + } ref = Mock() ref.get.return_value.to_dict.return_value = row updates = age_job_row(ref, 120) @@ -197,6 +204,8 @@ def test_age_job_preserves_evidence_and_shifts_numeric_and_datetime_deadlines(): 'enqueued_at': 880.0, 'available_at': 940.0, 'committed_at': datetime.fromtimestamp(880, timezone.utc), + 'held_deadline_at': 1880.0, + 'held_expired_at': 880.0, } ref.update.assert_called_once_with(updates) assert row['enqueued_at'] == 1000.0 diff --git a/backend/tests/unit/test_monitoring_alert_rule_contract.py b/backend/tests/unit/test_monitoring_alert_rule_contract.py index f6d7adc4e25..e20d4168a44 100644 --- a/backend/tests/unit/test_monitoring_alert_rule_contract.py +++ b/backend/tests/unit/test_monitoring_alert_rule_contract.py @@ -2169,3 +2169,21 @@ def test_pair_page_uses_recent_jobs_and_stock_stays_agent_tier(): assert repair['data'][0]['model']['expr'] == 'max(omi_enrichment_summary_repair_count)' assert repair['data'][2]['model']['conditions'][0]['evaluator']['params'] == [100] assert 'not retry attempts or a retry cap' in repair['annotations']['scope'] + + +def test_held_dead_letter_volume_is_agent_only_and_uses_reasoned_stock_plus_unique_expiry_rate(): + for export_name, rules in _all_rule_exports().items(): + rule = rules['omi-batch-held-expired-volume'] + nodes = {node['refId']: node['model'] for node in rule['data']} + assert nodes['A']['expr'] == ( + 'max(omi_batch_held_expired_units and ' + '(time() - omi_batch_held_expired_observed_at_seconds < 900)) or vector(0)' + ), export_name + assert nodes['B']['expr'] == 'sum(increase(omi_batch_held_expired_total[30m])) or vector(0)' + assert nodes['C']['expr'] == '(time() - max(omi_batch_held_expired_observed_at_seconds)) or vector(1e9)' + assert nodes['D']['expression'] == '($A > 100 && $C < 900) || $B > 10' + assert rule['labels']['tier'] == rule['labels']['audience'] == 'agent' + assert rule['notification_settings']['mute_time_intervals'] == ['omi-tier-silent'] + assert rule['for'] == '5m' and not rule['isPaused'] + assert 'legacy' in rule['annotations']['scope'] + assert 'Never delete' in rule['annotations']['safe_next_action'] diff --git a/backend/tests/unit/test_monitoring_telemetry_contract.py b/backend/tests/unit/test_monitoring_telemetry_contract.py index 21eb6f85e1b..d096c0dfe50 100644 --- a/backend/tests/unit/test_monitoring_telemetry_contract.py +++ b/backend/tests/unit/test_monitoring_telemetry_contract.py @@ -205,6 +205,9 @@ def test_stackdriver_exporter_ingests_firestore_read_count(path): 'prometheus.googleapis.com/omi_finalization_pair_', 'prometheus.googleapis.com/omi_enrichment_summary_repair_', 'prometheus.googleapis.com/omi_batch_record_pending_', + 'prometheus.googleapis.com/omi_batch_held_expired_total/', + 'prometheus.googleapis.com/omi_batch_held_expired_units/', + 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/', 'prometheus.googleapis.com/omi_record_seal_nojob_', 'prometheus.googleapis.com/omi_enrichment_obligation_', 'prometheus.googleapis.com/omi_finalization_byok_', @@ -263,6 +266,9 @@ def test_stackdriver_exporter_ingests_firestore_read_count(path): 'prometheus.googleapis.com/omi_finalization_pair_', 'prometheus.googleapis.com/omi_enrichment_summary_repair_', 'prometheus.googleapis.com/omi_batch_record_pending_', + 'prometheus.googleapis.com/omi_batch_held_expired_total/', + 'prometheus.googleapis.com/omi_batch_held_expired_units/', + 'prometheus.googleapis.com/omi_batch_held_expired_observed_at_seconds/', 'prometheus.googleapis.com/omi_record_seal_nojob_', 'prometheus.googleapis.com/omi_enrichment_obligation_', 'prometheus.googleapis.com/omi_finalization_byok_', @@ -608,3 +614,29 @@ def test_custom_metric_api_aggregates_are_renamed_back_too(env): (Path(__file__).resolve().parents[3] / "backend/charts/monitoring/metric-inventory.prod.json").read_text() ) assert not [m for m in inventory["metrics"] if re.fullmatch(r"stackdriver_.*_omi_.*", m)] + + +@pytest.mark.parametrize( + 'path', [CLOUD_RUN_EXPORTER, CLOUD_RUN_EXPORTER_DEV, CLOUD_RUN_SYNC_EXPORTER, CLOUD_RUN_SYNC_EXPORTER_DEV] +) +def test_held_dead_letter_producer_families_reach_the_volume_alert_through_each_cloud_run_bridge(path): + from prometheus_client import CollectorRegistry + from utils.batch_pipeline.metrics import QueueMetrics + + registry = CollectorRegistry() + producer = QueueMetrics(registry) + producer.held_expired.inc() + producer.held_expired_stock.set(101) + producer.held_expired_observed_at.set(1000) + families = { + 'omi_batch_held_expired_total', + 'omi_batch_held_expired_units', + 'omi_batch_held_expired_observed_at_seconds', + } + emitted = {sample.name for metric in registry.collect() for sample in metric.samples} + assert families <= emitted + imported = set(yaml.safe_load(path.read_text())['stackdriver']['metrics']['prefixes']) + assert {f'prometheus.googleapis.com/{name}/' for name in families} <= imported + rule = next(r for r in json.loads(ALERT_RULES.read_text()) if r['uid'] == 'omi-batch-held-expired-volume') + queries = ' '.join(node['model'].get('expr', '') for node in rule['data']) + assert all(name in queries for name in families) diff --git a/backend/utils/batch_pipeline/held_dead_letter.py b/backend/utils/batch_pipeline/held_dead_letter.py new file mode 100644 index 00000000000..24452d984b7 --- /dev/null +++ b/backend/utils/batch_pipeline/held_dead_letter.py @@ -0,0 +1,110 @@ +"""Bounded held expiry and UTC-nightly re-drive on the serving maintenance leader. + +Durable keyset cursors survive restarts. CAS transitions, not the advisory leader +lease, own correctness. No audio/storage, conversation or retention mutations. +""" + +import logging +import math + +from config.batch_pipeline import batch_held_dead_letter_enabled +from database.batch_pipeline_privacy import BatchPrivacyBlocked +from database.batch_pipeline_projection_protocol import ProjectionBlocked +from database.batch_pipeline_quarantine_recovery import ( + RecoveryStore, + held_candidates, + dead_held_candidates, + held_expired_stock, +) + +PAGE_SIZE = 32 +INTERVAL_SECONDS = 300 +CURSOR_KEY = 'held_dead_letter' +logger = logging.getLogger(__name__) + + +class HeldDeadLetter: + def __init__(self, store, *, metrics=None): + self.store = store + self.recovery = RecoveryStore(store) + self.metrics = metrics + + def _record(self, event): + if self.metrics is not None: + if event == 'expired': + self.metrics.held_expired.inc() + elif event == 'requeued': + self.metrics.record('requeued') + + def tick(self, now): + # Off has no queries, cursor writes, deadline stamps or expirations. + # The separate RecoveryStore.requeue_quarantined tool remains usable. + if not batch_held_dead_letter_enabled(): + return {'expired': 0, 'requeued': 0} + state = self.store.run(lambda tx: tx.get('leases', CURSOR_KEY)) + if state.get('next_at', 0) > now: + return {'expired': 0, 'requeued': 0} + counts = {'expired': 0, 'requeued': 0} + page = held_candidates(self.store.client, self.store.namespace, PAGE_SIZE, after_id=state.get('held_after')) + for sid, _ in page: + try: + outcome = self.recovery.expire_held(sid, now) + except (BatchPrivacyBlocked, ProjectionBlocked): + continue # Revocation/content-owner loss never authorizes a transition. + except Exception as error: + logger.warning( + 'event=batch_held_dead_letter action=expire outcome=retry exception_type=%s', type(error).__name__ + ) + continue # One malformed/unreadable row cannot pin the page forever. + if outcome == 'expired': + counts['expired'] += 1 + self._record('expired') + updated = {**state, 'held_after': page[-1][0] if len(page) == PAGE_SIZE else None} + day = int(now // 86400) + # First tick after UTC midnight starts a pass. Full pages continue on + # later maintenance ticks; a finished pass sleeps until the next day. + if state.get('redrive_day') != day or not state.get('redrive_done'): + after = state.get('dead_after') if state.get('redrive_day') == day else None + dead = dead_held_candidates(self.store.client, self.store.namespace, PAGE_SIZE, after_id=after) + for sid, job in dead: + # A row expired today gets until the next night; stale snapshots + # are rechecked by the sanctioned content/privacy/lease CAS. + expired_at = job.get('held_expired_at') + if ( + isinstance(expired_at, bool) + or not isinstance(expired_at, (int, float)) + or not math.isfinite(expired_at) + or expired_at >= day * 86400 + ): + continue + try: + changed = self.recovery.requeue_quarantined(sid, now, dead_before=day * 86400) + except (BatchPrivacyBlocked, ProjectionBlocked): + continue + except Exception as error: + logger.warning( + 'event=batch_held_dead_letter action=requeue outcome=retry exception_type=%s', + type(error).__name__, + ) + continue + if changed: + counts['requeued'] += 1 + self._record('requeued') + updated.update( + redrive_day=day, redrive_done=len(dead) < PAGE_SIZE, dead_after=dead[-1][0] if dead else None + ) + updated['next_at'] = now + INTERVAL_SECONDS + + def checkpoint(tx): + if tx.get('leases', CURSOR_KEY) == state: + tx.put('leases', CURSOR_KEY, updated) + + # Checkpoint after work: a crash replays an idempotent page. A competing + # leader cannot overwrite a newer cursor with this older snapshot. + self.store.run(checkpoint) + if self.metrics is not None: + stock = held_expired_stock(self.store.client, self.store.namespace) + self.metrics.held_expired_stock.set(stock) + self.metrics.held_expired_observed_at.set(now) + logger.info('event=batch_held_dead_letter expired=%d requeued=%d', counts['expired'], counts['requeued']) + return counts diff --git a/backend/utils/batch_pipeline/intake.py b/backend/utils/batch_pipeline/intake.py index 6fae85b172e..a43a0e26de3 100644 --- a/backend/utils/batch_pipeline/intake.py +++ b/backend/utils/batch_pipeline/intake.py @@ -159,7 +159,9 @@ def _terminally_held(job: dict) -> bool: Its encrypted artifacts stay server-side for a reviewed operator redrive, so a re-upload can neither change the outcome nor add a recovery path. """ - return job.get('state') == 'quarantined' and job.get('recovery_state') == 'failed_held' + return (job.get('state') == 'quarantined' and job.get('recovery_state') == 'failed_held') or ( + job.get('state') == 'dead' and job.get('dead_letter_reason') == 'held_expired' + ) def _held_status(jobs: list[dict]) -> tuple[str, str | None] | None: @@ -169,7 +171,8 @@ def _held_status(jobs: list[dict]) -> tuple[str, str | None] | None: failed = [ job for job in jobs - if job.get('state') == 'quarantined' and job.get('recovery_state') not in {'held', 'prepared', 'running'} + if _terminally_held(job) + or (job.get('state') == 'quarantined' and job.get('recovery_state') not in {'held', 'prepared', 'running'}) ] if not failed or not all(_terminally_held(job) for job in failed) or not held_terminal_status_enabled(): return None @@ -302,7 +305,8 @@ def read(tx): for job in jobs ) failed = sum( - job.get('state') == 'quarantined' and job.get('recovery_state') not in {'held', 'prepared', 'running'} + _terminally_held(job) + or (job.get('state') == 'quarantined' and job.get('recovery_state') not in {'held', 'prepared', 'running'}) for job in jobs ) held = None if completed else _held_status(jobs) diff --git a/backend/utils/batch_pipeline/metrics.py b/backend/utils/batch_pipeline/metrics.py index f26445899fc..c0018374af0 100644 --- a/backend/utils/batch_pipeline/metrics.py +++ b/backend/utils/batch_pipeline/metrics.py @@ -49,6 +49,15 @@ def __init__( min_refresh_seconds: float = 0.0, ): self.scope = (worker_release, namespace) + self.held_expired = Counter( + 'omi_batch_held_expired_total', 'Unique fenced held-to-dead transitions; audio retained', registry=registry + ) + self.held_expired_stock = Gauge( + 'omi_batch_held_expired_units', 'Retained dead(held_expired) stock across all ages', registry=registry + ) + self.held_expired_observed_at = Gauge( + 'omi_batch_held_expired_observed_at_seconds', 'Last successful held-expired stock sample', registry=registry + ) # Serving instances share one global queue; workers keep 0 for the HPA. self.min_refresh_seconds = min_refresh_seconds self._refresh_due = 0.0 diff --git a/backend/utils/batch_pipeline/retry_policy.py b/backend/utils/batch_pipeline/retry_policy.py index 657f6eea73a..c601267fc77 100644 --- a/backend/utils/batch_pipeline/retry_policy.py +++ b/backend/utils/batch_pipeline/retry_policy.py @@ -2,6 +2,8 @@ import logging +from config.batch_pipeline import batch_held_dead_letter_enabled, HELD_DEAD_LETTER_SECONDS + from config.policy import SchedulerPolicy from utils.batch_pipeline.correctness_metrics import record as record_correctness from utils.capture.retry_cap import observe, retry_cap @@ -29,6 +31,8 @@ def quarantine_job(tx, job: dict, now: float, reason: str) -> None: recovery_state='held', ) job.pop('retry_started_at', None) + if batch_held_dead_letter_enabled(): + job['held_deadline_at'] = now + HELD_DEAD_LETTER_SECONDS tx.put('owners', job['owner_key'], owner) diff --git a/backend/utils/batch_pipeline/scheduler.py b/backend/utils/batch_pipeline/scheduler.py index 5c6e0c6e641..7fa65cf75c7 100644 --- a/backend/utils/batch_pipeline/scheduler.py +++ b/backend/utils/batch_pipeline/scheduler.py @@ -936,6 +936,58 @@ def restore_audio_in_transaction(tx: Transaction, job: dict, now: float) -> None _append(tx, control, job) tx.put('jobs', job['segment_id'], job) + @staticmethod + def held_requeue_eligible(tx: Transaction, job: dict, now: float, *, allow_dead=True) -> bool: + """Dormant worker holds only; never reopen an active paid/operator intent.""" + held_dead = allow_dead and job.get('state') == 'dead' and job.get('dead_letter_reason') == 'held_expired' + state_ok = job.get('state') == 'quarantined' or held_dead + lease_until = job.get('lease_until', 0) + if ( + not state_ok + or job.get('quarantine_reason') != 'worker_failed' + # Provider-STT units do not belong to the GPU DRR ring. Their + # vendor-attempt custody needs its own sanctioned re-drive path. + or (job.get('stt_route') or {}).get('stt_route') == 'provider' + or job.get('recovery_state') not in {'held', 'failed_held'} + or job.get('result') + # Expiry retains the legacy dead re-drive contract. Also accept + # earlier held-expired rows written with merge_pending=False. + or not (job.get('merge_pending') is False or (held_dead and job.get('merge_pending') is True)) + or job.get('token') + or isinstance(lease_until, bool) + or not isinstance(lease_until, (int, float)) + or not math.isfinite(lease_until) + or lease_until > now + ): + return False + recovery = job.get('recovery') or {} + if recovery and not recovery.get('settled_at'): + return False + if any( + job.get(key) and (job.get(key) or {}).get('state') not in {'completed', 'failed_held'} + for key in ('silence_disposition', 'terminal_disposition', 'speech_redrive') + ): + return False + return not tx.get('results', job['segment_id']) + + @staticmethod + def requeue_quarantined_in_transaction(tx: Transaction, job: dict, now: float) -> bool: + """CAS requeue without deleting audio, results, intent or attempt history. + + RecoveryStore owns content/ledger/privacy checks, as the missing-input + store does for restore_audio_in_transaction. The commit fence still runs. + """ + if not Scheduler.held_requeue_eligible(tx, job, now): + return False + control = read_control(tx, scheduler_control_key(job['owner_key'])) + owner = tx.get('owners', job['owner_key']) + owner['revision'] = owner.get('revision', 0) + 1 + job.update(state='ready', available_at=now, merge_pending=True, token='', lease_until=0.0, held_requeued_at=now) + _append(tx, control, job) + tx.put('owners', job['owner_key'], owner) + tx.put('jobs', job['segment_id'], job) + return True + @staticmethod def redrive_in_transaction(tx: Transaction, job: dict, now: float) -> bool: if job.get('state') != 'dead' or not job.get('merge_pending'): diff --git a/backend/utils/batch_pipeline/serving.py b/backend/utils/batch_pipeline/serving.py index da1e474d5e0..fd0bb0463d2 100644 --- a/backend/utils/batch_pipeline/serving.py +++ b/backend/utils/batch_pipeline/serving.py @@ -32,6 +32,7 @@ from utils.executors import run_blocking, start_background_task, sync_executor from utils.batch_pipeline.contracts import PoolCapacity, WorkResult from utils.batch_pipeline.leader import SERVING_MAINTENANCE, LeaderLease +from utils.batch_pipeline.held_dead_letter import HeldDeadLetter from utils.batch_pipeline.merge import OrderedMerge from utils.batch_pipeline.metrics import QueueMetrics from utils.batch_pipeline.scheduler import Scheduler @@ -82,6 +83,7 @@ def __init__( self.recovery = recovery # Optional single-owner lease for periodic maintenance (see reconcile). self.leader = leader + self.held_dead_letter = HeldDeadLetter(store, metrics=metrics) self.merge = OrderedMerge(store, metrics=metrics) # Held units change only on a recovery or operator settle: scan them # at a slow cadence, not every maintenance pass. @@ -322,6 +324,12 @@ def _maintain(self, *, recover_quarantine: bool): logging.getLogger(__name__).error( 'event=batch_projection action=release_held outcome=retry exception_type=%s', type(error).__name__ ) + try: + self.held_dead_letter.tick(time.time()) + except Exception as error: + logging.getLogger(__name__).error( + 'event=batch_held_dead_letter outcome=retry exception_type=%s', type(error).__name__ + ) self.queue().recover(time.time()) owners = set() for segment in self.store.due('result_ready', time.time(), 100): diff --git a/config/deployment-setting-classification.json b/config/deployment-setting-classification.json index 784ffa2a889..6c116b56ae7 100644 --- a/config/deployment-setting-classification.json +++ b/config/deployment-setting-classification.json @@ -92,6 +92,7 @@ "LISTEN_EDGE_PRIVATE_VALUES_YAML" ], "config": [ + "BATCH_HELD_DEAD_LETTER_ENABLED", "HOME_TRANSCRIPTION_PRESENTATION_GATE", "HOME_TRANSCRIPTION_DEADLINE_SECONDS", "PUSHER_ABANDON_EVICTED_TOKENS", diff --git a/config/feature-flags.yaml b/config/feature-flags.yaml index f8957dc91cc..48c39c99f37 100644 --- a/config/feature-flags.yaml +++ b/config/feature-flags.yaml @@ -1,4 +1,5 @@ flags: + - {"key":"BATCH_HELD_DEAD_LETTER_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; seven-day held deadlines, retained dead letters and bounded nightly re-drive","fail":"open","owner":"uicleanup","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"False stops new stamps, expiry and automatic maintenance; private OIDC requeue-quarantined remains usable. Legacy rows get seven days at first sweep. Dead letters retain encrypted audio, reason held_expired and all intent/attempt history; no TTL, deletion or conversation writes. Nightly UTC keyset pass on existing serving maintenance leader; 32 rows per five minutes. Stock and rate alert to agent tier. See backend/docs/runbooks/batch-held-dead-letter.md."} - {"key":"PUSHER_PROJECTION_DEBT_DURABLE_ENABLED","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; persist exhausted or evicted playback projection debt and rebuild it in the record repair CronJob","fail":"open","owner":"uicleanup / omi-lifecycle","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"0, false or off disables both marker handoff and repair consumption. Disable recordSealRepair.projectionDebtDurableEnabled first, then the Pusher runtime ConfigMap flag. Consumer requires the existing CronJob to be enabled and unsuspended; no activation here. Content-free record_projection_rebuild_jobs markers remain inert. No hot-path writes, seal authority, token, PCM or audio-object changes. See backend/docs/listen_pusher_pipeline/changes/2026-11-06-durable-projection-debt.md."} - {"key":"BATCH_PIPELINE_SWEEP_AGE_GATE","aliases":[],"kind":"env","lifecycle":"ops_kill","surfaces":["backend"],"summary":"Default on; only off disables non-mutating age limits on provider-stt and enrichment-fresh sweep re-wakes","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."} - {"key":"BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS","aliases":[],"kind":"env","lifecycle":"config_switch","surfaces":["backend"],"summary":"Maximum durable obligation age for sweep re-wakes; default 14 days; invalid values use 14","fail":"closed","owner":"omi-pipeline","created":"2026-11-06","terminal":"keep","decision":"keep","notes":"Wave4 t6: exclude months-old retries while retaining two weeks for normal recovery. Provider uses created_at; fresh enrichment uses first_pending_at. Older rows remain unchanged for a separately bounded backfill. Initial enqueue and consumers are unchanged; off restores the existing sweeps including the independent enrichment parking policy."}