Repository navigation
fix(pipeline): age-gate ancient provider and fresh-enrichment sweep wakes - #24714
Conversation
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. |
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
2 issues found across 9 files
Confidence score: 4/5
- The repaired-destination branch in
test_sweep_age_gate.pyis still untested, so the age-gate contract could regress there unnoticed. Add a separate case for that branch. - The 15-day fixture in
test_provider_stt_outbox.pycan depend on a provider-specific age override, making the test non-hermetic. Clear the override so it uses the 14-day default.
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/tests/unit/test_provider_stt_outbox.py">
<violation number="1" location="backend/tests/unit/test_provider_stt_outbox.py:253">
P3: Clear the provider-specific age override in this test so its 15-day fixture consistently uses the 14-day default and remains hermetic.</violation>
</file>
<file name="backend/tests/unit/test_sweep_age_gate.py">
<violation number="1" location="backend/tests/unit/test_sweep_age_gate.py:45">
P2: This test never covers the repaired-destination branch promised by the age-gate contract. Configure an enrichment-queue binding and an invalid/non-fresh stored binding in a separate case, then assert an ancient row is still skipped after the sweep repairs it to `enrichment-fresh`; otherwise a regression in the post-repair age check can pass this suite.</violation>
</file>
Reply with feedback, questions, or to request a fix.
View guided diff | Turn on auto-fix | Re-trigger cubic
| monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh') | ||
| outbox.mark('u', 'c', 'en', now=now - age * 86400) | ||
| row_id = outbox.key('u', 'c') | ||
| rows[row_id]['dispatch']['queue'] = 'enrichment-fresh' |
There was a problem hiding this comment.
P2: This test never covers the repaired-destination branch promised by the age-gate contract. Configure an enrichment-queue binding and an invalid/non-fresh stored binding in a separate case, then assert an ancient row is still skipped after the sweep repairs it to enrichment-fresh; otherwise a regression in the post-repair age check can pass this suite.
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_sweep_age_gate.py, line 45:
<comment>This test never covers the repaired-destination branch promised by the age-gate contract. Configure an enrichment-queue binding and an invalid/non-fresh stored binding in a separate case, then assert an ancient row is still skipped after the sweep repairs it to `enrichment-fresh`; otherwise a regression in the post-repair age check can pass this suite.</comment>
<file context>
@@ -0,0 +1,139 @@
+ monkeypatch.setenv('ENRICHMENT_FRESH_TASKS_QUEUE', 'enrichment-fresh')
+ outbox.mark('u', 'c', 'en', now=now - age * 86400)
+ row_id = outbox.key('u', 'c')
+ rows[row_id]['dispatch']['queue'] = 'enrichment-fresh'
+ original = deepcopy(rows)
+ tasks.clear()
</file context>
| def test_ancient_sweep_keeps_existing_privacy_discard(box, monkeypatch): | ||
| from utils.observability.sweep_age import SKIPPED | ||
|
|
||
| monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') |
There was a problem hiding this comment.
P3: Clear the provider-specific age override in this test so its 15-day fixture consistently uses the 14-day default and remains hermetic.
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_provider_stt_outbox.py, line 253:
<comment>Clear the provider-specific age override in this test so its 15-day fixture consistently uses the 14-day default and remains hermetic.</comment>
<file context>
@@ -220,3 +220,43 @@ def test_wake_task_shape(monkeypatch):
+def test_ancient_sweep_keeps_existing_privacy_discard(box, monkeypatch):
+ from utils.observability.sweep_age import SKIPPED
+
+ monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on')
+ enqueue(box, now=100)
+ box.blocked.add(ROW)
</file context>
| monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') | |
| monkeypatch.setenv('BATCH_PIPELINE_SWEEP_AGE_GATE', 'on') | |
| monkeypatch.delenv('BATCH_PIPELINE_PROVIDER_STT_SWEEP_MAX_AGE_DAYS', raising=False) |
![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: ece6d2cb9d
ℹ️ 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".
| if skip_ancient(row, 'provider-stt', now=now): | ||
| continue |
There was a problem hiding this comment.
When the first 100 open provider rows are older than the configured age limit, this new skip leaves them permanently occupying the entire sweep page. The query above filters only on state == OPEN, applies limit(100) by default, and has no pagination or persisted cursor, so every maintenance pass reads those same skipped rows and never reaches younger obligations behind them. Consequently, fresh provider work whose initial wake was lost can remain unprocessed indefinitely; backend/routers/enrichment.py::maintain invokes this sweep with the default limit. Add bounded scan progress or otherwise exclude ancient rows from the limited candidate page while preserving their durable state.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 4235431350
ℹ️ 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".
| fields = ('created_at',) if queue == 'provider-stt' else ('first_pending_at', 'latest_pending_at') | ||
| timestamps = [] | ||
| for field in fields: | ||
| if field == 'latest_pending_at' and field not in row: | ||
| continue # Existing rows retain their original acceptance clock until a new mark. |
There was a problem hiding this comment.
When historical parking is disabled (ENRICHMENT_MAX_AGE_HOURS=0), redirecting a recent obligation into an ancient canonical fresh-queue row can make the recent work permanently ineligible for sweep recovery. custody.transfer_redirect copies the canonical row and takes the minimum first_pending_at, but never merges the donor's new latest_pending_at. This age check therefore sees only the canonical row's ancient timestamps, even though the donor contains newly accepted work. If the post-transfer wake fails or is exhausted, subsequent sweeps skip the combined obligation indefinitely. Merge the donor and canonical freshness timestamps during custody transfer, just as mark_update refreshes freshness for coalesced work.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 3ae885e3de
ℹ️ 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".
| if skip_ancient(row, 'provider-stt', now=now): | ||
| continue |
There was a problem hiding this comment.
This changes provider-STT recovery and enrichment lifecycle behavior by enabling an age gate by default, but the commit adds no pipeline change note. backend/AGENTS.md explicitly requires a note under backend/docs/listen_pusher_pipeline/changes/ for transcription or conversation-lifecycle changes. Searching the pipeline documentation for the new flag names and freshness field also found no coverage. Add the required note describing the rollout, rollback, and recovery expectations for obligations that are now skipped; the generated flag registry does not replace that operational documentation.
AGENTS.md reference: backend/AGENTS.md:L164-L164
Useful? React with 👍 / 👎.
Problem
Maintenance sweeps repeatedly re-enqueue months-old provider-STT and enrichment-fresh obligations ahead of current work. Wave 4 t6 identifies stale provider wakes and ancient enrichment retries as the queue contaminant; increasing dispatch alone does not stop that generator.
Fix
Skip sweep re-wakes strictly older than 14 days, independently configurable per queue. Provider-STT uses durable
created_at; enrichment-fresh uses the most recent obligation timestamp (latest_pending_at, withfirst_pending_atfor existing rows) and checks its stored or repaired destination. Fresh coalesced marks advance that timestamp without changing the original acceptance or deadline. Two weeks preserves a normal recovery window while excluding t6's months-old obligations. Missing, zero, negative, or otherwise unknown age metadata remains eligible. Initial enqueue and consumers are unchanged.Age-skipped rows are left byte-for-byte unchanged: no deletion, parking, token rotation or lease write. They remain durable for a separately bounded backfill. Existing privacy/deletion handling takes precedence over age skipping in both sweeps. Provider-STT advances an in-process document-ID cursor across bounded pages, including skipped rows, and wraps at exhaustion; a full ancient page cannot strand later fresh rows. The existing state single-field index suffices; no index is added. The existing independent 72-hour enrichment parking policy runs first; the new gate applies only to obligations that policy leaves open. Existing parking removes historical rows from the due scan without adding any new retirement or dead-letter behavior. Setting the new gate off restores legacy re-wake admission; the existing parking policy remains in force.
omi_batch_pipeline_sweep_skipped_total{queue,reason="max_age"}counts skipped candidates, not unique rows. Both Cloud Run GMP profiles already retain it; a test pins that coverage. Accessors live in config/policy, with budget ownership and generated registry entries.Tests
BACKEND_UNIT_TEST_FILE_LIST+bash backend/test.shviaomi-heavy: 321 passed across 7 affected files (provider outbox, sweep age contract, enrichment coalescing, enrichment route, existing enrichment age policy, batch-stage env ratchet, flag budget).Regressions cover an all-ancient provider page followed by fresh lost-wake work, cursor wrap, a fresh coalesced mark on an aged open enrichment row (including an older delayed mark), the actual enrichment adapter privacy-retirement callback preceding age skipping, existing historical parking preceding age skipping, and unknown non-positive epoch timestamps.
Final pre-push validation passed: 41 repository checks, backend typecheck (0 errors), selected unit hooks, and formatting.
Risk
Existing queued tasks continue to run; this only prevents new sweep re-wakes. Historical rows remain durable but will need a separately bounded backfill. The existing enrichment parking policy still applies to other eligible rows. No production access, merge, or deploy in this lane.
Rollback
Set
BATCH_PIPELINE_SWEEP_AGE_GATE=offto restore existing sweeps. Only explicitoffdisables the new gate; malformed ages fall back to 14 days.Failure-Class: FC-bounded-discovery-pinned-by-blocked-page