agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: use git-backed workspace diffs for memory consolidation (#18982)

Browse Source 
## Why

This PR make the `morpheus` agent (memory phase 2) use a git diff to
start it's consolidation. The workflow is the following:
1. The agent acquire a lock
2. If `.codex/memories` does not exist or is not a git root, initialize
everything (and make a first empty commit)
3. Update `raw_memories.md` and `rollout_summaries/` as before.
Basically we select max N phase 1 memories based on a given policy
4. We use git (`gix`) to get a diff between the current state of
`.codex/memories` and the last commit.
5. Dump the diff in `phase2_workspace_diff.md`
6. Spawn `morpheus` and point it to `phase2_workspace_diff.md`
7. Wait for `morpheus` to be done
8. Re-create a new `.git` and make one single commit on it. We do this
because we don't want to preserve history through `.git` and this is
cheap anyway
9. We release the lock
On top of this, we keep the retry policies etc etc

The goals of this new workflow are:
* Better support of any memory extensions such as `chronicle`
* Allow the user to manually edit memories and this will be considered
by the phase 2 agent
 
As a follow-up we will need to add support for user's edition while
`morpheus` is running

## What Changed

- Added memory workspace helpers that prepare the git baseline, compute
the diff, write `phase2_workspace_diff.md`, and reset the baseline after
successful consolidation.
- Updated Phase 2 to sync current inputs into `raw_memories.md` and
`rollout_summaries/`, prune old extension resources, skip clean
workspaces, and run the consolidation subagent only when the workspace
has changes.
- Tightened Phase 2 job ownership around long-running consolidation with
heartbeats and an ownership check before resetting the baseline.
- Simplified the prompt and state APIs so DB watermarks are bookkeeping,
while workspace dirtiness decides whether consolidation work exists.
- Updated the memory pipeline README and tests for workspace diffs,
extension-resource cleanup, pollution-driven forgetting, selection
ranking, and baseline persistence.

## Verification

- Added/updated coverage in `core/src/memories/tests.rs`,
`core/src/memories/workspace_tests.rs`, `state/src/runtime/memories.rs`,
and `core/tests/suite/memories.rs`.

---------

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
jif-oai
2026-04-27 14:32:44 +02:00
committed by GitHub
co-authored by Codex
parent f8c527e529
commit 01ab25dbb5
21 changed files with 1079 additions and 1058 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/state/src/runtime/memories.rs
+145 -196
View File
@@ -3,7 +3,6 @@ use super::threads::push_thread_filters;
use super::threads::push_thread_order_and_limit;
use super::*;
use crate::SortDirection;
use crate::model::Phase2InputSelection;
use crate::model::Phase2JobClaimOutcome;
use crate::model::Stage1JobClaim;
use crate::model::Stage1JobClaimOutcome;
@@ -11,12 +10,10 @@ use crate::model::Stage1Output;
use crate::model::Stage1OutputRow;
use crate::model::Stage1StartupClaimParams;
use crate::model::ThreadRow;
use crate::model::stage1_output_ref_from_parts;
use chrono::Duration;
use sqlx::Executor;
use sqlx::QueryBuilder;
use sqlx::Sqlite;
use std::collections::HashSet;
use uuid::Uuid;
const JOB_KIND_MEMORY_STAGE1: &str = "memory_stage1";
@@ -331,8 +328,7 @@ WHERE thread_id IN (
Ok(rows_affected as usize)
}
/// Returns the current phase-2 input set along with its diff against the
/// last successful phase-2 selection.
/// Returns the current phase-2 input set.
///
/// Query behavior:
/// - current selection keeps only non-empty stage-1 outputs whose
@@ -342,22 +338,17 @@ WHERE thread_id IN (
/// - eligible rows are ordered by `usage_count DESC`,
/// `COALESCE(last_usage, source_updated_at) DESC`, `source_updated_at DESC`,
/// `thread_id DESC`
/// - previously selected rows are identified by `selected_for_phase2 = 1`
/// - `previous_selected` contains the current persisted rows that belonged
/// to the last successful phase-2 baseline, even if those threads are no
/// longer memory-eligible
/// - `retained_thread_ids` records which current rows still match the exact
/// snapshot selected in the last successful phase-2 run
/// - removed rows are previously selected rows that are still present in
/// `stage1_outputs` but are no longer in the current selection, including
/// threads that are no longer memory-eligible
///
/// The returned rows are the complete Phase 2 filesystem input. Phase 2
/// syncs these rows directly; deletions are represented by the workspace
/// diff against the previous successful memory baseline.
pub async fn get_phase2_input_selection(
&self,
n: usize,
max_unused_days: i64,
) -> anyhow::Result<Phase2InputSelection> {
) -> anyhow::Result<Vec<Stage1Output>> {
if n == 0 {
return Ok(Phase2InputSelection::default());
return Ok(Vec::new());
}
let cutoff = (Utc::now() - Duration::days(max_unused_days.max(0))).timestamp();
@@ -372,9 +363,7 @@ SELECT
so.rollout_slug,
so.generated_at,
COALESCE(t.cwd, '') AS cwd,
t.git_branch AS git_branch,
so.selected_for_phase2,
so.selected_for_phase2_source_updated_at
t.git_branch AS git_branch
FROM stage1_outputs AS so
LEFT JOIN threads AS t
ON t.id = so.thread_id
@@ -398,70 +387,14 @@ LIMIT ?
.fetch_all(self.pool.as_ref())
.await?;
let mut current_thread_ids = HashSet::with_capacity(current_rows.len());
let mut selected = Vec::with_capacity(current_rows.len());
let mut retained_thread_ids = Vec::new();
for row in current_rows {
let thread_id = row.try_get::<String, _>("thread_id")?;
current_thread_ids.insert(thread_id.clone());
let source_updated_at = row.try_get::<i64, _>("source_updated_at")?;
if row.try_get::<i64, _>("selected_for_phase2")? != 0
&& row.try_get::<Option<i64>, _>("selected_for_phase2_source_updated_at")?
== Some(source_updated_at)
{
retained_thread_ids.push(ThreadId::try_from(thread_id.clone())?);
}
selected.push(Stage1Output::try_from(Stage1OutputRow::try_from_row(
&row,
)?)?);
}
let previous_rows = sqlx::query(
r#"
SELECT
so.thread_id,
COALESCE(t.rollout_path, '') AS rollout_path,
so.source_updated_at,
so.raw_memory,
so.rollout_summary,
so.rollout_slug,
so.generated_at,
COALESCE(t.cwd, '') AS cwd,
t.git_branch AS git_branch
FROM stage1_outputs AS so
LEFT JOIN threads AS t
ON t.id = so.thread_id
WHERE so.selected_for_phase2 = 1
ORDER BY so.source_updated_at DESC, so.thread_id DESC
"#,
)
.fetch_all(self.pool.as_ref())
.await?;
let previous_selected = previous_rows
.iter()
.map(Stage1OutputRow::try_from_row)
.map(|row| row.and_then(Stage1Output::try_from))
.collect::<Result<Vec<_>, _>>()?;
let mut removed = Vec::new();
for row in previous_rows {
let thread_id = row.try_get::<String, _>("thread_id")?;
if current_thread_ids.contains(thread_id.as_str()) {
continue;
}
removed.push(stage1_output_ref_from_parts(
thread_id,
row.try_get("source_updated_at")?,
row.try_get("rollout_slug")?,
)?);
}
Ok(Phase2InputSelection {
selected,
previous_selected,
retained_thread_ids,
removed,
})
Ok(selected)
}
/// Marks a thread as polluted and enqueues phase-2 forgetting when the
@@ -909,19 +842,22 @@ WHERE kind = ? AND job_key = ?
/// Enqueues or advances the global phase-2 consolidation job watermark.
///
/// The underlying upsert keeps the job `running` when already running, resets
/// `pending/error` jobs to `pending`, and advances `input_watermark` so each
/// enqueue marks new consolidation work even when `source_updated_at` is
/// older than prior maxima.
/// `pending/error` jobs to `pending`, and advances `input_watermark` as
/// bookkeeping even when `source_updated_at` is older than prior maxima.
/// Phase 2 does not use this watermark as a dirty check; git workspace diffing
/// decides whether consolidation work exists after the lock is claimed.
pub async fn enqueue_global_consolidation(&self, input_watermark: i64) -> anyhow::Result<()> {
enqueue_global_consolidation_with_executor(self.pool.as_ref(), input_watermark).await
}
/// Attempts to claim the global phase-2 consolidation job.
/// Attempts to claim the global phase-2 consolidation lock.
///
/// Claim semantics:
/// - reads the singleton global job row (`kind='memory_consolidate_global'`)
/// - returns `SkippedNotDirty` when `input_watermark <= last_success_watermark`
/// - returns `SkippedNotDirty` when retries are exhausted or retry backoff is active
/// - 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 `SkippedRunning` when an active running lease exists
/// - otherwise updates the row to `running`, sets ownership + lease, and
/// returns `Claimed`
@@ -939,7 +875,7 @@ WHERE kind = ? AND job_key = ?
let existing_job = sqlx::query(
r#"
SELECT status, lease_until, retry_at, retry_remaining, input_watermark, last_success_watermark
SELECT status, lease_until, retry_at, retry_remaining, input_watermark
FROM jobs
WHERE kind = ? AND job_key = ?
"#,
@@ -950,18 +886,49 @@ WHERE kind = ? AND job_key = ?
.await?;
let Some(existing_job) = existing_job else {
let rows_affected = sqlx::query(
r#"
INSERT INTO jobs (
kind,
job_key,
status,
worker_id,
ownership_token,
started_at,
finished_at,
lease_until,
retry_at,
retry_remaining,
last_error,
input_watermark,
last_success_watermark
) VALUES (?, ?, 'running', ?, ?, ?, NULL, ?, NULL, ?, NULL, 0, 0)
"#,
)
.bind(JOB_KIND_MEMORY_CONSOLIDATE_GLOBAL)
.bind(MEMORY_CONSOLIDATION_JOB_KEY)
.bind(worker_id.as_str())
.bind(ownership_token.as_str())
.bind(now)
.bind(lease_until)
.bind(DEFAULT_RETRY_REMAINING)
.execute(&mut *tx)
.await?
.rows_affected();
tx.commit().await?;
return Ok(Phase2JobClaimOutcome::SkippedNotDirty);
return if rows_affected == 0 {
Ok(Phase2JobClaimOutcome::SkippedRunning)
} else {
Ok(Phase2JobClaimOutcome::Claimed {
ownership_token,
input_watermark: 0,
})
};
};
let input_watermark: Option<i64> = existing_job.try_get("input_watermark")?;
let input_watermark_value = input_watermark.unwrap_or(0);
let last_success_watermark: Option<i64> = existing_job.try_get("last_success_watermark")?;
if input_watermark_value <= last_success_watermark.unwrap_or(0) {
tx.commit().await?;
return Ok(Phase2JobClaimOutcome::SkippedNotDirty);
}
let status: String = existing_job.try_get("status")?;
let existing_lease_until: Option<i64> = existing_job.try_get("lease_until")?;
let retry_at: Option<i64> = existing_job.try_get("retry_at")?;
@@ -969,11 +936,11 @@ WHERE kind = ? AND job_key = ?
if retry_remaining <= 0 {
tx.commit().await?;
return Ok(Phase2JobClaimOutcome::SkippedNotDirty);
return Ok(Phase2JobClaimOutcome::SkippedRetryUnavailable);
}
if retry_at.is_some_and(|retry_at| retry_at > now) {
tx.commit().await?;
return Ok(Phase2JobClaimOutcome::SkippedNotDirty);
return Ok(Phase2JobClaimOutcome::SkippedRetryUnavailable);
}
if status == "running" && existing_lease_until.is_some_and(|lease_until| lease_until > now)
{
@@ -994,7 +961,6 @@ SET
retry_at = NULL,
last_error = NULL
WHERE kind = ? AND job_key = ?
AND input_watermark > COALESCE(last_success_watermark, 0)
AND (status != 'running' OR lease_until IS NULL OR lease_until <= ?)
AND (retry_at IS NULL OR retry_at <= ?)
AND retry_remaining > 0
@@ -1063,8 +1029,7 @@ WHERE kind = ? AND job_key = ?
/// `max(existing_last_success_watermark, completed_watermark)`
/// - rewrites `selected_for_phase2` so only the exact selected stage-1
/// snapshots remain marked as part of the latest successful phase-2
/// selection, and persists each selected snapshot's
/// `source_updated_at` for future retained-vs-added diffing
/// selection, and persists each selected snapshot's `source_updated_at`
pub async fn mark_global_phase2_job_succeeded(
&self,
ownership_token: &str,
@@ -2271,16 +2236,6 @@ WHERE kind = 'memory_stage1'
"no-output without an existing stage1 output should not enqueue phase2"
);
let claim_phase2 = runtime
.try_claim_global_phase2_job(owner, /*lease_seconds*/ 3600)
.await
.expect("claim phase2");
assert_eq!(
claim_phase2,
Phase2JobClaimOutcome::SkippedNotDirty,
"phase2 should remain clean when no-output deleted nothing"
);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
@@ -2350,7 +2305,7 @@ WHERE kind = 'memory_stage1'
)
.await
.expect("mark initial phase2 succeeded"),
"initial phase2 success should clear global dirty state"
"initial phase2 success should finalize the global job"
);
let no_output_claim = runtime
@@ -2505,7 +2460,7 @@ WHERE kind = 'memory_stage1'
}
#[tokio::test]
async fn phase2_global_consolidation_reruns_when_watermark_advances() {
async fn phase2_global_lock_can_be_reclaimed_after_success_without_new_watermark() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -2537,24 +2492,13 @@ WHERE kind = 'memory_stage1'
"phase2 success should finalize for current token"
);
let claim_up_to_date = runtime
let claim_after_success = runtime
.try_claim_global_phase2_job(owner, /*lease_seconds*/ 3600)
.await
.expect("claim phase2 up-to-date");
assert_eq!(claim_up_to_date, Phase2JobClaimOutcome::SkippedNotDirty);
runtime
.enqueue_global_consolidation(/*input_watermark*/ 101)
.await
.expect("enqueue global consolidation again");
let claim_rerun = runtime
.try_claim_global_phase2_job(owner, /*lease_seconds*/ 3600)
.await
.expect("claim phase2 rerun");
.expect("claim phase2 after success");
assert!(
matches!(claim_rerun, Phase2JobClaimOutcome::Claimed { .. }),
"advanced watermark should be claimable"
matches!(claim_after_success, Phase2JobClaimOutcome::Claimed { .. }),
"the DB claim is only a lock; git workspace diff decides whether there is work"
);
let _ = tokio::fs::remove_dir_all(codex_home).await;
@@ -2801,7 +2745,7 @@ VALUES (?, ?, ?, ?, ?)
}
#[tokio::test]
async fn get_phase2_input_selection_reports_added_retained_and_removed_rows() {
async fn get_phase2_input_selection_returns_current_selected_rows() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -2895,28 +2839,19 @@ VALUES (?, ?, ?, ?, ?)
.await
.expect("load phase2 input selection");
assert_eq!(selection.selected.len(), 2);
assert_eq!(selection.previous_selected.len(), 2);
assert_eq!(selection.selected[0].thread_id, thread_id_c);
assert_eq!(selection.len(), 2);
assert_eq!(selection[0].thread_id, thread_id_c);
assert_eq!(
selection.selected[0].rollout_path,
selection[0].rollout_path,
codex_home.join(format!("rollout-{thread_id_c}.jsonl"))
);
assert_eq!(selection.selected[1].thread_id, thread_id_b);
assert_eq!(selection.retained_thread_ids, vec![thread_id_c]);
assert_eq!(selection.removed.len(), 1);
assert_eq!(selection.removed[0].thread_id, thread_id_a);
assert_eq!(
selection.removed[0].rollout_slug.as_deref(),
Some("rollout-a")
);
assert_eq!(selection[1].thread_id, thread_id_b);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
#[tokio::test]
async fn get_phase2_input_selection_marks_polluted_previous_selection_as_removed() {
async fn get_phase2_input_selection_excludes_polluted_previous_selection() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -3002,24 +2937,8 @@ VALUES (?, ?, ?, ?, ?)
.await
.expect("load phase2 input selection");
assert_eq!(selection.selected.len(), 1);
assert_eq!(selection.selected[0].thread_id, thread_id_enabled);
assert_eq!(selection.previous_selected.len(), 2);
assert!(
selection
.previous_selected
.iter()
.any(|item| item.thread_id == thread_id_enabled)
);
assert!(
selection
.previous_selected
.iter()
.any(|item| item.thread_id == thread_id_polluted)
);
assert_eq!(selection.retained_thread_ids, vec![thread_id_enabled]);
assert_eq!(selection.removed.len(), 1);
assert_eq!(selection.removed[0].thread_id, thread_id_polluted);
assert_eq!(selection.len(), 1);
assert_eq!(selection[0].thread_id, thread_id_enabled);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
@@ -3113,7 +3032,7 @@ VALUES (?, ?, ?, ?, ?)
}
#[tokio::test]
async fn get_phase2_input_selection_treats_regenerated_selected_rows_as_added() {
async fn get_phase2_input_selection_returns_regenerated_selected_rows() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -3213,12 +3132,9 @@ VALUES (?, ?, ?, ?, ?)
.get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 36_500)
.await
.expect("load phase2 input selection");
assert_eq!(selection.selected.len(), 1);
assert_eq!(selection.previous_selected.len(), 1);
assert_eq!(selection.selected[0].thread_id, thread_id);
assert_eq!(selection.selected[0].source_updated_at.timestamp(), 101);
assert!(selection.retained_thread_ids.is_empty());
assert!(selection.removed.is_empty());
assert_eq!(selection.len(), 1);
assert_eq!(selection[0].thread_id, thread_id);
assert_eq!(selection[0].source_updated_at.timestamp(), 101);
let (selected_for_phase2, selected_for_phase2_source_updated_at) =
sqlx::query_as::<_, (i64, Option<i64>)>(
@@ -3235,7 +3151,7 @@ VALUES (?, ?, ?, ?, ?)
}
#[tokio::test]
async fn get_phase2_input_selection_reports_regenerated_previous_selection_as_removed() {
async fn get_phase2_input_selection_uses_current_ranking_after_refreshes() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -3368,29 +3284,11 @@ VALUES (?, ?, ?, ?, ?)
.expect("load phase2 input selection");
assert_eq!(
selection
.selected
.iter()
.map(|output| output.thread_id)
.collect::<Vec<_>>(),
vec![thread_id_d, thread_id_c]
);
assert_eq!(
selection
.previous_selected
.iter()
.map(|output| output.thread_id)
.collect::<Vec<_>>(),
vec![thread_id_a, thread_id_b]
);
assert!(selection.retained_thread_ids.is_empty());
assert_eq!(
selection
.removed
.iter()
.map(|output| (output.thread_id, output.source_updated_at.timestamp()))
.collect::<Vec<_>>(),
vec![(thread_id_a, 102), (thread_id_b, 101)]
);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
@@ -3527,7 +3425,8 @@ VALUES (?, ?, ?, ?, ?)
.get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 36_500)
.await
.expect("load phase2 input selection after refresh");
assert_eq!(selection.retained_thread_ids, vec![thread_id]);
assert_eq!(selection.len(), 1);
assert_eq!(selection[0].thread_id, thread_id);
let (selected_for_phase2, selected_for_phase2_source_updated_at) =
sqlx::query_as::<_, (i64, Option<i64>)>(
@@ -3657,9 +3556,8 @@ VALUES (?, ?, ?, ?, ?)
.get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 36_500)
.await
.expect("load phase2 input selection");
assert_eq!(selection.selected.len(), 1);
assert_eq!(selection.selected[0].source_updated_at.timestamp(), 101);
assert!(selection.retained_thread_ids.is_empty());
assert_eq!(selection.len(), 1);
assert_eq!(selection[0].source_updated_at.timestamp(), 101);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
@@ -3870,7 +3768,6 @@ VALUES (?, ?, ?, ?, ?)
assert_eq!(
selection
.selected
.iter()
.map(|output| output.thread_id)
.collect::<Vec<_>>(),
@@ -3967,7 +3864,6 @@ VALUES (?, ?, ?, ?, ?)
assert_eq!(
selection
.selected
.iter()
.map(|output| output.thread_id)
.collect::<Vec<_>>(),
@@ -4056,9 +3952,9 @@ VALUES (?, ?, ?, ?, ?)
.await
.expect("load phase2 input selection");
assert_eq!(selection.selected.len(), 1);
assert_eq!(selection.selected[0].thread_id, newer_thread);
assert_eq!(selection.selected[0].source_updated_at.timestamp(), 200);
assert_eq!(selection.len(), 1);
assert_eq!(selection[0].thread_id, newer_thread);
assert_eq!(selection[0].source_updated_at.timestamp(), 200);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
@@ -4411,6 +4307,59 @@ VALUES (?, ?, ?, ?, ?)
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
#[tokio::test]
async fn phase2_global_lock_creates_missing_job_row() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
.expect("initialize runtime");
let owner_a = ThreadId::from_string(&Uuid::new_v4().to_string()).expect("owner a");
let owner_b = ThreadId::from_string(&Uuid::new_v4().to_string()).expect("owner b");
let claim = runtime
.try_claim_global_phase2_job(owner_a, /*lease_seconds*/ 3_600)
.await
.expect("claim global phase2 lock");
let ownership_token = match claim {
Phase2JobClaimOutcome::Claimed {
ownership_token,
input_watermark,
} => {
assert_eq!(input_watermark, 0);
ownership_token
}
other => panic!("unexpected phase2 lock claim outcome: {other:?}"),
};
let second_claim = runtime
.try_claim_global_phase2_job(owner_b, /*lease_seconds*/ 3_600)
.await
.expect("claim global phase2 lock from second owner");
assert_eq!(second_claim, Phase2JobClaimOutcome::SkippedRunning);
assert!(
runtime
.mark_global_phase2_job_succeeded(
ownership_token.as_str(),
/*completed_watermark*/ 0,
&[]
)
.await
.expect("mark phase2 lock success")
);
let claim_after_success = runtime
.try_claim_global_phase2_job(owner_b, /*lease_seconds*/ 3_600)
.await
.expect("claim global phase2 lock after success");
assert!(
matches!(claim_after_success, Phase2JobClaimOutcome::Claimed { .. }),
"git workspace diff, not the DB watermark, decides whether the claimed lock has work"
);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
#[tokio::test]
async fn phase2_global_lock_stale_lease_allows_takeover() {
let codex_home = unique_temp_dir();
@@ -4487,7 +4436,7 @@ VALUES (?, ?, ?, ?, ?)
}
#[tokio::test]
async fn phase2_backfilled_inputs_below_last_success_still_become_dirty() {
async fn enqueue_global_consolidation_keeps_phase2_input_watermark_monotonic() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
@@ -4527,23 +4476,23 @@ VALUES (?, ?, ?, ?, ?)
runtime
.enqueue_global_consolidation(/*input_watermark*/ 400)
.await
.expect("enqueue backfilled consolidation");
.expect("enqueue lower-watermark consolidation");
let owner_b = ThreadId::from_string(&Uuid::new_v4().to_string()).expect("owner b");
let claim_b = runtime
.try_claim_global_phase2_job(owner_b, /*lease_seconds*/ 3_600)
.await
.expect("claim backfilled consolidation");
.expect("claim lower-watermark consolidation");
match claim_b {
Phase2JobClaimOutcome::Claimed {
input_watermark, ..
} => {
assert!(
input_watermark > 500,
"backfilled enqueue should advance dirty watermark beyond last success"
"lower-watermark enqueue should still advance the bookkeeping watermark"
);
}
other => panic!("unexpected backfilled phase2 claim outcome: {other:?}"),
other => panic!("unexpected lower-watermark phase2 claim outcome: {other:?}"),
}
let _ = tokio::fs::remove_dir_all(codex_home).await;
@@ -4608,7 +4557,7 @@ VALUES (?, ?, ?, ?, ?)
.try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
.await
.expect("claim after fallback failure");
assert_eq!(claim, Phase2JobClaimOutcome::SkippedNotDirty);
assert_eq!(claim, Phase2JobClaimOutcome::SkippedRetryUnavailable);
let _ = tokio::fs::remove_dir_all(codex_home).await;
}