diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 8d6321e41..9ee784117 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -566,6 +566,9 @@ pub struct AgentsToml { /// Default maximum runtime in seconds for agent job workers. #[schemars(range(min = 1))] pub job_max_runtime_seconds: Option, + /// Whether to record a model-visible message when an agent turn is interrupted. + /// Defaults to true. + pub interrupt_message: Option, /// User-defined role declarations keyed by role name. /// diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index a88cc21b8..6cac42908 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -36,6 +36,10 @@ "$ref": "#/definitions/AgentRoleToml" }, "properties": { + "interrupt_message": { + "description": "Whether to record a model-visible message when an agent turn is interrupted. Defaults to true.", + "type": "boolean" + }, "job_max_runtime_seconds": { "description": "Default maximum runtime in seconds for agent job workers.", "format": "uint64", diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index edc6cc829..c5eb1d1fa 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -3941,6 +3941,7 @@ async fn load_config_rejects_missing_agent_role_config_file() -> std::io::Result max_threads: None, max_depth: None, job_max_runtime_seconds: None, + interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), AgentRoleToml { @@ -4856,6 +4857,29 @@ model = "gpt-5-mini" Ok(()) } +#[tokio::test] +async fn load_config_resolves_agent_interrupt_message() -> std::io::Result<()> { + let codex_home = TempDir::new()?; + let cfg = ConfigToml { + agents: Some(AgentsToml { + interrupt_message: Some(false), + ..Default::default() + }), + ..Default::default() + }; + + let config = Config::load_from_base_config_with_overrides( + cfg, + ConfigOverrides::default(), + codex_home.abs(), + ) + .await?; + + assert!(!config.agent_interrupt_message_enabled); + + Ok(()) +} + #[tokio::test] async fn load_config_normalizes_agent_role_nickname_candidates() -> std::io::Result<()> { let codex_home = TempDir::new()?; @@ -4864,6 +4888,7 @@ async fn load_config_normalizes_agent_role_nickname_candidates() -> std::io::Res max_threads: None, max_depth: None, job_max_runtime_seconds: None, + interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), AgentRoleToml { @@ -4906,6 +4931,7 @@ async fn load_config_rejects_empty_agent_role_nickname_candidates() -> std::io:: max_threads: None, max_depth: None, job_max_runtime_seconds: None, + interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), AgentRoleToml { @@ -4942,6 +4968,7 @@ async fn load_config_rejects_duplicate_agent_role_nickname_candidates() -> std:: max_threads: None, max_depth: None, job_max_runtime_seconds: None, + interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), AgentRoleToml { @@ -4978,6 +5005,7 @@ async fn load_config_rejects_unsafe_agent_role_nickname_candidates() -> std::io: max_threads: None, max_depth: None, job_max_runtime_seconds: None, + interrupt_message: None, roles: BTreeMap::from([( "researcher".to_string(), AgentRoleToml { @@ -5234,6 +5262,7 @@ async fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> { agent_roles: BTreeMap::new(), memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, + agent_interrupt_message_enabled: true, codex_home: fixture.codex_home(), sqlite_home: fixture.codex_home().to_path_buf(), log_dir: fixture.codex_home().join("log").to_path_buf(), @@ -5431,6 +5460,7 @@ async fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> { agent_roles: BTreeMap::new(), memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, + agent_interrupt_message_enabled: true, codex_home: fixture.codex_home(), sqlite_home: fixture.codex_home().to_path_buf(), log_dir: fixture.codex_home().join("log").to_path_buf(), @@ -5582,6 +5612,7 @@ async fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> { agent_roles: BTreeMap::new(), memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, + agent_interrupt_message_enabled: true, codex_home: fixture.codex_home(), sqlite_home: fixture.codex_home().to_path_buf(), log_dir: fixture.codex_home().join("log").to_path_buf(), @@ -5718,6 +5749,7 @@ async fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> { agent_roles: BTreeMap::new(), memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, + agent_interrupt_message_enabled: true, codex_home: fixture.codex_home(), sqlite_home: fixture.codex_home().to_path_buf(), log_dir: fixture.codex_home().join("log").to_path_buf(), diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 13d4a5d0a..cfd316753 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -432,6 +432,9 @@ pub struct Config { /// Maximum runtime in seconds for agent job workers before they are failed. pub agent_job_max_runtime_seconds: Option, + /// Whether to record a model-visible message when an agent turn is interrupted. + pub agent_interrupt_message_enabled: bool, + /// Maximum nesting depth allowed for spawned agent threads. pub agent_max_depth: i32, @@ -2002,6 +2005,11 @@ impl Config { "agents.job_max_runtime_seconds must fit within a 64-bit signed integer", )); } + let agent_interrupt_message_enabled = cfg + .agents + .as_ref() + .and_then(|agents| agents.interrupt_message) + .unwrap_or(true); let background_terminal_max_timeout = cfg .background_terminal_max_timeout .unwrap_or(DEFAULT_MAX_BACKGROUND_TERMINAL_TIMEOUT_MS) @@ -2366,6 +2374,7 @@ impl Config { agent_roles, memories: cfg.memories.unwrap_or_default().into(), agent_job_max_runtime_seconds, + agent_interrupt_message_enabled, codex_home, sqlite_home, log_dir, diff --git a/codex-rs/core/src/tasks/mod.rs b/codex-rs/core/src/tasks/mod.rs index 92da8c07f..b0ec96cfe 100644 --- a/codex-rs/core/src/tasks/mod.rs +++ b/codex-rs/core/src/tasks/mod.rs @@ -19,6 +19,7 @@ use tracing::info_span; use tracing::trace; use tracing::warn; +use crate::config::Config; use crate::context::ContextualUserFragment; use crate::hook_runtime::PendingInputHookDisposition; use crate::hook_runtime::inspect_pending_input; @@ -62,27 +63,50 @@ pub(crate) use user_shell::execute_user_shell_command; const GRACEFULL_INTERRUPTION_TIMEOUT_MS: u64 = 100; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum InterruptedTurnHistoryMarker { + Disabled, + ContextualUser, + Developer, +} + +impl InterruptedTurnHistoryMarker { + pub(crate) fn from_config(config: &Config) -> Self { + if !config.agent_interrupt_message_enabled { + return Self::Disabled; + } + if config.features.enabled(Feature::MultiAgentV2) { + Self::Developer + } else { + Self::ContextualUser + } + } +} + /// Shared model-visible marker used by both the real interrupt path and /// interrupted fork snapshots. -pub(crate) fn interrupted_turn_history_marker(multi_agent_v2_enabled: bool) -> ResponseItem { - let guidance = if multi_agent_v2_enabled { - crate::context::TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE - } else { - crate::context::TurnAborted::INTERRUPTED_GUIDANCE - }; - let marker = crate::context::TurnAborted::new(guidance); - if multi_agent_v2_enabled { - ResponseItem::Message { - id: None, - role: "developer".to_string(), - content: vec![ContentItem::InputText { - text: marker.render(), - }], - end_turn: None, - phase: None, +pub(crate) fn interrupted_turn_history_marker( + marker: InterruptedTurnHistoryMarker, +) -> Option { + match marker { + InterruptedTurnHistoryMarker::Disabled => None, + InterruptedTurnHistoryMarker::ContextualUser => Some(ContextualUserFragment::into( + crate::context::TurnAborted::new(crate::context::TurnAborted::INTERRUPTED_GUIDANCE), + )), + InterruptedTurnHistoryMarker::Developer => { + let marker = crate::context::TurnAborted::new( + crate::context::TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE, + ); + Some(ResponseItem::Message { + id: None, + role: "developer".to_string(), + content: vec![ContentItem::InputText { + text: marker.render(), + }], + end_turn: None, + phase: None, + }) } - } else { - ContextualUserFragment::into(marker) } } @@ -692,15 +716,20 @@ impl Session { if reason == TurnAbortReason::Interrupted { self.cleanup_after_interrupt(&task.turn_context).await; - let marker = interrupted_turn_history_marker(self.enabled(Feature::MultiAgentV2)); - self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref()) - .await; - self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)]) - .await; - // Ensure the marker is durably visible before emitting TurnAborted: some clients - // synchronously re-read the rollout on receipt of the abort event. - if let Err(err) = self.flush_rollout().await { - warn!("failed to flush interrupted-turn marker before emitting TurnAborted: {err}"); + if let Some(marker) = interrupted_turn_history_marker( + InterruptedTurnHistoryMarker::from_config(task.turn_context.config.as_ref()), + ) { + self.record_into_history(std::slice::from_ref(&marker), task.turn_context.as_ref()) + .await; + self.persist_rollout_items(&[RolloutItem::ResponseItem(marker)]) + .await; + // Ensure the marker is durably visible before emitting TurnAborted: some clients + // synchronously re-read the rollout on receipt of the abort event. + if let Err(err) = self.flush_rollout().await { + warn!( + "failed to flush interrupted-turn marker before emitting TurnAborted: {err}" + ); + } } } diff --git a/codex-rs/core/src/thread_manager.rs b/codex-rs/core/src/thread_manager.rs index 2509f3b0c..73dd09113 100644 --- a/codex-rs/core/src/thread_manager.rs +++ b/codex-rs/core/src/thread_manager.rs @@ -17,12 +17,12 @@ use crate::session::INITIAL_SUBMIT_ID; use crate::shell_snapshot::ShellSnapshot; use crate::skills_watcher::SkillsWatcher; use crate::skills_watcher::SkillsWatcherEvent; +use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; use codex_analytics::AnalyticsEventsClient; use codex_app_server_protocol::ThreadHistoryBuilder; use codex_app_server_protocol::TurnStatus; use codex_exec_server::EnvironmentManager; -use codex_features::Feature; use codex_login::AuthManager; use codex_login::CodexAuth; use codex_model_provider::create_model_provider; @@ -746,7 +746,7 @@ impl ThreadManager { let snapshot = snapshot.into(); let history = RolloutRecorder::get_rollout_history(&path).await?; let snapshot_state = snapshot_turn_state(&history); - let multi_agent_v2_enabled = config.features.enabled(Feature::MultiAgentV2); + let interrupted_marker = InterruptedTurnHistoryMarker::from_config(&config); let history = match snapshot { ForkSnapshot::TruncateBeforeNthUserMessage(nth_user_message) => { truncate_before_nth_user_message(history, nth_user_message, &snapshot_state) @@ -762,7 +762,7 @@ impl ThreadManager { append_interrupted_boundary( history, snapshot_state.active_turn_id, - multi_agent_v2_enabled, + interrupted_marker, ) } else { history @@ -1234,7 +1234,7 @@ fn snapshot_turn_state(history: &InitialHistory) -> SnapshotTurnState { fn append_interrupted_boundary( history: InitialHistory, turn_id: Option, - multi_agent_v2_enabled: bool, + interrupted_marker: InterruptedTurnHistoryMarker, ) -> InitialHistory { let aborted_event = RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id, @@ -1244,23 +1244,25 @@ fn append_interrupted_boundary( })); match history { - InitialHistory::New | InitialHistory::Cleared => InitialHistory::Forked(vec![ - RolloutItem::ResponseItem(interrupted_turn_history_marker(multi_agent_v2_enabled)), - aborted_event, - ]), + InitialHistory::New | InitialHistory::Cleared => { + let mut history = Vec::new(); + if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) { + history.push(RolloutItem::ResponseItem(marker)); + } + history.push(aborted_event); + InitialHistory::Forked(history) + } InitialHistory::Forked(mut history) => { - history.push(RolloutItem::ResponseItem(interrupted_turn_history_marker( - multi_agent_v2_enabled, - ))); + if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) { + history.push(RolloutItem::ResponseItem(marker)); + } history.push(aborted_event); InitialHistory::Forked(history) } InitialHistory::Resumed(mut resumed) => { - resumed - .history - .push(RolloutItem::ResponseItem(interrupted_turn_history_marker( - multi_agent_v2_enabled, - ))); + if let Some(marker) = interrupted_turn_history_marker(interrupted_marker) { + resumed.history.push(RolloutItem::ResponseItem(marker)); + } resumed.history.push(aborted_event); InitialHistory::Forked(resumed.history) } diff --git a/codex-rs/core/src/thread_manager_tests.rs b/codex-rs/core/src/thread_manager_tests.rs index 53a09ce84..540b48f6e 100644 --- a/codex-rs/core/src/thread_manager_tests.rs +++ b/codex-rs/core/src/thread_manager_tests.rs @@ -3,6 +3,7 @@ use crate::config::test_config; use crate::rollout::RolloutRecorder; use crate::session::session::SessionSettingsUpdate; use crate::session::tests::make_session_and_context; +use crate::tasks::InterruptedTurnHistoryMarker; use crate::tasks::interrupted_turn_history_marker; use codex_models_manager::collaboration_mode_presets::CollaborationModesConfig; use codex_models_manager::manager::RefreshStrategy; @@ -58,6 +59,16 @@ fn disabled_environment_manager_for_tests() -> Arc ResponseItem { + interrupted_turn_history_marker(InterruptedTurnHistoryMarker::ContextualUser) + .expect("contextual-user interrupted marker should be enabled") +} + +fn developer_interrupted_marker() -> ResponseItem { + interrupted_turn_history_marker(InterruptedTurnHistoryMarker::Developer) + .expect("developer interrupted marker should be enabled") +} + #[test] fn truncates_before_requested_user_message() { let items = [ @@ -449,16 +460,14 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() { append_interrupted_boundary( committed_history, /*turn_id*/ None, - /*multi_agent_v2_enabled*/ false, + InterruptedTurnHistoryMarker::ContextualUser, ) .get_rollout_items() ) .expect("serialize interrupted fork history"), serde_json::to_value(vec![ RolloutItem::ResponseItem(user_msg("hello")), - RolloutItem::ResponseItem(interrupted_turn_history_marker( - /*multi_agent_v2_enabled*/ false, - )), + RolloutItem::ResponseItem(contextual_user_interrupted_marker()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: None, reason: TurnAbortReason::Interrupted, @@ -473,15 +482,13 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() { append_interrupted_boundary( InitialHistory::New, /*turn_id*/ None, - /*multi_agent_v2_enabled*/ false, + InterruptedTurnHistoryMarker::ContextualUser, ) .get_rollout_items() ) .expect("serialize interrupted empty fork history"), serde_json::to_value(vec![ - RolloutItem::ResponseItem(interrupted_turn_history_marker( - /*multi_agent_v2_enabled*/ false, - )), + RolloutItem::ResponseItem(contextual_user_interrupted_marker()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: None, reason: TurnAbortReason::Interrupted, @@ -493,14 +500,60 @@ fn interrupted_fork_snapshot_appends_interrupt_boundary() { ); } +#[test] +fn disabled_interrupted_fork_snapshot_appends_only_interrupt_event() { + let committed_history = + InitialHistory::Forked(vec![RolloutItem::ResponseItem(user_msg("hello"))]); + + assert_eq!( + serde_json::to_value( + append_interrupted_boundary( + committed_history, + /*turn_id*/ None, + InterruptedTurnHistoryMarker::Disabled, + ) + .get_rollout_items() + ) + .expect("serialize disabled interrupted fork history"), + serde_json::to_value(vec![ + RolloutItem::ResponseItem(user_msg("hello")), + RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { + turn_id: None, + reason: TurnAbortReason::Interrupted, + completed_at: None, + duration_ms: None, + })), + ]) + .expect("serialize expected disabled interrupted fork history"), + ); + assert_eq!( + serde_json::to_value( + append_interrupted_boundary( + InitialHistory::New, + /*turn_id*/ None, + InterruptedTurnHistoryMarker::Disabled, + ) + .get_rollout_items() + ) + .expect("serialize disabled interrupted empty fork history"), + serde_json::to_value(vec![RolloutItem::EventMsg(EventMsg::TurnAborted( + TurnAbortedEvent { + turn_id: None, + reason: TurnAbortReason::Interrupted, + completed_at: None, + duration_ms: None, + }, + ))]) + .expect("serialize expected disabled interrupted empty fork history"), + ); +} + #[test] fn interrupted_snapshot_is_not_mid_turn() { let interrupted_history = InitialHistory::Forked(vec![ RolloutItem::ResponseItem(user_msg("hello")), RolloutItem::ResponseItem(assistant_msg("partial")), - RolloutItem::ResponseItem(interrupted_turn_history_marker( - /*multi_agent_v2_enabled*/ false, - )), + RolloutItem::ResponseItem(contextual_user_interrupted_marker()), RolloutItem::EventMsg(EventMsg::TurnAborted(TurnAbortedEvent { turn_id: Some("turn-1".to_string()), reason: TurnAbortReason::Interrupted, @@ -521,7 +574,7 @@ fn interrupted_snapshot_is_not_mid_turn() { #[test] fn multi_agent_v2_interrupted_marker_uses_developer_input_message() { - let marker = interrupted_turn_history_marker(/*multi_agent_v2_enabled*/ true); + let marker = developer_interrupted_marker(); let ResponseItem::Message { role, content, .. } = marker else { panic!("expected interrupted marker to be a message"); @@ -653,7 +706,7 @@ async fn interrupted_fork_snapshot_does_not_synthesize_turn_id_for_legacy_histor .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem( - interrupted_turn_history_marker(/*multi_agent_v2_enabled*/ false), + contextual_user_interrupted_marker(), )) .expect("serialize interrupted marker"); let interrupted_abort_json = serde_json::to_value(RolloutItem::EventMsg( @@ -845,7 +898,7 @@ async fn interrupted_fork_snapshot_uses_persisted_mid_turn_history_without_live_ .filter(|item| !matches!(item, RolloutItem::SessionMeta(_))) .collect(); let interrupted_marker_json = serde_json::to_value(RolloutItem::ResponseItem( - interrupted_turn_history_marker(/*multi_agent_v2_enabled*/ false), + contextual_user_interrupted_marker(), )) .expect("serialize interrupted marker"); assert_eq!( diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 82db5f13b..a08f4b1bf 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1631,6 +1631,104 @@ async fn multi_agent_v2_followup_task_interrupts_busy_child_without_losing_messa .expect("shutdown should submit"); } +#[tokio::test] +async fn multi_agent_v2_followup_task_can_disable_interrupted_marker() { + let (mut session, mut turn) = make_session_and_context().await; + let manager = thread_manager(); + let root = manager + .start_thread((*turn.config).clone()) + .await + .expect("root thread should start"); + session.services.agent_control = manager.agent_control(); + session.conversation_id = root.thread_id; + let mut config = turn.config.as_ref().clone(); + let _ = config.features.enable(Feature::MultiAgentV2); + config.agent_interrupt_message_enabled = false; + turn.config = Arc::new(config); + let session = Arc::new(session); + let turn = Arc::new(turn); + + let worker_path = AgentPath::try_from("/root/worker").expect("worker path"); + let agent_id = session + .services + .agent_control + .spawn_agent_with_metadata( + (*turn.config).clone(), + Op::CleanBackgroundTerminals, + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: root.thread_id, + depth: 1, + agent_path: Some(worker_path), + agent_nickname: None, + agent_role: None, + })), + crate::agent::control::SpawnAgentOptions::default(), + ) + .await + .expect("worker spawn should succeed") + .thread_id; + let thread = manager + .get_thread(agent_id) + .await + .expect("worker thread should exist"); + + let active_turn = thread.codex.session.new_default_turn().await; + let interrupted_turn_id = active_turn.sub_id.clone(); + thread + .codex + .session + .spawn_task( + Arc::clone(&active_turn), + vec![UserInput::Text { + text: "working".to_string(), + text_elements: Vec::new(), + }], + NeverEndingTask, + ) + .await; + + FollowupTaskHandlerV2 + .handle(invocation( + session, + turn, + "followup_task", + function_payload(json!({ + "target": agent_id.to_string(), + "message": "continue", + "interrupt": true + })), + )) + .await + .expect("interrupting v2 followup_task should succeed"); + + wait_for_turn_aborted(&thread, &interrupted_turn_id, TurnAbortReason::Interrupted).await; + let history_items = thread + .codex + .session + .clone_history() + .await + .raw_items() + .to_vec(); + assert!( + !history_items.iter().any(|item| matches!( + item, + ResponseItem::Message { content, .. } + if content.iter().any(|content_item| matches!( + content_item, + ContentItem::InputText { text } | ContentItem::OutputText { text } + if text.contains(TurnAborted::INTERRUPTED_GUIDANCE) + || text.contains(TurnAborted::INTERRUPTED_DEVELOPER_GUIDANCE) + )) + )), + "disabled interrupted-turn marker should not be recorded in history" + ); + + let _ = thread + .submit(Op::Shutdown {}) + .await + .expect("shutdown should submit"); +} + #[tokio::test] async fn multi_agent_v2_followup_task_completion_notifies_parent_on_every_turn() { let (mut session, mut turn) = make_session_and_context().await;