mirror of
https://github.com/openai/codex.git
synced 2026-11-08 08:37:30 +00:00
Distinguish forked sessions in session-start hooks (#44349)
## Why Forked threads reported `startup`, causing startup hooks to run again even when their context was inherited from the parent. Resuming with supplied history also reported `startup` instead of `resume`. ## What changed - Add `fork` as a `SessionStart` source and expose it in the hook input schema. - Report `fork` for histories with a fork parent and `resume` for supplied histories without one. - Continue dispatching `SubagentStart` for spawned subagents with either fresh or forked context. ## Testing Add regression coverage for fork hook matching without duplicate startup context, resume hooks with supplied history, and `SubagentStart` context injection for fresh and forked subagents. GitOrigin-RevId: 5ac1ae4ab8adcbbbceb8cb0ffb9b32b6402c585b
This commit is contained in:
1 parent
c6a59ef923
commit
e444aa99d7
7 files changed
+197
-16
No files matched your search
@@ -127,14 +127,14 @@ pub(crate) async fn run_pending_session_start_hooks(
|
|||||||
turn_context: &Arc<TurnContext>,
|
turn_context: &Arc<TurnContext>,
|
||||||
) -> bool {
|
) -> bool {
|
||||||
while let Some(session_start_source) = sess.take_pending_session_start_source().await {
|
while let Some(session_start_source) = sess.take_pending_session_start_source().await {
|
||||||
// Pending session-start hooks are reused to dispatch thread-spawn subagent
|
// Spawned subagents can start fresh or fork their parent's history, so both
|
||||||
// starts. Other subagent sessions are internal/system work and do not run
|
// sources dispatch SubagentStart. Internal/system subagents skip start hooks.
|
||||||
// start hooks.
|
|
||||||
let target = match &turn_context.session_source {
|
let target = match &turn_context.session_source {
|
||||||
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_role, .. })
|
SessionSource::SubAgent(SubAgentSource::ThreadSpawn { agent_role, .. })
|
||||||
if matches!(
|
if matches!(
|
||||||
session_start_source,
|
session_start_source,
|
||||||
codex_hooks::SessionStartSource::Startup
|
codex_hooks::SessionStartSource::Startup
|
||||||
|
| codex_hooks::SessionStartSource::Fork
|
||||||
) =>
|
) =>
|
||||||
{
|
{
|
||||||
let context = subagent_hook_context(sess, agent_role);
|
let context = subagent_hook_context(sess, agent_role);
|
||||||
|
|||||||
@@ -1682,10 +1682,15 @@ impl Session {
|
|||||||
sess.schedule_startup_prewarm(sess.get_prompt_base_instructions().await.text)
|
sess.schedule_startup_prewarm(sess.get_prompt_base_instructions().await.text)
|
||||||
.await;
|
.await;
|
||||||
let session_start_source = match &initial_history {
|
let session_start_source = match &initial_history {
|
||||||
InitialHistory::Resumed(_) => codex_hooks::SessionStartSource::Resume,
|
InitialHistory::Forked(_) if forked_from_id.is_some() => {
|
||||||
InitialHistory::New | InitialHistory::Forked(_) => {
|
codex_hooks::SessionStartSource::Fork
|
||||||
codex_hooks::SessionStartSource::Startup
|
|
||||||
}
|
}
|
||||||
|
// `thread/resume` with supplied history uses `Forked` internally
|
||||||
|
// without a fork parent, so it should still report `resume`.
|
||||||
|
InitialHistory::Resumed(_) | InitialHistory::Forked(_) => {
|
||||||
|
codex_hooks::SessionStartSource::Resume
|
||||||
|
}
|
||||||
|
InitialHistory::New => codex_hooks::SessionStartSource::Startup,
|
||||||
InitialHistory::Cleared => codex_hooks::SessionStartSource::Clear,
|
InitialHistory::Cleared => codex_hooks::SessionStartSource::Clear,
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ use anyhow::Result;
|
|||||||
use codex_config::HookStateToml;
|
use codex_config::HookStateToml;
|
||||||
use codex_config::McpServerConfig;
|
use codex_config::McpServerConfig;
|
||||||
use codex_config::test_support::CloudConfigBundleFixture;
|
use codex_config::test_support::CloudConfigBundleFixture;
|
||||||
|
use codex_core::ForkSnapshot;
|
||||||
use codex_core::StartThreadOptions;
|
use codex_core::StartThreadOptions;
|
||||||
use codex_core::TurnInput;
|
use codex_core::TurnInput;
|
||||||
use codex_core::TurnInputRequest;
|
use codex_core::TurnInputRequest;
|
||||||
@@ -15,12 +16,14 @@ use codex_core::config::Config;
|
|||||||
use codex_core::config::Constrained;
|
use codex_core::config::Constrained;
|
||||||
use codex_core::config::ThreadStoreConfig;
|
use codex_core::config::ThreadStoreConfig;
|
||||||
use codex_features::Feature;
|
use codex_features::Feature;
|
||||||
|
use codex_history::InitialHistory;
|
||||||
use codex_history::RolloutItem;
|
use codex_history::RolloutItem;
|
||||||
use codex_model_provider_info::ModelProviderInfo;
|
use codex_model_provider_info::ModelProviderInfo;
|
||||||
use codex_model_provider_info::built_in_model_providers;
|
use codex_model_provider_info::built_in_model_providers;
|
||||||
use codex_plugin::PluginHookSource;
|
use codex_plugin::PluginHookSource;
|
||||||
use codex_plugin::PluginId;
|
use codex_plugin::PluginId;
|
||||||
use codex_protocol::items::parse_hook_prompt_fragment;
|
use codex_protocol::items::parse_hook_prompt_fragment;
|
||||||
|
use codex_protocol::mcp::ClientMcpExtensions;
|
||||||
use codex_protocol::models::ContentItem;
|
use codex_protocol::models::ContentItem;
|
||||||
use codex_protocol::models::NetworkPermissions;
|
use codex_protocol::models::NetworkPermissions;
|
||||||
use codex_protocol::models::PermissionProfile;
|
use codex_protocol::models::PermissionProfile;
|
||||||
@@ -1612,6 +1615,159 @@ async fn session_start_runs_before_user_prompt_submit_on_first_turn() -> Result<
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn forked_thread_matches_fork_session_start_without_repeating_startup_context() -> Result<()>
|
||||||
|
{
|
||||||
|
skip_if_no_network!(Ok(()));
|
||||||
|
|
||||||
|
let server = start_mock_server().await;
|
||||||
|
let responses = mount_sse_sequence(
|
||||||
|
&server,
|
||||||
|
vec![
|
||||||
|
sse(vec![ev_completed("resp-1")]),
|
||||||
|
sse(vec![ev_completed("resp-2")]),
|
||||||
|
sse(vec![ev_completed("resp-fork")]),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
.await;
|
||||||
|
let mut builder = test_codex()
|
||||||
|
.with_pre_build_hook(|home| {
|
||||||
|
let script_path = home.join("session_start_hook.py");
|
||||||
|
fs::write(
|
||||||
|
&script_path,
|
||||||
|
r#"import json
|
||||||
|
import sys
|
||||||
|
|
||||||
|
source = json.load(sys.stdin)["source"]
|
||||||
|
print(json.dumps({"hookSpecificOutput": {
|
||||||
|
"hookEventName": "SessionStart",
|
||||||
|
"additionalContext": source + " hook context"
|
||||||
|
}}))
|
||||||
|
"#,
|
||||||
|
)
|
||||||
|
.expect("write session start hook");
|
||||||
|
let groups = ["^startup$", "^fork$"].map(|matcher| {
|
||||||
|
serde_json::json!({
|
||||||
|
"matcher": matcher,
|
||||||
|
"hooks": [{
|
||||||
|
"type": "command",
|
||||||
|
"command": format!("python3 {}", script_path.display()),
|
||||||
|
}],
|
||||||
|
})
|
||||||
|
});
|
||||||
|
fs::write(
|
||||||
|
home.join("hooks.json"),
|
||||||
|
serde_json::json!({"hooks": {"SessionStart": groups}}).to_string(),
|
||||||
|
)
|
||||||
|
.expect("write hooks.json");
|
||||||
|
})
|
||||||
|
.with_config(trust_discovered_hooks);
|
||||||
|
// Command hooks run on the host and require a host-native working directory.
|
||||||
|
let test = builder.build(&server).await?;
|
||||||
|
test.submit_turn("first prompt").await?;
|
||||||
|
test.submit_turn("second prompt").await?;
|
||||||
|
test.codex.flush_rollout().await?;
|
||||||
|
|
||||||
|
let forked = test
|
||||||
|
.thread_manager
|
||||||
|
.fork_thread(
|
||||||
|
ForkSnapshot::TruncateBeforeNthUserMessage(1),
|
||||||
|
test.config.clone(),
|
||||||
|
test.codex.rollout_path().expect("parent rollout path"),
|
||||||
|
/*thread_source*/ None,
|
||||||
|
/*parent_trace*/ None,
|
||||||
|
)
|
||||||
|
.await?
|
||||||
|
.thread;
|
||||||
|
forked
|
||||||
|
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
|
||||||
|
text: "edited second prompt".to_string(),
|
||||||
|
text_elements: Vec::new(),
|
||||||
|
}]))
|
||||||
|
.await?;
|
||||||
|
wait_for_event(&forked, |event| matches!(event, EventMsg::TurnComplete(_))).await;
|
||||||
|
|
||||||
|
let requests = responses.requests();
|
||||||
|
assert_eq!(requests.len(), 3);
|
||||||
|
let hook_contexts = requests[2]
|
||||||
|
.message_input_texts("developer")
|
||||||
|
.into_iter()
|
||||||
|
.filter(|message| {
|
||||||
|
matches!(
|
||||||
|
message.as_str(),
|
||||||
|
"startup hook context" | "fork hook context"
|
||||||
|
)
|
||||||
|
})
|
||||||
|
.collect::<Vec<_>>();
|
||||||
|
assert_eq!(
|
||||||
|
hook_contexts,
|
||||||
|
vec!["startup hook context", "fork hook context"]
|
||||||
|
);
|
||||||
|
|
||||||
|
forked.shutdown_and_wait().await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn explicit_history_runs_resume_session_start_hook() -> Result<()> {
|
||||||
|
skip_if_no_network!(Ok(()));
|
||||||
|
|
||||||
|
let server = start_mock_server().await;
|
||||||
|
let response = mount_sse_once(&server, sse(vec![ev_completed("resp-resume")])).await;
|
||||||
|
let mut builder = test_codex()
|
||||||
|
.with_pre_build_hook(|home| {
|
||||||
|
write_resume_and_compact_session_start_hook_with_context(
|
||||||
|
home,
|
||||||
|
"resume hook context",
|
||||||
|
"compact hook context",
|
||||||
|
)
|
||||||
|
.expect("write resume session start hook");
|
||||||
|
})
|
||||||
|
.with_config(trust_discovered_hooks);
|
||||||
|
// Command hooks run on the host and require a host-native working directory.
|
||||||
|
let test = builder.build(&server).await?;
|
||||||
|
let history = InitialHistory::Forked(vec![RolloutItem::ResponseItem(
|
||||||
|
responses::user_message_item("supplied history").into(),
|
||||||
|
)]);
|
||||||
|
let resumed = test
|
||||||
|
.thread_manager
|
||||||
|
.resume_thread_with_history(
|
||||||
|
test.config.clone(),
|
||||||
|
history,
|
||||||
|
test.thread_manager.auth_manager(),
|
||||||
|
/*parent_trace*/ None,
|
||||||
|
ClientMcpExtensions::default(),
|
||||||
|
)
|
||||||
|
.await?
|
||||||
|
.thread;
|
||||||
|
resumed
|
||||||
|
.start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
|
||||||
|
text: "continue".to_string(),
|
||||||
|
text_elements: Vec::new(),
|
||||||
|
}]))
|
||||||
|
.await?;
|
||||||
|
wait_for_event(&resumed, |event| matches!(event, EventMsg::TurnComplete(_))).await;
|
||||||
|
|
||||||
|
let hook_inputs = read_session_start_hook_inputs(test.codex_home_path())?;
|
||||||
|
assert_eq!(
|
||||||
|
hook_inputs
|
||||||
|
.iter()
|
||||||
|
.filter_map(|input| input.get("source").and_then(Value::as_str))
|
||||||
|
.collect::<Vec<_>>(),
|
||||||
|
vec!["resume"],
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
response
|
||||||
|
.single_request()
|
||||||
|
.message_input_texts("developer")
|
||||||
|
.iter()
|
||||||
|
.any(|message| message == "resume hook context"),
|
||||||
|
);
|
||||||
|
|
||||||
|
resumed.shutdown_and_wait().await?;
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn async_hook_context_is_injected_into_the_active_turn() -> Result<()> {
|
async fn async_hook_context_is_injected_into_the_active_turn() -> Result<()> {
|
||||||
skip_if_no_network!(Ok(()));
|
skip_if_no_network!(Ok(()));
|
||||||
|
|||||||
@@ -279,7 +279,7 @@ print(json.dumps({{"systemMessage": "root stop complete"}}))
|
|||||||
}]
|
}]
|
||||||
}],
|
}],
|
||||||
"SubagentStart": [{
|
"SubagentStart": [{
|
||||||
"matcher": "worker",
|
"matcher": "worker|default",
|
||||||
"hooks": [{
|
"hooks": [{
|
||||||
"type": "command",
|
"type": "command",
|
||||||
"command": format!("python3 {}", start_script_path.display()),
|
"command": format!("python3 {}", start_script_path.display()),
|
||||||
@@ -607,20 +607,29 @@ async fn spawned_agent_uses_multi_agent_reasoning_effort_for_requests(
|
|||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test_case(false; "fresh context")]
|
||||||
|
#[test_case(true; "forked context")]
|
||||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||||
async fn subagent_start_replaces_session_start_and_injects_context() -> Result<()> {
|
async fn subagent_start_replaces_session_start_and_injects_context(
|
||||||
|
fork_context: bool,
|
||||||
|
) -> Result<()> {
|
||||||
skip_if_no_network!(Ok(()));
|
skip_if_no_network!(Ok(()));
|
||||||
|
|
||||||
let server = start_mock_server().await;
|
let server = start_mock_server().await;
|
||||||
|
let agent_type = (!fork_context).then_some("worker");
|
||||||
|
let expected_agent_type = agent_type.unwrap_or("default");
|
||||||
let spawn_args = serde_json::to_string(&json!({
|
let spawn_args = serde_json::to_string(&json!({
|
||||||
"message": CHILD_PROMPT,
|
"message": CHILD_PROMPT,
|
||||||
"task_name": "child",
|
"task_name": "child",
|
||||||
"agent_type": "worker",
|
"agent_type": agent_type,
|
||||||
|
"fork_context": fork_context,
|
||||||
}))?;
|
}))?;
|
||||||
|
|
||||||
mount_sse_once_match(
|
mount_sse_once_match(
|
||||||
&server,
|
&server,
|
||||||
|req: &wiremock::Request| body_contains(req, TURN_1_PROMPT),
|
|req: &wiremock::Request| {
|
||||||
|
body_contains(req, TURN_1_PROMPT) && !body_contains(req, SPAWN_CALL_ID)
|
||||||
|
},
|
||||||
sse(vec![
|
sse(vec![
|
||||||
ev_response_created("resp-turn1-1"),
|
ev_response_created("resp-turn1-1"),
|
||||||
ev_function_call_with_namespace(
|
ev_function_call_with_namespace(
|
||||||
@@ -640,7 +649,6 @@ async fn subagent_start_replaces_session_start_and_injects_context() -> Result<(
|
|||||||
body_contains(req, CHILD_PROMPT)
|
body_contains(req, CHILD_PROMPT)
|
||||||
&& body_contains(req, SUBAGENT_START_CONTEXT)
|
&& body_contains(req, SUBAGENT_START_CONTEXT)
|
||||||
&& !body_contains(req, "<subagent_notification>")
|
&& !body_contains(req, "<subagent_notification>")
|
||||||
&& !body_contains(req, SPAWN_CALL_ID)
|
|
||||||
},
|
},
|
||||||
sse(vec![
|
sse(vec![
|
||||||
ev_response_created("resp-child-1"),
|
ev_response_created("resp-child-1"),
|
||||||
@@ -652,7 +660,9 @@ async fn subagent_start_replaces_session_start_and_injects_context() -> Result<(
|
|||||||
|
|
||||||
let _turn1_followup = mount_sse_once_match(
|
let _turn1_followup = mount_sse_once_match(
|
||||||
&server,
|
&server,
|
||||||
|req: &wiremock::Request| body_contains(req, SPAWN_CALL_ID),
|
|req: &wiremock::Request| {
|
||||||
|
body_contains(req, SPAWN_CALL_ID) && !body_contains(req, SUBAGENT_START_CONTEXT)
|
||||||
|
},
|
||||||
sse(vec![
|
sse(vec![
|
||||||
ev_response_created("resp-turn1-2"),
|
ev_response_created("resp-turn1-2"),
|
||||||
ev_assistant_message("msg-turn1-2", "parent done"),
|
ev_assistant_message("msg-turn1-2", "parent done"),
|
||||||
@@ -673,6 +683,7 @@ async fn subagent_start_replaces_session_start_and_injects_context() -> Result<(
|
|||||||
.enable(Feature::Collab)
|
.enable(Feature::Collab)
|
||||||
.expect("test config should allow feature update");
|
.expect("test config should allow feature update");
|
||||||
})
|
})
|
||||||
|
// Command hooks run on the host and require a host-native working directory.
|
||||||
.build(&server)
|
.build(&server)
|
||||||
.await?;
|
.await?;
|
||||||
|
|
||||||
@@ -686,7 +697,10 @@ async fn subagent_start_replaces_session_start_and_injects_context() -> Result<(
|
|||||||
)
|
)
|
||||||
.await?;
|
.await?;
|
||||||
assert_eq!(start_inputs.len(), 1);
|
assert_eq!(start_inputs.len(), 1);
|
||||||
assert_eq!(start_inputs[0]["agent_type"].as_str(), Some("worker"));
|
assert_eq!(
|
||||||
|
start_inputs[0]["agent_type"].as_str(),
|
||||||
|
Some(expected_agent_type)
|
||||||
|
);
|
||||||
let spawned_id = wait_for_spawned_thread_id(&test).await?;
|
let spawned_id = wait_for_spawned_thread_id(&test).await?;
|
||||||
assert_eq!(
|
assert_eq!(
|
||||||
start_inputs[0]["agent_id"].as_str(),
|
start_inputs[0]["agent_id"].as_str(),
|
||||||
@@ -714,7 +728,10 @@ async fn subagent_start_replaces_session_start_and_injects_context() -> Result<(
|
|||||||
child_prompt_input["agent_id"].as_str(),
|
child_prompt_input["agent_id"].as_str(),
|
||||||
Some(spawned_id.as_str())
|
Some(spawned_id.as_str())
|
||||||
);
|
);
|
||||||
assert_eq!(child_prompt_input["agent_type"].as_str(), Some("worker"));
|
assert_eq!(
|
||||||
|
child_prompt_input["agent_type"].as_str(),
|
||||||
|
Some(expected_agent_type)
|
||||||
|
);
|
||||||
|
|
||||||
let session_start_inputs = wait_for_hook_log(
|
let session_start_inputs = wait_for_hook_log(
|
||||||
test.codex_home_path(),
|
test.codex_home_path(),
|
||||||
|
|||||||
@@ -38,7 +38,8 @@
|
|||||||
"startup",
|
"startup",
|
||||||
"resume",
|
"resume",
|
||||||
"clear",
|
"clear",
|
||||||
"compact"
|
"compact",
|
||||||
|
"fork"
|
||||||
],
|
],
|
||||||
"type": "string"
|
"type": "string"
|
||||||
},
|
},
|
||||||
|
|||||||
@@ -26,6 +26,7 @@ pub enum SessionStartSource {
|
|||||||
Resume,
|
Resume,
|
||||||
Clear,
|
Clear,
|
||||||
Compact,
|
Compact,
|
||||||
|
Fork,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl SessionStartSource {
|
impl SessionStartSource {
|
||||||
@@ -35,6 +36,7 @@ impl SessionStartSource {
|
|||||||
Self::Resume => "resume",
|
Self::Resume => "resume",
|
||||||
Self::Clear => "clear",
|
Self::Clear => "clear",
|
||||||
Self::Compact => "compact",
|
Self::Compact => "compact",
|
||||||
|
Self::Fork => "fork",
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -851,7 +851,7 @@ fn permission_mode_schema(_gen: &mut SchemaGenerator) -> Schema {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn session_start_source_schema(_gen: &mut SchemaGenerator) -> Schema {
|
fn session_start_source_schema(_gen: &mut SchemaGenerator) -> Schema {
|
||||||
string_enum_schema(&["startup", "resume", "clear", "compact"])
|
string_enum_schema(&["startup", "resume", "clear", "compact", "fork"])
|
||||||
}
|
}
|
||||||
|
|
||||||
fn compaction_trigger_schema(_gen: &mut SchemaGenerator) -> Schema {
|
fn compaction_trigger_schema(_gen: &mut SchemaGenerator) -> Schema {
|
||||||
|
|||||||
Reference in new issue
Block a user