merge upstream/dev/friel/watchdog-runtime-and-prompts into collab stack

This commit is contained in:
Friel
2026-03-30 01:53:08 +00:00
58 changed files with 4043 additions and 426 deletions

View File

@@ -205,11 +205,16 @@ fn message_input_texts(body: &Value) -> Vec<String> {
};
items
.iter()
.filter(|item| item.get("type").and_then(Value::as_str) == Some("message"))
.filter(|item| {
item.get("role").and_then(Value::as_str).is_some()
|| item.get("type").and_then(Value::as_str) == Some("message")
})
.filter_map(|item| item.get("content").and_then(Value::as_array))
.flatten()
.filter(|span| span.get("type").and_then(Value::as_str) == Some("input_text"))
.filter_map(|span| span.get("text").and_then(Value::as_str))
.filter_map(|span| match span.get("type").and_then(Value::as_str) {
Some("input_text") | None => span.get("text").and_then(Value::as_str),
_ => None,
})
.map(str::to_string)
.collect()
}
@@ -320,9 +325,31 @@ async fn spawn_agents_on_csv_runs_and_exports() -> Result<()> {
test.submit_turn("run batch job").await?;
let output = fs::read_to_string(&output_path)?;
assert!(output.contains("result_json"));
assert!(output.contains("item_id"));
assert!(output.contains("\"item_id\""));
let mut lines = output.lines();
let headers = lines.next().expect("csv headers");
let header_cols = parse_simple_csv_line(headers);
let status_index = header_cols
.iter()
.position(|header| header == "status")
.expect("status column");
let result_json_index = header_cols
.iter()
.position(|header| header == "result_json")
.expect("result_json column");
assert!(header_cols.iter().any(|header| header == "result_json"));
assert!(header_cols.iter().any(|header| header == "item_id"));
let rows: Vec<Vec<String>> = lines.map(parse_simple_csv_line).collect();
assert_eq!(rows.len(), 2);
assert_eq!(
rows.iter()
.map(|cols| cols[status_index].as_str())
.collect::<Vec<_>>(),
vec!["completed", "completed"]
);
assert!(
rows.iter()
.all(|cols| !cols[result_json_index].trim().is_empty())
);
Ok(())
}
@@ -423,21 +450,28 @@ async fn spawn_agents_on_csv_stop_halts_future_items() -> Result<()> {
test.submit_turn("run job").await?;
let output = fs::read_to_string(&output_path)?;
let rows: Vec<&str> = output.lines().skip(1).collect();
let mut lines = output.lines();
let headers = lines.next().expect("csv headers");
let header_cols = parse_simple_csv_line(headers);
let job_id_index = header_cols
.iter()
.position(|header| header == "job_id")
.expect("job_id column");
let rows: Vec<&str> = lines.collect();
assert_eq!(rows.len(), 3);
let job_id = rows
let job_id: String = rows
.first()
.and_then(|line| {
parse_simple_csv_line(line)
.iter()
.find(|value| value.len() == 36)
.cloned()
})
.map(|line| parse_simple_csv_line(line))
.and_then(|cols| cols.get(job_id_index).cloned())
.expect("job_id from csv");
let db = test.codex.state_db().expect("state db");
let job = db.get_agent_job(job_id.as_str()).await?.expect("job");
assert_eq!(job.status, codex_state::AgentJobStatus::Cancelled);
let progress = db.get_agent_job_progress(job_id.as_str()).await?;
assert_eq!(
job.status,
codex_state::AgentJobStatus::Cancelled,
"unexpected final job state: job={job:?} progress={progress:?} output={output}"
);
assert_eq!(progress.total_items, 3);
assert_eq!(progress.completed_items, 1);
assert_eq!(progress.failed_items, 0);

View File

@@ -1746,12 +1746,18 @@ async fn includes_developer_instructions_message_in_request() {
.iter()
.filter(|item| item.get("role").and_then(|role| role.as_str()) == Some("developer"))
.collect();
let developer_contents: Vec<&str> = developer_messages
.iter()
.filter_map(|item| item.get("content").and_then(serde_json::Value::as_array))
.flat_map(|content| content.iter())
.filter(|span| span.get("type").and_then(serde_json::Value::as_str) == Some("input_text"))
.filter_map(|span| span.get("text").and_then(serde_json::Value::as_str))
.collect();
assert!(
developer_messages
developer_contents
.iter()
.any(|item| message_input_texts(item).contains(&"be useful")),
"expected developer instructions in a developer message, got {:?}",
request_body["input"]
.any(|content| content.contains("be useful")),
"expected developer instructions in a developer message, got {developer_contents:?}",
);
assert_message_role(&request_body["input"][1], "user");

View File

@@ -180,7 +180,8 @@ async fn prompt_tools_are_consistent_across_requests() -> anyhow::Result<()> {
"spawn_agent",
"send_input",
"resume_agent",
"wait_agent",
"list_agents",
"wait",
"close_agent",
]);
let body0 = req1.single_request().body_json();

View File

@@ -328,7 +328,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,7 +362,9 @@ async fn spawned_child_receives_forked_parent_context() -> Result<()> {
let _ = seed_turn.single_request();
test.submit_turn(TURN_1_PROMPT).await?;
let _ = spawn_turn.single_request();
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 deadline = Instant::now() + Duration::from_secs(2);
let child_request = loop {
@@ -389,6 +391,23 @@ 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| {
@@ -465,8 +484,12 @@ async fn spawn_agent_role_overrides_requested_model_and_reasoning_settings() ->
"custom".to_string(),
AgentRoleConfig {
description: Some("Custom role".to_string()),
model: None,
config_file: Some(role_path),
spawn_mode: None,
watchdog_interval_s: None,
nickname_candidates: None,
fork_context: None,
},
);
})
@@ -513,8 +536,12 @@ async fn spawn_agent_tool_description_mentions_role_locked_settings() -> Result<
"custom".to_string(),
AgentRoleConfig {
description: Some("Custom role".to_string()),
model: None,
config_file: Some(role_path),
spawn_mode: None,
watchdog_interval_s: None,
nickname_candidates: None,
fork_context: None,
},
);
});