openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-08 20:40:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Test exec catalog preservation across session and compaction changes (#60173)

Browse Source 
## What changed

- Cover incremental exec tool settings and catalog changes across resume, fork, tool mode switches, and compaction.
- Verify that adopting background compaction preserves live exec declarations, tool metadata updates, and retained conversation order after a model catalog refresh.
- Include an `exec` namespace in the top-level tool catalog ordering and snapshot round-trip test.

GitOrigin-RevId: 3df603b98a006abb55a5095b299080f036576242
This commit is contained in:
Alex Kira
2026-11-07 21:44:47 +00:00
committed by copyberry
parent 202b590e84
commit aa9b9cc94e
3 changed files with 601 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/context/world_state/top_level_tools_tests.rs
+9 -1
View File
@@ -267,6 +267,7 @@ fn initial_catalog_preserves_order_and_unchanged_catalog_emits_nothing() {
let definitions = vec![
namespace("zebra", vec![declaration("lookup")]),
namespace("alpha", vec![declaration("lookup")]),
namespace("exec", vec![declaration("lookup")]),
];
let state = TopLevelToolsState::new(
/*messages*/ Default::default(),
@@ -293,7 +294,14 @@ fn initial_catalog_preserves_order_and_unchanged_catalog_emits_nothing() {
.keys()
.map(String::as_str)
.collect::<Vec<_>>(),
vec!["alpha", "alpha.lookup", "zebra", "zebra.lookup"]
vec![
"alpha",
"alpha.lookup",
"exec",
"exec.lookup",
"zebra",
"zebra.lookup"
]
);
let snapshot: TopLevelToolsSnapshot = serde_json::from_value(saved).unwrap();
let (unchanged_snapshot, updates) = state.render_diff(PreviousSectionState::Known(&snapshot));
codex-rs/core/tests/suite/scenarios.rs
+292
View File
@@ -4026,6 +4026,298 @@ async fn astra_async_compaction_keeps_work_added_after_its_snapshot(
Ok(())
}
#[derive(Clone, Copy)]
enum ExecCatalogRefresh {
Hybrid,
Metadata,
}
#[test_case::test_case(ExecCatalogRefresh::Hybrid; "hybrid mode")]
#[test_case::test_case(ExecCatalogRefresh::Metadata; "exec metadata refresh")]
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn astra_async_compaction_preserves_exec_catalog(change: ExecCatalogRefresh) -> Result<()> {
use codex_models_manager::manager::RefreshStrategy;
use core_test_support::responses::mount_models_once_with_etag;
skip_if_no_network!(Ok(()));
let (release_background, background_gate) = oneshot::channel();
let (release_foreground, foreground_gate) = oneshot::channel();
let (streaming, _completions) = start_streaming_sse_server(vec![
vec![compaction_chunk(
/*gate*/ None,
vec![
ev_assistant_message("draft", "I drafted the launch notes."),
ev_completed_with_tokens("draft-response", /*total_tokens*/ 70_000),
],
)],
vec![compaction_chunk(
Some(background_gate),
vec![
compaction_output("ASYNC_LAUNCH_CHECKPOINT"),
ev_completed_with_tokens("compact-response", /*total_tokens*/ 3_000),
],
)],
vec![
compaction_chunk(
/*gate*/ None,
vec![
ev_response_created("team-response"),
ev_assistant_message("team-check", "I will check who is available."),
],
),
compaction_chunk(
Some(foreground_gate),
vec![
ev_function_call_with_namespace("team", "collaboration", "list_agents", "{}"),
ev_completed_with_tokens("team-response", /*total_tokens*/ 75_000),
],
),
],
vec![compaction_chunk(
/*gate*/ None,
vec![
ev_assistant_message("answer", "I can finish the notes myself."),
ev_completed_with_tokens("answer-response", /*total_tokens*/ 3_000),
],
)],
])
.await;
let model_server = wiremock::MockServer::start().await;
let mut catalog = bundled_models_response()?;
let model = catalog
.models
.iter_mut()
.find(|model| model.slug == "gpt-6-astra")
.expect("bundled Astra model");
model.tool_mode = Some(ToolMode::CodeModeOnly);
model.supports_image_detail_original = false;
mount_models_once_with_etag(&model_server, catalog.clone(), "\"exec-mode-v1\"").await;
let model_catalog_url = format!("{}/models", model_server.uri());
let (_config_server, test) = async_compaction_test(
test_codex(),
streaming.uri(),
vec![
Feature::IncrementalTools,
Feature::IncrementalExecTools,
Feature::ViewImage,
],
move |config| {
config
.features
.disable(Feature::UnifiedImageBudget)
.expect("use the model's image detail capability");
config.model_catalog = None;
config.model_provider.model_catalog_url = Some(model_catalog_url.clone().into());
config.code_mode.disable_in_process_fallback = true;
},
)
.await?;
test.codex.ensure_rollout_materialized().await;
test.submit_text_turn("Draft the launch notes.").await?;
let timeout = Duration::from_secs(/*secs*/ 20);
tokio::time::timeout(timeout, streaming.wait_for_request_count(/*count*/ 2)).await?;
// Refresh the same model while the prepared result is still pending.
let model = catalog
.models
.iter_mut()
.find(|model| model.slug == "gpt-6-astra")
.expect("bundled Astra model");
match change {
ExecCatalogRefresh::Hybrid => model.tool_mode = Some(ToolMode::CodeMode),
ExecCatalogRefresh::Metadata => {
model.supports_image_detail_original = true;
let messages = model
.model_messages
.get_or_insert_default()
.tools
.get_or_insert_default()
.code_mode
.get_or_insert_default();
messages.exec.get_or_insert_default().description =
Some("Previous exec instructions.".to_string());
messages.mcp_typescript_preamble = Some("type PreviousResult = string;".to_string());
}
}
mount_models_once_with_etag(&model_server, catalog.clone(), "\"exec-mode-v2\"").await;
test.thread_manager
.get_models_manager()
.list_models(
RefreshStrategy::Online,
codex_core::test_support::default_http_client_factory(),
)
.await;
test.codex
.start_or_steer_turn(TurnInputRequest::user_input(vec![text(
"Check who can finish the notes.",
)]))
.await?;
tokio::time::timeout(timeout, streaming.wait_for_request_count(/*count*/ 3)).await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::ItemCompleted(event)
if matches!(&event.item, TurnItem::AgentMessage(message) if message.id == "team-check"))
})
.await;
release_background
.send(())
.expect("release background compaction");
// Keep foreground work gated until the prepared result is available for optional adoption.
tokio::time::timeout(
timeout,
test.codex.wait_for_background_compaction_for_test(),
)
.await?;
release_foreground
.send(())
.expect("release foreground work");
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
tokio::time::timeout(
timeout,
test.codex.wait_for_background_compaction_for_test(),
)
.await?;
let requests = streaming
.requests()
.await
.iter()
.map(|body| serde_json::from_slice::<serde_json::Value>(body))
.collect::<serde_json::Result<Vec<_>>>()?;
assert_eq!(requests.len(), 4);
let history = test
.codex
.load_history(/*include_archived*/ false)
.await?
.items;
let input = requests[3]["input"].as_array().expect("request input");
let compaction = input
.iter()
.position(|item| item["encrypted_content"] == "ASYNC_LAUNCH_CHECKPOINT")
.expect("prepared summary was adopted");
let user = input
.iter()
.position(|item| {
item["role"] == "user"
&& item["content"][0]["text"] == "Check who can finish the notes."
})
.expect("retained user message");
let assistant = input
.iter()
.position(|item| {
item["role"] == "assistant"
&& item["content"][0]["text"] == "I will check who is available."
})
.expect("retained assistant message");
let call = input
.iter()
.position(|item| item["type"] == "function_call" && item["call_id"] == "team")
.expect("retained tool call");
let output = input
.iter()
.position(|item| item["type"] == "function_call_output" && item["call_id"] == "team")
.expect("retained tool output");
assert!(user < assistant && assistant < call && call < output);
let [before_adoption, after_adoption] = [&requests[2], &requests[3]].map(|request| {
request["input"]
.as_array()
.expect("request input")
.iter()
.filter(|item| item["type"] == "additional_tools")
.map(|item| item["tools"].clone())
.collect::<Vec<_>>()
});
assert_eq!(after_adoption, before_adoption);
let (exec_positions, exec_declarations): (Vec<_>, Vec<_>) = input
.iter()
.enumerate()
.filter_map(|(index, item)| {
core_test_support::responses::namespace_child_tool(item, "functions", "exec")
.map(|declaration| (index, declaration))
})
.unzip();
let (adoption_index, adopted) = history
.windows(2)
.enumerate()
.find_map(|(index, pair)| match pair {
[RolloutItem::Compacted(_), RolloutItem::WorldState(baseline)] if baseline.full => {
Some((index, &baseline.state["top_level_tools"]))
}
_ => None,
})
.expect("adopted full checkpoint");
let changed_hashes: &[&str] = match change {
ExecCatalogRefresh::Hybrid => &["/functions.exec"],
ExecCatalogRefresh::Metadata => &["/tools/functions.exec", "/exec/Tool: tools.view_image"],
};
for path in changed_hashes {
let live = history[..adoption_index]
.iter()
.rev()
.find_map(|item| match item {
RolloutItem::WorldState(patch) if !patch.full => {
patch.state.get("top_level_tools")?.pointer(path)
}
_ => None,
})
.and_then(serde_json::Value::as_str)
.expect("live exec update before adoption");
assert_eq!(
adopted.pointer(path).and_then(serde_json::Value::as_str),
Some(live),
"{path}"
);
}
if matches!(change, ExecCatalogRefresh::Metadata) {
let initial_exec = requests[0]["input"]
.as_array()
.expect("initial input")
.iter()
.find_map(|item| {
core_test_support::responses::namespace_child_tool(item, "functions", "exec")
})
.expect("initial exec declaration");
assert_eq!(exec_declarations, vec![initial_exec]);
assert!(exec_positions[0] < compaction && compaction < user);
let is_view_image_update = |item: &&serde_json::Value| {
item["role"] == "developer"
&& item["content"][0]["text"]
.as_str()
.is_some_and(|text| text.lines().any(|line| line == "### `view_image`"))
};
let live_update = requests[2]["input"]
.as_array()
.expect("input before adoption")
.iter()
.find(is_view_image_update)
.expect("view_image redefinition during background compaction");
assert!(
live_update["content"][0]["text"]
.as_str()
.expect("view_image update text")
.contains("detail?: \"high\" | \"original\"")
);
assert_eq!(
input
.iter()
.filter(is_view_image_update)
.collect::<Vec<_>>(),
vec![live_update]
);
assert!(input[compaction + 1..].contains(live_update));
return Ok(());
}
assert_eq!(exec_declarations.len(), 2);
assert_ne!(exec_declarations[0], exec_declarations[1]);
assert!(
exec_positions[0] < compaction
&& compaction < exec_positions[1]
&& exec_positions[1] < user
);
Ok(())
}
#[derive(PartialEq)]
enum PendingCompactionOutcome {
Absent,
codex-rs/core/tests/suite/scenarios_incremental_tools_resume.rs
+300 -1
View File
@@ -1,11 +1,15 @@
//! Request-history coverage for adopting incremental tools when initial context is rebuilt.
//! Request-history coverage for incremental tool modes across resume, fork, and window rebuilds.
use anyhow::Result;
use codex_core::ForkSnapshot;
use codex_core::StartThreadOptions;
use codex_core::TurnInputRequest;
use codex_features::Feature;
use codex_login::CodexAuth;
use codex_protocol::openai_models::ToolMode;
use codex_protocol::protocol::EventMsg;
use codex_protocol::protocol::Op;
use codex_protocol::user_input::UserInput;
use core_test_support::context_snapshot;
use core_test_support::context_snapshot::ContextSnapshotOptions;
use core_test_support::responses;
@@ -13,6 +17,8 @@ use core_test_support::skip_if_no_network;
use core_test_support::test_codex::TestCodexBuilder;
use core_test_support::test_codex::test_codex;
use core_test_support::wait_for_event;
use core_test_support::wait_for_successful_turn;
use pretty_assertions::assert_eq;
use serde_json::json;
use test_case::test_case;
@@ -215,3 +221,296 @@ async fn disable_incremental_tools_at_next_window() -> Result<()> {
);
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn exec_tools_follow_current_settings_across_history() -> Result<()> {
skip_if_no_network!(Ok(()));
let server = responses::start_mock_server().await;
let replies = (0..13)
.map(|index| {
let mut events = Vec::new();
if index == 9 {
events.push(json!({"type": "response.output_item.done", "item": {"type": "compaction", "encrypted_content": "COMPACTED_HISTORY"}}));
}
events.push(responses::ev_completed(&format!("lifecycle-{index}")));
responses::sse(events)
})
.collect();
let mock = responses::mount_sse_sequence(&server, replies).await;
let session_builder = |enabled, mode, planning, changed_catalog| {
builder(/*incremental_tools*/ true, /*token_budget*/ false)
.with_model_info_override("gpt-5.4", move |model| {
model.use_responses_lite = true;
model.tool_mode = Some(mode);
})
.with_config(move |config| {
config
.features
.set_enabled(Feature::IncrementalExecTools, enabled)
.expect("set incremental exec tools");
config.update_plan_enabled = planning;
if changed_catalog {
config.features.disable(Feature::ShellTool).unwrap();
}
})
};
let initial = session_builder(
/*enabled*/ true,
ToolMode::CodeModeOnly,
/*planning*/ false,
/*changed_catalog*/ false,
)
.build_with_auto_env(&server)
.await?;
initial
.submit_turn("Begin with an incremental exec catalog.")
.await?;
let before_updates = session_builder(
/*enabled*/ false,
ToolMode::CodeModeOnly,
/*planning*/ false,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &initial)
.await?;
before_updates
.submit_turn("Disable granular updates before any catalog change.")
.await?;
let enabled = session_builder(
/*enabled*/ true,
ToolMode::CodeModeOnly,
/*planning*/ false,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &before_updates)
.await?;
enabled
.submit_turn("Re-enable granular updates with the same catalog.")
.await?;
let mixed = session_builder(
/*enabled*/ true,
ToolMode::CodeModeOnly,
/*planning*/ true,
/*changed_catalog*/ true,
)
.restart_with_auto_env(&server, &enabled)
.await?;
mixed
.submit_turn("Add planning and remove shell tools.")
.await?;
let disabled_after_changes = session_builder(
/*enabled*/ false,
ToolMode::CodeModeOnly,
/*planning*/ true,
/*changed_catalog*/ true,
)
.restart_with_auto_env(&server, &mixed)
.await?;
disabled_after_changes
.submit_turn("Disable granular updates while retaining those catalog changes.")
.await?;
let hybrid = session_builder(
/*enabled*/ true,
ToolMode::CodeMode,
/*planning*/ false,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &disabled_after_changes)
.await?;
hybrid
.submit_turn("Resume in hybrid mode with the catalog flag enabled.")
.await?;
let returned = session_builder(
/*enabled*/ false,
ToolMode::CodeModeOnly,
/*planning*/ false,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &hybrid)
.await?;
returned
.submit_turn("Return to CodeModeOnly with planning still removed.")
.await?;
returned.codex.flush_rollout().await?;
let mut fork_config = returned.config.clone();
fork_config.update_plan_enabled = true;
let forked = returned
.thread_manager
.fork_legacy_thread(
ForkSnapshot::Interrupted,
StartThreadOptions::new(fork_config),
returned
.codex
.rollout_path()
.expect("persisted parent rollout"),
)
.await?;
forked
.thread
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
text: "Fork with planning enabled and granular updates disabled.".to_string(),
text_elements: Vec::new(),
}]))
.await?;
wait_for_successful_turn(&forked.thread).await;
forked.thread.shutdown_and_wait().await?;
let disabled = session_builder(
/*enabled*/ false,
ToolMode::CodeModeOnly,
/*planning*/ true,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &returned)
.await?;
disabled
.submit_turn("Resume with planning enabled and granular updates disabled.")
.await?;
disabled.codex.submit(Op::Compact).await?;
wait_for_successful_turn(&disabled.codex).await;
disabled
.submit_turn("Rebuild the complete catalog with granular updates disabled.")
.await?;
let enabled = session_builder(
/*enabled*/ true,
ToolMode::CodeModeOnly,
/*planning*/ false,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &disabled)
.await?;
enabled
.submit_turn("Enable granular updates and remove planning.")
.await?;
let resumed = session_builder(
/*enabled*/ true,
ToolMode::CodeModeOnly,
/*planning*/ true,
/*changed_catalog*/ false,
)
.restart_with_auto_env(&server, &enabled)
.await?;
resumed
.submit_turn("Resume with planning enabled and granular updates enabled.")
.await?;
let requests = mock.requests();
let declarations = requests
.iter()
.map(|request| {
request
.inputs_of_type("additional_tools")
.iter()
.filter_map(|item| responses::namespace_child_tool(item, "functions", "exec"))
.cloned()
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
assert_eq!(
declarations.iter().map(Vec::len).collect::<Vec<_>>(),
vec![
1, // Initial catalog.
1, // Disabling before any delta keeps the existing declaration.
2, // Enabling initializes a full catalog and its comparison state.
2, // Added and removed entries use developer messages.
2, // Disabling keeps those messages without a forced declaration.
3, // Hybrid mode uses ordinary declarations even with the flag enabled.
4, // Returning with the flag disabled uses a whole declaration.
5, // Fork adds planning through a whole declaration.
5, // Resume adds planning through the same whole declaration.
5, // Compaction samples the retained history.
1, // Full replacement uses the current disabled flag.
2, // Enabling initializes a full catalog before further deltas.
2, // Resume adds planning in a developer message.
]
);
for (before, after) in [(0, 1), (3, 4)] {
assert_eq!(declarations[before], declarations[after]);
assert!(
requests[after]
.input()
.starts_with(&requests[before].input())
);
}
assert_eq!(
declarations[0].last(),
declarations[6].last(),
"returning from hybrid restores the complete current catalog"
);
assert_eq!(
declarations[7].last(),
declarations[8].last(),
"fork and resume publish the same complete catalog while granular updates are disabled"
);
let has_update_plan = [6, 8].map(|index| {
declarations[index].last().expect("full exec declaration")["description"]
.as_str()
.expect("exec description")
.lines()
.any(|line| line == "### `update_plan`")
});
assert_eq!(has_update_plan, [false, true]);
assert_eq!(
declarations[10].last(),
declarations[8].last(),
"compaction while disabled preserves the complete current catalog"
);
assert_eq!(declarations[11].last(), declarations[0].last());
assert_eq!(declarations[12], declarations[11]);
let exec_definition_updates = requests
.iter()
.map(|request| {
request
.message_input_texts("developer")
.into_iter()
.filter(|text| {
text.starts_with(
"Use these definitions and namespace instructions with `functions.exec`.",
)
})
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
assert_eq!(
exec_definition_updates
.iter()
.map(Vec::len)
.collect::<Vec<_>>(),
vec![0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 1]
);
for index in [3, 12] {
let update = &exec_definition_updates[index][0];
assert_eq!(
update
.lines()
.filter(|line| *line != "## Namespace: functions")
.filter(|line| line.starts_with("## ") || line.starts_with("### "))
.collect::<Vec<_>>(),
vec!["### `update_plan`"]
);
}
assert_eq!(
exec_definition_updates[3], exec_definition_updates[4],
"disabling granular updates retains the previously published definitions"
);
assert_eq!(
exec_definition_updates[8], exec_definition_updates[9],
"compaction receives retained definitions without repeating them"
);
let removal_notices = requests
.iter()
.map(|request| {
request
.message_input_texts("developer")
.into_iter()
.filter(|text| text.contains("- Tool: tools.exec_command"))
.collect::<Vec<_>>()
})
.collect::<Vec<_>>();
assert_eq!(removal_notices[3].len(), 1);
assert_eq!(removal_notices[3], removal_notices[4]);
assert!(removal_notices[10].is_empty());
assert!(removal_notices[12].is_empty());
Ok(())
}