agent/codex
1
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

Moving updated-at timestamps to unique millisecond times (#17489)

Browse Source 
To allow the ability to have guaranteed-unique cursors, we make two
important updates:
* Add new updated_at_ms and created_at_ms columns that are in
millisecond precision
* Guarantee uniqueness -- if multiple items are inserted at the same
millisecond, bump the new one by one millisecond until it becomes unique

This lets us use single-number cursors for forwards and backwards paging
through resultsets and guarantee that the cursor is a fixed point to do
(timestamp > cursor) and get new items only.

This updated implementation is backwards-compatible since multiple
appservers can be running and won't handle the previous method well.
This commit is contained in:
David de Regt
2026-04-14 11:55:34 -04:00
committed by GitHub
Unverified
parent 61fe23159e
commit 4f2fc3e3fa
13 changed files with 470 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/src/codex_message_processor.rs
+18 -2
View File
@@ -9703,7 +9703,7 @@ async fn read_updated_at(path: &Path, created_at: Option<&str>) -> Option<String
.and_then(|meta| meta.modified().ok())
.map(|modified| {
let updated_at: DateTime<Utc> = modified.into();
updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)
updated_at.to_rfc3339_opts(SecondsFormat::Millis, true)
});
updated_at.or_else(|| created_at.map(str::to_string))
}
@@ -9954,6 +9954,22 @@ mod tests {
Ok(metadata)
}
#[test]
fn summary_from_thread_metadata_formats_protocol_timestamps_as_seconds() -> Result<()> {
let mut metadata =
test_thread_metadata(/*model*/ None, /*reasoning_effort*/ None)?;
metadata.created_at =
DateTime::parse_from_rfc3339("2025-09-05T16:53:11.123Z")?.with_timezone(&Utc);
metadata.updated_at =
DateTime::parse_from_rfc3339("2025-09-05T16:53:12.456Z")?.with_timezone(&Utc);
let summary = summary_from_thread_metadata(&metadata);
assert_eq!(summary.timestamp, Some("2025-09-05T16:53:11Z".to_string()));
assert_eq!(summary.updated_at, Some("2025-09-05T16:53:12Z".to_string()));
Ok(())
}
#[test]
fn merge_persisted_resume_metadata_prefers_persisted_model_and_reasoning_effort() -> Result<()>
{
@@ -10213,7 +10229,7 @@ mod tests {
let expected = ConversationSummary {
conversation_id,
timestamp: Some(timestamp.clone()),
updated_at: Some("2025-09-05T16:53:11Z".to_string()),
updated_at: Some(timestamp),
path: path.clone(),
preview: String::new(),
model_provider: "fallback".to_string(),
codex-rs/app-server/tests/suite/conversation_summary.rs
+2 -1
View File
@@ -18,6 +18,7 @@ use tokio::time::timeout;
const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10);
const FILENAME_TS: &str = "2025-01-02T12-00-00";
const META_RFC3339: &str = "2025-01-02T12:00:00Z";
const UPDATED_AT_RFC3339: &str = "2025-01-02T12:00:00.000Z";
const PREVIEW: &str = "Summarize this conversation";
const MODEL_PROVIDER: &str = "openai";
@@ -27,7 +28,7 @@ fn expected_summary(conversation_id: ThreadId, path: PathBuf) -> ConversationSum
path,
preview: PREVIEW.to_string(),
timestamp: Some(META_RFC3339.to_string()),
updated_at: Some(META_RFC3339.to_string()),
updated_at: Some(UPDATED_AT_RFC3339.to_string()),
model_provider: MODEL_PROVIDER.to_string(),
cwd: PathBuf::from("/"),
cli_version: "0.0.0".to_string(),
codex-rs/rollout/src/list.rs
+8 -7
View File
@@ -71,7 +71,6 @@ pub struct ThreadItem {
/// created_at comes from the filename timestamp with second precision.
pub created_at: Option<String>,
/// RFC3339 timestamp string for the most recent update (from file mtime).
/// updated_at is truncated to second precision to match created_at.
pub updated_at: Option<String>,
}
@@ -292,7 +291,10 @@ impl<'de> serde::Deserialize<'de> for Cursor {
impl From<codex_state::Anchor> for Cursor {
fn from(anchor: codex_state::Anchor) -> Self {
let ts = OffsetDateTime::from_unix_timestamp(anchor.ts.timestamp())
let ts = anchor
.ts
.timestamp_nanos_opt()
.and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(nanos as i128).ok())
.unwrap_or(OffsetDateTime::UNIX_EPOCH);
Self::new(ts, anchor.id)
}
@@ -1156,17 +1158,16 @@ async fn file_modified_time(path: &Path) -> io::Result<Option<OffsetDateTime>> {
return Ok(None);
};
let dt = OffsetDateTime::from(modified);
// Truncate to seconds so ordering and cursor comparisons align with the
// cursor timestamp format (which exposes seconds), keeping pagination stable.
Ok(truncate_to_seconds(dt))
Ok(truncate_to_millis(dt))
}
fn format_rfc3339(dt: OffsetDateTime) -> Option<String> {
dt.format(&Rfc3339).ok()
}
fn truncate_to_seconds(dt: OffsetDateTime) -> Option<OffsetDateTime> {
dt.replace_nanosecond(0).ok()
fn truncate_to_millis(dt: OffsetDateTime) -> Option<OffsetDateTime> {
let millis_nanos = (dt.nanosecond() / 1_000_000) * 1_000_000;
dt.replace_nanosecond(millis_nanos).ok()
}
async fn find_thread_path_by_id_str_in_subdir(
codex-rs/rollout/src/metadata.rs
+2 -2
View File
@@ -371,7 +371,7 @@ fn backfill_watermark_for_path(codex_home: &Path, path: &Path) -> String {
async fn file_modified_time_utc(path: &Path) -> Option<DateTime<Utc>> {
let modified = tokio::fs::metadata(path).await.ok()?.modified().ok()?;
let updated_at: DateTime<Utc> = modified.into();
updated_at.with_nanosecond(0)
Some(updated_at)
}
fn parse_timestamp_to_utc(ts: &str) -> Option<DateTime<Utc>> {
@@ -381,7 +381,7 @@ fn parse_timestamp_to_utc(ts: &str) -> Option<DateTime<Utc>> {
return dt.with_nanosecond(0);
}
if let Ok(dt) = DateTime::parse_from_rfc3339(ts) {
return dt.with_timezone(&Utc).with_nanosecond(0);
return Some(dt.with_timezone(&Utc));
}
None
}
codex-rs/rollout/src/recorder.rs
+1 -1
View File
@@ -1252,7 +1252,7 @@ impl From<codex_state::ThreadsPage> for ThreadsPage {
model_provider: Some(item.model_provider),
cli_version: Some(item.cli_version),
created_at: Some(item.created_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
updated_at: Some(item.updated_at.to_rfc3339_opts(SecondsFormat::Secs, true)),
updated_at: Some(item.updated_at.to_rfc3339_opts(SecondsFormat::Millis, true)),
})
.collect();
Self {
codex-rs/rollout/src/state_db.rs
+1 -3
View File
@@ -5,7 +5,6 @@ use crate::list::ThreadSortKey;
use crate::metadata;
use chrono::DateTime;
use chrono::NaiveDateTime;
use chrono::Timelike;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::dynamic_tools::DynamicToolSpec;
@@ -131,8 +130,7 @@ fn cursor_to_anchor(cursor: Option<&Cursor>) -> Option<codex_state::Anchor> {
dt.with_timezone(&Utc)
} else {
return None;
}
.with_nanosecond(0)?;
};
Some(codex_state::Anchor { ts, id })
}
codex-rs/state/migrations/0025_thread_timestamps_millis.sql
+112
View File
@@ -0,0 +1,112 @@
ALTER TABLE threads ADD COLUMN created_at_ms INTEGER;
ALTER TABLE threads ADD COLUMN updated_at_ms INTEGER;
CREATE TEMP TABLE thread_timestamp_migration AS
SELECT
id,
CASE
WHEN created_at < 1577836800000 THEN created_at * 1000
ELSE created_at
END AS created_at_base_ms,
CASE
WHEN updated_at < 1577836800000 THEN updated_at * 1000
ELSE updated_at
END AS updated_at_base_ms
FROM threads;
WITH RECURSIVE
ordered_created AS (
SELECT
id,
created_at_base_ms,
ROW_NUMBER() OVER (ORDER BY created_at_base_ms, id) AS row_number
FROM thread_timestamp_migration
),
assigned_created(row_number, id, created_at_ms) AS (
SELECT row_number, id, created_at_base_ms
FROM ordered_created
WHERE row_number = 1
UNION ALL
SELECT
ordered_created.row_number,
ordered_created.id,
MAX(ordered_created.created_at_base_ms, assigned_created.created_at_ms + 1)
FROM ordered_created
JOIN assigned_created ON ordered_created.row_number = assigned_created.row_number + 1
)
UPDATE threads
SET created_at_ms = (
SELECT created_at_ms
FROM assigned_created
WHERE assigned_created.id = threads.id
);
WITH RECURSIVE
ordered_updated AS (
SELECT
id,
updated_at_base_ms,
ROW_NUMBER() OVER (ORDER BY updated_at_base_ms, id) AS row_number
FROM thread_timestamp_migration
),
assigned_updated(row_number, id, updated_at_ms) AS (
SELECT row_number, id, updated_at_base_ms
FROM ordered_updated
WHERE row_number = 1
UNION ALL
SELECT
ordered_updated.row_number,
ordered_updated.id,
MAX(ordered_updated.updated_at_base_ms, assigned_updated.updated_at_ms + 1)
FROM ordered_updated
JOIN assigned_updated ON ordered_updated.row_number = assigned_updated.row_number + 1
)
UPDATE threads
SET updated_at_ms = (
SELECT updated_at_ms
FROM assigned_updated
WHERE assigned_updated.id = threads.id
);
DROP TABLE thread_timestamp_migration;
CREATE TRIGGER threads_created_at_ms_after_insert
AFTER INSERT ON threads
WHEN NEW.created_at_ms IS NULL
BEGIN
UPDATE threads
SET created_at_ms = NEW.created_at * 1000
WHERE id = NEW.id;
END;
CREATE TRIGGER threads_updated_at_ms_after_insert
AFTER INSERT ON threads
WHEN NEW.updated_at_ms IS NULL
BEGIN
UPDATE threads
SET updated_at_ms = NEW.updated_at * 1000
WHERE id = NEW.id;
END;
CREATE TRIGGER threads_created_at_ms_after_update
AFTER UPDATE OF created_at ON threads
WHEN NEW.created_at != OLD.created_at
AND NEW.created_at_ms IS OLD.created_at_ms
BEGIN
UPDATE threads
SET created_at_ms = NEW.created_at * 1000
WHERE id = NEW.id;
END;
CREATE TRIGGER threads_updated_at_ms_after_update
AFTER UPDATE OF updated_at ON threads
WHEN NEW.updated_at != OLD.updated_at
AND NEW.updated_at_ms IS OLD.updated_at_ms
BEGIN
UPDATE threads
SET updated_at_ms = NEW.updated_at * 1000
WHERE id = NEW.id;
END;
CREATE INDEX idx_threads_created_at_ms ON threads(created_at_ms DESC, id DESC);
CREATE INDEX idx_threads_updated_at_ms ON threads(updated_at_ms DESC, id DESC);
codex-rs/state/src/model/mod.rs
+2
View File
@@ -39,4 +39,6 @@ pub(crate) use memories::Stage1OutputRow;
pub(crate) use memories::stage1_output_ref_from_parts;
pub(crate) use thread_metadata::ThreadRow;
pub(crate) use thread_metadata::anchor_from_item;
pub(crate) use thread_metadata::datetime_to_epoch_millis;
pub(crate) use thread_metadata::datetime_to_epoch_seconds;
pub(crate) use thread_metadata::epoch_millis_to_datetime;
codex-rs/state/src/model/thread_metadata.rs
+23 -7
View File
@@ -1,6 +1,5 @@
use anyhow::Result;
use chrono::DateTime;
use chrono::Timelike;
use chrono::Utc;
use codex_protocol::ThreadId;
use codex_protocol::openai_models::ReasoningEffort;
@@ -300,7 +299,7 @@ impl ThreadMetadata {
}
fn canonicalize_datetime(dt: DateTime<Utc>) -> DateTime<Utc> {
dt.with_nanosecond(0).unwrap_or(dt)
epoch_millis_to_datetime(datetime_to_epoch_millis(dt)).unwrap_or(dt)
}
#[derive(Debug)]
@@ -389,8 +388,8 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
Ok(Self {
id: ThreadId::try_from(id)?,
rollout_path: PathBuf::from(rollout_path),
created_at: epoch_seconds_to_datetime(created_at)?,
updated_at: epoch_seconds_to_datetime(updated_at)?,
created_at: epoch_millis_to_datetime(created_at)?,
updated_at: epoch_millis_to_datetime(updated_at)?,
source,
agent_nickname,
agent_role,
@@ -423,13 +422,30 @@ pub(crate) fn anchor_from_item(item: &ThreadMetadata, sort_key: SortKey) -> Opti
Some(Anchor { ts, id })
}
pub(crate) fn datetime_to_epoch_millis(dt: DateTime<Utc>) -> i64 {
dt.timestamp_millis()
}
pub(crate) fn datetime_to_epoch_seconds(dt: DateTime<Utc>) -> i64 {
dt.timestamp()
}
pub(crate) fn epoch_seconds_to_datetime(secs: i64) -> Result<DateTime<Utc>> {
DateTime::<Utc>::from_timestamp(secs, 0)
.ok_or_else(|| anyhow::anyhow!("invalid unix timestamp: {secs}"))
pub(crate) fn epoch_millis_to_datetime(value: i64) -> Result<DateTime<Utc>> {
// Values older than 2020 if interpreted as milliseconds are legacy second-precision rows.
// Convert them in memory so old state DBs keep ordering correctly after new writes use ms.
const MIN_EPOCH_MILLIS: i64 = 1_577_836_800_000;
let millis = if value < MIN_EPOCH_MILLIS {
value.saturating_mul(1000)
} else {
value
};
DateTime::<Utc>::from_timestamp_millis(millis)
.ok_or_else(|| anyhow::anyhow!("invalid unix timestamp millis: {value}"))
}
pub(crate) fn epoch_seconds_to_datetime(value: i64) -> Result<DateTime<Utc>> {
DateTime::<Utc>::from_timestamp(value, 0)
.ok_or_else(|| anyhow::anyhow!("invalid unix timestamp seconds: {value}"))
}
/// Statistics about a backfill operation.
codex-rs/state/src/paths.rs
+1 -2
View File
@@ -1,10 +1,9 @@
use chrono::DateTime;
use chrono::Timelike;
use chrono::Utc;
use std::path::Path;
pub(crate) async fn file_modified_time_utc(path: &Path) -> Option<DateTime<Utc>> {
let modified = tokio::fs::metadata(path).await.ok()?.modified().ok()?;
let updated_at: DateTime<Utc> = modified.into();
Some(updated_at.with_nanosecond(0).unwrap_or(updated_at))
Some(updated_at)
}
codex-rs/state/src/runtime.rs
+10
View File
@@ -22,7 +22,9 @@ use crate::migrations::runtime_state_migrator;
use crate::model::AgentJobRow;
use crate::model::ThreadRow;
use crate::model::anchor_from_item;
use crate::model::datetime_to_epoch_millis;
use crate::model::datetime_to_epoch_seconds;
use crate::model::epoch_millis_to_datetime;
use crate::paths::file_modified_time_utc;
use chrono::DateTime;
use chrono::Utc;
@@ -47,6 +49,7 @@ use std::collections::BTreeSet;
use std::path::Path;
use std::path::PathBuf;
use std::sync::Arc;
use std::sync::atomic::AtomicI64;
use std::time::Duration;
use tracing::warn;
@@ -76,6 +79,7 @@ pub struct StateRuntime {
default_provider: String,
pool: Arc<sqlx::SqlitePool>,
logs_pool: Arc<sqlx::SqlitePool>,
thread_updated_at_millis: Arc<AtomicI64>,
}
impl StateRuntime {
@@ -120,11 +124,17 @@ impl StateRuntime {
return Err(err);
}
};
let thread_updated_at_millis: Option<i64> =
sqlx::query_scalar("SELECT MAX(threads.updated_at_ms) FROM threads")
.fetch_one(pool.as_ref())
.await?;
let thread_updated_at_millis = thread_updated_at_millis.unwrap_or(0);
let runtime = Arc::new(Self {
pool,
logs_pool,
codex_home,
default_provider,
thread_updated_at_millis: Arc::new(AtomicI64::new(thread_updated_at_millis)),
});
if let Err(err) = runtime.run_logs_startup_maintenance().await {
warn!(
codex-rs/state/src/runtime/memories.rs
+42 -35
View File
@@ -126,12 +126,9 @@ WHERE thread_id = ?
/// (`push_thread_filters`)
/// - excludes threads with `memory_mode != 'enabled'`
/// - excludes the current thread id
/// - keeps only threads in the age window:
/// `updated_at >= now - max_age_days` and `updated_at <= now - min_rollout_idle_hours`
/// - keeps only threads whose memory is stale:
/// `COALESCE(stage1_outputs.source_updated_at, -1) < threads.updated_at` and
/// `COALESCE(jobs.last_success_watermark, -1) < threads.updated_at`
/// - orders by `updated_at DESC, id DESC` and applies `scan_limit`
/// - keeps only threads whose millisecond `updated_at` is in the age window
/// - keeps only threads whose memory is stale compared to millisecond `updated_at`
/// - orders by `updated_at_ms DESC, id DESC` and applies `scan_limit`
///
/// For each selected thread, this function calls [`Self::try_claim_stage1_job`]
/// with `source_updated_at = thread.updated_at.timestamp()` and returns up to
@@ -155,34 +152,35 @@ WHERE thread_id = ?
let worker_id = current_thread_id;
let current_thread_id = worker_id.to_string();
let max_age_cutoff = (Utc::now() - Duration::days(max_age_days.max(0))).timestamp();
let idle_cutoff = (Utc::now() - Duration::hours(min_rollout_idle_hours.max(0))).timestamp();
let max_age_cutoff = (Utc::now() - Duration::days(max_age_days.max(0))).timestamp_millis();
let idle_cutoff =
(Utc::now() - Duration::hours(min_rollout_idle_hours.max(0))).timestamp_millis();
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
SELECT
id,
rollout_path,
created_at,
updated_at,
source,
agent_path,
agent_nickname,
agent_role,
model_provider,
model,
reasoning_effort,
cwd,
cli_version,
title,
sandbox_policy,
approval_mode,
tokens_used,
first_user_message,
archived_at,
git_sha,
git_branch,
git_origin_url
threads.id,
threads.rollout_path,
threads.created_at_ms AS created_at,
threads.updated_at_ms AS updated_at,
threads.source,
threads.agent_path,
threads.agent_nickname,
threads.agent_role,
threads.model_provider,
threads.model,
threads.reasoning_effort,
threads.cwd,
threads.cli_version,
threads.title,
threads.sandbox_policy,
threads.approval_mode,
threads.tokens_used,
threads.first_user_message,
threads.archived_at,
threads.git_sha,
threads.git_branch,
threads.git_origin_url
FROM threads
LEFT JOIN stage1_outputs
ON stage1_outputs.thread_id = threads.id
@@ -207,14 +205,23 @@ LEFT JOIN jobs
);
builder.push(" AND threads.memory_mode = 'enabled'");
builder
.push(" AND id != ")
.push(" AND threads.id != ")
.push_bind(current_thread_id.as_str());
builder
.push(" AND updated_at >= ")
.push(" AND ")
.push("threads.updated_at_ms")
.push(" >= ")
.push_bind(max_age_cutoff);
builder.push(" AND updated_at <= ").push_bind(idle_cutoff);
builder.push(" AND COALESCE(stage1_outputs.source_updated_at, -1) < updated_at");
builder.push(" AND COALESCE(jobs.last_success_watermark, -1) < updated_at");
builder
.push(" AND ")
.push("threads.updated_at_ms")
.push(" <= ")
.push_bind(idle_cutoff);
let updated_at = "threads.updated_at_ms";
builder.push(" AND ((COALESCE(stage1_outputs.source_updated_at, -1) + 1) * 1000) <= ");
builder.push(updated_at);
builder.push(" AND ((COALESCE(jobs.last_success_watermark, -1) + 1) * 1000) <= ");
builder.push(updated_at);
push_thread_order_and_limit(&mut builder, SortKey::UpdatedAt, scan_limit);
let items = builder
codex-rs/state/src/runtime/threads.rs
+248 -107
View File
@@ -1,35 +1,36 @@
use super::*;
use codex_protocol::protocol::SessionSource;
use std::sync::atomic::Ordering;
impl StateRuntime {
pub async fn get_thread(&self, id: ThreadId) -> anyhow::Result<Option<crate::ThreadMetadata>> {
let row = sqlx::query(
r#"
SELECT
id,
rollout_path,
created_at,
updated_at,
source,
agent_nickname,
agent_role,
agent_path,
model_provider,
model,
reasoning_effort,
cwd,
cli_version,
title,
sandbox_policy,
approval_mode,
tokens_used,
first_user_message,
archived_at,
git_sha,
git_branch,
git_origin_url
threads.id,
threads.rollout_path,
threads.created_at_ms AS created_at,
threads.updated_at_ms AS updated_at,
threads.source,
threads.agent_nickname,
threads.agent_role,
threads.agent_path,
threads.model_provider,
threads.model,
threads.reasoning_effort,
threads.cwd,
threads.cli_version,
threads.title,
threads.sandbox_policy,
threads.approval_mode,
threads.tokens_used,
threads.first_user_message,
threads.archived_at,
threads.git_sha,
threads.git_branch,
threads.git_origin_url
FROM threads
WHERE id = ?
WHERE threads.id = ?
"#,
)
.bind(id.to_string())
@@ -336,34 +337,9 @@ ON CONFLICT(child_thread_id) DO NOTHING
archived_only: bool,
cwd: Option<&Path>,
) -> anyhow::Result<Option<crate::ThreadMetadata>> {
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
SELECT
id,
rollout_path,
created_at,
updated_at,
source,
agent_nickname,
agent_role,
agent_path,
model_provider,
model,
reasoning_effort,
cwd,
cli_version,
title,
sandbox_policy,
approval_mode,
tokens_used,
first_user_message,
archived_at,
git_sha,
git_branch,
git_origin_url
FROM threads
"#,
);
let mut builder = QueryBuilder::<Sqlite>::new("");
push_thread_select_columns(&mut builder);
builder.push(" FROM threads");
push_thread_filters(
&mut builder,
archived_only,
@@ -373,10 +349,10 @@ FROM threads
crate::SortKey::UpdatedAt,
/*search_term*/ None,
);
builder.push(" AND title = ");
builder.push(" AND threads.title = ");
builder.push_bind(title);
if let Some(cwd) = cwd {
builder.push(" AND cwd = ");
builder.push(" AND threads.cwd = ");
builder.push_bind(cwd.display().to_string());
}
push_thread_order_and_limit(&mut builder, crate::SortKey::UpdatedAt, /*limit*/ 1);
@@ -400,34 +376,9 @@ FROM threads
) -> anyhow::Result<crate::ThreadsPage> {
let limit = page_size.saturating_add(1);
let mut builder = QueryBuilder::<Sqlite>::new(
r#"
SELECT
id,
rollout_path,
created_at,
updated_at,
source,
agent_nickname,
agent_role,
agent_path,
model_provider,
model,
reasoning_effort,
cwd,
cli_version,
title,
sandbox_policy,
approval_mode,
tokens_used,
first_user_message,
archived_at,
git_sha,
git_branch,
git_origin_url
FROM threads
"#,
);
let mut builder = QueryBuilder::<Sqlite>::new("");
push_thread_select_columns(&mut builder);
builder.push(" FROM threads");
push_thread_filters(
&mut builder,
archived_only,
@@ -470,7 +421,7 @@ FROM threads
model_providers: Option<&[String]>,
archived_only: bool,
) -> anyhow::Result<Vec<ThreadId>> {
let mut builder = QueryBuilder::<Sqlite>::new("SELECT id FROM threads");
let mut builder = QueryBuilder::<Sqlite>::new("SELECT threads.id FROM threads");
push_thread_filters(
&mut builder,
archived_only,
@@ -501,6 +452,7 @@ FROM threads
&self,
metadata: &crate::ThreadMetadata,
) -> anyhow::Result<bool> {
let updated_at = self.allocate_thread_updated_at(metadata.updated_at)?;
let result = sqlx::query(
r#"
INSERT INTO threads (
@@ -508,6 +460,8 @@ INSERT INTO threads (
rollout_path,
created_at,
updated_at,
created_at_ms,
updated_at_ms,
source,
agent_nickname,
agent_role,
@@ -528,14 +482,16 @@ INSERT INTO threads (
git_branch,
git_origin_url,
memory_mode
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO NOTHING
"#,
)
.bind(metadata.id.to_string())
.bind(metadata.rollout_path.display().to_string())
.bind(datetime_to_epoch_seconds(metadata.created_at))
.bind(datetime_to_epoch_seconds(metadata.updated_at))
.bind(datetime_to_epoch_seconds(updated_at))
.bind(datetime_to_epoch_millis(metadata.created_at))
.bind(datetime_to_epoch_millis(updated_at))
.bind(metadata.source.as_str())
.bind(metadata.agent_nickname.as_deref())
.bind(metadata.agent_role.as_deref())
@@ -599,14 +555,63 @@ ON CONFLICT(id) DO NOTHING
thread_id: ThreadId,
updated_at: DateTime<Utc>,
) -> anyhow::Result<bool> {
let result = sqlx::query("UPDATE threads SET updated_at = ? WHERE id = ?")
.bind(datetime_to_epoch_seconds(updated_at))
.bind(thread_id.to_string())
.execute(self.pool.as_ref())
.await?;
let updated_at = self.allocate_thread_updated_at(updated_at)?;
let result =
sqlx::query("UPDATE threads SET updated_at = ?, updated_at_ms = ? WHERE id = ?")
.bind(datetime_to_epoch_seconds(updated_at))
.bind(datetime_to_epoch_millis(updated_at))
.bind(thread_id.to_string())
.execute(self.pool.as_ref())
.await?;
Ok(result.rows_affected() > 0)
}
/// Allocate a persisted `updated_at` value for thread-list cursor ordering.
///
/// We keep a process-local high-water mark so hot rollout writes can get unique,
/// monotonic millisecond timestamps without querying SQLite on every update. Older
/// backfill/repair timestamps are allowed through unchanged so historical ordering
/// remains tied to the rollout file mtimes.
fn allocate_thread_updated_at(
&self,
updated_at: DateTime<Utc>,
) -> anyhow::Result<DateTime<Utc>> {
let candidate = datetime_to_epoch_millis(updated_at);
let allocated = loop {
let current = self.thread_updated_at_millis.load(Ordering::Relaxed);
// New wall-clock time: advance the process-local high-water mark and use it as-is.
if candidate > current {
if self
.thread_updated_at_millis
.compare_exchange(current, candidate, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
break candidate;
}
continue;
}
// Older timestamps come from backfill/repair paths that preserve rollout mtimes.
// Do not drag historical rows forward just because this process has seen newer writes.
if candidate.saturating_add(1000) <= current {
break candidate;
}
// Same hot one-second bucket as the current high-water mark. Allocate the next
// millisecond so updated_at remains unique and cursor-orderable inside the process.
let bumped = current.saturating_add(1);
if self
.thread_updated_at_millis
.compare_exchange(current, bumped, Ordering::Relaxed, Ordering::Relaxed)
.is_ok()
{
break bumped;
}
};
epoch_millis_to_datetime(allocated)
}
pub async fn update_thread_git_info(
&self,
thread_id: ThreadId,
@@ -641,6 +646,7 @@ WHERE id = ?
metadata: &crate::ThreadMetadata,
creation_memory_mode: Option<&str>,
) -> anyhow::Result<()> {
let updated_at = self.allocate_thread_updated_at(metadata.updated_at)?;
sqlx::query(
r#"
INSERT INTO threads (
@@ -648,6 +654,8 @@ INSERT INTO threads (
rollout_path,
created_at,
updated_at,
created_at_ms,
updated_at_ms,
source,
agent_nickname,
agent_role,
@@ -668,11 +676,13 @@ INSERT INTO threads (
git_branch,
git_origin_url,
memory_mode
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET
rollout_path = excluded.rollout_path,
created_at = excluded.created_at,
updated_at = excluded.updated_at,
created_at_ms = excluded.created_at_ms,
updated_at_ms = excluded.updated_at_ms,
source = excluded.source,
agent_nickname = excluded.agent_nickname,
agent_role = excluded.agent_role,
@@ -697,7 +707,9 @@ ON CONFLICT(id) DO UPDATE SET
.bind(metadata.id.to_string())
.bind(metadata.rollout_path.display().to_string())
.bind(datetime_to_epoch_seconds(metadata.created_at))
.bind(datetime_to_epoch_seconds(metadata.updated_at))
.bind(datetime_to_epoch_seconds(updated_at))
.bind(datetime_to_epoch_millis(metadata.created_at))
.bind(datetime_to_epoch_millis(updated_at))
.bind(metadata.source.as_str())
.bind(metadata.agent_nickname.as_deref())
.bind(metadata.agent_role.as_deref())
@@ -909,6 +921,36 @@ fn one_thread_id_from_rows(
}
}
pub(super) fn push_thread_select_columns(builder: &mut QueryBuilder<'_, Sqlite>) {
builder.push(
r#"
SELECT
threads.id,
threads.rollout_path,
threads.created_at_ms AS created_at,
threads.updated_at_ms AS updated_at,
threads.source,
threads.agent_nickname,
threads.agent_role,
threads.agent_path,
threads.model_provider,
threads.model,
threads.reasoning_effort,
threads.cwd,
threads.cli_version,
threads.title,
threads.sandbox_policy,
threads.approval_mode,
threads.tokens_used,
threads.first_user_message,
threads.archived_at,
threads.git_sha,
threads.git_branch,
threads.git_origin_url
"#,
);
}
pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option<Option<Vec<DynamicToolSpec>>> {
items.iter().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.dynamic_tools.clone()),
@@ -952,13 +994,13 @@ pub(super) fn push_thread_filters<'a>(
) {
builder.push(" WHERE 1 = 1");
if archived_only {
builder.push(" AND archived = 1");
builder.push(" AND threads.archived = 1");
} else {
builder.push(" AND archived = 0");
builder.push(" AND threads.archived = 0");
}
builder.push(" AND first_user_message <> ''");
builder.push(" AND threads.first_user_message <> ''");
if !allowed_sources.is_empty() {
builder.push(" AND source IN (");
builder.push(" AND threads.source IN (");
let mut separated = builder.separated(", ");
for source in allowed_sources {
separated.push_bind(source);
@@ -968,7 +1010,7 @@ pub(super) fn push_thread_filters<'a>(
if let Some(model_providers) = model_providers
&& !model_providers.is_empty()
{
builder.push(" AND model_provider IN (");
builder.push(" AND threads.model_provider IN (");
let mut separated = builder.separated(", ");
for provider in model_providers {
separated.push_bind(provider);
@@ -976,15 +1018,15 @@ pub(super) fn push_thread_filters<'a>(
separated.push_unseparated(")");
}
if let Some(search_term) = search_term {
builder.push(" AND instr(title, ");
builder.push(" AND instr(threads.title, ");
builder.push_bind(search_term);
builder.push(") > 0");
}
if let Some(anchor) = anchor {
let anchor_ts = datetime_to_epoch_seconds(anchor.ts);
let anchor_ts = datetime_to_epoch_millis(anchor.ts);
let column = match sort_key {
SortKey::CreatedAt => "created_at",
SortKey::UpdatedAt => "updated_at",
SortKey::CreatedAt => "threads.created_at_ms",
SortKey::UpdatedAt => "threads.updated_at_ms",
};
builder.push(" AND (");
builder.push(column);
@@ -1006,8 +1048,8 @@ pub(super) fn push_thread_order_and_limit(
limit: usize,
) {
let order_column = match sort_key {
SortKey::CreatedAt => "created_at",
SortKey::UpdatedAt => "updated_at",
SortKey::CreatedAt => "threads.created_at_ms",
SortKey::UpdatedAt => "threads.updated_at_ms",
};
builder.push(" ORDER BY ");
builder.push(order_column);
@@ -1207,12 +1249,13 @@ mod tests {
.await
.expect("initial upsert should succeed");
let updated_at = datetime_to_epoch_seconds(
let updated_at = datetime_to_epoch_millis(
DateTime::<Utc>::from_timestamp(1_700_000_100, 0).expect("timestamp"),
);
sqlx::query(
"UPDATE threads SET updated_at = ?, tokens_used = ?, first_user_message = ? WHERE id = ?",
"UPDATE threads SET updated_at = ?, updated_at_ms = ?, tokens_used = ?, first_user_message = ? WHERE id = ?",
)
.bind(updated_at / 1000)
.bind(updated_at)
.bind(123_i64)
.bind("newer preview")
@@ -1242,7 +1285,7 @@ mod tests {
persisted.first_user_message.as_deref(),
Some("newer preview")
);
assert_eq!(datetime_to_epoch_seconds(persisted.updated_at), updated_at);
assert_eq!(datetime_to_epoch_millis(persisted.updated_at), updated_at);
assert_eq!(persisted.git_sha.as_deref(), Some("abc123"));
assert_eq!(persisted.git_branch.as_deref(), Some("feature/branch"));
assert_eq!(
@@ -1291,8 +1334,8 @@ mod tests {
Some("newer preview")
);
assert_eq!(
datetime_to_epoch_seconds(persisted.updated_at),
datetime_to_epoch_seconds(existing.updated_at)
datetime_to_epoch_millis(persisted.updated_at),
datetime_to_epoch_millis(existing.updated_at)
);
}
@@ -1367,6 +1410,104 @@ mod tests {
);
}
#[tokio::test]
async fn thread_updated_at_uses_unique_epoch_millis_and_reads_legacy_seconds() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await
.expect("state db should initialize");
let first_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000901").expect("valid thread id");
let second_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000902").expect("valid thread id");
let older_id =
ThreadId::from_string("00000000-0000-0000-0000-000000000903").expect("valid thread id");
let updated_at =
DateTime::<Utc>::from_timestamp_millis(1_700_001_111_123).expect("timestamp millis");
let mut first = test_thread_metadata(&codex_home, first_id, codex_home.clone());
first.updated_at = updated_at;
let mut second = test_thread_metadata(&codex_home, second_id, codex_home.clone());
second.updated_at = updated_at;
runtime
.upsert_thread(&first)
.await
.expect("first upsert should succeed");
runtime
.upsert_thread(&second)
.await
.expect("second upsert should succeed");
let first = runtime
.get_thread(first_id)
.await
.expect("thread should load")
.expect("thread should exist");
let second = runtime
.get_thread(second_id)
.await
.expect("thread should load")
.expect("thread should exist");
assert_eq!(
datetime_to_epoch_millis(first.updated_at),
1_700_001_111_123
);
assert_eq!(
datetime_to_epoch_millis(second.updated_at),
1_700_001_111_124
);
let second_row: (i64, i64, Option<i64>, Option<i64>) = sqlx::query_as(
"SELECT created_at, updated_at, created_at_ms, updated_at_ms FROM threads WHERE id = ?",
)
.bind(second_id.to_string())
.fetch_one(runtime.pool.as_ref())
.await
.expect("thread timestamp row should load");
assert_eq!(
second_row,
(
datetime_to_epoch_seconds(second.created_at),
1_700_001_111,
Some(datetime_to_epoch_millis(second.created_at)),
Some(1_700_001_111_124)
)
);
let older_updated_at =
DateTime::<Utc>::from_timestamp_millis(1_700_001_100_123).expect("timestamp millis");
let mut older = test_thread_metadata(&codex_home, older_id, codex_home.clone());
older.updated_at = older_updated_at;
runtime
.upsert_thread(&older)
.await
.expect("older upsert should succeed");
let older = runtime
.get_thread(older_id)
.await
.expect("thread should load")
.expect("thread should exist");
assert_eq!(
datetime_to_epoch_millis(older.updated_at),
1_700_001_100_123
);
sqlx::query("UPDATE threads SET updated_at = ? WHERE id = ?")
.bind(1_700_001_112_i64)
.bind(first_id.to_string())
.execute(runtime.pool.as_ref())
.await
.expect("legacy timestamp write should succeed");
let legacy = runtime
.get_thread(first_id)
.await
.expect("thread should load")
.expect("thread should exist");
assert_eq!(
datetime_to_epoch_millis(legacy.updated_at),
1_700_001_112_000
);
}
#[tokio::test]
async fn apply_rollout_items_uses_override_updated_at_when_provided() {
let codex_home = unique_temp_dir();