From f44de610db5d9c22ccf1f18c1b5993b4c65f4dad Mon Sep 17 00:00:00 2001 From: Friel Date: Sat, 18 Apr 2026 07:09:08 +0000 Subject: [PATCH] Add watchdog runtime handles --- codex-rs/config/src/config_toml.rs | 5 + codex-rs/core/config.schema.json | 12 + codex-rs/core/src/agent/control.rs | 211 +++++- codex-rs/core/src/agent/control_tests.rs | 417 ++++++++++- codex-rs/core/src/agent/mod.rs | 4 + codex-rs/core/src/agent/role.rs | 28 +- codex-rs/core/src/agent/role_tests.rs | 42 ++ codex-rs/core/src/agent/watchdog.rs | 645 ++++++++++++++++++ codex-rs/core/src/config/agent_roles.rs | 9 + codex-rs/core/src/config/config_tests.rs | 4 + codex-rs/core/src/config/mod.rs | 14 + .../core/src/tools/handlers/multi_agents.rs | 4 + .../src/tools/handlers/multi_agents/spawn.rs | 154 ++++- .../src/tools/handlers/multi_agents/wait.rs | 25 + .../multi_agents/watchdog_self_close.rs | 99 +++ .../handlers/multi_agents/watchdog_snooze.rs | 68 ++ .../src/tools/handlers/multi_agents_tests.rs | 288 +++++++- codex-rs/core/src/tools/spec.rs | 8 + codex-rs/features/src/lib.rs | 8 + codex-rs/thread-manager-sample/src/main.rs | 1 + codex-rs/tools/src/agent_tool.rs | 91 +++ codex-rs/tools/src/lib.rs | 3 + codex-rs/tools/src/tool_config.rs | 3 + codex-rs/tools/src/tool_registry_plan.rs | 22 + .../tools/src/tool_registry_plan_types.rs | 2 + 25 files changed, 2134 insertions(+), 33 deletions(-) create mode 100644 codex-rs/core/src/agent/watchdog.rs create mode 100644 codex-rs/core/src/tools/handlers/multi_agents/watchdog_self_close.rs create mode 100644 codex-rs/core/src/tools/handlers/multi_agents/watchdog_snooze.rs diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index 6406f9e1ff..4ec2728366 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -366,6 +366,10 @@ pub struct ConfigToml { /// Agent-related settings (thread limits, etc.). pub agents: Option, + /// Default interval in seconds for idle-time watchdog check-ins. + #[schemars(range(min = 1))] + pub watchdog_interval_s: Option, + /// Memories subsystem settings. pub memories: Option, @@ -665,6 +669,7 @@ pub struct AgentsToml { } #[derive(Serialize, Deserialize, Debug, Clone, Default, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] #[schemars(deny_unknown_fields)] pub struct AgentRoleToml { /// Human-facing role documentation used in spawn tool guidance. diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index a30a3ed925..4bf6d496dd 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -358,6 +358,9 @@ "default": null, "description": "Optional feature toggles scoped to this profile.", "properties": { + "agent_watchdog": { + "type": "boolean" + }, "apply_patch_freeform": { "type": "boolean" }, @@ -3824,6 +3827,9 @@ "default": null, "description": "Centralized feature flags (new). Prefer this over individual toggles.", "properties": { + "agent_watchdog": { + "type": "boolean" + }, "apply_patch_freeform": { "type": "boolean" }, @@ -4465,6 +4471,12 @@ ], "description": "Collection of settings that are specific to the TUI." }, + "watchdog_interval_s": { + "description": "Default interval in seconds for idle-time watchdog check-ins.", + "format": "int64", + "minimum": 1.0, + "type": "integer" + }, "web_search": { "allOf": [ { diff --git a/codex-rs/core/src/agent/control.rs b/codex-rs/core/src/agent/control.rs index 6f56328d03..3f10004370 100644 --- a/codex-rs/core/src/agent/control.rs +++ b/codex-rs/core/src/agent/control.rs @@ -1,4 +1,7 @@ use crate::agent::AgentStatus; +use crate::agent::RemovedWatchdog; +use crate::agent::WatchdogManager; +use crate::agent::WatchdogRegistration; use crate::agent::registry::AgentMetadata; use crate::agent::registry::AgentRegistry; use crate::agent::role::DEFAULT_ROLE_NAME; @@ -128,32 +131,68 @@ fn keep_forked_rollout_item(item: &RolloutItem) -> bool { } } +fn is_watchdog_helper_source(session_source: &SessionSource) -> bool { + matches!( + session_source, + SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + agent_role: Some(agent_role), + .. + }) if agent_role == "watchdog" + ) +} + /// Control-plane handle for multi-agent operations. /// `AgentControl` is held by each session (via `SessionServices`). It provides capability to /// spawn new agents and the inter-agent communication layer. /// An `AgentControl` instance is intended to be created at most once per root thread/session /// tree. That same `AgentControl` is then shared with every sub-agent spawned from that root, /// which keeps the registry scoped to that root thread rather than the entire `ThreadManager`. -#[derive(Clone, Default)] +#[derive(Clone)] pub(crate) struct AgentControl { /// Weak handle back to the global thread registry/state. /// This is `Weak` to avoid reference cycles and shadow persistence of the form /// `ThreadManagerState -> CodexThread -> Session -> SessionServices -> ThreadManagerState`. manager: Weak, state: Arc, + watchdogs: Option>, +} + +impl Default for AgentControl { + fn default() -> Self { + Self { + manager: Weak::new(), + state: Arc::new(AgentRegistry::default()), + watchdogs: None, + } + } } impl AgentControl { /// Construct a new `AgentControl` that can spawn/message agents via the given manager state. pub(crate) fn new(manager: Weak) -> Self { + let state = Arc::new(AgentRegistry::default()); + let watchdogs = WatchdogManager::new(manager.clone(), Arc::clone(&state)); + watchdogs.start(); Self { manager, - ..Default::default() + state, + watchdogs: Some(watchdogs), + } + } + + pub(crate) fn from_parts( + manager: Weak, + state: Arc, + watchdogs: Arc, + ) -> Self { + Self { + manager, + state, + watchdogs: Some(watchdogs), } } /// Spawn a new agent thread and submit the initial prompt. - #[cfg(test)] pub(crate) async fn spawn_agent( &self, config: crate::config::Config, @@ -322,7 +361,9 @@ impl AgentControl { self.send_input(new_thread.thread_id, initial_operation) .await?; - if !new_thread.thread.enabled(Feature::MultiAgentV2) { + if !new_thread.thread.enabled(Feature::MultiAgentV2) + && !matches!(agent_metadata.agent_role.as_deref(), Some("watchdog")) + { let child_reference = agent_metadata .agent_path .as_ref() @@ -354,7 +395,9 @@ impl AgentControl { inherited_exec_policy: Option>, inherited_thread_state: InheritedThreadState, ) -> CodexResult { - if options.fork_parent_spawn_call_id.is_none() { + if options.fork_parent_spawn_call_id.is_none() + && !is_watchdog_helper_source(&session_source) + { return Err(CodexErr::Fatal( "spawn_agent fork requires a parent spawn call id".to_string(), )); @@ -714,6 +757,30 @@ impl AgentControl { result } + pub(crate) async fn send_watchdog_wakeup( + &self, + owner_thread_id: ThreadId, + message: String, + ) -> CodexResult { + let Some(message) = sanitize_watchdog_wakeup_message(message) else { + return Ok(String::new()); + }; + let watchdog_path = AgentPath::root() + .join("watchdog") + .unwrap_or_else(|_| AgentPath::root()); + self.send_inter_agent_communication( + owner_thread_id, + InterAgentCommunication::new( + watchdog_path, + AgentPath::root(), + Vec::new(), + message, + /*trigger_turn*/ true, + ), + ) + .await + } + /// Interrupt the current task for an existing agent thread. pub(crate) async fn interrupt_agent(&self, agent_id: ThreadId) -> CodexResult { let state = self.upgrade()?; @@ -753,10 +820,106 @@ impl AgentControl { result } + pub(crate) async fn register_watchdog( + &self, + registration: WatchdogRegistration, + ) -> CodexResult> { + self.watchdog_manager()?.register(registration).await + } + + pub(crate) async fn unregister_watchdogs_for_owner( + &self, + owner_thread_id: ThreadId, + ) -> Vec { + let Some(watchdogs) = self.watchdogs.as_ref() else { + return Vec::new(); + }; + watchdogs.unregister_for_owner(owner_thread_id).await + } + + pub(crate) async fn unregister_watchdog_handle( + &self, + target_thread_id: ThreadId, + ) -> Option { + let watchdogs = self.watchdogs.as_ref()?; + watchdogs.unregister_handle(target_thread_id).await + } + + pub(crate) async fn is_watchdog_handle(&self, target_thread_id: ThreadId) -> bool { + let Some(watchdogs) = self.watchdogs.as_ref() else { + return false; + }; + watchdogs.is_watchdog_handle(target_thread_id).await + } + + #[cfg(test)] + pub(crate) async fn set_watchdog_active_helper_for_tests( + &self, + target_thread_id: ThreadId, + helper_thread_id: ThreadId, + ) { + if let Some(watchdogs) = self.watchdogs.as_ref() { + watchdogs + .set_active_helper_for_tests(target_thread_id, helper_thread_id) + .await; + } + } + + #[cfg(test)] + pub(crate) async fn watchdog_helper_is_suppressed_for_tests( + &self, + helper_thread_id: ThreadId, + ) -> bool { + let Some(watchdogs) = self.watchdogs.as_ref() else { + return false; + }; + watchdogs + .helper_is_suppressed_for_tests(helper_thread_id) + .await + } + + pub(crate) async fn watchdog_owner_for_active_helper( + &self, + helper_thread_id: ThreadId, + ) -> Option { + let watchdogs = self.watchdogs.as_ref()?; + watchdogs.owner_for_active_helper(helper_thread_id).await + } + + pub(crate) async fn watchdog_target_for_active_helper( + &self, + helper_thread_id: ThreadId, + ) -> Option { + let watchdogs = self.watchdogs.as_ref()?; + watchdogs.target_for_active_helper(helper_thread_id).await + } + + pub(crate) async fn snooze_watchdog_helper( + &self, + helper_thread_id: ThreadId, + delay_seconds: Option, + ) -> Option { + let watchdogs = self.watchdogs.as_ref()?; + watchdogs + .snooze_active_helper(helper_thread_id, delay_seconds) + .await + } + + fn watchdog_manager(&self) -> CodexResult<&Arc> { + self.watchdogs.as_ref().ok_or_else(|| { + CodexErr::UnsupportedOperation("watchdog manager unavailable".to_string()) + }) + } + /// Mark `agent_id` as explicitly closed in persisted spawn-edge state, then shut down the /// agent and any live descendants reached from the in-memory tree. pub(crate) async fn close_agent(&self, agent_id: ThreadId) -> CodexResult { let state = self.upgrade()?; + if let Some(removed_watchdog) = self.unregister_watchdog_handle(agent_id).await + && let Some(helper_id) = removed_watchdog.active_helper_id + { + let _ = self.shutdown_live_agent(helper_id).await; + } if let Ok(thread) = state.get_thread(agent_id).await && let Some(state_db_ctx) = thread.state_db() && let Err(err) = state_db_ctx @@ -1343,6 +1506,44 @@ fn thread_spawn_parent_thread_id(session_source: &SessionSource) -> Option Option { + let Some(stripped_message) = strip_leading_watchdog_prompt_scaffold(&message) else { + let message = message.trim(); + return (!message.is_empty()).then(|| message.to_string()); + }; + + let stripped_message = stripped_message.trim(); + (!stripped_message.is_empty()).then(|| stripped_message.to_string()) +} + +fn strip_leading_watchdog_prompt_scaffold(message: &str) -> Option<&str> { + let mut lines = message.split_inclusive('\n').scan(0, |offset, line| { + let line_start = *offset; + *offset += line.len(); + Some((line_start, line)) + }); + + let mut saw_watchdog_scaffold = false; + let mut report_start = None; + for (line_start, line) in &mut lines { + let trimmed = line.trim(); + if trimmed.contains("watchdog check-in agent") + || trimmed.starts_with("Read AGENTS.watchdog.md") + || trimmed.starts_with("Target agent id:") + { + saw_watchdog_scaffold = true; + } + if trimmed.starts_with("AUTOPLAN_WATCHDOG_REPORT") + || trimmed.starts_with("Watchdog report:") + { + report_start = Some(line_start); + break; + } + } + + saw_watchdog_scaffold.then(|| &message[report_start.unwrap_or(message.len())..]) +} + fn agent_matches_prefix(agent_path: Option<&AgentPath>, prefix: &AgentPath) -> bool { if prefix.is_root() { return true; diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index b047202cd8..9dea3aba76 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -60,6 +60,31 @@ async fn test_config() -> (TempDir, Config) { test_config_with_cli_overrides(Vec::new()).await } +fn mcp_server_config(command: &str) -> McpServerConfig { + McpServerConfig { + transport: McpServerTransportConfig::Stdio { + command: command.to_string(), + args: Vec::new(), + env: None, + env_vars: Vec::new(), + cwd: None, + }, + experimental_environment: None, + enabled: true, + required: false, + supports_parallel_tool_calls: false, + disabled_reason: None, + startup_timeout_sec: None, + tool_timeout_sec: None, + default_tools_approval_mode: None, + enabled_tools: None, + disabled_tools: None, + scopes: None, + oauth_resource: None, + tools: std::collections::HashMap::new(), + } +} + fn text_input(text: &str) -> Op { vec![UserInput::Text { text: text.to_string(), @@ -401,6 +426,295 @@ async fn resume_agent_errors_when_manager_dropped() { ); } +#[tokio::test] +async fn watchdog_spawns_helper_after_owner_completes() { + let harness = AgentControlHarness::new().await; + let (owner_thread_id, owner_thread) = harness.start_thread().await; + let (target_thread_id, _) = harness.start_thread().await; + let mut config = harness.config.clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + + harness + .control + .register_watchdog(WatchdogRegistration { + owner_thread_id, + target_thread_id, + child_depth: 0, + interval_s: 60, + prompt: "check in".to_string(), + config, + }) + .await + .expect("watchdog registration should succeed"); + + let owner_turn = owner_thread.codex.session.new_default_turn().await; + owner_thread + .codex + .session + .send_event( + owner_turn.as_ref(), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: owner_turn.sub_id.clone(), + last_agent_message: Some("root done".to_string()), + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ) + .await; + + timeout(Duration::from_secs(5), async { + loop { + let helper_spawned = harness.manager.captured_ops().into_iter().any(|(thread_id, op)| { + thread_id != owner_thread_id + && thread_id != target_thread_id + && matches!(op, Op::UserInput { items, .. } if items.iter().any(|item| match item { + UserInput::Text { text, .. } => text.contains("check in"), + UserInput::Image { .. } + | UserInput::LocalImage { .. } + | UserInput::Skill { .. } + | UserInput::Mention { .. } => false, + _ => false, + })) + }); + if helper_spawned { + break; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("watchdog should spawn a helper after the owner turn completes"); +} + +#[tokio::test] +async fn watchdog_helper_forks_owner_history() { + let harness = AgentControlHarness::new().await; + let (owner_thread_id, owner_thread) = harness.start_thread().await; + let (target_thread_id, _) = harness.start_thread().await; + let mut config = harness.config.clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + config + .mcp_servers + .set(std::collections::HashMap::from([( + "slow".to_string(), + mcp_server_config("missing-watchdog-mcp"), + )])) + .expect("test config should allow MCP servers"); + + let owner_turn = owner_thread.codex.session.new_default_turn().await; + owner_thread + .codex + .session + .record_conversation_items( + owner_turn.as_ref(), + &[assistant_message( + "previous owner response: pong 81 (118)", + Some(MessagePhase::FinalAnswer), + )], + ) + .await; + + harness + .control + .register_watchdog(WatchdogRegistration { + owner_thread_id, + target_thread_id, + child_depth: 0, + interval_s: 60, + prompt: "check in".to_string(), + config, + }) + .await + .expect("watchdog registration should succeed"); + + owner_thread + .codex + .session + .send_event( + owner_turn.as_ref(), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: owner_turn.sub_id.clone(), + last_agent_message: Some("root done".to_string()), + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ) + .await; + + let helper_thread_id = timeout(Duration::from_secs(5), async { + loop { + if let Some((thread_id, _)) = harness.manager.captured_ops().into_iter().find( + |(thread_id, op)| { + *thread_id != owner_thread_id + && *thread_id != target_thread_id + && matches!(op, Op::UserInput { items, .. } if items.iter().any(|item| match item { + UserInput::Text { text, .. } => text.contains("check in"), + UserInput::Image { .. } + | UserInput::LocalImage { .. } + | UserInput::Skill { .. } + | UserInput::Mention { .. } => false, + _ => false, + })) + }, + ) { + break thread_id; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("watchdog should spawn a helper"); + + let helper_thread = harness + .manager + .get_thread(helper_thread_id) + .await + .expect("helper thread should be registered"); + let history_items = helper_thread + .codex + .session + .clone_history() + .await + .raw_items() + .to_vec(); + assert!(history_contains_text( + &history_items, + "previous owner response: pong 81 (118)" + )); + assert!( + !helper_thread + .codex + .session + .services + .mcp_connection_manager + .read() + .await + .has_servers(), + "watchdog helpers should not start their own MCP clients" + ); +} + +#[tokio::test] +async fn watchdog_forwards_completed_helper_without_waiting_for_interval() { + let harness = AgentControlHarness::new().await; + let (owner_thread_id, owner_thread) = harness.start_thread().await; + let (target_thread_id, _) = harness.start_thread().await; + let mut config = harness.config.clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + + harness + .control + .register_watchdog(WatchdogRegistration { + owner_thread_id, + target_thread_id, + child_depth: 0, + interval_s: 3600, + prompt: "check in".to_string(), + config, + }) + .await + .expect("watchdog registration should succeed"); + + let owner_turn = owner_thread.codex.session.new_default_turn().await; + owner_thread + .codex + .session + .send_event( + owner_turn.as_ref(), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: owner_turn.sub_id.clone(), + last_agent_message: Some("root done".to_string()), + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ) + .await; + + let helper_thread_id = timeout(Duration::from_secs(5), async { + loop { + if let Some((thread_id, _)) = harness.manager.captured_ops().into_iter().find( + |(thread_id, op)| { + *thread_id != owner_thread_id + && *thread_id != target_thread_id + && matches!(op, Op::UserInput { items, .. } if items.iter().any(|item| match item { + UserInput::Text { text, .. } => text.contains("check in"), + UserInput::Image { .. } + | UserInput::LocalImage { .. } + | UserInput::Skill { .. } + | UserInput::Mention { .. } => false, + _ => false, + })) + }, + ) { + break thread_id; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("watchdog should spawn a helper"); + + let helper_thread = harness + .manager + .get_thread(helper_thread_id) + .await + .expect("helper thread should be registered"); + let helper_turn = helper_thread.codex.session.new_default_turn().await; + helper_thread + .codex + .session + .send_event( + helper_turn.as_ref(), + EventMsg::TurnComplete(TurnCompleteEvent { + turn_id: helper_turn.sub_id.clone(), + last_agent_message: Some("ping 5 (5)".to_string()), + completed_at: None, + duration_ms: None, + time_to_first_token_ms: None, + }), + ) + .await; + + let expected = InterAgentCommunication::new( + AgentPath::try_from("/root/watchdog").expect("watchdog path"), + AgentPath::root(), + Vec::new(), + "ping 5 (5)".to_string(), + /*trigger_turn*/ true, + ); + timeout(Duration::from_secs(5), async { + loop { + let captured = harness.manager.captured_ops().into_iter().any(|entry| { + entry + == ( + owner_thread_id, + Op::InterAgentCommunication { + communication: expected.clone(), + }, + ) + }); + if captured { + break; + } + sleep(Duration::from_millis(50)).await; + } + }) + .await + .expect("watchdog should forward completed helper output without interval delay"); +} + #[tokio::test] async fn send_input_errors_when_thread_missing() { let harness = AgentControlHarness::new().await; @@ -561,6 +875,105 @@ async fn send_inter_agent_communication_without_turn_queues_message_without_trig )); } +#[tokio::test] +async fn send_watchdog_wakeup_queues_mailbox_message_for_root() { + let harness = AgentControlHarness::new().await; + let (thread_id, _thread) = harness.start_thread().await; + + let submission_id = harness + .control + .send_watchdog_wakeup(thread_id, "Watchdog report: checks are green.".to_string()) + .await + .expect("send_watchdog_wakeup should succeed"); + assert!(!submission_id.is_empty()); + + let expected = InterAgentCommunication::new( + AgentPath::try_from("/root/watchdog").expect("watchdog path"), + AgentPath::root(), + Vec::new(), + "Watchdog report: checks are green.".to_string(), + /*trigger_turn*/ true, + ); + let captured = harness.manager.captured_ops().into_iter().find(|entry| { + *entry + == ( + thread_id, + Op::InterAgentCommunication { + communication: expected.clone(), + }, + ) + }); + assert_eq!( + captured, + Some(( + thread_id, + Op::InterAgentCommunication { + communication: expected, + }, + )) + ); +} + +#[tokio::test] +async fn send_watchdog_wakeup_strips_helper_prompt_scaffold() { + let harness = AgentControlHarness::new().await; + let (thread_id, _thread) = harness.start_thread().await; + let message = "# You are a Subagent\n\n\ + More importantly, you are a **watchdog check-in agent**.\n\ + Keep the root agent unblocked.\n\n\ + Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6\n\n\ + AUTOPLAN_WATCHDOG_REPORT\n\ + required_action: rerun CI"; + + let submission_id = harness + .control + .send_watchdog_wakeup(thread_id, message.to_string()) + .await + .expect("send_watchdog_wakeup should succeed"); + assert!(!submission_id.is_empty()); + + let expected = InterAgentCommunication::new( + AgentPath::try_from("/root/watchdog").expect("watchdog path"), + AgentPath::root(), + Vec::new(), + "AUTOPLAN_WATCHDOG_REPORT\nrequired_action: rerun CI".to_string(), + /*trigger_turn*/ true, + ); + let captured = harness.manager.captured_ops().into_iter().any(|entry| { + entry + == ( + thread_id, + Op::InterAgentCommunication { + communication: expected.clone(), + }, + ) + }); + assert!(captured); +} + +#[tokio::test] +async fn send_watchdog_wakeup_ignores_scaffold_without_report() { + let harness = AgentControlHarness::new().await; + let (thread_id, _thread) = harness.start_thread().await; + let message = "# You are a Subagent\n\n\ + More importantly, you are a **watchdog check-in agent**.\n\ + Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6"; + + let submission_id = harness + .control + .send_watchdog_wakeup(thread_id, message.to_string()) + .await + .expect("send_watchdog_wakeup should succeed"); + assert!(submission_id.is_empty()); + assert!( + !harness + .manager + .captured_ops() + .into_iter() + .any(|(id, op)| id == thread_id && matches!(op, Op::InterAgentCommunication { .. })) + ); +} + #[tokio::test] async fn append_message_records_assistant_message() { let harness = AgentControlHarness::new().await; @@ -747,7 +1160,6 @@ async fn spawn_agent_can_fork_parent_thread_history_with_sanitized_items() { agent_role: None, })), SpawnAgentOptions { - fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, @@ -1067,7 +1479,6 @@ async fn spawn_agent_fork_flushes_parent_rollout_before_loading_history() { agent_role: None, })), SpawnAgentOptions { - fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::FullHistory), ..Default::default() }, @@ -1177,8 +1588,8 @@ async fn spawn_agent_fork_last_n_turns_keeps_only_recent_turns() { agent_role: None, })), SpawnAgentOptions { - fork_parent_spawn_call_id: Some(parent_spawn_call_id.clone()), fork_mode: Some(SpawnAgentForkMode::LastNTurns(2)), + fork_parent_spawn_call_id: Some(parent_spawn_call_id), ..Default::default() }, ) diff --git a/codex-rs/core/src/agent/mod.rs b/codex-rs/core/src/agent/mod.rs index a60fc3004a..2855a51cbb 100644 --- a/codex-rs/core/src/agent/mod.rs +++ b/codex-rs/core/src/agent/mod.rs @@ -4,6 +4,7 @@ pub(crate) mod mailbox; mod registry; pub(crate) mod role; pub(crate) mod status; +pub(crate) mod watchdog; pub(crate) use codex_protocol::protocol::AgentStatus; pub(crate) use control::AgentControl; @@ -12,3 +13,6 @@ pub(crate) use mailbox::MailboxReceiver; pub(crate) use registry::exceeds_thread_spawn_depth_limit; pub(crate) use registry::next_thread_spawn_depth; pub(crate) use status::agent_status_from_event; +pub(crate) use watchdog::RemovedWatchdog; +pub(crate) use watchdog::WatchdogManager; +pub(crate) use watchdog::WatchdogRegistration; diff --git a/codex-rs/core/src/agent/role.rs b/codex-rs/core/src/agent/role.rs index 2ab16cd22a..33dbaecf49 100644 --- a/codex-rs/core/src/agent/role.rs +++ b/codex-rs/core/src/agent/role.rs @@ -128,6 +128,15 @@ pub(crate) fn resolve_role_config<'a>( .or_else(|| built_in::configs().get(role_name)) } +pub(crate) fn watchdog_interval_for_role(config: &Config, role_name: Option<&str>) -> Option { + let role_name = role_name.unwrap_or(DEFAULT_ROLE_NAME); + if role_name == "watchdog" { + Some(config.watchdog_interval_s) + } else { + None + } +} + fn preservation_policy(config: &Config, role_layer_toml: &TomlValue) -> (bool, bool) { let role_selects_provider = role_layer_toml.get("model_provider").is_some(); let role_selects_profile = role_layer_toml.get("profile").is_some(); @@ -342,7 +351,13 @@ pub(crate) mod spawn_tool_spec { } }) .unwrap_or_default(); - format!("{name}: {{\n{description}{locked_settings_note}\n}}") + let watchdog_note = if name == "watchdog" { + "\n- This role creates an idle-time watchdog with the configured watchdog interval." + .to_string() + } else { + String::new() + }; + format!("{name}: {{\n{description}{locked_settings_note}{watchdog_note}\n}}") } else { format!("{name}: no description") } @@ -393,6 +408,17 @@ Rules: nickname_candidates: None, } ), + ( + "watchdog".to_string(), + AgentRoleConfig { + description: Some( + "Creates an idle-time watchdog handle instead of a normal worker." + .to_string(), + ), + config_file: None, + nickname_candidates: None, + } + ), // Awaiter is temp removed // ( // "awaiter".to_string(), diff --git a/codex-rs/core/src/agent/role_tests.rs b/codex-rs/core/src/agent/role_tests.rs index 2550d58f82..aa05cf5307 100644 --- a/codex-rs/core/src/agent/role_tests.rs +++ b/codex-rs/core/src/agent/role_tests.rs @@ -773,3 +773,45 @@ fn built_in_config_file_contents_resolves_explorer_only() { None ); } + +#[tokio::test] +async fn watchdog_role_uses_builtin_interval() { + let (_home, config) = test_config_with_cli_overrides(Vec::new()).await; + + assert_eq!( + watchdog_interval_for_role(&config, Some("watchdog")), + Some(60) + ); +} + +#[tokio::test] +async fn watchdog_role_uses_top_level_interval() { + let (_home, config) = test_config_with_cli_overrides(vec![( + "watchdog_interval_s".to_string(), + TomlValue::Integer(3), + )]) + .await; + + assert_eq!( + watchdog_interval_for_role(&config, Some("watchdog")), + Some(3) + ); +} + +#[tokio::test] +async fn custom_role_cannot_define_watchdog_interval() { + let (_home, mut config) = test_config_with_cli_overrides(Vec::new()).await; + config.agent_roles.insert( + "slow_watch".to_string(), + AgentRoleConfig { + description: Some("Slow watchdog".to_string()), + config_file: None, + nickname_candidates: None, + }, + ); + + assert_eq!( + watchdog_interval_for_role(&config, Some("slow_watch")), + None + ); +} diff --git a/codex-rs/core/src/agent/watchdog.rs b/codex-rs/core/src/agent/watchdog.rs new file mode 100644 index 0000000000..8bb37aaa74 --- /dev/null +++ b/codex-rs/core/src/agent/watchdog.rs @@ -0,0 +1,645 @@ +use super::control::AgentControl; +use super::control::SpawnAgentForkMode; +use super::control::SpawnAgentOptions; +use super::registry::AgentRegistry; +use super::registry::exceeds_thread_spawn_depth_limit; +use super::status::is_final; +use crate::config::Config; +use crate::thread_manager::ThreadManagerState; +use codex_protocol::ThreadId; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; +use codex_protocol::protocol::AgentStatus; +use codex_protocol::protocol::Op; +use codex_protocol::protocol::SessionSource; +use codex_protocol::protocol::SubAgentSource; +use codex_protocol::user_input::UserInput; +use std::collections::HashMap; +use std::collections::HashSet; +use std::sync::Arc; +use std::sync::Weak; +use std::sync::atomic::AtomicBool; +use std::sync::atomic::AtomicI64; +use std::sync::atomic::Ordering; +use tokio::sync::Mutex; +use tokio::time::Duration; +use tokio::time::Instant; +use tracing::warn; + +const WATCHDOG_TICK_SECONDS: i64 = 1; +const WATCHDOG_MIN_SNOOZE_SECONDS: u64 = 30; +const WATCHDOG_MAX_SNOOZE_SECONDS: u64 = 60 * 60; + +#[derive(Clone)] +pub(crate) struct WatchdogRegistration { + pub(crate) owner_thread_id: ThreadId, + pub(crate) target_thread_id: ThreadId, + pub(crate) child_depth: i32, + pub(crate) interval_s: i64, + pub(crate) prompt: String, + pub(crate) config: Config, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct RemovedWatchdog { + pub(crate) target_thread_id: ThreadId, + pub(crate) active_helper_id: Option, +} + +struct WatchdogEntry { + registration: WatchdogRegistration, + interval: Duration, + last_trigger: Instant, + active_helper_id: Option, + snoozed_until: Option, + owner_idle_since: Option, + owner_was_running: bool, + force_due_once: bool, + generation: i64, +} + +pub(crate) struct WatchdogManager { + manager: Weak, + state: Arc, + registrations: Mutex>, + suppressed_helpers: Mutex>, + started: AtomicBool, + next_generation: AtomicI64, +} + +impl WatchdogManager { + pub(crate) fn new(manager: Weak, state: Arc) -> Arc { + Arc::new(Self { + manager, + state, + registrations: Mutex::new(HashMap::new()), + suppressed_helpers: Mutex::new(HashSet::new()), + started: AtomicBool::new(false), + next_generation: AtomicI64::new(1), + }) + } + + pub(crate) fn start(self: &Arc) { + if self + .started + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_err() + { + return; + } + + let manager = Arc::clone(self); + tokio::spawn(async move { + manager.run_loop().await; + }); + } + + pub(crate) async fn register( + self: &Arc, + registration: WatchdogRegistration, + ) -> CodexResult> { + if exceeds_thread_spawn_depth_limit( + registration.child_depth, + registration.config.agent_max_depth, + ) { + let max_depth = registration.config.agent_max_depth; + return Err(CodexErr::UnsupportedOperation(format!( + "agent depth limit reached: max depth is {max_depth}" + ))); + } + + let interval = interval_duration(registration.interval_s)?; + let generation = self.next_generation.fetch_add(1, Ordering::AcqRel); + let now = Instant::now(); + let entry = WatchdogEntry { + registration, + interval, + last_trigger: now, + active_helper_id: None, + snoozed_until: None, + owner_idle_since: Some(now), + owner_was_running: false, + force_due_once: true, + generation, + }; + + let (superseded, helper_ids_to_unsuppress) = { + let mut registrations = self.registrations.lock().await; + let superseded_targets = registrations + .iter() + .filter_map(|(target_thread_id, existing_entry)| { + (existing_entry.registration.owner_thread_id + == entry.registration.owner_thread_id + && *target_thread_id != entry.registration.target_thread_id) + .then_some(*target_thread_id) + }) + .collect::>(); + let mut superseded = Vec::new(); + let mut helper_ids_to_unsuppress = Vec::new(); + for superseded_target in superseded_targets { + if let Some(removed) = registrations.remove(&superseded_target) { + if let Some(helper_id) = removed.active_helper_id { + helper_ids_to_unsuppress.push(helper_id); + } + superseded.push(RemovedWatchdog { + target_thread_id: superseded_target, + active_helper_id: removed.active_helper_id, + }); + } + } + registrations.insert(entry.registration.target_thread_id, entry); + (superseded, helper_ids_to_unsuppress) + }; + if !helper_ids_to_unsuppress.is_empty() { + let mut suppressed_helpers = self.suppressed_helpers.lock().await; + for helper_id in helper_ids_to_unsuppress { + suppressed_helpers.remove(&helper_id); + } + } + Ok(superseded) + } + + pub(crate) async fn unregister_for_owner( + &self, + owner_thread_id: ThreadId, + ) -> Vec { + let mut registrations = self.registrations.lock().await; + let mut removed = Vec::new(); + registrations.retain(|target_thread_id, entry| { + if entry.registration.owner_thread_id == owner_thread_id { + removed.push(RemovedWatchdog { + target_thread_id: *target_thread_id, + active_helper_id: entry.active_helper_id, + }); + false + } else { + true + } + }); + removed + } + + pub(crate) async fn unregister_handle( + &self, + target_thread_id: ThreadId, + ) -> Option { + let mut registrations = self.registrations.lock().await; + registrations + .remove(&target_thread_id) + .map(|removed| RemovedWatchdog { + target_thread_id, + active_helper_id: removed.active_helper_id, + }) + } + + pub(crate) async fn is_watchdog_handle(&self, target_thread_id: ThreadId) -> bool { + self.registrations + .lock() + .await + .contains_key(&target_thread_id) + } + + #[cfg(test)] + pub(crate) async fn set_active_helper_for_tests( + &self, + target_thread_id: ThreadId, + helper_thread_id: ThreadId, + ) { + if let Some(entry) = self.registrations.lock().await.get_mut(&target_thread_id) { + entry.active_helper_id = Some(helper_thread_id); + } + } + + pub(crate) async fn owner_for_active_helper( + &self, + helper_thread_id: ThreadId, + ) -> Option { + let registrations = self.registrations.lock().await; + registrations.values().find_map(|entry| { + (entry.active_helper_id == Some(helper_thread_id)) + .then_some(entry.registration.owner_thread_id) + }) + } + + pub(crate) async fn target_for_active_helper( + &self, + helper_thread_id: ThreadId, + ) -> Option { + let registrations = self.registrations.lock().await; + registrations.iter().find_map(|(target_thread_id, entry)| { + (entry.active_helper_id == Some(helper_thread_id)).then_some(*target_thread_id) + }) + } + + async fn run_loop(self: Arc) { + let tick = tick_duration(); + loop { + self.run_once().await; + if self.manager.upgrade().is_none() { + break; + } + tokio::time::sleep(tick).await; + } + } + + pub(crate) async fn run_once(self: &Arc) { + let Some(manager_state) = self.manager.upgrade() else { + self.registrations.lock().await.clear(); + return; + }; + let snapshots = { + let registrations = self.registrations.lock().await; + registrations + .iter() + .map(|(target_id, entry)| (*target_id, entry.generation)) + .collect::>() + }; + let now = Instant::now(); + for (target_id, generation) in snapshots { + self.evaluate(&manager_state, target_id, generation, now) + .await; + } + } + + async fn evaluate( + self: &Arc, + manager_state: &Arc, + target_thread_id: ThreadId, + generation: i64, + now: Instant, + ) { + let Some(snapshot) = self.snapshot(target_thread_id, generation).await else { + return; + }; + + let owner_thread = manager_state.get_thread(snapshot.owner_thread_id).await; + let owner_status = match owner_thread.as_ref() { + Ok(thread) => thread.agent_status().await, + Err(_) => AgentStatus::NotFound, + }; + let control_for_spawn = AgentControl::from_parts( + self.manager.clone(), + Arc::clone(&self.state), + Arc::clone(self), + ); + if is_watchdog_terminated(&owner_status) { + let _ = control_for_spawn + .shutdown_live_agent(target_thread_id) + .await; + return; + } + + let owner_running = is_running(&owner_status); + let owner_idle_since = self + .update_owner_idle_state_if_generation(target_thread_id, generation, owner_running, now) + .await; + if owner_running { + return; + } + let owner_idle_since = owner_idle_since.or(snapshot.owner_idle_since); + let Some(owner_idle_since) = owner_idle_since else { + return; + }; + + if let Some(helper_id) = snapshot.active_helper_id { + let helper_status = get_status(manager_state, helper_id).await; + if !is_final(&helper_status) { + return; + } + let helper_suppressed = self.take_suppressed_helper(helper_id).await; + if let AgentStatus::Completed(Some(message)) = helper_status + && !helper_suppressed + && let Err(err) = control_for_spawn + .send_watchdog_wakeup(snapshot.owner_thread_id, message) + .await + { + warn!( + helper_id = %helper_id, + owner_thread_id = %snapshot.owner_thread_id, + "watchdog helper forward failed: {err}" + ); + } + let _ = control_for_spawn.shutdown_live_agent(helper_id).await; + self.update_after_spawn( + target_thread_id, + generation, + now, + /*active_helper_id*/ None, + ) + .await; + return; + } + + let force_due = self + .take_force_due_if_generation(target_thread_id, generation) + .await; + if !force_due && now.duration_since(owner_idle_since) < snapshot.interval { + return; + } + + if let Some(snoozed_until) = snapshot.snoozed_until { + if now < snoozed_until { + return; + } + self.clear_snooze_if_generation(target_thread_id, generation) + .await; + } + + if !force_due && now.duration_since(snapshot.last_trigger) < snapshot.interval { + return; + } + + let session_source = SessionSource::SubAgent(SubAgentSource::ThreadSpawn { + parent_thread_id: snapshot.owner_thread_id, + depth: snapshot.child_depth, + agent_path: None, + agent_nickname: None, + agent_role: Some("watchdog".to_string()), + }); + 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, + "watchdog helper MCP server clearing failed: {err}" + ); + self.update_after_spawn( + target_thread_id, + generation, + now, + /*active_helper_id*/ None, + ) + .await; + return; + } + let helper_prompt = watchdog_helper_prompt(snapshot.owner_thread_id, &snapshot.prompt); + let spawn_result = control_for_spawn + .spawn_agent_with_metadata( + helper_config, + Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: helper_prompt, + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + }, + Some(session_source), + SpawnAgentOptions { + fork_parent_spawn_call_id: None, + fork_mode: Some(SpawnAgentForkMode::FullHistory), + environments: None, + }, + ) + .await; + + match spawn_result { + Ok(helper) => { + self.update_after_spawn(target_thread_id, generation, now, Some(helper.thread_id)) + .await; + } + Err(err) => { + warn!("watchdog spawn failed for target {target_thread_id}: {err}"); + self.update_after_spawn( + target_thread_id, + generation, + now, + /*active_helper_id*/ None, + ) + .await; + } + } + } + + async fn snapshot( + &self, + target_thread_id: ThreadId, + generation: i64, + ) -> Option { + let registrations = self.registrations.lock().await; + let entry = registrations.get(&target_thread_id)?; + if entry.generation != generation { + return None; + } + Some(WatchdogSnapshot { + owner_thread_id: entry.registration.owner_thread_id, + child_depth: entry.registration.child_depth, + prompt: entry.registration.prompt.clone(), + config: entry.registration.config.clone(), + interval: entry.interval, + last_trigger: entry.last_trigger, + active_helper_id: entry.active_helper_id, + snoozed_until: entry.snoozed_until, + owner_idle_since: entry.owner_idle_since, + }) + } + + async fn update_owner_idle_state_if_generation( + &self, + target_thread_id: ThreadId, + generation: i64, + owner_running: bool, + now: Instant, + ) -> Option { + let mut registrations = self.registrations.lock().await; + let entry = registrations.get_mut(&target_thread_id)?; + if entry.generation != generation { + return None; + } + if owner_running { + entry.owner_idle_since = None; + entry.snoozed_until = None; + entry.owner_was_running = true; + return None; + } + if entry.owner_was_running || entry.owner_idle_since.is_none() { + entry.owner_idle_since = Some(now); + } + entry.owner_was_running = false; + entry.owner_idle_since + } + + async fn take_force_due_if_generation( + &self, + target_thread_id: ThreadId, + generation: i64, + ) -> bool { + let mut registrations = self.registrations.lock().await; + let Some(entry) = registrations.get_mut(&target_thread_id) else { + return false; + }; + if entry.generation != generation || !entry.force_due_once { + return false; + } + entry.force_due_once = false; + true + } + + async fn clear_snooze_if_generation(&self, target_thread_id: ThreadId, generation: i64) { + let mut registrations = self.registrations.lock().await; + let Some(entry) = registrations.get_mut(&target_thread_id) else { + return; + }; + if entry.generation == generation { + entry.snoozed_until = None; + } + } + + pub(crate) async fn take_suppressed_helper(&self, helper_thread_id: ThreadId) -> bool { + self.suppressed_helpers + .lock() + .await + .remove(&helper_thread_id) + } + + #[cfg(test)] + pub(crate) async fn helper_is_suppressed_for_tests(&self, helper_thread_id: ThreadId) -> bool { + self.suppressed_helpers + .lock() + .await + .contains(&helper_thread_id) + } + + pub(crate) async fn snooze_active_helper( + &self, + helper_thread_id: ThreadId, + requested_delay_seconds: Option, + ) -> Option { + let result = { + let mut registrations = self.registrations.lock().await; + let (target_thread_id, entry) = + registrations + .iter_mut() + .find_map(|(target_thread_id, entry)| { + (entry.active_helper_id == Some(helper_thread_id)) + .then_some((*target_thread_id, entry)) + })?; + let delay_seconds = requested_delay_seconds + .map(|seconds| { + seconds.clamp(WATCHDOG_MIN_SNOOZE_SECONDS, WATCHDOG_MAX_SNOOZE_SECONDS) + }) + .unwrap_or_else(|| entry.interval.as_secs().max(1)); + entry.snoozed_until = Some(Instant::now() + Duration::from_secs(delay_seconds)); + entry.active_helper_id = None; + WatchdogSnoozeResult { + target_thread_id, + delay_seconds, + } + }; + self.suppressed_helpers + .lock() + .await + .insert(helper_thread_id); + Some(result) + } + + async fn update_after_spawn( + &self, + target_thread_id: ThreadId, + generation: i64, + now: Instant, + active_helper_id: Option, + ) { + let mut registrations = self.registrations.lock().await; + let Some(entry) = registrations.get_mut(&target_thread_id) else { + return; + }; + if entry.generation != generation { + return; + } + entry.last_trigger = now; + entry.active_helper_id = active_helper_id; + entry.owner_idle_since = Some(now); + entry.snoozed_until = None; + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct WatchdogSnoozeResult { + pub(crate) target_thread_id: ThreadId, + pub(crate) delay_seconds: u64, +} + +#[derive(Clone)] +struct WatchdogSnapshot { + owner_thread_id: ThreadId, + child_depth: i32, + prompt: String, + config: Config, + interval: Duration, + last_trigger: Instant, + active_helper_id: Option, + snoozed_until: Option, + owner_idle_since: Option, +} + +fn is_running(status: &AgentStatus) -> bool { + matches!(status, AgentStatus::PendingInit | AgentStatus::Running) +} + +fn is_watchdog_terminated(status: &AgentStatus) -> bool { + matches!(status, AgentStatus::Shutdown | AgentStatus::NotFound) +} + +async fn get_status(manager_state: &Arc, thread_id: ThreadId) -> AgentStatus { + match manager_state.get_thread(thread_id).await { + Ok(thread) => thread.agent_status().await, + Err(_) => AgentStatus::NotFound, + } +} + +fn interval_duration(interval_s: i64) -> CodexResult { + if interval_s <= 0 { + return Err(CodexErr::UnsupportedOperation( + "watchdog interval must be greater than zero".to_string(), + )); + } + Ok(Duration::from_secs(interval_s as u64)) +} + +fn tick_duration() -> Duration { + Duration::from_secs(WATCHDOG_TICK_SECONDS as u64) +} + +fn watchdog_helper_prompt(owner_thread_id: ThreadId, prompt: &str) -> String { + if prompt.trim().is_empty() { + format!("Target agent id: {owner_thread_id}") + } else { + format!("Target agent id: {owner_thread_id}\n\n{prompt}") + } +} + +#[cfg(test)] +mod tests { + use super::watchdog_helper_prompt; + use codex_protocol::ThreadId; + + #[test] + fn watchdog_helper_prompt_includes_owner_and_task() { + let owner_thread_id = ThreadId::default(); + assert_eq!( + watchdog_helper_prompt(owner_thread_id, "check in"), + format!("Target agent id: {owner_thread_id}\n\ncheck in") + ); + } + + #[test] + fn owner_completed_status_does_not_terminate_watchdog() { + assert!(!super::is_watchdog_terminated( + &codex_protocol::protocol::AgentStatus::Completed(None) + )); + assert!(!super::is_watchdog_terminated( + &codex_protocol::protocol::AgentStatus::Interrupted + )); + assert!(!super::is_watchdog_terminated( + &codex_protocol::protocol::AgentStatus::Errored("turn failed".to_string()) + )); + assert!(super::is_watchdog_terminated( + &codex_protocol::protocol::AgentStatus::Shutdown + )); + assert!(super::is_watchdog_terminated( + &codex_protocol::protocol::AgentStatus::NotFound + )); + } +} diff --git a/codex-rs/core/src/config/agent_roles.rs b/codex-rs/core/src/config/agent_roles.rs index abdef33e7d..0a5d83bdfd 100644 --- a/codex-rs/core/src/config/agent_roles.rs +++ b/codex-rs/core/src/config/agent_roles.rs @@ -266,6 +266,15 @@ pub(crate) fn parse_agent_role_file_contents( parsed.config.developer_instructions.as_deref(), role_name_hint.is_none(), )?; + if parsed.config.watchdog_interval_s.is_some() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!( + "agent role file {} cannot set watchdog_interval_s; set it at the top level of config.toml", + role_file_label.display() + ), + )); + } let role_name = parsed .name diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 1352de991e..1a3386245c 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -6381,6 +6381,7 @@ async fn test_precedence_fixture_with_o3_profile() -> std::io::Result<()> { agent_max_threads: DEFAULT_AGENT_MAX_THREADS, agent_max_depth: DEFAULT_AGENT_MAX_DEPTH, agent_roles: BTreeMap::new(), + watchdog_interval_s: DEFAULT_WATCHDOG_INTERVAL_S, memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, agent_interrupt_message_enabled: true, @@ -6583,6 +6584,7 @@ async fn test_precedence_fixture_with_gpt3_profile() -> std::io::Result<()> { agent_max_threads: DEFAULT_AGENT_MAX_THREADS, agent_max_depth: DEFAULT_AGENT_MAX_DEPTH, agent_roles: BTreeMap::new(), + watchdog_interval_s: DEFAULT_WATCHDOG_INTERVAL_S, memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, agent_interrupt_message_enabled: true, @@ -6739,6 +6741,7 @@ async fn test_precedence_fixture_with_zdr_profile() -> std::io::Result<()> { agent_max_threads: DEFAULT_AGENT_MAX_THREADS, agent_max_depth: DEFAULT_AGENT_MAX_DEPTH, agent_roles: BTreeMap::new(), + watchdog_interval_s: DEFAULT_WATCHDOG_INTERVAL_S, memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, agent_interrupt_message_enabled: true, @@ -6880,6 +6883,7 @@ async fn test_precedence_fixture_with_gpt5_profile() -> std::io::Result<()> { agent_max_threads: DEFAULT_AGENT_MAX_THREADS, agent_max_depth: DEFAULT_AGENT_MAX_DEPTH, agent_roles: BTreeMap::new(), + watchdog_interval_s: DEFAULT_WATCHDOG_INTERVAL_S, memories: MemoriesConfig::default(), agent_job_max_runtime_seconds: DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS, agent_interrupt_message_enabled: true, diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 20b2a923f8..f449bd5035 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -175,6 +175,7 @@ pub(crate) const DEFAULT_MULTI_AGENT_V2_MIN_WAIT_TIMEOUT_MS: i64 = 10_000; pub(crate) const MAX_MULTI_AGENT_V2_WAIT_TIMEOUT_MS: i64 = 3600 * 1000; pub(crate) const DEFAULT_AGENT_MAX_DEPTH: i32 = 1; pub(crate) const DEFAULT_AGENT_JOB_MAX_RUNTIME_SECONDS: Option = None; +pub(crate) const DEFAULT_WATCHDOG_INTERVAL_S: i64 = 60; const LOCAL_DEV_BUILD_VERSION: &str = "0.0.0"; pub const CONFIG_TOML_FILE: &str = "config.toml"; @@ -616,6 +617,9 @@ pub struct Config { /// User-defined role declarations keyed by role name. pub agent_roles: BTreeMap, + /// Singleton interval in seconds for the built-in watchdog role. + pub watchdog_interval_s: i64, + /// Memories subsystem settings. pub memories: MemoriesConfig, @@ -2492,6 +2496,15 @@ impl Config { let agent_roles = agent_roles::load_agent_roles(fs, &cfg, &config_layer_stack, &mut startup_warnings) .await?; + let watchdog_interval_s = cfg + .watchdog_interval_s + .unwrap_or(DEFAULT_WATCHDOG_INTERVAL_S); + if watchdog_interval_s <= 0 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "watchdog_interval_s must be greater than zero", + )); + } let openai_base_url = cfg .openai_base_url @@ -2959,6 +2972,7 @@ impl Config { agent_max_threads, agent_max_depth, agent_roles, + watchdog_interval_s, memories: cfg.memories.unwrap_or_default().into(), agent_job_max_runtime_seconds, agent_interrupt_message_enabled, diff --git a/codex-rs/core/src/tools/handlers/multi_agents.rs b/codex-rs/core/src/tools/handlers/multi_agents.rs index 2d70d3e92d..9c3ae75c1d 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents.rs @@ -62,12 +62,16 @@ pub(crate) use resume_agent::Handler as ResumeAgentHandler; pub(crate) use send_input::Handler as SendInputHandler; pub(crate) use spawn::Handler as SpawnAgentHandler; pub(crate) use wait::Handler as WaitAgentHandler; +pub(crate) use watchdog_self_close::Handler as WatchdogSelfCloseHandler; +pub(crate) use watchdog_snooze::Handler as WatchdogSnoozeHandler; pub(crate) mod close_agent; mod resume_agent; mod send_input; mod spawn; pub(crate) mod wait; +mod watchdog_self_close; +mod watchdog_snooze; #[cfg(test)] #[path = "multi_agents_tests.rs"] 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 777cb9be1c..c36f7b8941 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/spawn.rs @@ -1,4 +1,7 @@ use super::*; +use crate::agent::RemovedWatchdog; +use crate::agent::WatchdogRegistration; +use crate::agent::control::LiveAgent; use crate::agent::control::SpawnAgentForkMode; use crate::agent::control::SpawnAgentOptions; use crate::agent::control::render_input_preview; @@ -6,7 +9,14 @@ use crate::agent::exceeds_thread_spawn_depth_limit; use crate::agent::next_thread_spawn_depth; use crate::agent::role::DEFAULT_ROLE_NAME; use crate::agent::role::apply_role_to_config; +use crate::agent::role::watchdog_interval_for_role; +use crate::config::Config; use crate::session::turn_context::TurnEnvironment; +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; @@ -41,6 +51,18 @@ impl ToolHandler for Handler { let session_source = turn.session_source.clone(); let child_depth = next_thread_spawn_depth(&session_source); let max_depth = turn.config.agent_max_depth; + let watchdog_interval_s = watchdog_interval_for_role(&turn.config, role_name); + let is_watchdog = watchdog_interval_s.is_some(); + if is_watchdog && !turn.config.features.enabled(Feature::AgentWatchdog) { + return Err(FunctionCallError::RespondToModel( + "watchdogs are disabled".to_string(), + )); + } + if is_watchdog && matches!(session_source, SessionSource::SubAgent(_)) { + return Err(FunctionCallError::RespondToModel( + "watchdogs can only be spawned by root agents".to_string(), + )); + } if exceeds_thread_spawn_depth_limit(child_depth, max_depth) { return Err(FunctionCallError::RespondToModel( "Agent depth limit reached. Solve the task yourself.".to_string(), @@ -83,30 +105,54 @@ impl ToolHandler for Handler { apply_spawn_agent_runtime_overrides(&mut config, turn.as_ref())?; apply_spawn_agent_overrides(&mut config, child_depth); - let result = Box::pin( - session.services.agent_control.spawn_agent_with_metadata( + let spawn_source = thread_spawn_source( + session.conversation_id, + &turn.session_source, + child_depth, + role_name, + /*task_name*/ None, + )?; + let result = if let Some(watchdog_interval_s) = watchdog_interval_s { + let thread_id = spawn_watchdog( + &session.services.agent_control, config, - input_items, - Some(thread_spawn_source( - session.conversation_id, - &turn.session_source, - child_depth, - role_name, - /*task_name*/ None, - )?), - SpawnAgentOptions { - fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), - fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), - environments: Some( - turn.environments - .iter() - .map(TurnEnvironment::selection) - .collect(), - ), - }, - ), - ) - .await + prompt.clone(), + session.conversation_id, + child_depth, + watchdog_interval_s, + spawn_source, + ) + .await + .map_err(collab_spawn_error)?; + Ok(LiveAgent { + thread_id, + metadata: session + .services + .agent_control + .get_agent_metadata(thread_id) + .unwrap_or_default(), + status: session.services.agent_control.get_status(thread_id).await, + }) + } else { + Box::pin( + session.services.agent_control.spawn_agent_with_metadata( + config, + input_items, + Some(spawn_source), + SpawnAgentOptions { + fork_parent_spawn_call_id: args.fork_context.then(|| call_id.clone()), + fork_mode: args.fork_context.then_some(SpawnAgentForkMode::FullHistory), + environments: Some( + turn.environments + .iter() + .map(TurnEnvironment::selection) + .collect(), + ), + }, + ), + ) + .await + } .map_err(collab_spawn_error); let (new_thread_id, new_agent_metadata, status) = match &result { Ok(spawned_agent) => ( @@ -215,3 +261,65 @@ impl ToolOutput for SpawnAgentResult { tool_output_code_mode_result(self, "spawn_agent") } } + +async fn spawn_watchdog( + agent_control: &crate::agent::AgentControl, + config: Config, + prompt: String, + owner_thread_id: ThreadId, + child_depth: i32, + interval_s: i64, + spawn_source: SessionSource, +) -> codex_protocol::error::Result { + let mut handle_config = config.clone(); + handle_config + .mcp_servers + .set(HashMap::new()) + .map_err(|err| { + codex_protocol::error::CodexErr::UnsupportedOperation(format!( + "failed to clear watchdog MCP servers: {err}" + )) + })?; + let target_thread_id = agent_control + .spawn_agent(handle_config, Op::Interrupt, Some(spawn_source)) + .await?; + let superseded_before_register = agent_control + .unregister_watchdogs_for_owner(owner_thread_id) + .await; + shutdown_removed_watchdogs(agent_control, superseded_before_register).await; + let registration = WatchdogRegistration { + owner_thread_id, + target_thread_id, + child_depth, + interval_s, + prompt, + config, + }; + let superseded_after_register = match agent_control.register_watchdog(registration).await { + Ok(removed) => removed, + Err(err) => { + let _ = agent_control.close_agent(target_thread_id).await; + return Err(err); + } + }; + shutdown_removed_watchdogs(agent_control, superseded_after_register).await; + Ok(target_thread_id) +} + +async fn shutdown_removed_watchdogs( + agent_control: &crate::agent::AgentControl, + removed_watchdogs: Vec, +) { + let mut thread_ids = HashSet::new(); + for removed in removed_watchdogs { + thread_ids.insert(removed.target_thread_id); + if let Some(helper_id) = removed.active_helper_id { + thread_ids.insert(helper_id); + } + } + let mut thread_ids = thread_ids.into_iter().collect::>(); + thread_ids.sort_by_key(ToString::to_string); + for thread_id in thread_ids { + let _ = agent_control.close_agent(thread_id).await; + } +} diff --git a/codex-rs/core/src/tools/handlers/multi_agents/wait.rs b/codex-rs/core/src/tools/handlers/multi_agents/wait.rs index 77fa5f83a2..9f2d05fd62 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents/wait.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents/wait.rs @@ -58,6 +58,31 @@ impl ToolHandler for Handler { agent_role: agent_metadata.agent_role, }); } + let mut watchdog_statuses = HashMap::new(); + for receiver_thread_id in &receiver_thread_ids { + if session + .services + .agent_control + .is_watchdog_handle(*receiver_thread_id) + .await + { + watchdog_statuses.insert( + *receiver_thread_id, + session + .services + .agent_control + .get_status(*receiver_thread_id) + .await, + ); + } + } + if watchdog_statuses.len() == receiver_thread_ids.len() { + let agent_statuses = build_wait_agent_statuses(&watchdog_statuses, &receiver_agents); + let content = serde_json::to_string(&agent_statuses).unwrap_or_default(); + return Err(FunctionCallError::RespondToModel(format!( + "wait_agent cannot be used to wait for watchdog check-ins. You passed only watchdog handle ids. Watchdog check-ins only happen after the current turn ends and the owner thread is idle for at least watchdog_interval_s. `wait_agent` on a watchdog handle is status-only and cannot confirm a new check-in. Current watchdog handle statuses: {content}" + ))); + } let timeout_ms = args.timeout_ms.unwrap_or(DEFAULT_WAIT_TIMEOUT_MS); let timeout_ms = match timeout_ms { diff --git a/codex-rs/core/src/tools/handlers/multi_agents/watchdog_self_close.rs b/codex-rs/core/src/tools/handlers/multi_agents/watchdog_self_close.rs new file mode 100644 index 0000000000..d1e1101cfd --- /dev/null +++ b/codex-rs/core/src/tools/handlers/multi_agents/watchdog_self_close.rs @@ -0,0 +1,99 @@ +use super::*; + +pub(crate) struct Handler; + +impl ToolHandler for Handler { + type Output = WatchdogSelfCloseResult; + + fn kind(&self) -> ToolKind { + ToolKind::Function + } + + fn matches_kind(&self, payload: &ToolPayload) -> bool { + matches!(payload, ToolPayload::Function { .. }) + } + + async fn handle(&self, invocation: ToolInvocation) -> Result { + let ToolInvocation { + session, payload, .. + } = invocation; + let arguments = function_arguments(payload)?; + let args: WatchdogSelfCloseArgs = parse_arguments(&arguments)?; + let helper_thread_id = session.conversation_id; + let Some(owner_thread_id) = session + .services + .agent_control + .watchdog_owner_for_active_helper(helper_thread_id) + .await + else { + return Err(FunctionCallError::RespondToModel( + "watchdog_self_close is only available in watchdog check-in threads.".to_string(), + )); + }; + let Some(target_thread_id) = session + .services + .agent_control + .watchdog_target_for_active_helper(helper_thread_id) + .await + else { + return Err(FunctionCallError::RespondToModel( + "watchdog_self_close is only available in watchdog check-in threads.".to_string(), + )); + }; + + let status = session + .services + .agent_control + .get_status(target_thread_id) + .await; + + if let Some(message) = args.message + && !message.trim().is_empty() + { + let _ = session + .services + .agent_control + .send_watchdog_wakeup(owner_thread_id, message) + .await; + } + + session + .services + .agent_control + .close_agent(target_thread_id) + .await + .map_err(|err| collab_agent_error(target_thread_id, err))?; + + Ok(WatchdogSelfCloseResult { + previous_status: status, + }) + } +} + +#[derive(Debug, Deserialize)] +struct WatchdogSelfCloseArgs { + message: Option, +} + +#[derive(Debug, Serialize)] +pub(crate) struct WatchdogSelfCloseResult { + previous_status: AgentStatus, +} + +impl ToolOutput for WatchdogSelfCloseResult { + fn log_preview(&self) -> String { + tool_output_json_text(self, "watchdog_self_close") + } + + fn success_for_logging(&self) -> bool { + true + } + + fn to_response_item(&self, call_id: &str, payload: &ToolPayload) -> ResponseInputItem { + tool_output_response_item(call_id, payload, self, Some(true), "watchdog_self_close") + } + + fn code_mode_result(&self, _payload: &ToolPayload) -> JsonValue { + tool_output_code_mode_result(self, "watchdog_self_close") + } +} diff --git a/codex-rs/core/src/tools/handlers/multi_agents/watchdog_snooze.rs b/codex-rs/core/src/tools/handlers/multi_agents/watchdog_snooze.rs new file mode 100644 index 0000000000..739fa08fde --- /dev/null +++ b/codex-rs/core/src/tools/handlers/multi_agents/watchdog_snooze.rs @@ -0,0 +1,68 @@ +use super::*; + +pub(crate) struct Handler; + +impl ToolHandler for Handler { + type Output = WatchdogSnoozeResult; + + fn kind(&self) -> ToolKind { + ToolKind::Function + } + + fn matches_kind(&self, payload: &ToolPayload) -> bool { + matches!(payload, ToolPayload::Function { .. }) + } + + async fn handle(&self, invocation: ToolInvocation) -> Result { + let ToolInvocation { + session, payload, .. + } = invocation; + let arguments = function_arguments(payload)?; + let args: WatchdogSnoozeArgs = parse_arguments(&arguments)?; + let Some(result) = session + .services + .agent_control + .snooze_watchdog_helper(session.conversation_id, args.delay_seconds) + .await + else { + return Err(FunctionCallError::RespondToModel( + "watchdog.snooze is only available in watchdog check-in threads.".to_string(), + )); + }; + let _ = args.reason; + Ok(WatchdogSnoozeResult { + target_thread_id: result.target_thread_id.to_string(), + delay_seconds: result.delay_seconds, + }) + } +} + +#[derive(Debug, Deserialize)] +struct WatchdogSnoozeArgs { + delay_seconds: Option, + reason: Option, +} + +#[derive(Debug, Serialize)] +pub(crate) struct WatchdogSnoozeResult { + target_thread_id: String, + delay_seconds: u64, +} + +impl ToolOutput for WatchdogSnoozeResult { + fn log_preview(&self) -> String { + tool_output_json_text(self, "snooze") + } + + fn success_for_logging(&self) -> bool { + true + } + + fn to_response_item(&self, call_id: &str, payload: &ToolPayload) -> ResponseInputItem { + tool_output_response_item(call_id, payload, self, Some(true), "snooze") + } + + fn code_mode_result(&self, _payload: &ToolPayload) -> JsonValue { + tool_output_code_mode_result(self, "snooze") + } +} 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 61dc77eb36..66821e9e29 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -1,5 +1,6 @@ use super::*; use crate::ThreadManager; +use crate::agent::WatchdogRegistration; use crate::config::AgentRoleConfig; use crate::config::DEFAULT_AGENT_MAX_DEPTH; use crate::function_tool::FunctionCallError; @@ -240,7 +241,8 @@ async fn spawn_agent_uses_explorer_role_and_preserves_approval_policy() { let (mut session, mut turn) = make_session_and_context().await; let manager = thread_manager(); - session.services.agent_control = manager.agent_control(); + let agent_control = manager.agent_control(); + session.services.agent_control = agent_control.clone(); let mut config = (*turn.config).clone(); let provider_info = built_in_model_providers(/* openai_base_url */ /*openai_base_url*/ None)["ollama"].clone(); @@ -525,6 +527,234 @@ async fn spawn_agent_returns_agent_id_without_task_name() { assert_eq!(success, Some(true)); } +#[tokio::test] +async fn spawn_agent_watchdog_role_returns_inert_handle() { + let (mut session, mut turn) = make_session_and_context().await; + let manager = thread_manager(); + let agent_control = manager.agent_control(); + session.services.agent_control = agent_control.clone(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + turn.config = Arc::new(config); + + let output = SpawnAgentHandler + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "spawn_agent", + function_payload(json!({ + "message": "check in later", + "agent_type": "watchdog" + })), + )) + .await + .expect("spawn_agent should succeed"); + let (content, success) = expect_text_output(output); + let result: serde_json::Value = + serde_json::from_str(&content).expect("spawn_agent result should be json"); + let agent_id = parse_agent_id( + result["agent_id"] + .as_str() + .expect("spawn_agent result should include agent_id"), + ); + + assert_eq!(success, Some(true)); + assert_eq!( + agent_control.get_status(agent_id).await, + AgentStatus::PendingInit + ); + let ops_for_agent = manager + .captured_ops() + .into_iter() + .filter_map(|(thread_id, op)| (thread_id == agent_id).then_some(op)) + .collect::>(); + assert_eq!(ops_for_agent, vec![Op::Interrupt]); + assert!(agent_control.is_watchdog_handle(agent_id).await); +} + +#[tokio::test] +async fn watchdog_snooze_rejects_non_watchdog_thread() { + let (session, turn) = make_session_and_context().await; + + let err = WatchdogSnoozeHandler + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "snooze", + function_payload(json!({"delay_seconds": 60})), + )) + .await + .expect_err("non-watchdog thread should not snooze"); + + assert_eq!( + err, + FunctionCallError::RespondToModel( + "watchdog.snooze is only available in watchdog check-in threads.".to_string(), + ) + ); +} + +#[tokio::test] +async fn watchdog_snooze_suppresses_helper_and_clears_active_helper() { + let (mut session, mut turn) = make_session_and_context().await; + let manager = thread_manager(); + let agent_control = manager.agent_control(); + let owner = manager + .start_thread((*turn.config).clone()) + .await + .expect("owner thread should start"); + let target = manager + .start_thread((*turn.config).clone()) + .await + .expect("watchdog handle should start"); + let helper_thread_id = session.conversation_id; + session.services.agent_control = agent_control.clone(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + turn.config = Arc::new(config.clone()); + + agent_control + .register_watchdog(WatchdogRegistration { + owner_thread_id: owner.thread_id, + target_thread_id: target.thread_id, + child_depth: 0, + interval_s: 60, + prompt: "check in".to_string(), + config, + }) + .await + .expect("watchdog registration should succeed"); + agent_control + .set_watchdog_active_helper_for_tests(target.thread_id, helper_thread_id) + .await; + + let output = WatchdogSnoozeHandler + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "snooze", + function_payload(json!({"delay_seconds": 5})), + )) + .await + .expect("watchdog helper should snooze"); + let (content, success) = expect_text_output(output); + let result: serde_json::Value = + serde_json::from_str(&content).expect("snooze result should be json"); + + assert_eq!(success, Some(true)); + assert_eq!(result["target_thread_id"], target.thread_id.to_string()); + assert_eq!(result["delay_seconds"], 30); + assert_eq!( + agent_control + .watchdog_target_for_active_helper(helper_thread_id) + .await, + None + ); + assert!( + agent_control + .watchdog_helper_is_suppressed_for_tests(helper_thread_id) + .await + ); +} + +#[tokio::test] +async fn watchdog_self_close_rejects_non_watchdog_thread() { + let (session, turn) = make_session_and_context().await; + + let err = WatchdogSelfCloseHandler + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "watchdog_self_close", + function_payload(json!({})), + )) + .await + .expect_err("non-watchdog thread should not self-close"); + + assert_eq!( + err, + FunctionCallError::RespondToModel( + "watchdog_self_close is only available in watchdog check-in threads.".to_string(), + ) + ); +} + +#[tokio::test] +async fn watchdog_self_close_notifies_owner_and_unregisters_handle() { + let (mut session, mut turn) = make_session_and_context().await; + let manager = thread_manager(); + let agent_control = manager.agent_control(); + let owner = manager + .start_thread((*turn.config).clone()) + .await + .expect("owner thread should start"); + let target = manager + .start_thread((*turn.config).clone()) + .await + .expect("watchdog handle should start"); + let helper_thread_id = session.conversation_id; + session.services.agent_control = agent_control.clone(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + turn.config = Arc::new(config.clone()); + + agent_control + .register_watchdog(WatchdogRegistration { + owner_thread_id: owner.thread_id, + target_thread_id: target.thread_id, + child_depth: 0, + interval_s: 60, + prompt: "check in".to_string(), + config, + }) + .await + .expect("watchdog registration should succeed"); + agent_control + .set_watchdog_active_helper_for_tests(target.thread_id, helper_thread_id) + .await; + + let output = WatchdogSelfCloseHandler + .handle(invocation( + Arc::new(session), + Arc::new(turn), + "watchdog_self_close", + function_payload(json!({"message": "watchdog done"})), + )) + .await + .expect("watchdog helper should self-close"); + let (content, success) = expect_text_output(output); + let result: serde_json::Value = + serde_json::from_str(&content).expect("self-close result should be json"); + + assert_eq!(success, Some(true)); + assert_eq!(result["previous_status"], json!("pending_init")); + assert!(!agent_control.is_watchdog_handle(target.thread_id).await); + assert_eq!( + agent_control.get_status(target.thread_id).await, + AgentStatus::NotFound + ); + let expected = InterAgentCommunication::new( + AgentPath::try_from("/root/watchdog").expect("watchdog path"), + AgentPath::root(), + Vec::new(), + "watchdog done".to_string(), + /*trigger_turn*/ true, + ); + assert!(manager.captured_ops().into_iter().any(|(thread_id, op)| { + thread_id == owner.thread_id + && matches!(op, Op::InterAgentCommunication { communication } if communication == expected) + })); +} + #[tokio::test] async fn multi_agent_v2_spawn_requires_task_name() { let (mut session, mut turn) = make_session_and_context().await; @@ -2524,6 +2754,62 @@ async fn wait_agent_returns_not_found_for_missing_agents() { assert_eq!(success, None); } +#[tokio::test] +async fn wait_agent_rejects_only_watchdog_handles() { + let (mut session, mut turn) = make_session_and_context().await; + let manager = thread_manager(); + let agent_control = manager.agent_control(); + let owner = manager + .start_thread((*turn.config).clone()) + .await + .expect("owner thread should start"); + let target = manager + .start_thread((*turn.config).clone()) + .await + .expect("watchdog handle should start"); + session.services.agent_control = agent_control.clone(); + let mut config = (*turn.config).clone(); + config + .features + .enable(Feature::AgentWatchdog) + .expect("test config should allow feature update"); + turn.config = Arc::new(config); + let session = Arc::new(session); + let turn = Arc::new(turn); + + agent_control + .register_watchdog(WatchdogRegistration { + owner_thread_id: owner.thread_id, + target_thread_id: target.thread_id, + child_depth: 0, + interval_s: 60, + prompt: "check in later".to_string(), + config: (*turn.config).clone(), + }) + .await + .expect("watchdog registration should succeed"); + + let Err(err) = WaitAgentHandler + .handle(invocation( + session, + turn, + "wait_agent", + function_payload(json!({ + "targets": [target.thread_id.to_string()], + "timeout_ms": 1000 + })), + )) + .await + else { + panic!("wait_agent should reject watchdog-only targets"); + }; + let FunctionCallError::RespondToModel(message) = err else { + panic!("expected model-facing error"); + }; + assert!(message.contains("watchdog handle ids")); + assert!(message.contains("pending_init")); +} + #[tokio::test] async fn wait_agent_times_out_when_status_is_not_final() { let (mut session, turn) = make_session_and_context().await; diff --git a/codex-rs/core/src/tools/spec.rs b/codex-rs/core/src/tools/spec.rs index 479dcc382f..e93bba3a50 100644 --- a/codex-rs/core/src/tools/spec.rs +++ b/codex-rs/core/src/tools/spec.rs @@ -1,6 +1,8 @@ use crate::shell::Shell; use crate::shell::ShellType; use crate::tools::handlers::agent_jobs::BatchJobHandler; +use crate::tools::handlers::multi_agents::WatchdogSelfCloseHandler; +use crate::tools::handlers::multi_agents::WatchdogSnoozeHandler; use crate::tools::handlers::multi_agents_common::DEFAULT_WAIT_TIMEOUT_MS; use crate::tools::handlers::multi_agents_common::MAX_WAIT_TIMEOUT_MS; use crate::tools::handlers::multi_agents_common::MIN_WAIT_TIMEOUT_MS; @@ -296,6 +298,12 @@ pub(crate) fn build_specs_with_discoverable_tools( ToolHandlerKind::WaitAgentV2 => { builder.register_handler(handler.name, Arc::new(WaitAgentHandlerV2)); } + ToolHandlerKind::WatchdogSelfClose => { + builder.register_handler(handler.name, Arc::new(WatchdogSelfCloseHandler)); + } + ToolHandlerKind::WatchdogSnooze => { + builder.register_handler(handler.name, Arc::new(WatchdogSnoozeHandler)); + } } } if let Some(deferred_mcp_tools) = deferred_mcp_tools.as_ref() { diff --git a/codex-rs/features/src/lib.rs b/codex-rs/features/src/lib.rs index bf384672a1..477fdd6088 100644 --- a/codex-rs/features/src/lib.rs +++ b/codex-rs/features/src/lib.rs @@ -144,6 +144,8 @@ pub enum Feature { Collab, /// Enable task-path-based multi-agent routing. MultiAgentV2, + /// Enable idle-time watchdog handles for root agents. + AgentWatchdog, /// Enable CSV-backed agent job tools. SpawnCsv, /// Enable apps. @@ -885,6 +887,12 @@ pub const FEATURES: &[FeatureSpec] = &[ stage: Stage::UnderDevelopment, default_enabled: false, }, + FeatureSpec { + id: Feature::AgentWatchdog, + key: "agent_watchdog", + stage: Stage::UnderDevelopment, + default_enabled: false, + }, FeatureSpec { id: Feature::SpawnCsv, key: "enable_fanout", diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index cc2262512d..c74e32f21e 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -212,6 +212,7 @@ fn new_config(model: Option, arg0_paths: Arg0DispatchPaths) -> anyhow::R agent_interrupt_message_enabled: false, agent_max_depth: 1, agent_roles: BTreeMap::new(), + watchdog_interval_s: 60, memories: MemoriesConfig::default(), sqlite_home: codex_home.to_path_buf(), log_dir: codex_home.join("log").to_path_buf(), diff --git a/codex-rs/tools/src/agent_tool.rs b/codex-rs/tools/src/agent_tool.rs index 7f83e6cada..bbaae5d2fc 100644 --- a/codex-rs/tools/src/agent_tool.rs +++ b/codex-rs/tools/src/agent_tool.rs @@ -1,4 +1,6 @@ use crate::JsonSchema; +use crate::ResponsesApiNamespace; +use crate::ResponsesApiNamespaceTool; use crate::ResponsesApiTool; use crate::ToolSpec; use codex_protocol::openai_models::ModelPreset; @@ -258,6 +260,95 @@ pub fn create_close_agent_tool_v1() -> ToolSpec { }) } +pub fn create_watchdog_self_close_tool() -> ToolSpec { + let properties = BTreeMap::from([( + "message".to_string(), + JsonSchema::string(Some( + "Optional final message to send to the parent/root thread before closing this watchdog handle and ending this check-in immediately." + .to_string(), + )), + )]); + + ToolSpec::Function(ResponsesApiTool { + name: "watchdog_self_close".to_string(), + description: "Watchdog-only: send an optional final message to the parent/root thread, close this watchdog's persistent handle, and end this check-in immediately." + .to_string(), + strict: false, + defer_loading: Some(true), + parameters: JsonSchema::object(properties, /*required*/ None, Some(false.into())), + output_schema: Some(close_agent_output_schema()), + }) +} + +pub fn create_watchdog_snooze_tool() -> ToolSpec { + let properties = BTreeMap::from([ + ( + "delay_seconds".to_string(), + JsonSchema::number(Some( + "Optional snooze delay in seconds. If omitted, the watchdog uses its configured interval." + .to_string(), + )), + ), + ( + "reason".to_string(), + JsonSchema::string(Some( + "Optional short reason for snoozing this check-in.".to_string(), + )), + ), + ]); + + ToolSpec::Function(ResponsesApiTool { + name: "snooze".to_string(), + description: "Watchdog-only: keep this watchdog running, skip reporting anything for this check-in, and wait before the next wakeup." + .to_string(), + strict: false, + defer_loading: Some(true), + parameters: JsonSchema::object(properties, /*required*/ None, Some(false.into())), + output_schema: Some(watchdog_snooze_output_schema()), + }) +} + +pub fn create_watchdog_tools_namespace(tools: Vec) -> ToolSpec { + let tools = tools + .into_iter() + .filter_map(|tool| match tool { + ToolSpec::Function(tool) => Some(ResponsesApiNamespaceTool::Function(tool)), + ToolSpec::Freeform(_) + | ToolSpec::ImageGeneration { .. } + | ToolSpec::LocalShell { .. } + | ToolSpec::Namespace(_) + | ToolSpec::ToolSearch { .. } + | ToolSpec::WebSearch { .. } => None, + }) + .collect(); + + ToolSpec::Namespace(ResponsesApiNamespace { + name: "watchdog".to_string(), + description: + "Watchdog-only tools for parent-thread recovery and watchdog check-in lifecycle control." + .to_string(), + tools, + }) +} + +fn watchdog_snooze_output_schema() -> Value { + json!({ + "type": "object", + "properties": { + "target_thread_id": { + "type": "string", + "description": "Watchdog handle that was snoozed." + }, + "delay_seconds": { + "type": "number", + "description": "Effective snooze delay in seconds." + } + }, + "required": ["target_thread_id", "delay_seconds"], + "additionalProperties": false + }) +} + pub fn create_close_agent_tool_v2() -> ToolSpec { let properties = BTreeMap::from([( "target".to_string(), diff --git a/codex-rs/tools/src/lib.rs b/codex-rs/tools/src/lib.rs index 40e5b03468..3df4a1a708 100644 --- a/codex-rs/tools/src/lib.rs +++ b/codex-rs/tools/src/lib.rs @@ -40,6 +40,9 @@ pub use agent_tool::create_spawn_agent_tool_v1; pub use agent_tool::create_spawn_agent_tool_v2; pub use agent_tool::create_wait_agent_tool_v1; pub use agent_tool::create_wait_agent_tool_v2; +pub use agent_tool::create_watchdog_self_close_tool; +pub use agent_tool::create_watchdog_snooze_tool; +pub use agent_tool::create_watchdog_tools_namespace; pub use apply_patch_tool::ApplyPatchToolArgs; pub use apply_patch_tool::create_apply_patch_freeform_tool; pub use apply_patch_tool::create_apply_patch_json_tool; diff --git a/codex-rs/tools/src/tool_config.rs b/codex-rs/tools/src/tool_config.rs index 32ee9e1e5c..cb2e4d385b 100644 --- a/codex-rs/tools/src/tool_config.rs +++ b/codex-rs/tools/src/tool_config.rs @@ -106,6 +106,7 @@ pub struct ToolsConfig { pub collab_tools: bool, pub goal_tools: bool, pub multi_agent_v2: bool, + pub agent_watchdog: bool, pub hide_spawn_agent_metadata: bool, pub spawn_agent_usage_hint: bool, pub spawn_agent_usage_hint_text: Option, @@ -145,6 +146,7 @@ impl ToolsConfig { let include_code_mode_only = include_code_mode && features.enabled(Feature::CodeModeOnly); let include_goal_tools = features.enabled(Feature::Goals); let include_multi_agent_v2 = features.enabled(Feature::MultiAgentV2); + let include_agent_watchdog = features.enabled(Feature::AgentWatchdog); let include_collab_tools = include_multi_agent_v2 || features.enabled(Feature::Collab); let include_agent_jobs = features.enabled(Feature::SpawnCsv); let include_search_tool = @@ -223,6 +225,7 @@ impl ToolsConfig { collab_tools: include_collab_tools, goal_tools: include_goal_tools, multi_agent_v2: include_multi_agent_v2, + agent_watchdog: include_agent_watchdog, hide_spawn_agent_metadata: false, spawn_agent_usage_hint: true, spawn_agent_usage_hint_text: None, diff --git a/codex-rs/tools/src/tool_registry_plan.rs b/codex-rs/tools/src/tool_registry_plan.rs index c00c7c8b13..26bb159b17 100644 --- a/codex-rs/tools/src/tool_registry_plan.rs +++ b/codex-rs/tools/src/tool_registry_plan.rs @@ -57,6 +57,9 @@ use crate::create_view_image_tool; use crate::create_wait_agent_tool_v1; use crate::create_wait_agent_tool_v2; use crate::create_wait_tool; +use crate::create_watchdog_self_close_tool; +use crate::create_watchdog_snooze_tool; +use crate::create_watchdog_tools_namespace; use crate::create_web_search_tool; use crate::create_write_stdin_tool; use crate::default_namespace_description; @@ -494,6 +497,25 @@ pub fn build_tool_registry_plan( } } + if config.agent_watchdog { + plan.push_spec( + create_watchdog_tools_namespace(vec![ + create_watchdog_self_close_tool(), + create_watchdog_snooze_tool(), + ]), + /*supports_parallel_tool_calls*/ false, + config.code_mode_enabled, + ); + plan.register_handler( + crate::ToolName::namespaced("watchdog", "watchdog_self_close"), + ToolHandlerKind::WatchdogSelfClose, + ); + plan.register_handler( + crate::ToolName::namespaced("watchdog", "snooze"), + ToolHandlerKind::WatchdogSnooze, + ); + } + if config.agent_jobs_tools { plan.push_spec( create_spawn_agents_on_csv_tool(), diff --git a/codex-rs/tools/src/tool_registry_plan_types.rs b/codex-rs/tools/src/tool_registry_plan_types.rs index f6ea215ac1..59b92e262e 100644 --- a/codex-rs/tools/src/tool_registry_plan_types.rs +++ b/codex-rs/tools/src/tool_registry_plan_types.rs @@ -40,6 +40,8 @@ pub enum ToolHandlerKind { ViewImage, WaitAgentV1, WaitAgentV2, + WatchdogSelfClose, + WatchdogSnooze, } #[derive(Debug, Clone, PartialEq, Eq)]