feat(TASK-028): escalation_timeout 0 waits indefinitely and is the default
Child escalations (coder plan approvals, deviation gates, user__* calls from subagents) previously expired after the escalating agent's escalation_timeout, defaulting to 300 s, and 0 timed out instantly. In autonomous runs the orchestrator often could not reply within that window, so approvals silently fell through to the best-judgment fallback. - Extract await_escalation_reply as the wait seam: 0 awaits the oneshot with no deadline; nonzero keeps tokio::time::timeout with the same timeout message. A dropped sender still resolves to the existing cancelled reply either way. - DEFAULT_ESCALATION_TIMEOUT_SECS and default_escalation_timeout() -> 0, so graph agents (which inherit the default via from_graph) are covered. - Document the new semantics in config.agent.example.yaml and drop the 5-minute-timeout claim from the default spawn instructions. - Tests: paused-clock unlimited wait (3600 s), nonzero expiry message, dropped-sender completion, default pin, and a grep pin that the stale timeout wording is gone from src/, assets/, and the example config. Plan: PLAN-review-retry §C (D3).
This commit is contained in:
4 files changed
+121
-14
No files matched your search
@@ -48,7 +48,7 @@ max_concurrent_jobs:
|
||||
inject_spawn_instructions: true # Inject the default agent spawning instructions into the agent's system prompt
|
||||
summarization_model: null # Model to use for summarizing sub-agent output (e.g. 'openai:gpt-4o-mini'); defaults to current model
|
||||
summarization_threshold: 4000 # Character threshold above which sub-agent output is summarized before returning to parent
|
||||
escalation_timeout: 300 # Seconds a sub-agent waits for a user interaction response before timing out (default: 5 minutes)
|
||||
escalation_timeout: 0 # 0 = wait indefinitely for the reply (default); N = give up after N seconds
|
||||
mcp_servers: # Optional list of MCP servers that the agent utilizes
|
||||
- github # Corresponds to the name of an MCP server in the `<coyote-config-dir>/mcp.json` file
|
||||
mcp_tools: # Optional per-server tool allowlist for the agent's MCP servers
|
||||
|
||||
+1
-1
@@ -889,7 +889,7 @@ fn default_summarization_threshold() -> usize {
|
||||
}
|
||||
|
||||
fn default_escalation_timeout() -> u64 {
|
||||
300
|
||||
0
|
||||
}
|
||||
|
||||
impl AgentConfig {
|
||||
|
||||
@@ -200,7 +200,7 @@ pub(in crate::config) const DEFAULT_SPAWN_INSTRUCTIONS: &str = indoc! {"
|
||||
1. Read the child's question and options.
|
||||
2. If you can answer from context, call `agent__reply_escalation` with your answer.
|
||||
3. If you need the user's input, call the appropriate `user__*` tool yourself, then relay the answer via `agent__reply_escalation`.
|
||||
4. **Respond promptly**; the child agent is blocked and waiting (5-minute timeout).
|
||||
4. **Respond promptly**; the child blocks until answered — no timeout by default (`escalation_timeout` per agent).
|
||||
"};
|
||||
|
||||
pub(in crate::config) const DEFAULT_JOB_INSTRUCTIONS: &str = indoc! {"
|
||||
|
||||
@@ -13,7 +13,7 @@ use tokio::sync::oneshot;
|
||||
|
||||
pub const USER_FUNCTION_PREFIX: &str = "user__";
|
||||
|
||||
const DEFAULT_ESCALATION_TIMEOUT_SECS: u64 = 300;
|
||||
const DEFAULT_ESCALATION_TIMEOUT_SECS: u64 = 0;
|
||||
const CUSTOM_MULTI_CHOICE_ANSWER_OPTION: &str = "Other (custom)";
|
||||
|
||||
pub fn user_interaction_function_declarations() -> Vec<FunctionDeclaration> {
|
||||
@@ -289,17 +289,32 @@ async fn handle_escalated(ctx: &RequestContext, action: &str, args: &Value) -> R
|
||||
|
||||
root_queue.submit(request);
|
||||
|
||||
let timeout = Duration::from_secs(timeout_secs);
|
||||
match tokio::time::timeout(timeout, rx).await {
|
||||
Ok(Ok(reply)) => Ok(json!({ "answer": reply })),
|
||||
Ok(Err(_)) => Ok(json!({
|
||||
"error": "Escalation was cancelled. The parent agent dropped the request",
|
||||
"fallback": "Make your best judgment and proceed",
|
||||
})),
|
||||
await_escalation_reply(rx, timeout_secs).await
|
||||
}
|
||||
|
||||
/// Waits for the parent's reply. `timeout_secs == 0` waits indefinitely; a
|
||||
/// dropped sender (the parent cancelled the request) resolves either way.
|
||||
async fn await_escalation_reply(rx: oneshot::Receiver<String>, timeout_secs: u64) -> Result<Value> {
|
||||
let reply = if timeout_secs == 0 {
|
||||
rx.await
|
||||
} else {
|
||||
match tokio::time::timeout(Duration::from_secs(timeout_secs), rx).await {
|
||||
Ok(reply) => reply,
|
||||
Err(_) => {
|
||||
return Ok(json!({
|
||||
"error": format!(
|
||||
"Escalation timed out after {timeout_secs} seconds waiting for user response"
|
||||
),
|
||||
"fallback": "Make your best judgment and proceed",
|
||||
}));
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
match reply {
|
||||
Ok(reply) => Ok(json!({ "answer": reply })),
|
||||
Err(_) => Ok(json!({
|
||||
"error": format!(
|
||||
"Escalation timed out after {timeout_secs} seconds waiting for user response"
|
||||
),
|
||||
"error": "Escalation was cancelled. The parent agent dropped the request",
|
||||
"fallback": "Make your best judgment and proceed",
|
||||
})),
|
||||
}
|
||||
@@ -351,4 +366,96 @@ mod tests {
|
||||
assert_eq!(v["action"], "confirm");
|
||||
assert_eq!(v["options"], json!([]));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn escalation_timeout_defaults_to_unlimited() {
|
||||
assert_eq!(DEFAULT_ESCALATION_TIMEOUT_SECS, 0);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn zero_timeout_waits_indefinitely_for_the_reply() {
|
||||
let (tx, rx) = oneshot::channel();
|
||||
let wait = tokio::spawn(await_escalation_reply(rx, 0));
|
||||
|
||||
tokio::time::advance(Duration::from_secs(3600)).await;
|
||||
assert!(!wait.is_finished());
|
||||
|
||||
tx.send("approved".to_string()).unwrap();
|
||||
let v = wait.await.unwrap().unwrap();
|
||||
assert_eq!(v, json!({ "answer": "approved" }));
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn nonzero_timeout_still_expires_with_the_existing_message() {
|
||||
let (tx, rx) = oneshot::channel::<String>();
|
||||
let wait = tokio::spawn(await_escalation_reply(rx, 5));
|
||||
|
||||
tokio::time::advance(Duration::from_secs(6)).await;
|
||||
let v = wait.await.unwrap().unwrap();
|
||||
assert_eq!(
|
||||
v["error"],
|
||||
"Escalation timed out after 5 seconds waiting for user response"
|
||||
);
|
||||
assert_eq!(v["fallback"], "Make your best judgment and proceed");
|
||||
assert!(tx.send("too late".to_string()).is_err());
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn zero_timeout_resolves_when_the_sender_is_dropped() {
|
||||
let (tx, rx) = oneshot::channel::<String>();
|
||||
drop(tx);
|
||||
|
||||
let v = tokio::time::timeout(Duration::from_secs(1), await_escalation_reply(rx, 0))
|
||||
.await
|
||||
.expect("the wait must complete once the sender is gone")
|
||||
.unwrap();
|
||||
assert!(
|
||||
v["error"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.starts_with("Escalation was cancelled"),
|
||||
"{v}"
|
||||
);
|
||||
assert_eq!(v["fallback"], "Make your best judgment and proceed");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn stale_timeout_wording_is_gone_from_shipped_text() {
|
||||
let needles = [
|
||||
concat!("(5-minute", " timeout)"),
|
||||
concat!("escalation_timeout:", " 300"),
|
||||
concat!("default:", " 5 minutes"),
|
||||
];
|
||||
let root = std::path::Path::new(env!("CARGO_MANIFEST_DIR"));
|
||||
let mut files = vec![root.join("config.agent.example.yaml")];
|
||||
collect_files(&root.join("src"), &mut files);
|
||||
collect_files(&root.join("assets"), &mut files);
|
||||
assert!(files.len() > 3, "walked only {} files", files.len());
|
||||
|
||||
let mut hits = Vec::new();
|
||||
for path in &files {
|
||||
let text = String::from_utf8_lossy(&std::fs::read(path).unwrap()).into_owned();
|
||||
for needle in needles {
|
||||
if text.contains(needle) {
|
||||
hits.push(format!("{}: {needle:?}", path.display()));
|
||||
}
|
||||
}
|
||||
}
|
||||
assert!(
|
||||
hits.is_empty(),
|
||||
"stale timeout wording found:\n{}",
|
||||
hits.join("\n")
|
||||
);
|
||||
}
|
||||
|
||||
fn collect_files(dir: &std::path::Path, out: &mut Vec<std::path::PathBuf>) {
|
||||
for entry in std::fs::read_dir(dir).unwrap().flatten() {
|
||||
let path = entry.path();
|
||||
if path.is_dir() {
|
||||
collect_files(&path, out);
|
||||
} else if path.is_file() {
|
||||
out.push(path);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user