diff --git a/src/app/agents.rs b/src/app/agents.rs index ccdd4af662..b2aac01cf4 100644 --- a/src/app/agents.rs +++ b/src/app/agents.rs @@ -6,8 +6,10 @@ use super::{terminal_targets::TerminalTargetError, App}; use crate::api::schema::AgentStartParams; const DEFAULT_AGENT_START_TIMEOUT: Duration = Duration::from_secs(30); -const MAX_AGENT_START_TIMEOUT: Duration = Duration::from_secs(300); -const AGENT_START_SETTLE_DELAY: Duration = Duration::from_secs(3); +pub(crate) const MAX_AGENT_START_TIMEOUT: Duration = Duration::from_secs(300); +pub(crate) const AGENT_START_SETTLE_DELAY: Duration = Duration::from_secs(3); +const INVALID_AGENT_TIMEOUT_MESSAGE: &str = + "agent start timeout must be greater than 3000ms and at most 300000ms"; const INVALID_AGENT_NAME_MESSAGE: &str = "agent name must start with a lowercase letter and contain only lowercase letters, digits, '-' or '_' (1-32 characters)"; fn valid_agent_name(name: &str) -> bool { @@ -243,8 +245,7 @@ impl App { }, AgentStartError::InvalidTimeout => crate::api::schema::ErrorBody { code: "invalid_agent_timeout".into(), - message: "agent start timeout must be greater than 3000ms and at most 300000ms" - .into(), + message: INVALID_AGENT_TIMEOUT_MESSAGE.into(), }, AgentStartError::TargetNotFound(target) => crate::api::schema::ErrorBody { code: "agent_pane_not_found".into(), diff --git a/src/app/mod.rs b/src/app/mod.rs index ef33c693d1..49a905ae32 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -8,6 +8,7 @@ pub(crate) mod actions; mod agent_resume; pub(crate) mod agent_view; mod agents; +pub(crate) use agents::{AGENT_START_SETTLE_DELAY, MAX_AGENT_START_TIMEOUT}; mod api; mod api_helpers; pub(crate) use api_helpers::limit_snapshot_lines; diff --git a/src/cli/agent.rs b/src/cli/agent.rs index 362c9ab181..306c3bd9fa 100644 --- a/src/cli/agent.rs +++ b/src/cli/agent.rs @@ -3,9 +3,12 @@ use std::time::{Duration, Instant}; use crate::api::schema::{ AgentPromptParams, AgentPromptWaitOptions, AgentReadParams, AgentRenameParams, AgentSendKeysParams, AgentStartParams, AgentTarget, AgentWaitParams, EmptyParams, Method, - ReadFormat, ReadSource, Request, + PaneProcessInfoParams, PaneTarget, ReadFormat, ReadSource, Request, }; +const AGENT_START_POLL_INTERVAL: Duration = Duration::from_millis(100); +const PANE_SHELL_READINESS_RETRY_TIMEOUT: Duration = Duration::from_secs(2); + pub(super) fn run_agent_command(args: &[String]) -> std::io::Result { let Some(subcommand) = args.first().map(|arg| arg.as_str()) else { print_agent_help(); @@ -330,24 +333,62 @@ fn agent_start(args: &[String]) -> std::io::Result { return Ok(2); }; let expected_kind = crate::detect::agent_label(expected_kind).to_string(); - let mut response = super::send_request(&Request { - id: "cli:agent:start".into(), - method: Method::AgentStart(AgentStartParams { - name: name.clone(), - kind, - pane_id: pane_id.clone(), - args: if separator < args.len() { - args[separator + 1..].to_vec() - } else { - Vec::new() - }, - timeout_ms, - }), - })?; - if response.get("error").is_some() { - return super::print_response(&response); - } + let agent_args = if separator < args.len() { + args[separator + 1..].to_vec() + } else { + Vec::new() + }; let timeout = Duration::from_millis(timeout_ms.unwrap_or(30_000)); + let retryable_timeout = timeout > crate::app::AGENT_START_SETTLE_DELAY + && timeout <= crate::app::MAX_AGENT_START_TIMEOUT; + let pinned_terminal_id = pane_terminal_id(&pane_id)?; + let mut retry_deadline = None; + let mut previous_busy_response = None; + let mut response = loop { + if let Some(previous_busy_response) = previous_busy_response.as_ref() { + let retry_expired = retry_deadline.is_some_and(|deadline| Instant::now() >= deadline); + if retry_expired + || pane_terminal_id(&pane_id)? != pinned_terminal_id + || !pane_shell_is_initializing(&pane_id)? + { + return super::print_response(previous_busy_response); + } + } + + let response = super::send_request(&Request { + id: "cli:agent:start".into(), + method: Method::AgentStart(AgentStartParams { + name: name.clone(), + kind: kind.clone(), + pane_id: pane_id.clone(), + args: agent_args.clone(), + timeout_ms, + }), + })?; + if response.get("error").is_none() { + break response; + } + if response["error"]["code"].as_str() != Some("agent_pane_busy") + || !retryable_timeout + || pinned_terminal_id.is_none() + || pane_terminal_id(&pane_id)? != pinned_terminal_id + || !pane_shell_is_initializing(&pane_id)? + { + return super::print_response(&response); + } + + let deadline = *retry_deadline + .get_or_insert_with(|| Instant::now() + PANE_SHELL_READINESS_RETRY_TIMEOUT); + previous_busy_response = Some(response); + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + if let Some(previous_busy_response) = previous_busy_response.as_ref() { + return super::print_response(previous_busy_response); + } + } + std::thread::sleep(AGENT_START_POLL_INTERVAL.min(remaining)); + }; + let Some(expected_terminal_id) = response["result"]["agent"]["terminal_id"].as_str() else { return super::print_response(&cli_agent_error( "cli:agent:start", @@ -355,6 +396,12 @@ fn agent_start(args: &[String]) -> std::io::Result { "agent start response did not include terminal_id", )); }; + if pinned_terminal_id + .as_deref() + .is_some_and(|pinned| pinned != expected_terminal_id) + { + return super::print_response(&agent_name_lost_error("cli:agent:start", name)); + } let waited = wait_for_named_agent( name, &pane_id, @@ -505,8 +552,7 @@ fn wait_for_named_agent( expected_kind: &str, expected_terminal_id: &str, ) -> std::io::Result> { - let started_at = Instant::now(); - let deadline = started_at.checked_add(timeout); + let deadline = Instant::now().checked_add(timeout); let mut first_poll = true; loop { if deadline.is_some_and(|deadline| Instant::now() >= deadline) { @@ -525,7 +571,7 @@ fn wait_for_named_agent( if response.get("error").is_some() { response = resolve_agent_target_unchecked(fallback_pane_id, poll_id)?; if response.get("error").is_some() { - std::thread::sleep(Duration::from_millis(100)); + std::thread::sleep(AGENT_START_POLL_INTERVAL); continue; } } @@ -567,10 +613,66 @@ fn wait_for_named_agent( if let Some(outcome) = outcome { return Ok(outcome); } - std::thread::sleep(Duration::from_millis(100)); + std::thread::sleep(AGENT_START_POLL_INTERVAL); } } +fn pane_terminal_id(pane_id: &str) -> std::io::Result> { + let response = super::send_request(&Request { + id: "cli:agent:start:pane".into(), + method: Method::PaneGet(PaneTarget { + pane_id: pane_id.to_owned(), + }), + })?; + Ok(response["result"]["pane"]["terminal_id"] + .as_str() + .map(str::to_owned)) +} + +fn pane_shell_is_initializing(pane_id: &str) -> std::io::Result { + let response = super::send_request(&Request { + id: "cli:agent:start:process_info".into(), + method: Method::PaneProcessInfo(PaneProcessInfoParams { + pane_id: Some(pane_id.to_owned()), + }), + })?; + Ok(process_info_shows_shell_initialization( + &response["result"]["process_info"], + )) +} + +#[cfg(unix)] +fn process_info_shows_shell_initialization(process_info: &serde_json::Value) -> bool { + let Some(shell_pid) = process_info["shell_pid"].as_u64() else { + return false; + }; + if process_info["foreground_process_group_id"].as_u64() != Some(shell_pid) { + return false; + } + process_info["foreground_processes"] + .as_array() + .is_some_and(|processes| { + processes.iter().any(|process| { + process["pid"].as_u64() == Some(shell_pid) + && (process["name"] + .as_str() + .is_some_and(crate::platform::is_pane_shell_process_name) + || process["argv"] + .as_array() + .and_then(|argv| argv.first()) + .and_then(serde_json::Value::as_str) + .is_some_and(crate::platform::is_pane_shell_process_name)) + }) + }) +} + +// Windows exposes no foreground process group, so shell initialization is not +// observable and a busy `agent.start` is not retried there. +#[cfg(not(unix))] +fn process_info_shows_shell_initialization(_process_info: &serde_json::Value) -> bool { + false +} + fn agent_name_lost_error(request_id: &str, expected_name: &str) -> serde_json::Value { cli_agent_error( request_id, diff --git a/tests/cli/agent_transport.rs b/tests/cli/agent_transport.rs index 85419e90ee..a052f69720 100644 --- a/tests/cli/agent_transport.rs +++ b/tests/cli/agent_transport.rs @@ -8,6 +8,24 @@ fn agent_start_waits_through_unknown_then_rejects_blocked() { let listener = UnixListener::bind(&socket_path).unwrap(); let server = thread::spawn(move || { + let (mut pane_stream, pane_line) = accept_fake_cli_operation(&listener); + let pane: serde_json::Value = serde_json::from_str(&pane_line).unwrap(); + assert_eq!(pane["method"], "pane.get"); + assert_eq!(pane["params"]["pane_id"], "w1:p1"); + writeln!( + pane_stream, + "{}", + serde_json::json!({ + "id": pane["id"], + "result": { + "type": "pane_info", + "pane": { "terminal_id": "term_1" } + } + }) + ) + .unwrap(); + pane_stream.flush().unwrap(); + let (mut start_stream, start_line) = accept_fake_cli_operation(&listener); let start: serde_json::Value = serde_json::from_str(&start_line).unwrap(); assert_eq!(start["method"], "agent.start"); @@ -98,6 +116,106 @@ fn agent_start_waits_through_unknown_then_rejects_blocked() { cleanup_test_base(&base); } +#[test] +fn agent_start_does_not_retry_after_the_target_terminal_changes() { + let base = unique_test_dir(); + fs::create_dir_all(&base).unwrap(); + let socket_path = base.join("herdr.sock"); + let listener = UnixListener::bind(&socket_path).unwrap(); + + let server = thread::spawn(move || { + for _ in 0..2 { + let (mut stream, line) = accept_fake_cli_operation(&listener); + let request: serde_json::Value = serde_json::from_str(&line).unwrap(); + assert_eq!(request["method"], "pane.get"); + writeln!( + stream, + "{}", + serde_json::json!({ + "id": request["id"], + "result": { + "type": "pane_info", + "pane": { "terminal_id": "term_1" } + } + }) + ) + .unwrap(); + stream.flush().unwrap(); + + let (mut stream, line) = accept_fake_cli_operation(&listener); + let request: serde_json::Value = serde_json::from_str(&line).unwrap(); + if request["method"] == "agent.start" { + writeln!( + stream, + "{}", + serde_json::json!({ + "id": request["id"], + "error": { + "code": "agent_pane_busy", + "message": "agent target pane w1:p1 is not an available shell" + } + }) + ) + .unwrap(); + stream.flush().unwrap(); + } else { + assert_eq!(request["method"], "pane.process_info"); + writeln!( + stream, + "{}", + serde_json::json!({ + "id": request["id"], + "result": { + "type": "pane_process_info", + "process_info": { + "pane_id": "w1:p1", + "shell_pid": 10, + "foreground_process_group_id": 10, + "foreground_processes": [ + { "pid": 10, "name": "bash" }, + { "pid": 11, "name": "startup-helper" } + ] + } + } + }) + ) + .unwrap(); + stream.flush().unwrap(); + } + } + + let (mut stream, line) = accept_fake_cli_operation(&listener); + let request: serde_json::Value = serde_json::from_str(&line).unwrap(); + assert_eq!(request["method"], "pane.get"); + writeln!( + stream, + "{}", + serde_json::json!({ + "id": request["id"], + "result": { + "type": "pane_info", + "pane": { "terminal_id": "term_2" } + } + }) + ) + .unwrap(); + stream.flush().unwrap(); + }); + + let started = run_cli( + &socket_path, + &[ + "agent", "start", "reviewer", "--kind", "pi", "--pane", "w1:p1", + ], + ); + assert_eq!(started.status.code(), Some(1)); + let error: serde_json::Value = serde_json::from_slice(&started.stderr).unwrap(); + assert_eq!(error["error"]["code"], "agent_pane_busy"); + + server.join().unwrap(); + cleanup_test_base(&base); +} + #[test] fn prompt_wait_is_sent_as_one_agent_request() { let base = unique_test_dir(); diff --git a/tests/cli/agents.rs b/tests/cli/agents.rs index 37c8d557c5..f5affc87af 100644 --- a/tests/cli/agents.rs +++ b/tests/cli/agents.rs @@ -1,5 +1,183 @@ use super::harness::*; +fn write_delayed_shell_and_fake_pi( + base: &Path, + shell_delay_seconds: &str, +) -> (PathBuf, PathBuf, PathBuf) { + use std::os::unix::fs::PermissionsExt; + + let bin = base.join("bin"); + let delayed_shell = bin.join("delayed-shell"); + let fake_pi = bin.join("pi"); + let invocations = base.join("pi-invocations"); + fs::create_dir_all(&bin).unwrap(); + fs::write( + &delayed_shell, + format!("#!/bin/sh\n/bin/sleep {shell_delay_seconds}\nexec /bin/sh\n"), + ) + .unwrap(); + fs::write( + &fake_pi, + format!( + "#!/bin/sh\nprintf '%s\\n' \"$@\" >> '{}'\nexport HERDR_AGENT=pi\n'{}' pane report-agent \"$HERDR_PANE_ID\" --source custom:delayed-shell-pi --agent pi --state idle >/dev/null\nwhile IFS= read -r _prompt; do :; done\n", + invocations.display(), + env!("CARGO_BIN_EXE_herdr"), + ), + ) + .unwrap(); + fs::set_permissions(&delayed_shell, fs::Permissions::from_mode(0o755)).unwrap(); + fs::set_permissions(&fake_pi, fs::Permissions::from_mode(0o755)).unwrap(); + (bin, delayed_shell, invocations) +} + +#[test] +fn agent_start_waits_for_a_new_pane_shell_to_finish_initializing() { + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let socket_path = runtime_dir.join("herdr.sock"); + let (bin, delayed_shell, invocations) = write_delayed_shell_and_fake_pi(&base, "0.4"); + let config = format!( + "onboarding = false\n[terminal]\ndefault_shell = {:?}\nshell_mode = \"non_login\"\n", + delayed_shell.to_str().unwrap() + ); + let herdr = spawn_herdr_with_config( + &config_home, + &runtime_dir, + &socket_path, + Some(&bin), + &config, + ); + wait_for_socket(&socket_path, Duration::from_secs(5)); + let seed = run_cli_json( + &socket_path, + &["workspace", "create", "--cwd", base.to_str().unwrap()], + ); + let seed_workspace = seed["result"]["workspace"]["workspace_id"] + .as_str() + .unwrap(); + let created = run_cli_json( + &socket_path, + &[ + "workspace", + "create", + "--cwd", + base.to_str().unwrap(), + "--no-focus", + ], + ); + let pane_id = created["result"]["root_pane"]["pane_id"].as_str().unwrap(); + let terminal_id = created["result"]["root_pane"]["terminal_id"] + .as_str() + .unwrap(); + assert!(!created["result"]["root_pane"]["focused"].as_bool().unwrap()); + + let started = run_cli_json( + &socket_path, + &[ + "agent", + "start", + "worker", + "--kind", + "pi", + "--pane", + pane_id, + "--timeout", + "8000", + "--", + "--no-context-files", + "--no-skills", + "--no-extensions", + ], + ); + + assert_eq!(started["result"]["agent"]["terminal_id"], terminal_id); + assert_eq!(started["result"]["agent"]["pane_id"], pane_id); + assert_eq!(started["result"]["agent"]["interactive_ready"], true); + assert_eq!( + fs::read_to_string(&invocations).unwrap(), + "--no-context-files\n--no-skills\n--no-extensions\n" + ); + assert_eq!( + run_cli_json(&socket_path, &["workspace", "list"])["result"]["workspaces"] + .as_array() + .unwrap() + .iter() + .find(|workspace| workspace["workspace_id"] == seed_workspace) + .unwrap()["focused"], + true + ); + + cleanup_spawned_herdr(herdr, base); +} + +#[test] +fn agent_start_stops_retrying_when_the_pane_shell_stays_busy() { + let base = unique_test_dir(); + let config_home = base.join("config"); + let runtime_dir = base.join("runtime"); + let socket_path = runtime_dir.join("herdr.sock"); + let (bin, delayed_shell, invocations) = write_delayed_shell_and_fake_pi(&base, "2.3"); + let config = format!( + "onboarding = false\n[terminal]\ndefault_shell = {:?}\nshell_mode = \"non_login\"\n", + delayed_shell.to_str().unwrap() + ); + let herdr = spawn_herdr_with_config( + &config_home, + &runtime_dir, + &socket_path, + Some(&bin), + &config, + ); + wait_for_socket(&socket_path, Duration::from_secs(5)); + let created = run_cli_json( + &socket_path, + &["workspace", "create", "--cwd", base.to_str().unwrap()], + ); + let pane_id = created["result"]["root_pane"]["pane_id"].as_str().unwrap(); + + let started_at = Instant::now(); + let unavailable = run_cli( + &socket_path, + &[ + "agent", + "start", + "worker", + "--kind", + "pi", + "--pane", + pane_id, + "--timeout", + "8000", + ], + ); + assert_eq!(unavailable.status.code(), Some(1)); + let error: serde_json::Value = serde_json::from_slice(&unavailable.stderr).unwrap(); + assert_eq!(error["error"]["code"], "agent_pane_busy"); + assert!(started_at.elapsed() >= Duration::from_secs(2)); + assert!(started_at.elapsed() < Duration::from_secs(4)); + assert!(!invocations.exists()); + + let retried = run_cli_json( + &socket_path, + &[ + "agent", + "start", + "worker", + "--kind", + "pi", + "--pane", + pane_id, + "--timeout", + "8000", + ], + ); + assert_eq!(retried["result"]["type"], "agent_started"); + assert_eq!(fs::read_to_string(&invocations).unwrap(), "\n"); + + cleanup_spawned_herdr(herdr, base); +} + #[test] fn agent_start_command_works() { use std::os::unix::fs::PermissionsExt; @@ -75,7 +253,7 @@ fn agent_start_command_works() { assert_eq!(error["error"]["code"], "invalid_agent_argument"); } - for invalid_timeout in ["3000", "300001"] { + for invalid_timeout in ["3000", "300001", "18446744073709551615"] { let rejected = run_cli( &socket_path, &[ @@ -259,7 +437,7 @@ fn agent_start_rejects_a_shell_replaced_by_a_foreground_program() { "--pane", &pane_id, "--timeout", - "1000", + "4000", ], ); assert_eq!(started.status.code(), Some(1)); @@ -310,6 +488,23 @@ fn agent_start_timeout_releases_the_name_for_reuse() { .unwrap() .to_string(); + assert!(run_cli( + &socket_path, + &[ + "pane", + "report-agent", + &reuse_pane_id, + "--source", + "custom:reuse", + "--agent", + "pi", + "--state", + "idle", + ], + ) + .status + .success()); + let started = run_cli( &socket_path, &[ @@ -328,22 +523,6 @@ fn agent_start_timeout_releases_the_name_for_reuse() { let error: serde_json::Value = serde_json::from_slice(&started.stderr).unwrap(); assert_eq!(error["error"]["code"], "timeout"); - assert!(run_cli( - &socket_path, - &[ - "pane", - "report-agent", - &reuse_pane_id, - "--source", - "custom:reuse", - "--agent", - "pi", - "--state", - "idle", - ], - ) - .status - .success()); let reused = run_cli(&socket_path, &["agent", "rename", &reuse_pane_id, "worker"]); assert!( reused.status.success(),