From 68b73e5eb93c9da10386e9e34e2c8dc2dbfb8d4c Mon Sep 17 00:00:00 2001 From: Friel Date: Thu, 2 Apr 2026 19:51:20 +0000 Subject: [PATCH] Fix watchdog wake/runtime plumbing --- codex-rs/core/src/agent/control_tests.rs | 61 +++++++++++++++++++ codex-rs/core/src/agent/role.rs | 6 +- codex-rs/core/src/agent/role_tests.rs | 10 ++- codex-rs/core/src/agent/watchdog.rs | 6 ++ codex-rs/core/src/codex.rs | 26 +++++++- .../src/tools/handlers/multi_agents/spawn.rs | 21 ++----- .../src/tools/handlers/multi_agents_tests.rs | 5 ++ 7 files changed, 112 insertions(+), 23 deletions(-) diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 6ce87384a4..841090a9ed 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -1037,6 +1037,67 @@ async fn spawn_agent_fork_injects_output_for_parent_spawn_call() { .expect("parent shutdown should submit"); } +#[tokio::test] +async fn spawn_agent_fork_skips_output_when_parent_spawn_call_is_missing() { + let harness = AgentControlHarness::new().await; + let (parent_thread_id, parent_thread) = harness.start_thread().await; + parent_thread + .inject_user_message_without_turn("parent seed context".to_string()) + .await; + parent_thread + .codex + .session + .ensure_rollout_materialized() + .await; + parent_thread.codex.session.flush_rollout().await; + + let synthetic_call_id = format!("watchdog_{parent_thread_id}"); + let child_thread_id = harness + .control + .spawn_agent_with_metadata( + harness.config.clone(), + text_input("child task"), + Some(SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id, + depth: 1, + agent_path: None, + agent_nickname: None, + agent_role: None, + })), + SpawnAgentOptions { + fork_parent_spawn_call_id: Some(synthetic_call_id.clone()), + post_fork_developer_message: None, + }, + ) + .await + .expect("forked spawn should succeed") + .thread_id; + + let child_thread = harness + .manager + .get_thread(child_thread_id) + .await + .expect("child thread should be registered"); + let history = child_thread.codex.session.clone_history().await; + + assert!(!history.raw_items().iter().any(|item| { + matches!( + item, + ResponseItem::FunctionCallOutput { call_id, .. } if call_id == &synthetic_call_id + ) + })); + + let _ = harness + .control + .shutdown_live_agent(child_thread_id) + .await + .expect("child shutdown should submit"); + let _ = parent_thread + .submit(Op::Shutdown {}) + .await + .expect("parent shutdown should submit"); +} + #[tokio::test] async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { let harness = AgentControlHarness::new().await; diff --git a/codex-rs/core/src/agent/role.rs b/codex-rs/core/src/agent/role.rs index b84382ae3c..a69df55eae 100644 --- a/codex-rs/core/src/agent/role.rs +++ b/codex-rs/core/src/agent/role.rs @@ -29,7 +29,9 @@ const AGENT_TYPE_UNAVAILABLE_ERROR: &str = "agent type is currently not availabl pub(crate) fn watchdog_interval_for_role(config: &Config, role_name: Option<&str>) -> Option { let role_name = role_name.unwrap_or(DEFAULT_ROLE_NAME); - resolve_role_config(config, role_name).and_then(|role| role.watchdog_interval_s) + let role = resolve_role_config(config, role_name)?; + role.watchdog_interval_s + .or_else(|| (role_name == "watchdog").then_some(config.watchdog_interval_s)) } /// Applies a named role layer to `config` while preserving caller-owned model selection. @@ -424,7 +426,7 @@ Rules: - Close the watchdog handle only when it is no longer needed or when replacing it with a new watchdog."#.to_string()), model: None, config_file: None, - watchdog_interval_s: Some(crate::config::DEFAULT_WATCHDOG_INTERVAL_S), + watchdog_interval_s: None, nickname_candidates: None, fork_context: Some(true), } diff --git a/codex-rs/core/src/agent/role_tests.rs b/codex-rs/core/src/agent/role_tests.rs index b7e1bc5cab..931cb34fd0 100644 --- a/codex-rs/core/src/agent/role_tests.rs +++ b/codex-rs/core/src/agent/role_tests.rs @@ -690,12 +690,16 @@ enabled = false } #[tokio::test] -async fn watchdog_interval_for_role_returns_built_in_watchdog_interval() { - let (_home, config) = test_config_with_cli_overrides(Vec::new()).await; +async fn watchdog_interval_for_role_uses_configured_watchdog_interval_for_builtin_watchdog() { + let (_home, config) = test_config_with_cli_overrides(vec![( + "watchdog_interval_s".to_string(), + TomlValue::Integer(1), + )]) + .await; assert_eq!( watchdog_interval_for_role(&config, Some("watchdog")), - Some(crate::config::DEFAULT_WATCHDOG_INTERVAL_S) + Some(1) ); assert_eq!(watchdog_interval_for_role(&config, Some("default")), None); } diff --git a/codex-rs/core/src/agent/watchdog.rs b/codex-rs/core/src/agent/watchdog.rs index b7e851172e..7fdc7ac82b 100644 --- a/codex-rs/core/src/agent/watchdog.rs +++ b/codex-rs/core/src/agent/watchdog.rs @@ -315,6 +315,12 @@ impl WatchdogManager { }); let mut helper_config = snapshot.config.clone(); helper_config.ephemeral = true; + if let Err(err) = helper_config.mcp_servers.set(HashMap::new()) { + warn!( + target_thread_id = %target_thread_id, + "failed to clear watchdog helper MCP servers: {err}" + ); + } if helper_config .features .enabled(Feature::AgentPromptInjection) diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 73c97d2d20..e2d4759fbc 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -4676,6 +4676,7 @@ mod handlers { use crate::tasks::UserShellCommandMode; use crate::tasks::UserShellCommandTask; use crate::tasks::execute_user_shell_command; + use codex_protocol::protocol::AGENT_INBOX_KIND; use codex_protocol::protocol::CodexErrorInfo; use codex_protocol::protocol::ErrorEvent; use codex_protocol::protocol::Event; @@ -4883,6 +4884,7 @@ mod handlers { text_elements: Vec::new(), }] }); + let skip_mcp_refresh = is_synthetic_agent_inbox_turn(&turn_input, &pending_items); let turn_sub_id = if attempts == 1 { sub_id.clone() } else { @@ -4892,8 +4894,10 @@ mod handlers { // Keep injected inbox wakeups visible to telemetry after the TurnContext field rename. current_context.session_telemetry.user_prompt(&turn_input); - sess.refresh_mcp_servers_if_requested(¤t_context) - .await; + if !skip_mcp_refresh { + sess.refresh_mcp_servers_if_requested(¤t_context) + .await; + } sess.spawn_task( Arc::clone(¤t_context), turn_input, @@ -4919,6 +4923,24 @@ mod handlers { Some(user_message.content) } + fn is_synthetic_agent_inbox_turn( + turn_input: &[UserInput], + pending_items: &[ResponseInputItem], + ) -> bool { + let is_empty_user_message = matches!( + turn_input, + [UserInput::Text { + text, + text_elements, + }] if text.is_empty() && text_elements.is_empty() + ); + is_empty_user_message + && matches!( + pending_items.first(), + Some(ResponseInputItem::FunctionCall { name, .. }) if name == AGENT_INBOX_KIND + ) + } + pub async fn run_user_shell_command(sess: &Arc, sub_id: String, command: String) { if let Some((turn_context, cancellation_token)) = sess.active_turn_context_and_cancellation_token().await diff --git a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs index e2aa941d3f..cd19c93d23 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -11,7 +11,9 @@ use crate::agent::role::default_fork_context_for_role; use crate::agent::role::watchdog_interval_for_role; use crate::config::Config; use codex_features::Feature; +use codex_protocol::protocol::Op; use codex_protocol::protocol::SessionSource; +use std::collections::HashMap; use std::collections::HashSet; pub(crate) struct Handler; @@ -66,24 +68,11 @@ impl ToolHandler for Handler { "Agent depth limit reached. Solve the task yourself.".to_string(), )); } - session - .send_event( - &turn, - CollabAgentSpawnBeginEvent { - call_id: call_id.clone(), - sender_thread_id: session.conversation_id, - prompt: prompt.clone(), - model: args.model.clone().unwrap_or_default(), - reasoning_effort: args.reasoning_effort.unwrap_or_default(), - } - .into(), - ) - .await; - let mut config = - build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?; let fork_context = args .fork_context .unwrap_or_else(|| default_fork_context_for_role(&turn.config, role_name)); + let mut config = + build_agent_spawn_config(&session.get_base_instructions().await, turn.as_ref())?; if !fork_context { apply_requested_spawn_agent_model_overrides( &session, @@ -265,7 +254,7 @@ async fn spawn_watchdog( spawn_source: SessionSource, ) -> crate::error::Result { let target_thread_id = agent_control - .spawn_agent_handle(config.clone(), Some(spawn_source)) + .spawn_agent(config.clone(), Op::Interrupt, Some(spawn_source)) .await?; let superseded_before_register = agent_control .unregister_watchdogs_for_owner(owner_thread_id) 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 653d0dd520..f1182d0a0e 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -438,6 +438,11 @@ async fn spawn_agent_watchdog_role_returns_handle_with_role_defaults() { manager.agent_control().get_status(agent_id).await, AgentStatus::PendingInit ); + let watchdog_targets = manager.agent_control().watchdog_targets(&[agent_id]).await; + assert_eq!( + watchdog_targets, + std::collections::HashSet::from([agent_id]) + ); } #[tokio::test]