Fix watchdog wake/runtime plumbing

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