openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-21 17:48:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

persist parent thread id in sqlite

Browse Source
This commit is contained in:
Owen Lin
2026-05-31 21:24:05 -07:00
parent 299bd50947
commit 2dbf0c558b
9 changed files with 158 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/rollout/src/metadata.rs
+1
View File
@@ -47,6 +47,7 @@ pub(crate) fn builder_from_session_meta(
session_meta.meta.source.clone(),
);
builder.model_provider = session_meta.meta.model_provider.clone();
builder.parent_thread_id = session_meta.meta.parent_thread_id;
builder.agent_nickname = session_meta.meta.agent_nickname.clone();
builder.agent_role = session_meta.meta.agent_role.clone();
builder.agent_path = session_meta.meta.agent_path.clone();
codex-rs/rollout/src/recorder.rs
+1 -1
View File
@@ -1699,7 +1699,7 @@ fn thread_item_from_state_metadata(item: codex_state::ThreadMetadata) -> ThreadI
.or_else(|_| serde_json::from_value(Value::String(item.source)))
.unwrap_or(SessionSource::Unknown),
),
parent_thread_id: None,
parent_thread_id: item.parent_thread_id,
agent_nickname: item.agent_nickname,
agent_role: item.agent_role,
model_provider: Some(item.model_provider),
codex-rs/rollout/src/recorder_tests.rs
+110
View File
@@ -14,6 +14,7 @@ use codex_protocol::protocol::SandboxPolicy;
use codex_protocol::protocol::SessionMeta;
use codex_protocol::protocol::SessionMetaLine;
use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::SubAgentSource;
use codex_protocol::protocol::TurnContextItem;
use codex_protocol::protocol::UserMessageEvent;
use pretty_assertions::assert_eq;
@@ -803,6 +804,115 @@ async fn list_threads_state_db_only_skips_jsonl_repair_scan() -> std::io::Result
Ok(())
}
#[tokio::test]
async fn list_threads_state_db_preserves_parent_thread_id() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init(
home.path().to_path_buf(),
config.model_provider_id.clone(),
)
.await
.expect("state db should initialize");
runtime
.mark_backfill_complete(/*last_watermark*/ None)
.await
.expect("backfill should be complete");
let uuid = Uuid::from_u128(9014);
let thread_id = ThreadId::from_string(&uuid.to_string()).expect("valid thread id");
let parent_thread_id = ThreadId::new();
let ts = "2025-01-03T15-00-00";
let day_dir = home.path().join("sessions/2025/01/03");
fs::create_dir_all(&day_dir)?;
let path = day_dir.join(format!("rollout-{ts}-{uuid}.jsonl"));
let rollout_line = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::SessionMeta(SessionMetaLine {
meta: SessionMeta {
id: thread_id,
forked_from_id: None,
parent_thread_id: Some(parent_thread_id),
timestamp: ts.to_string(),
cwd: home.path().to_path_buf(),
originator: "test_originator".to_string(),
cli_version: "test_version".to_string(),
source: SessionSource::SubAgent(SubAgentSource::Review),
thread_source: None,
agent_path: None,
agent_nickname: None,
agent_role: None,
model_provider: Some(config.model_provider_id.clone()),
base_instructions: None,
dynamic_tools: None,
memory_mode: None,
},
git: None,
}),
};
let mut file = File::create(&path)?;
writeln!(
file,
"{}",
serde_json::to_string(&rollout_line).expect("serialize rollout")
)?;
let user_event = RolloutLine {
timestamp: ts.to_string(),
item: RolloutItem::EventMsg(EventMsg::UserMessage(UserMessageEvent {
message: "Hello from user".to_string(),
..Default::default()
})),
};
writeln!(
file,
"{}",
serde_json::to_string(&user_event).expect("serialize user event")
)?;
file.flush()?;
let repaired_page = RolloutRecorder::list_threads(
Some(runtime.clone()),
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(repaired_page.items.len(), 1);
assert_eq!(
repaired_page.items[0].parent_thread_id,
Some(parent_thread_id)
);
let state_db_only_page = RolloutRecorder::list_threads_from_state_db(
Some(runtime.clone()),
&config,
/*page_size*/ 10,
/*cursor*/ None,
ThreadSortKey::CreatedAt,
SortDirection::Desc,
&[],
/*model_providers*/ None,
/*cwd_filters*/ None,
config.model_provider_id.as_str(),
/*search_term*/ None,
)
.await?;
assert_eq!(state_db_only_page.items.len(), 1);
assert_eq!(
state_db_only_page.items[0].parent_thread_id,
Some(parent_thread_id)
);
Ok(())
}
#[tokio::test]
async fn list_threads_default_filter_returns_filesystem_scan_results() -> std::io::Result<()> {
let home = TempDir::new().expect("temp dir");
codex-rs/state/migrations/0036_threads_parent_thread_id.sql
+1
View File
@@ -0,0 +1 @@
ALTER TABLE threads ADD COLUMN parent_thread_id TEXT;
codex-rs/state/src/extract.rs
+2
View File
@@ -51,6 +51,7 @@ fn apply_session_meta_from_item(metadata: &mut ThreadMetadata, meta_line: &Sessi
metadata.id = meta_line.meta.id;
metadata.source = enum_to_string(&meta_line.meta.source);
metadata.thread_source = meta_line.meta.thread_source;
metadata.parent_thread_id = meta_line.meta.parent_thread_id;
metadata.agent_nickname = meta_line.meta.agent_nickname.clone();
metadata.agent_role = meta_line.meta.agent_role.clone();
metadata.agent_path = meta_line.meta.agent_path.clone();
@@ -514,6 +515,7 @@ mod tests {
updated_at: created_at,
source: "cli".to_string(),
thread_source: None,
parent_thread_id: None,
agent_path: None,
agent_nickname: None,
agent_role: None,
codex-rs/state/src/model/thread_metadata.rs
+31
View File
@@ -71,6 +71,8 @@ pub struct ThreadMetadata {
pub source: String,
/// Optional analytics source classification for this thread.
pub thread_source: Option<ThreadSource>,
/// Immediate control/spawn parent thread id, if this thread is a subagent.
pub parent_thread_id: Option<ThreadId>,
/// Optional random unique nickname assigned to an AgentControl-spawned sub-agent.
pub agent_nickname: Option<String>,
/// Optional role (agent_role) assigned to an AgentControl-spawned sub-agent.
@@ -124,6 +126,8 @@ pub struct ThreadMetadataBuilder {
pub source: SessionSource,
/// Optional analytics source classification for this thread.
pub thread_source: Option<ThreadSource>,
/// Immediate control/spawn parent thread id, if this thread is a subagent.
pub parent_thread_id: Option<ThreadId>,
/// Optional random unique nickname assigned to the session.
pub agent_nickname: Option<String>,
/// Optional role (agent_role) assigned to the session.
@@ -165,6 +169,7 @@ impl ThreadMetadataBuilder {
updated_at: None,
source,
thread_source: None,
parent_thread_id: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
@@ -197,6 +202,7 @@ impl ThreadMetadataBuilder {
updated_at,
source,
thread_source: self.thread_source,
parent_thread_id: self.parent_thread_id,
agent_nickname: self.agent_nickname.clone(),
agent_role: self.agent_role.clone(),
agent_path: self
@@ -257,6 +263,9 @@ impl ThreadMetadata {
if self.source != other.source {
diffs.push("source");
}
if self.parent_thread_id != other.parent_thread_id {
diffs.push("parent_thread_id");
}
if self.agent_nickname != other.agent_nickname {
diffs.push("agent_nickname");
}
@@ -327,6 +336,7 @@ pub(crate) struct ThreadRow {
updated_at: i64,
source: String,
thread_source: Option<String>,
parent_thread_id: Option<String>,
agent_nickname: Option<String>,
agent_role: Option<String>,
agent_path: Option<String>,
@@ -356,6 +366,7 @@ impl ThreadRow {
updated_at: row.try_get("updated_at")?,
source: row.try_get("source")?,
thread_source: row.try_get("thread_source")?,
parent_thread_id: row.try_get("parent_thread_id")?,
agent_nickname: row.try_get("agent_nickname")?,
agent_role: row.try_get("agent_role")?,
agent_path: row.try_get("agent_path")?,
@@ -389,6 +400,7 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
updated_at,
source,
thread_source,
parent_thread_id,
agent_nickname,
agent_role,
agent_path,
@@ -412,6 +424,7 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
.map(|thread_source| thread_source.parse())
.transpose()
.map_err(anyhow::Error::msg)?;
let parent_thread_id = parent_thread_id.map(ThreadId::try_from).transpose()?;
Ok(Self {
id: ThreadId::try_from(id)?,
rollout_path: PathBuf::from(rollout_path),
@@ -419,6 +432,7 @@ impl TryFrom<ThreadRow> for ThreadMetadata {
updated_at: epoch_millis_to_datetime(updated_at)?,
source,
thread_source,
parent_thread_id,
agent_nickname,
agent_role,
agent_path,
@@ -506,6 +520,7 @@ mod tests {
updated_at: 1_700_000_100,
source: "cli".to_string(),
thread_source: None,
parent_thread_id: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
@@ -536,6 +551,7 @@ mod tests {
updated_at: DateTime::<Utc>::from_timestamp(1_700_000_100, 0).expect("timestamp"),
source: "cli".to_string(),
thread_source: None,
parent_thread_id: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
@@ -568,6 +584,21 @@ mod tests {
);
}
#[test]
fn thread_row_parses_parent_thread_id() {
let parent_thread_id = "00000000-0000-0000-0000-000000000124";
let mut row = thread_row(/*reasoning_effort*/ None);
row.parent_thread_id = Some(parent_thread_id.to_string());
let mut expected = expected_thread_metadata(/*reasoning_effort*/ None);
expected.parent_thread_id =
Some(ThreadId::from_string(parent_thread_id).expect("valid thread id"));
assert_eq!(
ThreadMetadata::try_from(row).expect("thread metadata should parse"),
expected
);
}
#[test]
fn thread_row_ignores_unknown_reasoning_effort_values() {
let metadata = ThreadMetadata::try_from(thread_row(Some("future")))
codex-rs/state/src/runtime/memories.rs
+2
View File
@@ -173,6 +173,7 @@ SELECT
threads.updated_at_ms AS updated_at,
threads.source,
threads.thread_source,
threads.parent_thread_id,
threads.agent_path,
threads.agent_nickname,
threads.agent_role,
@@ -543,6 +544,7 @@ SELECT
threads.updated_at_ms AS updated_at,
threads.source,
threads.thread_source,
threads.parent_thread_id,
threads.agent_nickname,
threads.agent_role,
threads.agent_path,
codex-rs/state/src/runtime/test_support.rs
+1
View File
@@ -49,6 +49,7 @@ pub(super) fn test_thread_metadata(
updated_at: now,
source: "cli".to_string(),
thread_source: None,
parent_thread_id: None,
agent_nickname: None,
agent_role: None,
agent_path: None,
codex-rs/state/src/runtime/threads.rs
+9 -2
View File
@@ -14,6 +14,7 @@ SELECT
threads.updated_at_ms AS updated_at,
threads.source,
threads.thread_source,
threads.parent_thread_id,
threads.agent_nickname,
threads.agent_role,
threads.agent_path,
@@ -487,6 +488,7 @@ INSERT INTO threads (
updated_at_ms,
source,
thread_source,
parent_thread_id,
agent_nickname,
agent_role,
agent_path,
@@ -507,7 +509,7 @@ INSERT INTO threads (
git_branch,
git_origin_url,
memory_mode
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO NOTHING
"#,
)
@@ -523,6 +525,7 @@ ON CONFLICT(id) DO NOTHING
.thread_source
.map(codex_protocol::protocol::ThreadSource::as_str),
)
.bind(metadata.parent_thread_id.as_ref().map(ToString::to_string))
.bind(metadata.agent_nickname.as_deref())
.bind(metadata.agent_role.as_deref())
.bind(metadata.agent_path.as_deref())
@@ -693,6 +696,7 @@ INSERT INTO threads (
updated_at_ms,
source,
thread_source,
parent_thread_id,
agent_nickname,
agent_role,
agent_path,
@@ -713,7 +717,7 @@ 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,
@@ -722,6 +726,7 @@ ON CONFLICT(id) DO UPDATE SET
updated_at_ms = excluded.updated_at_ms,
source = excluded.source,
thread_source = excluded.thread_source,
parent_thread_id = excluded.parent_thread_id,
agent_nickname = excluded.agent_nickname,
agent_role = excluded.agent_role,
agent_path = excluded.agent_path,
@@ -755,6 +760,7 @@ ON CONFLICT(id) DO UPDATE SET
.thread_source
.map(codex_protocol::protocol::ThreadSource::as_str),
)
.bind(metadata.parent_thread_id.as_ref().map(ToString::to_string))
.bind(metadata.agent_nickname.as_deref())
.bind(metadata.agent_role.as_deref())
.bind(metadata.agent_path.as_deref())
@@ -925,6 +931,7 @@ SELECT
threads.updated_at_ms AS updated_at,
threads.source,
threads.thread_source,
threads.parent_thread_id,
threads.agent_nickname,
threads.agent_role,
threads.agent_path,