From 5d314f324c7ffc54ac24ecf464c45f9c4bcfa861 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Mon, 27 Apr 2026 14:58:11 +0200 Subject: [PATCH] Allow Phase 2 memory claims after retry exhaustion (#19809) ## Why The Phase 2 memories job row is only the global lock for the git-backed memory workspace. Manual memory edits do not enqueue new Stage 1 work, so a Phase 2 row with `retry_remaining = 0` could be skipped before the worker ever claimed the lock and generated `phase2_workspace_diff.md`. That left workspace-only changes unconsolidated after repeated failures, even when retry backoff had elapsed and the filesystem had real diffable work. ## What Changed - Allow `try_claim_global_phase2_job` to claim the Phase 2 lock after the retry budget is exhausted, while still respecting active `retry_at` backoff and fresh running leases. - Treat `SkippedRetryUnavailable` for Phase 2 as backoff-only, and update the outcome docs to match. - Clamp Phase 2 retry bookkeeping at zero when failed attempts are recorded. ## Verification - Added `phase2_global_lock_can_be_claimed_after_retry_budget_is_exhausted` to cover the exhausted-budget lock claim path. - Ran `cargo test -p codex-state`. --- codex-rs/state/src/model/memories.rs | 2 +- codex-rs/state/src/runtime/memories.rs | 87 ++++++++++++++++++++++---- 2 files changed, 76 insertions(+), 13 deletions(-) diff --git a/codex-rs/state/src/model/memories.rs b/codex-rs/state/src/model/memories.rs index 006b51a0d..ada9d4e1e 100644 --- a/codex-rs/state/src/model/memories.rs +++ b/codex-rs/state/src/model/memories.rs @@ -115,7 +115,7 @@ pub enum Phase2JobClaimOutcome { /// Snapshot of `input_watermark` at claim time. input_watermark: i64, }, - /// The global job is in retry backoff or has exhausted its retry budget. + /// The global job is in retry backoff. SkippedRetryUnavailable, /// Another worker currently owns a fresh global consolidation lease. SkippedRunning, diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index 1e051720d..62175d15a 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -857,7 +857,7 @@ WHERE kind = ? AND job_key = ? /// - creates and claims the singleton row when it does not exist yet /// - does not use DB watermarks to decide whether Phase 2 has work; git workspace /// dirtiness is the source of truth after the caller materializes inputs - /// - returns `SkippedRetryUnavailable` when retries are exhausted or retry backoff is active + /// - returns `SkippedRetryUnavailable` when retry backoff is active /// - returns `SkippedRunning` when an active running lease exists /// - otherwise updates the row to `running`, sets ownership + lease, and /// returns `Claimed` @@ -875,7 +875,7 @@ WHERE kind = ? AND job_key = ? let existing_job = sqlx::query( r#" -SELECT status, lease_until, retry_at, retry_remaining, input_watermark +SELECT status, lease_until, retry_at, input_watermark FROM jobs WHERE kind = ? AND job_key = ? "#, @@ -932,12 +932,6 @@ INSERT INTO jobs ( let status: String = existing_job.try_get("status")?; let existing_lease_until: Option = existing_job.try_get("lease_until")?; let retry_at: Option = existing_job.try_get("retry_at")?; - let retry_remaining: i64 = existing_job.try_get("retry_remaining")?; - - if retry_remaining <= 0 { - tx.commit().await?; - return Ok(Phase2JobClaimOutcome::SkippedRetryUnavailable); - } if retry_at.is_some_and(|retry_at| retry_at > now) { tx.commit().await?; return Ok(Phase2JobClaimOutcome::SkippedRetryUnavailable); @@ -963,7 +957,6 @@ SET WHERE kind = ? AND job_key = ? AND (status != 'running' OR lease_until IS NULL OR lease_until <= ?) AND (retry_at IS NULL OR retry_at <= ?) - AND retry_remaining > 0 "#, ) .bind(worker_id.as_str()) @@ -1104,7 +1097,7 @@ WHERE thread_id = ? AND source_updated_at = ? /// - updates only the owned running singleton global row /// - sets `status='error'`, clears lease /// - writes failure reason and retry time - /// - decrements `retry_remaining` + /// - decrements `retry_remaining` without going below zero pub async fn mark_global_phase2_job_failed( &self, ownership_token: &str, @@ -1121,7 +1114,7 @@ SET finished_at = ?, lease_until = NULL, retry_at = ?, - retry_remaining = retry_remaining - 1, + retry_remaining = max(retry_remaining - 1, 0), last_error = ? WHERE kind = ? AND job_key = ? AND status = 'running' AND ownership_token = ? @@ -1162,7 +1155,7 @@ SET finished_at = ?, lease_until = NULL, retry_at = ?, - retry_remaining = retry_remaining - 1, + retry_remaining = max(retry_remaining - 1, 0), last_error = ? WHERE kind = ? AND job_key = ? AND status = 'running' @@ -2504,6 +2497,76 @@ WHERE kind = 'memory_stage1' let _ = tokio::fs::remove_dir_all(codex_home).await; } + #[tokio::test] + async fn phase2_global_lock_can_be_claimed_after_retry_budget_is_exhausted() { + let codex_home = unique_temp_dir(); + let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) + .await + .expect("initialize runtime"); + + runtime + .enqueue_global_consolidation(/*input_watermark*/ 100) + .await + .expect("enqueue global consolidation"); + + let owner = ThreadId::from_string(&Uuid::new_v4().to_string()).expect("owner id"); + for attempt in 0..3 { + let claim = runtime + .try_claim_global_phase2_job(owner, /*lease_seconds*/ 3_600) + .await + .expect("claim phase2 before retry exhaustion"); + let ownership_token = match claim { + Phase2JobClaimOutcome::Claimed { + ownership_token, .. + } => ownership_token, + other => panic!( + "attempt {} should claim phase2 before retries are exhausted: {other:?}", + attempt + 1 + ), + }; + assert!( + runtime + .mark_global_phase2_job_failed( + ownership_token.as_str(), + "boom", + /*retry_delay_seconds*/ 0, + ) + .await + .expect("mark phase2 failed"), + "attempt {} should decrement retry budget", + attempt + 1 + ); + } + + let job_row = + sqlx::query("SELECT retry_remaining FROM jobs WHERE kind = ? AND job_key = ?") + .bind("memory_consolidate_global") + .bind("global") + .fetch_one(runtime.pool.as_ref()) + .await + .expect("load phase2 job row after retry exhaustion"); + assert_eq!( + job_row + .try_get::("retry_remaining") + .expect("retry_remaining"), + 0 + ); + + let claim_after_exhaustion = runtime + .try_claim_global_phase2_job(owner, /*lease_seconds*/ 3_600) + .await + .expect("claim phase2 after retry exhaustion"); + assert!( + matches!( + claim_after_exhaustion, + Phase2JobClaimOutcome::Claimed { .. } + ), + "phase2 claim should only lock; workspace diffing decides whether there is work" + ); + + let _ = tokio::fs::remove_dir_all(codex_home).await; + } + #[tokio::test] async fn list_stage1_outputs_for_global_returns_latest_outputs() { let codex_home = unique_temp_dir();