openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add watchdog runtime handles

Browse Source
This commit is contained in:
Friel
2026-05-04 00:50:02 +00:00
parent d1ec95ec0e
commit f44de610db
25 changed files with 2134 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/config/src/config_toml.rs
+5
View File
@@ -366,6 +366,10 @@ pub struct ConfigToml {
/// Agent-related settings (thread limits, etc.).
pub agents: Option<AgentsToml>,
/// Default interval in seconds for idle-time watchdog check-ins.
#[schemars(range(min = 1))]
pub watchdog_interval_s: Option<i64>,
/// Memories subsystem settings.
pub memories: Option<MemoriesToml>,
@@ -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.
codex-rs/core/config.schema.json
+12
View File
@@ -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": [
{
codex-rs/core/src/agent/control.rs
+206 -5
View File
@@ -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<ThreadManagerState>,
state: Arc<AgentRegistry>,
watchdogs: Option<Arc<WatchdogManager>>,
}
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<ThreadManagerState>) -> 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<ThreadManagerState>,
state: Arc<AgentRegistry>,
watchdogs: Arc<WatchdogManager>,
) -> 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<Arc<crate::exec_policy::ExecPolicyManager>>,
inherited_thread_state: InheritedThreadState,
) -> CodexResult<crate::thread_manager::NewThread> {
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<String> {
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<String> {
let state = self.upgrade()?;
@@ -753,10 +820,106 @@ impl AgentControl {
result
}
pub(crate) async fn register_watchdog(
&self,
registration: WatchdogRegistration,
) -> CodexResult<Vec<RemovedWatchdog>> {
self.watchdog_manager()?.register(registration).await
}
pub(crate) async fn unregister_watchdogs_for_owner(
&self,
owner_thread_id: ThreadId,
) -> Vec<RemovedWatchdog> {
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<RemovedWatchdog> {
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<ThreadId> {
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<ThreadId> {
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<u64>,
) -> Option<crate::agent::watchdog::WatchdogSnoozeResult> {
let watchdogs = self.watchdogs.as_ref()?;
watchdogs
.snooze_active_helper(helper_thread_id, delay_seconds)
.await
}
fn watchdog_manager(&self) -> CodexResult<&Arc<WatchdogManager>> {
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<String> {
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<Threa
}
}
fn sanitize_watchdog_wakeup_message(message: String) -> Option<String> {
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;
codex-rs/core/src/agent/control_tests.rs
+414 -3
View File
@@ -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()
},
)
codex-rs/core/src/agent/mod.rs
+4
View File
@@ -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;
codex-rs/core/src/agent/role.rs
+27 -1
View File
@@ -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<i64> {
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(),
codex-rs/core/src/agent/role_tests.rs
+42
View File
@@ -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
);
}
codex-rs/core/src/agent/watchdog.rs
+645
View File
@@ -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<ThreadId>,
}
struct WatchdogEntry {
registration: WatchdogRegistration,
interval: Duration,
last_trigger: Instant,
active_helper_id: Option<ThreadId>,
snoozed_until: Option<Instant>,
owner_idle_since: Option<Instant>,
owner_was_running: bool,
force_due_once: bool,
generation: i64,
}
pub(crate) struct WatchdogManager {
manager: Weak<ThreadManagerState>,
state: Arc<AgentRegistry>,
registrations: Mutex<HashMap<ThreadId, WatchdogEntry>>,
suppressed_helpers: Mutex<HashSet<ThreadId>>,
started: AtomicBool,
next_generation: AtomicI64,
}
impl WatchdogManager {
pub(crate) fn new(manager: Weak<ThreadManagerState>, state: Arc<AgentRegistry>) -> Arc<Self> {
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<Self>) {
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<Self>,
registration: WatchdogRegistration,
) -> CodexResult<Vec<RemovedWatchdog>> {
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::<Vec<_>>();
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<RemovedWatchdog> {
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<RemovedWatchdog> {
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<ThreadId> {
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<ThreadId> {
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<Self>) {
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<Self>) {
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::<Vec<_>>()
};
let now = Instant::now();
for (target_id, generation) in snapshots {
self.evaluate(&manager_state, target_id, generation, now)
.await;
}
}
async fn evaluate(
self: &Arc<Self>,
manager_state: &Arc<ThreadManagerState>,
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<WatchdogSnapshot> {
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<Instant> {
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<u64>,
) -> Option<WatchdogSnoozeResult> {
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<ThreadId>,
) {
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<ThreadId>,
snoozed_until: Option<Instant>,
owner_idle_since: Option<Instant>,
}
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<ThreadManagerState>, 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<Duration> {
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
));
}
}
codex-rs/core/src/config/agent_roles.rs
+9
View File
@@ -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
codex-rs/core/src/config/config_tests.rs
+4
View File
@@ -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,
codex-rs/core/src/config/mod.rs
+14
View File
@@ -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<u64> = 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<String, AgentRoleConfig>,
/// 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,
codex-rs/core/src/tools/handlers/multi_agents.rs
+4
View File
@@ -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"]
codex-rs/core/src/tools/handlers/multi_agents/spawn.rs
+131 -23
View File
@@ -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<ThreadId> {
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<RemovedWatchdog>,
) {
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::<Vec<_>>();
thread_ids.sort_by_key(ToString::to_string);
for thread_id in thread_ids {
let _ = agent_control.close_agent(thread_id).await;
}
}
codex-rs/core/src/tools/handlers/multi_agents/wait.rs
+25
View File
@@ -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 {
codex-rs/core/src/tools/handlers/multi_agents/watchdog_self_close.rs
+99
View File
@@ -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<Self::Output, FunctionCallError> {
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<String>,
}
#[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")
}
}
codex-rs/core/src/tools/handlers/multi_agents/watchdog_snooze.rs
+68
View File
@@ -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<Self::Output, FunctionCallError> {
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<u64>,
reason: Option<String>,
}
#[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")
}
}
codex-rs/core/src/tools/handlers/multi_agents_tests.rs
+287 -1
View File
@@ -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::<Vec<_>>();
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;
codex-rs/core/src/tools/spec.rs
+8
View File
@@ -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() {
codex-rs/features/src/lib.rs
+8
View File
@@ -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",
codex-rs/thread-manager-sample/src/main.rs
+1
View File
@@ -212,6 +212,7 @@ fn new_config(model: Option<String>, 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(),
codex-rs/tools/src/agent_tool.rs
+91
View File
@@ -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>) -> 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(),
codex-rs/tools/src/lib.rs
+3
View File
@@ -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;
codex-rs/tools/src/tool_config.rs
+3
View File
@@ -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<String>,
@@ -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,
codex-rs/tools/src/tool_registry_plan.rs
+22
View File
@@ -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(),
codex-rs/tools/src/tool_registry_plan_types.rs
+2
View File
@@ -40,6 +40,8 @@ pub enum ToolHandlerKind {
ViewImage,
WaitAgentV1,
WaitAgentV2,
WatchdogSelfClose,
WatchdogSnooze,
}
#[derive(Debug, Clone, PartialEq, Eq)]