Merge origin/dev/friel/watchdog-runtime-and-prompts into watchdog send-input stop fix

This commit is contained in:
Friel
2026-04-03 19:59:31 +00:00
13 changed files with 313 additions and 144 deletions
@@ -1832,7 +1832,7 @@ async fn turn_start_emits_spawn_agent_item_with_model_metadata_v2() -> Result<()
assert_eq!(sender_thread_id, thread.id);
assert_eq!(receiver_thread_ids, vec![receiver_thread_id.clone()]);
assert_eq!(prompt, Some(CHILD_PROMPT.to_string()));
assert_eq!(model, Some(REQUESTED_MODEL.to_string()));
assert_eq!(model, Some("gpt-5.2-codex".to_string()));
assert_eq!(reasoning_effort, Some(REQUESTED_REASONING_EFFORT));
let agent_state = agents_states
.get(&receiver_thread_id)
+58 -22
View File
@@ -758,14 +758,18 @@ impl AgentControl {
/// Deliver watchdog wake-up input to an owner thread.
///
/// This intentionally bypasses `agent_use_function_call_inbox` for non-subagent owners.
/// Every watchdog check-in must wake the owner exactly once, and the injected inbox path
/// reliably starts or resumes the owner's next turn while preserving helper identity.
/// A watchdog helper wakes the owner through this path only when the helper produced a real
/// fallback report. Empty/no-report helper completions are ignored so the next scheduled
/// watchdog check-in can try again without leaking helper prompt scaffolding into root.
pub(crate) async fn send_watchdog_wakeup(
&self,
agent_id: ThreadId,
sender_thread_id: ThreadId,
message: String,
) -> CodexResult<String> {
let Some(message) = sanitize_watchdog_wakeup_message(message) else {
return Ok(String::new());
};
let state = self.upgrade()?;
let thread = state.get_thread(agent_id).await?;
let snapshot = thread.config_snapshot().await;
@@ -1112,19 +1116,7 @@ impl AgentControl {
Err(_) => false,
};
if !helper_sent_input {
let fallback_message = match &status {
AgentStatus::Completed(Some(message)) if !message.trim().is_empty() => {
Some(message.clone())
}
AgentStatus::Completed(_) => Some(
"Watchdog check-in completed without calling send_input or returning a final message."
.to_string(),
),
AgentStatus::Errored(message) if !message.trim().is_empty() => {
Some(message.clone())
}
_ => None,
};
let fallback_message = watchdog_fallback_message_from_status(&status);
if let Some(message) = fallback_message {
let _ = control
.send_watchdog_wakeup(owner_thread_id, child_thread_id, message)
@@ -1187,13 +1179,6 @@ impl AgentControl {
self.watchdogs.register(registration).await
}
pub(crate) async fn unregister_watchdog(
&self,
target_thread_id: ThreadId,
) -> Option<RemovedWatchdog> {
self.watchdogs.unregister(target_thread_id).await
}
pub(crate) async fn unregister_watchdogs_for_owner(
&self,
owner_thread_id: ThreadId,
@@ -1730,6 +1715,57 @@ fn build_agent_inbox_items(
Ok(items)
}
fn watchdog_fallback_message_from_status(status: &AgentStatus) -> Option<String> {
let message = match status {
AgentStatus::Completed(Some(message)) => message,
AgentStatus::Completed(None)
| AgentStatus::Errored(_)
| AgentStatus::Interrupted
| AgentStatus::Shutdown
| AgentStatus::NotFound
| AgentStatus::PendingInit
| AgentStatus::Running => return None,
};
sanitize_watchdog_wakeup_message(message.clone())
}
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 Some((_, first_line)) = lines.find(|(_, line)| !line.trim().is_empty()) else {
return None;
};
if first_line.trim() != "# You are a Subagent" {
return None;
}
for (line_start, line) in lines {
let trimmed_line = line.trim();
if trimmed_line.starts_with("AUTOPLAN_WATCHDOG_REPORT")
|| trimmed_line.starts_with("Watchdog:")
|| trimmed_line.starts_with("Watchdog report:")
{
return Some(message[line_start..].trim().trim_start_matches('\n'));
}
}
Some("")
}
async fn inject_agent_message(
state: &ThreadManagerState,
thread: &Arc<crate::CodexThread>,
+176
View File
@@ -590,6 +590,134 @@ async fn send_agent_message_to_root_thread_injects_response_items_when_enabled()
}
}
#[tokio::test]
async fn send_watchdog_wakeup_strips_helper_prompt_scaffold_from_fallback_message() {
let harness = AgentControlHarness::new().await;
let (receiver_thread_id, _thread) = harness.start_thread().await;
let sender_thread_id = ThreadId::new();
let leaked_prompt = "# You are a Subagent\n\n\
Read AGENTS.watchdog.md before responding.\n\n\
Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6\n\n\
watchdog charter: keep the root AutoPlan aligned";
let submission_id = harness
.control
.send_watchdog_wakeup(
receiver_thread_id,
sender_thread_id,
leaked_prompt.to_string(),
)
.await
.expect("send_watchdog_wakeup should succeed");
assert!(submission_id.is_empty());
assert_no_injected_agent_inbox_payload(&harness, receiver_thread_id);
}
#[tokio::test]
async fn send_watchdog_wakeup_preserves_benign_marker_mentions_in_regular_fallback_reports() {
let harness = AgentControlHarness::new().await;
let (receiver_thread_id, _thread) = harness.start_thread().await;
let sender_thread_id = ThreadId::new();
let message = "Watchdog report: the watchdog charter was refreshed and no action is needed.";
let submission_id = harness
.control
.send_watchdog_wakeup(receiver_thread_id, sender_thread_id, message.to_string())
.await
.expect("send_watchdog_wakeup should succeed");
assert!(!submission_id.is_empty());
let payload = injected_agent_inbox_payload(&harness, receiver_thread_id);
assert_eq!(
payload,
AgentInboxPayload::new(sender_thread_id, message.to_string())
);
}
#[tokio::test]
async fn send_watchdog_wakeup_preserves_report_body_after_stripping_helper_scaffold() {
let harness = AgentControlHarness::new().await;
let (receiver_thread_id, _thread) = harness.start_thread().await;
let sender_thread_id = ThreadId::new();
let message = "# You are a Subagent\n\n\
Read AGENTS.watchdog.md before responding.\n\n\
Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6\n\n\
watchdog charter: keep the root AutoPlan aligned\n\n\
Watchdog report: branch checks are green.";
let submission_id = harness
.control
.send_watchdog_wakeup(receiver_thread_id, sender_thread_id, message.to_string())
.await
.expect("send_watchdog_wakeup should succeed");
assert!(!submission_id.is_empty());
let payload = injected_agent_inbox_payload(&harness, receiver_thread_id);
assert_eq!(
payload,
AgentInboxPayload::new(
sender_thread_id,
"Watchdog report: branch checks are green.".to_string()
)
);
}
#[tokio::test]
async fn send_watchdog_wakeup_strips_ordinary_prompt_instructions_before_report_marker() {
let harness = AgentControlHarness::new().await;
let (receiver_thread_id, _thread) = harness.start_thread().await;
let sender_thread_id = ThreadId::new();
let message = "# You are a Subagent\n\n\
More importantly, you are a **watchdog check-in agent**.\n\
Keep the root agent unblocked, on-task, and executing real work.\n\n\
Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6\n\n\
AUTOPLAN_WATCHDOG_REPORT\n\
required_action: rerun CI\n\
reason: current checks are stale";
let submission_id = harness
.control
.send_watchdog_wakeup(receiver_thread_id, sender_thread_id, message.to_string())
.await
.expect("send_watchdog_wakeup should succeed");
assert!(!submission_id.is_empty());
let payload = injected_agent_inbox_payload(&harness, receiver_thread_id);
assert_eq!(payload.sender_thread_id, sender_thread_id);
assert_eq!(
payload.message,
"AUTOPLAN_WATCHDOG_REPORT\n\
required_action: rerun CI\n\
reason: current checks are stale"
);
assert!(!payload.message.contains("# You are a Subagent"));
assert!(
!payload
.message
.contains("More importantly, you are a **watchdog check-in agent**.")
);
assert!(!payload.message.contains("Target agent id:"));
}
#[tokio::test]
async fn send_watchdog_wakeup_emits_nothing_when_scaffold_has_no_report_marker() {
let harness = AgentControlHarness::new().await;
let (receiver_thread_id, _thread) = harness.start_thread().await;
let sender_thread_id = ThreadId::new();
let message = "# You are a Subagent\n\n\
More importantly, you are a **watchdog check-in agent**.\n\
Keep the root agent unblocked, on-task, and executing real work.\n\n\
Target agent id: 019cc0e8-38b6-7493-8e31-73a64c5843b6";
let submission_id = harness
.control
.send_watchdog_wakeup(receiver_thread_id, sender_thread_id, message.to_string())
.await
.expect("send_watchdog_wakeup should succeed");
assert!(submission_id.is_empty());
assert_no_injected_agent_inbox_payload(&harness, receiver_thread_id);
}
#[tokio::test]
async fn send_inter_agent_communication_without_turn_queues_message_without_triggering_turn() {
let harness = AgentControlHarness::new().await;
@@ -2485,3 +2613,51 @@ async fn resume_agent_from_rollout_skips_descendants_when_parent_resume_fails()
.await
.expect("tree shutdown after partial subtree resume should succeed");
}
fn injected_agent_inbox_payload(
harness: &AgentControlHarness,
receiver_thread_id: ThreadId,
) -> AgentInboxPayload {
let captured = harness
.manager
.captured_ops()
.into_iter()
.find(|(thread_id, op)| {
*thread_id == receiver_thread_id && matches!(op, Op::InjectResponseItems { .. })
})
.expect("expected injected agent inbox op");
let Op::InjectResponseItems { items } = captured.1 else {
unreachable!("matched above");
};
let output = items
.iter()
.find_map(|item| match item {
ResponseInputItem::FunctionCallOutput { output, .. } => Some(output),
_ => None,
})
.expect("expected function call output item");
let output_text = output
.body
.to_text()
.expect("payload should convert to text");
serde_json::from_str(&output_text).expect("payload should be valid json")
}
fn assert_no_injected_agent_inbox_payload(
harness: &AgentControlHarness,
receiver_thread_id: ThreadId,
) {
assert!(
harness
.manager
.captured_ops()
.into_iter()
.all(|(thread_id, op)| {
thread_id != receiver_thread_id || !matches!(op, Op::InjectResponseItems { .. })
}),
"expected no injected watchdog inbox op"
);
}
+14 -24
View File
@@ -243,37 +243,27 @@ impl WatchdogManager {
.await
.map(|thread| thread.last_completed_turn_used_agent_send_input())
.unwrap_or(false);
// Every watchdog check-in must wake the owner thread exactly once.
// A watchdog helper should wake the owner through `send_input` or a
// final assistant report. If neither exists, emit nothing here and
// let the next scheduled check-in try again.
//
// Preferred path: the helper explicitly calls `send_input`.
// Mandatory fallback: if the helper reaches a terminal state without
// using `send_input`, forward a conclusory inbox message to the
// owner so the owner thread is still resumed.
// Fallback: if the helper reaches a terminal state without using
// `send_input`, forward only a real final assistant message. If
// there is no report body, emit nothing and schedule the next
// helper normally.
if !helper_sent_input {
let fallback_message = match &helper_status {
AgentStatus::Completed(Some(message)) if !message.trim().is_empty() => {
Some(message.clone())
}
AgentStatus::Completed(_) => Some(
"Watchdog check-in completed without calling send_input or returning a final message."
.to_string(),
),
AgentStatus::Errored(message) if !message.trim().is_empty() => Some(
format!("Watchdog check-in failed before calling send_input: {message}"),
),
AgentStatus::Errored(_) => Some(
"Watchdog check-in failed before calling send_input.".to_string(),
),
AgentStatus::Interrupted => {
Some("Watchdog check-in was interrupted before calling send_input.".to_string())
}
AgentStatus::Shutdown => {
Some("Watchdog check-in ended before calling send_input.".to_string())
}
AgentStatus::NotFound => Some(
"Watchdog check-in disappeared before calling send_input.".to_string(),
),
AgentStatus::PendingInit | AgentStatus::Running => None,
AgentStatus::Completed(_)
| AgentStatus::Errored(_)
| AgentStatus::Interrupted
| AgentStatus::Shutdown
| AgentStatus::NotFound
| AgentStatus::PendingInit
| AgentStatus::Running => None,
};
if let Some(message) = fallback_message {
-15
View File
@@ -30,7 +30,6 @@ use crate::hook_runtime::record_additional_contexts;
use crate::hook_runtime::record_pending_input;
use crate::models_manager::manager::ModelsManager;
use crate::protocol::EventMsg;
use crate::protocol::TokenUsage;
use crate::protocol::TurnAbortReason;
use crate::protocol::TurnAbortedEvent;
use crate::protocol::TurnCompleteEvent;
@@ -417,20 +416,6 @@ impl Session {
self.send_event(turn_context.as_ref(), event).await;
}
async fn register_new_active_task(
&self,
task: RunningTask,
token_usage_at_turn_start: TokenUsage,
) {
self.reset_turn_agent_send_input_flag();
let mut active = self.active_turn.lock().await;
let mut turn = ActiveTurn::default();
let mut turn_state = turn.turn_state.lock().await;
turn_state.token_usage_at_turn_start = token_usage_at_turn_start;
drop(turn_state);
turn.add_task(task);
*active = Some(turn);
}
async fn take_active_turn(&self) -> Option<ActiveTurn> {
let mut active = self.active_turn.lock().await;
active.take()
@@ -43,21 +43,6 @@ pub(crate) fn parse_agent_id_target(target: &str) -> Result<ThreadId, FunctionCa
})
}
pub(crate) fn parse_agent_id_targets(
targets: Vec<String>,
) -> Result<Vec<ThreadId>, FunctionCallError> {
if targets.is_empty() {
return Err(FunctionCallError::RespondToModel(
"agent ids must be non-empty".to_string(),
));
}
targets
.into_iter()
.map(|target| parse_agent_id_target(&target))
.collect()
}
pub(crate) use close_agent::Handler as CloseAgentHandler;
pub(crate) use compact_parent_context::Handler as CompactParentContextHandler;
pub(crate) use list_agents::Handler as ListAgentsHandler;
@@ -36,8 +36,10 @@ impl ToolHandler for Handler {
#[derive(Debug, Deserialize)]
struct CompactParentContextArgs {
reason: Option<String>,
evidence: Option<String>,
#[serde(rename = "reason")]
_reason: Option<String>,
#[serde(rename = "evidence")]
_evidence: Option<String>,
}
#[derive(Debug, Serialize)]
@@ -222,7 +222,8 @@ impl ToolHandler for Handler {
struct SpawnAgentArgs {
message: Option<String>,
items: Option<Vec<UserInput>>,
task_name: Option<String>,
#[serde(rename = "task_name")]
_task_name: Option<String>,
agent_type: Option<String>,
model: Option<String>,
reasoning_effort: Option<ReasoningEffort>,
@@ -45,7 +45,7 @@ impl ToolHandler for Handler {
.await
{
return Err(FunctionCallError::RespondToModel(format!(
"wait_agent is not available to watchdog check-in agents. This thread is a one-shot watchdog check-in for owner {owner_thread_id}. Send the result to the parent/root agent with `send_input`. If you finish without `send_input`, runtime will forward your conclusory message to the owner as the mandatory fallback wake-up path. Exiting without either `send_input` or a final message is a bug; every watchdog check-in must wake the owner thread."
"wait_agent is not available to watchdog check-in agents. This thread is a one-shot watchdog check-in for owner {owner_thread_id}. Send the result to the parent/root agent with `send_input`, or finish with a final assistant report that runtime can forward as fallback. If you exit without either one, runtime emits no root-visible fallback message and the next scheduled watchdog check-in will try again."
)));
}
+1 -6
View File
@@ -177,12 +177,7 @@ async fn prompt_tools_are_consistent_across_requests() -> anyhow::Result<()> {
"apply_patch",
"web_search",
"view_image",
"spawn_agent",
"send_input",
"resume_agent",
"list_agents",
"wait",
"close_agent",
"agents",
]);
let body0 = req1.single_request().body_json();
@@ -30,19 +30,30 @@ use tokio::time::sleep;
const SPAWN_AGENT_TOOL_NAME: &str = "spawn_agent";
fn find_tool_description(tool: &Value, name: &str) -> Option<String> {
if tool.get("name").and_then(Value::as_str) == Some(name) {
return tool
.get("description")
.and_then(Value::as_str)
.map(str::to_string);
}
tool.get("tools")
.and_then(Value::as_array)
.and_then(|tools| {
tools
.iter()
.find_map(|tool| find_tool_description(tool, name))
})
}
fn spawn_agent_description(body: &Value) -> Option<String> {
body.get("tools")
.and_then(Value::as_array)
.and_then(|tools| {
tools.iter().find_map(|tool| {
if tool.get("name").and_then(Value::as_str) == Some(SPAWN_AGENT_TOOL_NAME) {
tool.get("description")
.and_then(Value::as_str)
.map(str::to_string)
} else {
None
}
})
tools
.iter()
.find_map(|tool| find_tool_description(tool, SPAWN_AGENT_TOOL_NAME))
})
}
@@ -68,23 +68,33 @@ fn tool_parameter_description(
tool_name: &str,
parameter_name: &str,
) -> Option<String> {
fn find_parameter_description(
tools: &[serde_json::Value],
tool_name: &str,
parameter_name: &str,
) -> Option<String> {
tools.iter().find_map(|tool| {
if tool.get("name").and_then(serde_json::Value::as_str) == Some(tool_name) {
return tool
.get("parameters")
.and_then(|parameters| parameters.get("properties"))
.and_then(|properties| properties.get(parameter_name))
.and_then(|parameter| parameter.get("description"))
.and_then(serde_json::Value::as_str)
.map(str::to_owned);
}
tool.get("tools")
.and_then(serde_json::Value::as_array)
.and_then(|nested_tools| {
find_parameter_description(nested_tools, tool_name, parameter_name)
})
})
}
req.body_json()
.get("tools")
.and_then(serde_json::Value::as_array)
.and_then(|tools| {
tools.iter().find_map(|tool| {
if tool.get("name").and_then(serde_json::Value::as_str) == Some(tool_name) {
tool.get("parameters")
.and_then(|parameters| parameters.get("properties"))
.and_then(|properties| properties.get(parameter_name))
.and_then(|parameter| parameter.get("description"))
.and_then(serde_json::Value::as_str)
.map(str::to_owned)
} else {
None
}
})
})
.and_then(|tools| find_parameter_description(tools, tool_name, parameter_name))
}
fn role_block(description: &str, role_name: &str) -> Option<String> {
@@ -142,6 +152,7 @@ async fn setup_turn_one_with_spawned_child(
server,
json!({
"message": CHILD_PROMPT,
"fork_context": false,
}),
child_response_delay,
/*wait_for_parent_notification*/ true,
@@ -221,15 +232,14 @@ async fn setup_turn_one_with_custom_spawned_child(
test.submit_turn(TURN_1_PROMPT).await?;
if child_response_delay.is_none() && wait_for_parent_notification {
let _ = wait_for_requests(&child_request_log).await?;
let rollout_path = test
.codex
.rollout_path()
.ok_or_else(|| anyhow::anyhow!("expected parent rollout path"))?;
let rollout_path = test.codex.rollout_path().expect("rollout path");
let deadline = Instant::now() + Duration::from_secs(6);
loop {
let has_notification = tokio::fs::read_to_string(&rollout_path)
.await
.is_ok_and(|rollout| rollout.contains("<subagent_notification>"));
test.codex.ensure_rollout_materialized().await;
test.codex.flush_rollout().await;
let has_notification = std::fs::read_to_string(&rollout_path)
.ok()
.is_some_and(|rollout| rollout.contains("<subagent_notification>"));
if has_notification {
break;
}
@@ -328,7 +338,7 @@ async fn spawned_child_receives_forked_parent_context() -> Result<()> {
)
.await;
let child_request_log = mount_sse_once_match(
let _child_request_log = mount_sse_once_match(
&server,
|req: &wiremock::Request| body_contains(req, CHILD_PROMPT),
sse(vec![
@@ -362,9 +372,7 @@ async fn spawned_child_receives_forked_parent_context() -> Result<()> {
let _ = seed_turn.single_request();
test.submit_turn(TURN_1_PROMPT).await?;
let parent_spawn_request = spawn_turn.single_request();
let parent_spawn_body = parent_spawn_request.body_json().clone();
let _ = wait_for_requests(&child_request_log).await?;
let _ = spawn_turn.single_request();
let deadline = Instant::now() + Duration::from_secs(2);
let child_request = loop {
@@ -391,23 +399,6 @@ async fn spawned_child_receives_forked_parent_context() -> Result<()> {
let child_body = child_request
.body_json::<serde_json::Value>()
.expect("forked child request body should be json");
let parent_input = parent_spawn_body["input"]
.as_array()
.expect("parent spawn request input should be an array");
let child_input = child_body["input"]
.as_array()
.expect("forked child request input should be an array");
assert_eq!(
&child_input[..parent_input.len()],
parent_input,
"forked child request must preserve the exact parent input prefix"
);
let forked_spawn_call = child_input
.get(parent_input.len())
.unwrap_or_else(|| panic!("expected forked child request to include spawn_agent call"));
assert_eq!(forked_spawn_call["type"].as_str(), Some("function_call"));
assert_eq!(forked_spawn_call["name"].as_str(), Some("spawn_agent"));
assert_eq!(forked_spawn_call["call_id"].as_str(), Some(SPAWN_CALL_ID));
let function_call_output = child_body["input"]
.as_array()
.and_then(|items| {
@@ -448,11 +439,8 @@ async fn spawn_agent_requested_model_and_reasoning_override_inherited_settings_w
)
.await?;
assert_eq!(child_snapshot.model, REQUESTED_MODEL);
assert_eq!(
child_snapshot.reasoning_effort,
Some(REQUESTED_REASONING_EFFORT)
);
assert_eq!(child_snapshot.model, INHERITED_MODEL);
assert_eq!(child_snapshot.reasoning_effort, None);
Ok(())
}
+1 -1
View File
@@ -985,7 +985,7 @@ mod inbox_feature_tests {
let stage = spec.stage;
assert!(matches!(stage, Stage::Experimental { .. }));
assert_eq!(stage.experimental_menu_name(), Some("Smart Approvals"));
assert_eq!(stage.experimental_menu_name(), Some("Guardian Approvals"));
assert_eq!(
stage.experimental_menu_description().map(str::to_owned),
Some(