Repository navigation
fix(batch): bound held dwell and re-drive retained dead letters - #24736
Conversation
Failure-Class: FC-batch-fault-domain-head-of-line-poison
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 1003714676
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| or (job.get('stt_route') or {}).get('stt_route') == 'provider' | ||
| or job.get('recovery_state') not in {'held', 'failed_held'} | ||
| or job.get('result') | ||
| or job.get('merge_pending') is not False | ||
| or job.get('token') |
There was a problem hiding this comment.
When a held row has artifact_disposition.state == 'missing', this eligibility check still permits expiry and automatic requeue. Expiry changes it to dead, but ArtifactMissingStore.reupload_snapshot() explicitly requires state == 'quarantined', so the original-upload repair invoked by intake._repair_existing_upload() now fails until another worker attempt fails and quarantines it again. The nightly requeue also sends known-missing input back to compute without restoring it. Exclude missing-artifact holds from these transitions, or update the existing restoration protocol to support the new states atomically.
Useful? React with 👍 / 👎.
| # 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. |
There was a problem hiding this comment.
This introduces default-on transcription lifecycle behavior, a new retained dead-letter disposition, and automatic requeue, but the commit adds no note under backend/docs/listen_pusher_pipeline/changes/ (checked the commit's changed-file list and searched the repository's existing notes). The runbook does not satisfy the separate same-PR change-note requirement in backend/AGENTS.md; add a dated note covering the lifecycle change, rollout/rollback state, and validation evidence.
AGENTS.md reference: backend/AGENTS.md:L164-L164
Useful? React with 👍 / 👎.
| from database.batch_pipeline_quarantine_recovery import RecoveryStore | ||
| from database.batch_pipeline_projection_protocol import ProjectionBlocked |
There was a problem hiding this comment.
The new handler introduces function-local imports, contrary to the explicit backend/AGENTS.md requirement that all imports be at module scope. Move these imports to the router's import block and likewise move the new local kill_switch import in config/batch_pipeline.py to module scope.
AGENTS.md reference: backend/AGENTS.md:L168-L168
Useful? React with 👍 / 👎.
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
13 issues found and verified against the latest diff
Confidence score: 2/5
- In
prod_omi_backend_listen_values.yaml, production enables the dead-letter lifecycle in the listen pod while the serving maintenance loop is off, so held rows may never be processed. Align the production switches with the loop that runs this work. held_requeue_eligiblecan send rows with missing artifacts back to the GPU without restoring their audio. Exclude missing-artifact rows from requeue eligibility.held_requeue_eligiblealso assumesstt_routeis a mapping, so legacy rows with the string value'provider'can raise instead of being handled. Guard the type before reading route fields.held_dead_letter.pycan repeatedly process retained dead rows and exceed the intended 32-row tick limit. Filter candidates to quarantined rows and share one budget across both passes.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="backend/database/batch_pipeline_projection.py">
<violation number="1" location="backend/database/batch_pipeline_projection.py:885">
P2: This broader predicate also authorizes the artifact-missing repair path for retained dead letters. Add an explicit `state == 'quarantined'` guard to `ArtifactMissingStore.apply()` and other repair entry points so expired retained rows remain untouched until requeue.</violation>
</file>
<file name="backend/utils/batch_pipeline/scheduler.py">
<violation number="1" location="backend/utils/batch_pipeline/scheduler.py:951">
P2: `held_requeue_eligible` assumes every `stt_route` is a mapping. Legacy/provider rows can store the route as the string `'provider'`, so this raises instead of returning `False` for provider-STT rows or requeueing eligible legacy GPU rows; normalize both persisted shapes before calling `.get`.</violation>
<violation number="2" location="backend/utils/batch_pipeline/scheduler.py:952">
P2: `held_requeue_eligible` admits rows with `artifact_disposition.state == 'missing'`, so expiry can make them `dead` and nightly requeue can send them to GPU without restoring audio. Exclude missing-artifact rows from these transitions or make restoration support these states atomically.</violation>
</file>
<file name="backend/tests/unit/test_flag_budget.py">
<violation number="1" location="backend/tests/unit/test_flag_budget.py:233">
P3: The new flag bypasses the common registry assertions, so important metadata can drift without this contract test failing. Add the same registry, kind, surface, fail, and default-on checks for `BATCH_HELD_DEAD_LETTER_ENABLED`.</violation>
</file>
<file name="backend/utils/batch_pipeline/held_dead_letter.py">
<violation number="1" location="backend/utils/batch_pipeline/held_dead_letter.py:48">
P2: This sweep reprocesses expired rows forever: `held_candidates` still returns `dead` rows because they retain `recovery_state='failed_held'`. Add a `state == 'quarantined'` filter to `held_candidates` so retained dead stock does not consume the expiry page and delay live holds.</violation>
<violation number="2" location="backend/utils/batch_pipeline/held_dead_letter.py:68">
P2: This tick can process 64 rows, not the specified 32-row bound, because it runs a full held page and a full dead-letter page back-to-back. Share one per-tick budget across both passes so maintenance load and re-drive throughput remain bounded.</violation>
</file>
<file name="backend/tests/integration/test_batch_held_dead_letter_firestore.py">
<violation number="1" location="backend/tests/integration/test_batch_held_dead_letter_firestore.py:19">
P2: This integration test depends on the caller's `BATCH_HELD_DEAD_LETTER_ENABLED` value. Set the flag to `true` in this test or the shared fixture so an operator/test-runner kill switch cannot make the enabled-path assertions fail before the lifecycle is exercised.</violation>
</file>
<file name="backend/charts/backend-listen/prod_omi_backend_listen_values.yaml">
<violation number="1" location="backend/charts/backend-listen/prod_omi_backend_listen_values.yaml:140">
P1: This enables the dead-letter lifecycle in the listen pod without enabling the serving maintenance loop that executes it. Because production leaves both serving and projection-pump switches off, held rows will not be swept into retained dead letters or nightly re-driven; enable the maintenance role/switch in the same rollout or run this loop from a dedicated deployment.</violation>
</file>
<file name="backend/tests/unit/test_lifecycle_invariants.py">
<violation number="1" location="backend/tests/unit/test_lifecycle_invariants.py:320">
P2: This aggregate xfail permanently masks regressions in the GPU quarantine path because the intentionally unstamped provider-STT writers keep the assertion failing. Add a separate non-xfailed assertion for `retry_policy.quarantine_job` and keep only the provider-STT paths in the expected gap.</violation>
</file>
<file name="backend/utils/batch_pipeline/retry_policy.py">
<violation number="1" location="backend/utils/batch_pipeline/retry_policy.py:34">
P2: When the flag is disabled for a requeue/retry cycle, this branch leaves the prior `held_deadline_at` intact. Re-enabling maintenance can therefore dead-letter the newly quarantined row immediately instead of granting a fresh seven-day window; clear the old field on the disabled path.</violation>
</file>
<file name="backend/docs/feature-flag-registry.md">
<violation number="1" location="backend/docs/feature-flag-registry.md:516">
P2: This registry row omits the `batch-pipeline-worker` chart from its deployment cells even though that chart injects the flag. Operators following the registry can disable serving hosts while leaving worker expiry and maintenance enabled; include the worker chart in the generated deployment inventory.</violation>
</file>
<file name="backend/utils/batch_pipeline/serving.py">
<violation number="1" location="backend/utils/batch_pipeline/serving.py:327">
P2: This call also runs for non-periodic `/reconcile` requests, which bypass the leader lease. Gate the held-dead-letter tick behind the periodic leader path so authenticated reconcile traffic cannot trigger the nightly maintenance work on every serving instance.</violation>
</file>
<file name="backend/routers/batch_pipeline.py">
<violation number="1" location="backend/routers/batch_pipeline.py:231">
P3: Move the handler imports into the module import block; the backend import policy forbids function-local imports. Move `kill_switch` in `config/batch_pipeline.py` to module scope as well.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Turn on auto-fix | Re-trigger cubic
| 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 |
There was a problem hiding this comment.
P1: This enables the dead-letter lifecycle in the listen pod without enabling the serving maintenance loop that executes it. Because production leaves both serving and projection-pump switches off, held rows will not be swept into retained dead letters or nightly re-driven; enable the maintenance role/switch in the same rollout or run this loop from a dedicated deployment.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/charts/backend-listen/prod_omi_backend_listen_values.yaml, line 140:
<comment>This enables the dead-letter lifecycle in the listen pod without enabling the serving maintenance loop that executes it. Because production leaves both serving and projection-pump switches off, held rows will not be swept into retained dead letters or nightly re-driven; enable the maintenance role/switch in the same rollout or run this loop from a dedicated deployment.</comment>
<file context>
@@ -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
</file context>
| 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' |
There was a problem hiding this comment.
P2: This broader predicate also authorizes the artifact-missing repair path for retained dead letters. Add an explicit state == 'quarantined' guard to ArtifactMissingStore.apply() and other repair entry points so expired retained rows remain untouched until requeue.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/database/batch_pipeline_projection.py, line 885:
<comment>This broader predicate also authorizes the artifact-missing repair path for retained dead letters. Add an explicit `state == 'quarantined'` guard to `ArtifactMissingStore.apply()` and other repair entry points so expired retained rows remain untouched until requeue.</comment>
<file context>
@@ -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
</file context>
| 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' |
There was a problem hiding this comment.
P2: held_requeue_eligible assumes every stt_route is a mapping. Legacy/provider rows can store the route as the string 'provider', so this raises instead of returning False for provider-STT rows or requeueing eligible legacy GPU rows; normalize both persisted shapes before calling .get.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/scheduler.py, line 951:
<comment>`held_requeue_eligible` assumes every `stt_route` is a mapping. Legacy/provider rows can store the route as the string `'provider'`, so this raises instead of returning `False` for provider-STT rows or requeueing eligible legacy GPU rows; normalize both persisted shapes before calling `.get`.</comment>
<file context>
@@ -936,6 +936,57 @@ def restore_audio_in_transaction(tx: Transaction, job: dict, now: float) -> None
+ 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')
</file context>
| or (job.get('stt_route') or {}).get('stt_route') == 'provider' | |
| or (job.get('stt_route') == 'provider' or (isinstance(job.get('stt_route'), dict) and job['stt_route'].get('stt_route') == 'provider')) |
| # 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) |
There was a problem hiding this comment.
P2: This tick can process 64 rows, not the specified 32-row bound, because it runs a full held page and a full dead-letter page back-to-back. Share one per-tick budget across both passes so maintenance load and re-drive throughput remain bounded.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/held_dead_letter.py, line 68:
<comment>This tick can process 64 rows, not the specified 32-row bound, because it runs a full held page and a full dead-letter page back-to-back. Share one per-tick budget across both passes so maintenance load and re-drive throughput remain bounded.</comment>
<file context>
@@ -0,0 +1,110 @@
+ # 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
</file context>
| pytestmark = pytest.mark.integration | ||
|
|
||
|
|
||
| def test_worker_fail_expire_and_nightly_redrive_keep_audio_and_cold_queue_integrity(lane): |
There was a problem hiding this comment.
P2: This integration test depends on the caller's BATCH_HELD_DEAD_LETTER_ENABLED value. Set the flag to true in this test or the shared fixture so an operator/test-runner kill switch cannot make the enabled-path assertions fail before the lifecycle is exercised.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/integration/test_batch_held_dead_letter_firestore.py, line 19:
<comment>This integration test depends on the caller's `BATCH_HELD_DEAD_LETTER_ENABLED` value. Set the flag to `true` in this test or the shared fixture so an operator/test-runner kill switch cannot make the enabled-path assertions fail before the lifecycle is exercised.</comment>
<file context>
@@ -0,0 +1,62 @@
+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]
</file context>
| 'event=batch_projection action=release_held outcome=retry exception_type=%s', type(error).__name__ | ||
| ) | ||
| try: | ||
| self.held_dead_letter.tick(time.time()) |
There was a problem hiding this comment.
P2: This call also runs for non-periodic /reconcile requests, which bypass the leader lease. Gate the held-dead-letter tick behind the periodic leader path so authenticated reconcile traffic cannot trigger the nightly maintenance work on every serving instance.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/serving.py, line 327:
<comment>This call also runs for non-periodic `/reconcile` requests, which bypass the leader lease. Gate the held-dead-letter tick behind the periodic leader path so authenticated reconcile traffic cannot trigger the nightly maintenance work on every serving instance.</comment>
<file context>
@@ -321,6 +323,12 @@ def _maintain(self, *, recover_quarantine: bool):
'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(
</file context>
| 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')) |
There was a problem hiding this comment.
P2: This sweep reprocesses expired rows forever: held_candidates still returns dead rows because they retain recovery_state='failed_held'. Add a state == 'quarantined' filter to held_candidates so retained dead stock does not consume the expiry page and delay live holds.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/held_dead_letter.py, line 48:
<comment>This sweep reprocesses expired rows forever: `held_candidates` still returns `dead` rows because they retain `recovery_state='failed_held'`. Add a `state == 'quarantined'` filter to `held_candidates` so retained dead stock does not consume the expiry page and delay live holds.</comment>
<file context>
@@ -0,0 +1,110 @@
+ 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:
</file context>
| # 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'} |
There was a problem hiding this comment.
P2: held_requeue_eligible admits rows with artifact_disposition.state == 'missing', so expiry can make them dead and nightly requeue can send them to GPU without restoring audio. Exclude missing-artifact rows from these transitions or make restoration support these states atomically.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/utils/batch_pipeline/scheduler.py, line 952:
<comment>`held_requeue_eligible` admits rows with `artifact_disposition.state == 'missing'`, so expiry can make them `dead` and nightly requeue can send them to GPU without restoring audio. Exclude missing-artifact rows from these transitions or make restoration support these states atomically.</comment>
<file context>
@@ -936,6 +936,57 @@ def restore_audio_in_transaction(tx: Transaction, job: dict, now: float) -> None
+ # 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')
+ or job.get('merge_pending') is not False
</file context>
| or job.get('recovery_state') not in {'held', 'failed_held'} | |
| or (job.get('artifact_disposition') or {}).get('state') == 'missing' | |
| or job.get('recovery_state') not in {'held', 'failed_held'} |
| 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' |
There was a problem hiding this comment.
P3: The new flag bypasses the common registry assertions, so important metadata can drift without this contract test failing. Add the same registry, kind, surface, fail, and default-on checks for BATCH_HELD_DEAD_LETTER_ENABLED.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_flag_budget.py, line 233:
<comment>The new flag bypasses the common registry assertions, so important metadata can drift without this contract test failing. Add the same registry, kind, surface, fail, and default-on checks for `BATCH_HELD_DEAD_LETTER_ENABLED`.</comment>
<file context>
@@ -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'
</file context>
| 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' | |
| held = names['BATCH_HELD_DEAD_LETTER_ENABLED'] | |
| held_flag = flags['BATCH_HELD_DEAD_LETTER_ENABLED'] | |
| assert held['registry'] == 'config/feature-flags.yaml' | |
| assert held['owner'] == 'uicleanup / omi-lifecycle' | |
| assert held['review_by'] == '2026-12-06' | |
| assert held_flag['owner'] == 'uicleanup' | |
| assert held_flag['kind'] == 'env' | |
| assert held_flag['lifecycle'] == 'ops_kill' | |
| assert 'backend' in held_flag['surfaces'] | |
| assert held_flag['fail'] == 'open' | |
| assert held_flag['summary'].startswith('Default on') |
| @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 |
There was a problem hiding this comment.
P3: Move the handler imports into the module import block; the backend import policy forbids function-local imports. Move kill_switch in config/batch_pipeline.py to module scope as well.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/routers/batch_pipeline.py, line 231:
<comment>Move the handler imports into the module import block; the backend import policy forbids function-local imports. Move `kill_switch` in `config/batch_pipeline.py` to module scope as well.</comment>
<file context>
@@ -225,6 +225,21 @@ async def redrive_dead(body: RedriveRequest):
+@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
+
</file context>
There was a problem hiding this comment.
3 issues found across 48 files
Confidence score: 3/5
test_monitoring_alert_rule_contract.pycan pass if Grafana loses the Prometheus datasource or evaluates the wrong node, leaving the alert unable to track the retained-stock/expiry threshold. Assert the datasource type and UID, and pin conditionEand its reduce expression.- In
batch-held-dead-letter.md,accepted=falsecan mean ordinary eligibility refusals, not just a dead-letter hold. Tell operators to check the row’s CAS eligibility before interpreting that result.
Prompt for AI agents (unresolved issues)
Check if these issues are valid — if so, understand the root cause of each and fix them. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. If appropriate, use sub-agents to investigate and fix each issue separately.
<file name="backend/docs/runbooks/batch-held-dead-letter.md">
<violation number="1" location="backend/docs/runbooks/batch-held-dead-letter.md:41">
P3: `accepted=false` also covers ordinary eligibility refusals, including active intents, provider-STT holds, leases, tokens, results, or missing receipts. Tell operators to inspect the row’s CAS eligibility rather than treating every false response as only fenced or already advanced.</violation>
</file>
<file name="backend/tests/unit/test_monitoring_alert_rule_contract.py">
<violation number="1" location="backend/tests/unit/test_monitoring_alert_rule_contract.py:2178">
P2: This contract does not pin the metric query datasource, so replacing the Prometheus datasource on `A`-`C` would still pass and could make the alert unable to evaluate these metrics. Assert the datasource type and UID for each metric node.</violation>
<violation number="2" location="backend/tests/unit/test_monitoring_alert_rule_contract.py:2184">
P2: This test never verifies which node Grafana evaluates, so the alert could stop using the retained-stock/expiry threshold while all these assertions still pass. Pin `rule['condition'] == 'E'` and node `E`'s reduce expression to `D`.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Turn on auto-fix | Re-trigger cubic
| ), 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' |
There was a problem hiding this comment.
P2: This test never verifies which node Grafana evaluates, so the alert could stop using the retained-stock/expiry threshold while all these assertions still pass. Pin rule['condition'] == 'E' and node E's reduce expression to D.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_monitoring_alert_rule_contract.py, line 2184:
<comment>This test never verifies which node Grafana evaluates, so the alert could stop using the retained-stock/expiry threshold while all these assertions still pass. Pin `rule['condition'] == 'E'` and node `E`'s reduce expression to `D`.</comment>
<file context>
@@ -2169,3 +2169,21 @@ def test_pair_page_uses_recent_jobs_and_stock_stays_agent_tier():
+ ), 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']
</file context>
| assert nodes['D']['expression'] == '($A > 100 && $C < 900) || $B > 10' | |
| assert rule['condition'] == 'E', export_name | |
| assert nodes['E']['type'] == 'reduce' and nodes['E']['expression'] == 'D', export_name | |
| assert nodes['D']['expression'] == '($A > 100 && $C < 900) || $B > 10' |
| 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'] == ( |
There was a problem hiding this comment.
P2: This contract does not pin the metric query datasource, so replacing the Prometheus datasource on A-C would still pass and could make the alert unable to evaluate these metrics. Assert the datasource type and UID for each metric node.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/tests/unit/test_monitoring_alert_rule_contract.py, line 2178:
<comment>This contract does not pin the metric query datasource, so replacing the Prometheus datasource on `A`-`C` would still pass and could make the alert unable to evaluate these metrics. Assert the datasource type and UID for each metric node.</comment>
<file context>
@@ -2169,3 +2169,21 @@ def test_pair_page_uses_recent_jobs_and_stock_stays_agent_tier():
+ 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)'
</file context>
| `POST /v2/batch-pipeline/requeue-quarantined` with | ||
| `{"stage":"prod|dev","segment_id":"<retained segment>"}` 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 |
There was a problem hiding this comment.
P3: accepted=false also covers ordinary eligibility refusals, including active intents, provider-STT holds, leases, tokens, results, or missing receipts. Tell operators to inspect the row’s CAS eligibility rather than treating every false response as only fenced or already advanced.
Prompt for AI agents
Check if this issue is valid — if so, understand the root cause and fix it. When an issue isn't valid or won't be fixed in this PR, reply in its thread with the reason and then resolve the thread. At backend/docs/runbooks/batch-held-dead-letter.md, line 41:
<comment>`accepted=false` also covers ordinary eligibility refusals, including active intents, provider-STT holds, leases, tokens, results, or missing receipts. Tell operators to inspect the row’s CAS eligibility rather than treating every false response as only fenced or already advanced.</comment>
<file context>
@@ -0,0 +1,66 @@
+`POST /v2/batch-pipeline/requeue-quarantined` with
+`{"stage":"prod|dev","segment_id":"<retained segment>"}` 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.
</file context>
| `accepted=false` is a fenced or already advanced row. It performs no provider | |
| +`accepted=false` means the row failed one or more CAS eligibility checks, such as a fence, active intent, provider-STT hold or already advanced state. |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 2332c5882b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| 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) |
There was a problem hiding this comment.
For an upload whose client continues polling, setting merge_pending=True makes the newly expired row eligible for the unconditional Scheduler.redrive_in_transaction() loop in intake.durable_status(). The next status request therefore requeues it immediately instead of retaining it until the next nightly pass—even when BATCH_HELD_DEAD_LETTER_ENABLED has subsequently been disabled. That legacy transition also bypasses the new requeue eligibility checks. Exclude held_expired rows from automatic status redrive and route their automatic retries through the fenced nightly path, while retaining the explicit operator rollback route.
Useful? React with 👍 / 👎.
| 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 |
There was a problem hiding this comment.
When a dead-letter pass takes longer than one day, this resets its cursor at midnight even though redrive_done is still false. Rows skipped because of content/privacy fences or read errors remain in the query, so a sufficiently large blocked prefix permanently starves eligible rows behind it. With the configured 32-row pages and five-minute interval, a synthetic run containing 9,216 blocked rows followed by one eligible row revisited the same prefix on both days and never reached the eligible row; maintenance delays lower that threshold further. Preserve the cursor and the pass's age cutoff until the unfinished pass reaches its end, then schedule the next nightly pass.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 30b59b4c95
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| 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) |
There was a problem hiding this comment.
When an expired hold precedes other pending captures for the same owner, setting merge_pending=True puts it back at the head of first_pending() (database/batch_pipeline_store.py), whose query does not exclude dead rows. OrderedMerge.commit_next() then blocks healthy result_ready successors behind this non-runnable row: they incur the default head wait, or remain blocked until requeue when BATCH_PIPELINE_MERGE_HEAD_WAIT_SECONDS=off selects strict ordering. Before expiry, the quarantined row had merge_pending=False and did not block them. Preserve rollback recovery without making retained dead letters participate in the merge frontier.
Useful? React with 👍 / 👎.
GPU batch holds had no bounded terminal disposition or automatic re-attempt. This adds a seven-day
held_deadline_at, retaineddead(held_expired)rows, a private-OIDC CAS requeue, a nightly UTC re-drive on the existing serving maintenance leader, and an agent-tier volume alert. No CronJob is added.Initial quarantine and failed-held recovery writers stamp a full window. Legacy rows receive seven days at their first eligible sweep. Expiry/requeue recheck content ownership, privacy generation, dormant held state, no result or result receipt, no worker token and an expired finite lease. Active paid/operator intents block the transition. Durable document-key cursors bound each pass to 32 rows per five-minute tick; crashes replay idempotently and malformed rows cannot pin the page. Nightly re-drive rechecks its previous-day cutoff inside the transaction before normal GPU admission.
BATCH_HELD_DEAD_LETTER_ENABLED/batch_held_dead_letter_enabled()defaults ON. It is registered in kill switches, flag budget, feature registry, deployment classification, common runtime declarations and worker overlays. Rollback: false on serving/workers stops new stamps and automatic maintenance; the private requeue endpoint remains usable. Expiry atomically setsmerge_pending=true, so after reverting the new code an operator can still use the unchangedPOST /v2/batch-pipeline/redrive-deadroute. No pre-revert drain or data rewrite is required for rows written by this implementation. The rolled-back service requires an operator re-drive rather than the new nightly helper. Seebackend/docs/runbooks/batch-held-dead-letter.md.Custody: nothing is deleted. Every new dead letter has reason
held_expired, retains its deadline, encrypted audio references and attempt/intent history, and has a sanctioned requeue path. No TTL, conversation-status writer or custody/privacy fence is changed. Status and partial-accounting readers retain the held answer, so age does not acknowledge phone custody or hide healthy siblings. Projection expiry (4b), BYOK/finalization tools (4c), andprivacy_blockedremain outside this change. Provider-STT holds remain fenced from the GPU queue; the state-table deadline gap remains explicitly strict-xfailed.The volume alert is agent tier, with
omi-tier-silentrouting so it never pages humans. It checks fresh retained stock >100 or >10 expirations in 30 minutes for five minutes. Each stock sample must be younger than 15 minutes before aggregation; unrelated legacy dead reasons do not count. Notification timing follows the warning contract. Split/combined rules, metric inventories, the sync GMP scrape allowlist and all four dev/prod main/sync exporter prefix lists include the three emitted metric families. Thresholds are initial operator defaults, not production-calibrated. Backend and metrics-egress/alert rollout must use the checked-in declarations; this PR performs none of those rollouts.Validation (offline/synthetic; suites overlap):
make preflightandscripts/pr-preflight: 52 checks passed. Both published fix and merge heads passed the normal pre-push hook, including backend tests, type checks, OpenAPI and Firestore guard; no bypass, cap override, force or timeout retry.exec-heavy-system-squeeze-major-install-point,exec-breeze-portion-boy-castle-gauge-marine,exec-blur-source-live-meadow-inject-damp). The last Luna P1 rollback finding is fixed inline with the atomic legacy marker and explicit rollback regressions; no further review round was requested.30b59b4c95631168a4a5fac9d7807cadd747327c.Main is merged through
d6112658e5fa, retaining #24735's cadence controls and #24733's 16-worker seal-repair rollback. The lane base already includes #24712/#24716. No.githubedits; unrelated Flutter work is preserved.Not measured: production stock/dwell, vendor spend, live nightly execution, live exporter/alert delivery, deployed revision or user-visible recovery. No GCP tier, production access, merge to main or deployment was used.
Product invariant: INV-MEM-4; canonical memory intake/promotion authority is unchanged.
Failure-Class: FC-batch-fault-domain-head-of-line-poison