openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-29 19:08:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge 615da1b670 into sapling-pr-archive-bolinfest

Browse Source
This commit is contained in:
Michael Bolin
2026-02-27 17:58:13 -08:00
committed by GitHub
parent 9c0185ebb8 615da1b670
commit 8caf978d0e
24 changed files with 452 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -1603,6 +1603,7 @@ dependencies = [
"codex-protocol",
"codex-responses-api-proxy",
"codex-rmcp-client",
"codex-state",
"codex-stdio-to-uds",
"codex-tui",
"codex-utils-cargo-bin",
@@ -1614,6 +1615,7 @@ dependencies = [
"pretty_assertions",
"regex-lite",
"serde_json",
"sqlx",
"supports-color 3.0.2",
"tempfile",
"tokio",
codex-rs/app-server-protocol/schema/json/ServerNotification.json
+5
View File
@@ -1655,6 +1655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -1724,6 +1728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/codex_app_server_protocol.schemas.json
+5
View File
@@ -12498,6 +12498,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -12567,6 +12571,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadForkResponse.json
+5
View File
@@ -882,6 +882,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -951,6 +955,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadListResponse.json
+5
View File
@@ -655,6 +655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -724,6 +728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadReadResponse.json
+5
View File
@@ -655,6 +655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -724,6 +728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadResumeResponse.json
+5
View File
@@ -882,6 +882,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -951,6 +955,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadRollbackResponse.json
+5
View File
@@ -655,6 +655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -724,6 +728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadStartResponse.json
+5
View File
@@ -882,6 +882,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -951,6 +955,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadStartedNotification.json
+5
View File
@@ -655,6 +655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -724,6 +728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/json/v2/ThreadUnarchiveResponse.json
+5
View File
@@ -655,6 +655,10 @@
"description": "Working directory captured for the thread.",
"type": "string"
},
"ephemeral": {
"description": "Whether the thread is ephemeral and should not be materialized on disk.",
"type": "boolean"
},
"gitInfo": {
"anyOf": [
{
@@ -724,6 +728,7 @@
"cliVersion",
"createdAt",
"cwd",
"ephemeral",
"id",
"modelProvider",
"preview",
codex-rs/app-server-protocol/schema/typescript/v2/Thread.ts
+4
View File
@@ -11,6 +11,10 @@ export type Thread = { id: string,
* Usually the first user message in the thread, if available.
*/
preview: string,
/**
* Whether the thread is ephemeral and should not be materialized on disk.
*/
ephemeral: boolean,
/**
* Model provider used for this thread (for example, 'openai').
*/
codex-rs/app-server-protocol/src/protocol/v2.rs
+2
View File
@@ -2517,6 +2517,8 @@ pub struct Thread {
pub id: String,
/// Usually the first user message in the thread, if available.
pub preview: String,
/// Whether the thread is ephemeral and should not be materialized on disk.
pub ephemeral: bool,
/// Model provider used for this thread (for example, 'openai').
pub model_provider: String,
/// Unix timestamp (in seconds) when the thread was created.
codex-rs/app-server/README.md
+1
View File
@@ -62,6 +62,7 @@ Use the thread APIs to create, list, or archive conversations. Drive a conversat
- Initialize once per connection: Immediately after opening a transport connection, send an `initialize` request with your client metadata, then emit an `initialized` notification. Any other request on that connection before this handshake gets rejected.
- Start (or resume) a thread: Call `thread/start` to open a fresh conversation. The response returns the thread object and you’ll also get a `thread/started` notification. If you’re continuing an existing conversation, call `thread/resume` with its ID instead. If you want to branch from an existing conversation, call `thread/fork` to create a new thread id with copied history.
The returned `thread.ephemeral` flag tells you whether the session is intentionally in-memory only; when it is `true`, `thread.path` is `null`.
- Begin a turn: To send user input, call `turn/start` with the target `threadId` and the user's input. Optional fields let you override model, cwd, sandbox policy, etc. This immediately returns the new turn object and triggers a `turn/started` notification.
- Stream events: After `turn/start`, keep reading JSON-RPC notifications on stdout. You’ll see `item/started`, `item/completed`, deltas like `item/agentMessage/delta`, tool progress, etc. These represent streaming model output plus any side effects (commands, tool calls, reasoning notes).
- Finish the turn: When the model is done (or the turn is interrupted via making the `turn/interrupt` call), the server sends `turn/completed` with the final turn state and token usage.
codex-rs/app-server/src/codex_message_processor.rs
+2
View File
@@ -7869,6 +7869,7 @@ fn build_thread_from_snapshot(
Thread {
id: thread_id.to_string(),
preview: String::new(),
ephemeral: config_snapshot.ephemeral,
model_provider: config_snapshot.model_provider_id.clone(),
created_at: now,
updated_at: now,
@@ -7910,6 +7911,7 @@ pub(crate) fn summary_to_thread(summary: ConversationSummary) -> Thread {
Thread {
id: conversation_id.to_string(),
preview,
ephemeral: false,
model_provider,
created_at: created_at.map(|dt| dt.timestamp()).unwrap_or(0),
updated_at: updated_at.map(|dt| dt.timestamp()).unwrap_or(0),
codex-rs/app-server/src/thread_status.rs
+1
View File
@@ -733,6 +733,7 @@ mod tests {
Thread {
id: thread_id.to_string(),
preview: String::new(),
ephemeral: false,
model_provider: "mock-provider".to_string(),
created_at: 0,
updated_at: 0,
codex-rs/app-server/tests/suite/v2/thread_read.rs
+16
View File
@@ -78,6 +78,7 @@ async fn thread_read_returns_summary_without_turns() -> Result<()> {
assert_eq!(thread.id, conversation_id);
assert_eq!(thread.preview, preview);
assert_eq!(thread.model_provider, "mock_provider");
assert!(!thread.ephemeral, "stored rollouts should not be ephemeral");
assert!(thread.path.as_ref().expect("thread path").is_absolute());
assert_eq!(thread.cwd, PathBuf::from("/"));
assert_eq!(thread.cli_version, "0.0.0");
@@ -278,6 +279,11 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
Some(new_name),
"thread/read must serialize `thread.name` on the wire"
);
assert_eq!(
thread_json.get("ephemeral").and_then(Value::as_bool),
Some(false),
"thread/read must serialize `thread.ephemeral` on the wire"
);
// List should also surface the name.
let list_id = mcp
@@ -317,6 +323,11 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
Some(new_name),
"thread/list must serialize `thread.name` on the wire"
);
assert_eq!(
listed_json.get("ephemeral").and_then(Value::as_bool),
Some(false),
"thread/list must serialize `thread.ephemeral` on the wire"
);
// Resume should also surface the name.
let resume_id = mcp
@@ -345,6 +356,11 @@ async fn thread_name_set_is_reflected_in_read_list_and_resume() -> Result<()> {
Some(new_name),
"thread/resume must serialize `thread.name` on the wire"
);
assert_eq!(
resumed_json.get("ephemeral").and_then(Value::as_bool),
Some(false),
"thread/resume must serialize `thread.ephemeral` on the wire"
);
Ok(())
}
codex-rs/app-server/tests/suite/v2/thread_start.rs
+30
View File
@@ -62,6 +62,10 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
thread.created_at > 0,
"created_at should be a positive UNIX timestamp"
);
assert!(
!thread.ephemeral,
"new persistent threads should not be ephemeral"
);
assert_eq!(thread.status, ThreadStatus::Idle);
let thread_path = thread.path.clone().expect("thread path should be present");
assert!(thread_path.is_absolute(), "thread path should be absolute");
@@ -80,6 +84,11 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Some(&Value::Null),
"new threads should serialize `name: null`"
);
assert_eq!(
thread_json.get("ephemeral").and_then(Value::as_bool),
Some(false),
"new persistent threads should serialize `ephemeral: false`"
);
assert_eq!(thread.name, None);
// A corresponding thread/started notification should arrive.
@@ -98,6 +107,13 @@ async fn thread_start_creates_thread_and_emits_started() -> Result<()> {
Some(&Value::Null),
"thread/started should serialize `name: null` for new threads"
);
assert_eq!(
started_thread_json
.get("ephemeral")
.and_then(Value::as_bool),
Some(false),
"thread/started should serialize `ephemeral: false` for new persistent threads"
);
let started: ThreadStartedNotification =
serde_json::from_value(notif.params.expect("params must be present"))?;
assert_eq!(started.thread, thread);
@@ -196,11 +212,25 @@ async fn thread_start_ephemeral_remains_pathless() -> Result<()> {
mcp.read_stream_until_response_message(RequestId::Integer(req_id)),
)
.await??;
let resp_result = resp.result.clone();
let ThreadStartResponse { thread, .. } = to_response::<ThreadStartResponse>(resp)?;
assert!(
thread.ephemeral,
"ephemeral threads should be marked explicitly"
);
assert_eq!(
thread.path, None,
"ephemeral threads should not expose a path"
);
let thread_json = resp_result
.get("thread")
.and_then(Value::as_object)
.expect("thread/start result.thread must be an object");
assert_eq!(
thread_json.get("ephemeral").and_then(Value::as_bool),
Some(true),
"ephemeral threads should serialize `ephemeral: true`"
);
Ok(())
}
codex-rs/cli/Cargo.toml
+2
View File
@@ -35,6 +35,7 @@ codex-mcp-server = { workspace = true }
codex-protocol = { workspace = true }
codex-responses-api-proxy = { workspace = true }
codex-rmcp-client = { workspace = true }
codex-state = { workspace = true }
codex-stdio-to-uds = { workspace = true }
codex-tui = { workspace = true }
libc = { workspace = true }
@@ -62,3 +63,4 @@ assert_matches = { workspace = true }
codex-utils-cargo-bin = { workspace = true }
predicates = { workspace = true }
pretty_assertions = { workspace = true }
sqlx = { workspace = true }
codex-rs/cli/src/main.rs
+63
View File
@@ -22,6 +22,8 @@ use codex_exec::Command as ExecCommand;
use codex_exec::ReviewArgs;
use codex_execpolicy::ExecPolicyCheckCommand;
use codex_responses_api_proxy::Args as ResponsesApiProxyArgs;
use codex_state::StateRuntime;
use codex_state::state_db_path;
use codex_tui::AppExitInfo;
use codex_tui::Cli as TuiCli;
use codex_tui::ExitReason;
@@ -163,6 +165,10 @@ struct DebugCommand {
enum DebugSubcommand {
/// Tooling: helps debug the app server.
AppServer(DebugAppServerCommand),
/// Internal: reset local memory state for a fresh start.
#[clap(hide = true)]
ClearMemories,
}
#[derive(Debug, Parser)]
@@ -751,6 +757,9 @@ async fn cli_main(arg0_paths: Arg0DispatchPaths) -> anyhow::Result<()> {
DebugSubcommand::AppServer(cmd) => {
run_debug_app_server_command(cmd)?;
}
DebugSubcommand::ClearMemories => {
run_debug_clear_memories_command(&root_config_overrides, &interactive).await?;
}
},
Some(Subcommand::Execpolicy(ExecpolicyCommand { sub })) => match sub {
ExecpolicySubcommand::Check(cmd) => run_execpolicycheck(cmd)?,
@@ -877,6 +886,60 @@ fn maybe_print_under_development_feature_warning(
);
}
async fn run_debug_clear_memories_command(
root_config_overrides: &CliConfigOverrides,
interactive: &TuiCli,
) -> anyhow::Result<()> {
let cli_kv_overrides = root_config_overrides
.parse_overrides()
.map_err(anyhow::Error::msg)?;
let overrides = ConfigOverrides {
config_profile: interactive.config_profile.clone(),
..Default::default()
};
let config =
Config::load_with_cli_overrides_and_harness_overrides(cli_kv_overrides, overrides).await?;
let state_path = state_db_path(config.sqlite_home.as_path());
let mut cleared_state_db = false;
if tokio::fs::try_exists(&state_path).await? {
let state_db = StateRuntime::init(
config.sqlite_home.clone(),
config.model_provider_id.clone(),
None,
)
.await?;
state_db.reset_memory_data_for_fresh_start().await?;
cleared_state_db = true;
}
let memory_root = config.codex_home.join("memories");
let removed_memory_root = match tokio::fs::remove_dir_all(&memory_root).await {
Ok(()) => true,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => false,
Err(err) => return Err(err.into()),
};
let mut message = if cleared_state_db {
format!("Cleared memory state from {}.", state_path.display())
} else {
format!("No state db found at {}.", state_path.display())
};
if removed_memory_root {
message.push_str(&format!(" Removed {}.", memory_root.display()));
} else {
message.push_str(&format!(
" No memory directory found at {}.",
memory_root.display()
));
}
println!("{message}");
Ok(())
}
/// Prepend root-level overrides so they have lower precedence than
/// CLI-specific ones specified after the subcommand (if any).
fn prepend_config_flags(
codex-rs/cli/tests/debug_clear_memories.rs
+141
View File
@@ -0,0 +1,141 @@
use std::path::Path;
use anyhow::Result;
use codex_state::StateRuntime;
use codex_state::state_db_path;
use predicates::str::contains;
use sqlx::SqlitePool;
use tempfile::TempDir;
fn codex_command(codex_home: &Path) -> Result<assert_cmd::Command> {
let mut cmd = assert_cmd::Command::new(codex_utils_cargo_bin::cargo_bin("codex")?);
cmd.env("CODEX_HOME", codex_home);
Ok(cmd)
}
#[tokio::test]
async fn debug_clear_memories_resets_state_and_removes_memory_dir() -> Result<()> {
let codex_home = TempDir::new()?;
let runtime = StateRuntime::init(
codex_home.path().to_path_buf(),
"test-provider".to_string(),
None,
)
.await?;
drop(runtime);
let thread_id = "00000000-0000-0000-0000-000000000123";
let db_path = state_db_path(codex_home.path());
let pool = SqlitePool::connect(&format!("sqlite://{}", db_path.display())).await?;
sqlx::query(
r#"
INSERT INTO threads (
id,
rollout_path,
created_at,
updated_at,
source,
agent_nickname,
agent_role,
model_provider,
cwd,
cli_version,
title,
sandbox_policy,
approval_mode,
tokens_used,
first_user_message,
archived,
archived_at,
git_sha,
git_branch,
git_origin_url,
memory_mode
) VALUES (?, ?, 1, 1, 'cli', NULL, NULL, 'test-provider', ?, '', '', 'read-only', 'on-request', 0, '', 0, NULL, NULL, NULL, NULL, 'enabled')
"#,
)
.bind(thread_id)
.bind(codex_home.path().join("session.jsonl").display().to_string())
.bind(codex_home.path().display().to_string())
.execute(&pool)
.await?;
sqlx::query(
r#"
INSERT INTO stage1_outputs (
thread_id,
source_updated_at,
raw_memory,
rollout_summary,
generated_at,
rollout_slug,
usage_count,
last_usage,
selected_for_phase2,
selected_for_phase2_source_updated_at
) VALUES (?, 1, 'raw', 'summary', 1, NULL, 0, NULL, 0, NULL)
"#,
)
.bind(thread_id)
.execute(&pool)
.await?;
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
('memory_stage1', ?, 'completed', NULL, NULL, NULL, NULL, NULL, NULL, 3, NULL, NULL, 1),
('memory_consolidate_global', 'global', 'completed', NULL, NULL, NULL, NULL, NULL, NULL, 3, NULL, NULL, 1)
"#,
)
.bind(thread_id)
.execute(&pool)
.await?;
let memory_root = codex_home.path().join("memories");
std::fs::create_dir_all(&memory_root)?;
std::fs::write(memory_root.join("memory_summary.md"), "stale memory")?;
drop(pool);
let mut cmd = codex_command(codex_home.path())?;
cmd.args(["debug", "clear-memories"])
.assert()
.success()
.stdout(contains("Cleared memory state"));
let pool = SqlitePool::connect(&format!("sqlite://{}", db_path.display())).await?;
let stage1_outputs_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stage1_outputs")
.fetch_one(&pool)
.await?;
assert_eq!(stage1_outputs_count, 0);
let memory_jobs_count: i64 = sqlx::query_scalar(
"SELECT COUNT(*) FROM jobs WHERE kind = 'memory_stage1' OR kind = 'memory_consolidate_global'",
)
.fetch_one(&pool)
.await?;
assert_eq!(memory_jobs_count, 0);
let memory_mode: String = sqlx::query_scalar("SELECT memory_mode FROM threads WHERE id = ?")
.bind(thread_id)
.fetch_one(&pool)
.await?;
assert_eq!(memory_mode, "disabled");
assert!(!memory_root.exists());
Ok(())
}
codex-rs/core/src/codex.rs
+1
View File
@@ -851,6 +851,7 @@ impl SessionConfiguration {
approval_policy: self.approval_policy.value(),
sandbox_policy: self.sandbox_policy.get().clone(),
cwd: self.cwd.clone(),
ephemeral: self.original_config_do_not_use.ephemeral,
reasoning_effort: self.collaboration_mode.reasoning_effort(),
personality: self.personality,
session_source: self.session_source.clone(),
codex-rs/core/src/codex_thread.rs
+1
View File
@@ -30,6 +30,7 @@ pub struct ThreadConfigSnapshot {
pub approval_policy: AskForApproval,
pub sandbox_policy: SandboxPolicy,
pub cwd: PathBuf,
pub ephemeral: bool,
pub reasoning_effort: Option<ReasoningEffort>,
pub personality: Option<Personality>,
pub session_source: SessionSource,
codex-rs/state/src/runtime/memories.rs
+136
View File
@@ -30,6 +30,19 @@ impl StateRuntime {
/// stage-1 (`memory_stage1`) and phase-2 (`memory_consolidate_global`)
/// memory pipelines.
pub async fn clear_memory_data(&self) -> anyhow::Result<()> {
self.clear_memory_data_inner(false).await
}
/// Resets persisted memory state for a clean-slate local start.
///
/// In addition to clearing persisted stage-1 outputs and memory pipeline
/// jobs, this disables memory generation for all existing threads so
/// historical rollouts are not immediately picked up again.
pub async fn reset_memory_data_for_fresh_start(&self) -> anyhow::Result<()> {
self.clear_memory_data_inner(true).await
}
async fn clear_memory_data_inner(&self, disable_existing_threads: bool) -> anyhow::Result<()> {
let mut tx = self.pool.begin().await?;
sqlx::query(
@@ -51,6 +64,18 @@ WHERE kind = ? OR kind = ?
.execute(&mut *tx)
.await?;
if disable_existing_threads {
sqlx::query(
r#"
UPDATE threads
SET memory_mode = 'disabled'
WHERE memory_mode = 'enabled'
"#,
)
.execute(&mut *tx)
.await?;
}
tx.commit().await?;
Ok(())
}
@@ -1158,6 +1183,8 @@ ON CONFLICT(kind, job_key) DO UPDATE SET
#[cfg(test)]
mod tests {
use super::JOB_KIND_MEMORY_CONSOLIDATE_GLOBAL;
use super::JOB_KIND_MEMORY_STAGE1;
use super::StateRuntime;
use super::test_support::test_thread_metadata;
use super::test_support::unique_temp_dir;
@@ -1664,6 +1691,115 @@ mod tests {
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
#[tokio::test]
async fn reset_memory_data_for_fresh_start_clears_rows_and_disables_threads() {
let codex_home = unique_temp_dir();
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string(), None)
.await
.expect("initialize runtime");
let now = Utc::now() - Duration::hours(13);
let worker_id = ThreadId::from_string(&Uuid::new_v4().to_string()).expect("worker id");
let enabled_thread_id =
ThreadId::from_string(&Uuid::new_v4().to_string()).expect("enabled thread id");
let disabled_thread_id =
ThreadId::from_string(&Uuid::new_v4().to_string()).expect("disabled thread id");
let mut enabled =
test_thread_metadata(&codex_home, enabled_thread_id, codex_home.join("enabled"));
enabled.created_at = now;
enabled.updated_at = now;
runtime
.upsert_thread(&enabled)
.await
.expect("upsert enabled thread");
let claim = runtime
.try_claim_stage1_job(
enabled_thread_id,
worker_id,
enabled.updated_at.timestamp(),
3600,
64,
)
.await
.expect("claim enabled thread");
let ownership_token = match claim {
Stage1JobClaimOutcome::Claimed { ownership_token } => ownership_token,
other => panic!("unexpected claim outcome: {other:?}"),
};
assert!(
runtime
.mark_stage1_job_succeeded(
enabled_thread_id,
ownership_token.as_str(),
enabled.updated_at.timestamp(),
"raw",
"summary",
None,
)
.await
.expect("mark enabled thread succeeded"),
"stage1 success should be recorded"
);
runtime
.enqueue_global_consolidation(enabled.updated_at.timestamp())
.await
.expect("enqueue global consolidation");
let mut disabled =
test_thread_metadata(&codex_home, disabled_thread_id, codex_home.join("disabled"));
disabled.created_at = now;
disabled.updated_at = now;
runtime
.upsert_thread(&disabled)
.await
.expect("upsert disabled thread");
sqlx::query("UPDATE threads SET memory_mode = 'disabled' WHERE id = ?")
.bind(disabled_thread_id.to_string())
.execute(runtime.pool.as_ref())
.await
.expect("disable existing thread");
runtime
.reset_memory_data_for_fresh_start()
.await
.expect("reset memory data");
let stage1_outputs_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stage1_outputs")
.fetch_one(runtime.pool.as_ref())
.await
.expect("count stage1 outputs");
assert_eq!(stage1_outputs_count, 0);
let memory_jobs_count: i64 =
sqlx::query_scalar("SELECT COUNT(*) FROM jobs WHERE kind = ? OR kind = ?")
.bind(JOB_KIND_MEMORY_STAGE1)
.bind(JOB_KIND_MEMORY_CONSOLIDATE_GLOBAL)
.fetch_one(runtime.pool.as_ref())
.await
.expect("count memory jobs");
assert_eq!(memory_jobs_count, 0);
let enabled_memory_mode: String =
sqlx::query_scalar("SELECT memory_mode FROM threads WHERE id = ?")
.bind(enabled_thread_id.to_string())
.fetch_one(runtime.pool.as_ref())
.await
.expect("read enabled thread memory mode");
assert_eq!(enabled_memory_mode, "disabled");
let disabled_memory_mode: String =
sqlx::query_scalar("SELECT memory_mode FROM threads WHERE id = ?")
.bind(disabled_thread_id.to_string())
.fetch_one(runtime.pool.as_ref())
.await
.expect("read disabled thread memory mode");
assert_eq!(disabled_memory_mode, "disabled");
let _ = tokio::fs::remove_dir_all(codex_home).await;
}
#[tokio::test]
async fn claim_stage1_jobs_enforces_global_running_cap() {
let codex_home = unique_temp_dir();