254
},
255
"run_in_background": {
256
"type": "boolean",
257
- "description": "Start the command as a background task and return a task id immediately instead of waiting for it to finish. Set this to true for servers, watchers, builds, test suites, and anything that may run longer than two minutes (foreground commands are killed after 120 seconds). Poll the task's output and status with command_output, and stop it with kill_command. Background tasks are killed when sigit exits. Defaults to false."
257
+ "description": "Start the command as a background task and return a task id immediately instead of waiting for it to finish. Set this to true for servers, watchers, builds, test suites, and anything that may run longer than two minutes (foreground commands are killed after 120 seconds). Wait for it with monitor_command (or take a quick look with command_output), and stop it with kill_command. Background tasks are killed when sigit exits. Defaults to false."
258
}
259
},
260
"required": ["command"],
432
"additionalProperties": false
433
}),
434
},
435
+ AgentTool {
436
+ name: "monitor_command",
437
+ description: "Follow a background task (started with run_command's \
438
+ run_in_background) until it stops: it exits, a line of new output \
439
+ matches `until`, or `timeout_seconds` pass. One call covers the whole \
440
+ wait and shows up as a single live-updating card, so use this to wait \
441
+ for a build, test run or release instead of calling command_output \
442
+ over and over. Reports the exit code (or that the task is still \
443
+ running) and the output gathered while watching. The task keeps \
444
+ running if the watch ends first. Use command_output only for a quick \
445
+ look at a server or watcher.",
446
+ parameters_schema: json!({
447
+ "type": "object",
448
+ "properties": {
449
+ "task_id": {
450
+ "type": "integer",
451
+ "description": "Id of the background task, as returned by run_command with run_in_background."
452
+ },
453
+ "until": {
454
+ "type": "string",
455
+ "description": "Optional regular expression. The watch ends as soon as a line of new output matches it, e.g. \"Finished|error\\\\[\"."
456
+ },
457
+ "timeout_seconds": {
458
+ "type": "integer",
459
+ "minimum": 1,
460
+ "maximum": 600,
461
+ "description": "Longest the watch may last. Defaults to 300."
462
+ }
463
+ },
464
+ "required": ["task_id"],
465
+ "additionalProperties": false
466
+ }),
467
+ },
468
]
469
}
470
558
"command_output" => exec_command_output_wait(arguments, active_session())
559
.await
560
.into(),
561
+ "monitor_command" => exec_monitor_command(arguments, active_session(), tool_call_id)
562
+ .await
563
+ .into(),
564
// Tools discovered from MCP servers are namespaced `mcp__<server>__<tool>`
565
// and forwarded to the owning server.
566
_ if crate::mcp::is_mcp_tool(name) => crate::mcp::call_tool(name, arguments).await.into(),
3081
);
3082
3083
format!(
3048
- "Started background task {task_id}: `{command_str}`. Poll its output and status \
3049
- with command_output (task_id: {task_id}); stop it with kill_command. The task is \
3050
- killed when sigit exits."
3084
+ "Started background task {task_id}: `{command_str}`. Wait for it with \
3085
+ monitor_command (task_id: {task_id}), or take a quick look with command_output; \
3086
+ stop it with kill_command. The task is killed when sigit exits."
3087
)
3088
}
3089
3142
poll_command_output(arguments, owner, false).text
3143
}
3144
3145
+/// The sentence both `command_output` and `monitor_command` open with.
3146
+fn task_status_line(task_id: u64, task: &BackgroundTask, exit_code: Option<i32>) -> String {
3147
+ let command = &task.command;
3148
+ match exit_code {
3149
+ None => format!("Task {task_id} (`{command}`) is still running."),
3150
+ Some(_) if task.killed => format!("Task {task_id} (`{command}`) was killed."),
3151
+ Some(code) => format!("Task {task_id} (`{command}`) exited with code {code}."),
3152
+ }
3153
+}
3154
+
3155
+/// The command of `owner`'s background task `task_id`, for a card title.
3156
+pub fn background_task_command(task_id: u64) -> Option<String> {
3157
+ let owner = active_session();
3158
+ lock_tasks()
3159
+ .get(&task_id)
3160
+ .filter(|task| task.owner == owner)
3161
+ .map(|task| task.command.clone())
3162
+}
3163
+
3164
/// One `command_output` check, with the tool text and whether the task is
3165
/// still running, so the long-poll loop knows when to keep waiting.
3166
struct CommandPoll {
3216
3217
let (new_output, dropped) = drain_task_output(task);
3218
3164
- let command = &task.command;
3165
- let status = match exit_code {
3166
- None => format!("Task {task_id} (`{command}`) is still running."),
3167
- Some(_) if task.killed => format!("Task {task_id} (`{command}`) was killed."),
3168
- Some(code) => format!("Task {task_id} (`{command}`) exited with code {code}."),
3169
- };
3219
+ let status = task_status_line(task_id, task, exit_code);
3220
3221
let running = exit_code.is_none();
3222
let text = if new_output.is_empty() {
3270
}
3271
}
3272
3273
+// ── monitor_command ──────────────────────────────────────────────────────────
3274
+
3275
+/// Where a long-running tool reports what it has so far, so the client's card
3276
+/// for the call can update before the call returns. The seam has `client_fs`'s
3277
+/// shape: `main.rs` registers an ACP implementation at `initialize`, and with
3278
+/// nothing registered (TUI, headless) updates go nowhere.
3279
+pub trait ToolProgress: Send + Sync {
3280
+ /// Replace the content of tool call `tool_call_id` with `text`.
3281
+ fn update(&self, session_id: &str, tool_call_id: &str, text: &str);
3282
+}
3283
+
3284
+static TOOL_PROGRESS: RwLock<Option<Arc<dyn ToolProgress>>> = RwLock::new(None);
3285
+
3286
+pub fn register_tool_progress(progress: Arc<dyn ToolProgress>) {
3287
+ if let Ok(mut guard) = TOOL_PROGRESS.write() {
3288
+ *guard = Some(progress);
3289
+ }
3290
+}
3291
+
3292
+fn report_progress(session: Option<&str>, tool_call_id: Option<&str>, text: &str) {
3293
+ let (Some(session), Some(tool_call_id)) = (session, tool_call_id) else {
3294
+ return;
3295
+ };
3296
+ let progress = TOOL_PROGRESS.read().ok().and_then(|guard| guard.clone());
3297
+ if let Some(progress) = progress {
3298
+ progress.update(session, tool_call_id, text);
3299
+ }
3300
+}
3301
+
3302
+/// One look at a task for `monitor_command`.
3303
+struct MonitorStep {
3304
+ status: String,
3305
+ exit_code: Option<i32>,
3306
+ output: String,
3307
+ dropped: bool,
3308
+}
3309
+
3310
+fn monitor_step(task_id: u64, owner: Option<&str>) -> Option<MonitorStep> {
3311
+ let mut map = lock_tasks();
3312
+ let task = map
3313
+ .get_mut(&task_id)
3314
+ .filter(|task| task.owner.as_deref() == owner)?;
3315
+ let was_running = task.exit_code.is_none();
3316
+ let exit_code = poll_exit_code(task);
3317
+ if was_running && exit_code.is_some() {
3318
+ // Let the reader threads flush the final output (see poll_command_output).
3319
+ std::thread::sleep(std::time::Duration::from_millis(100));
3320
+ }
3321
+ let (output, dropped) = drain_task_output(task);
3322
+ Some(MonitorStep {
3323
+ status: task_status_line(task_id, task, exit_code),
3324
+ exit_code,
3325
+ output,
3326
+ dropped,
3327
+ })
3328
+}
3329
+
3330
+/// Keep the tail of `buf` within COMMAND_OUTPUT_LIMIT, like the reader threads.
3331
+fn cap_tail(buf: &mut String) -> bool {
3332
+ if buf.len() <= COMMAND_OUTPUT_LIMIT {
3333
+ return false;
3334
+ }
3335
+ let mut cut = buf.len() - COMMAND_OUTPUT_LIMIT;
3336
+ while !buf.is_char_boundary(cut) {
3337
+ cut += 1;
3338
+ }
3339
+ buf.drain(..cut);
3340
+ true
3341
+}
3342
+
3343
+/// `monitor_command` tool: follow a background task to a stopping point.
3344
+async fn exec_monitor_command(
3345
+ arguments: &str,
3346
+ owner: Option<String>,
3347
+ tool_call_id: Option<&str>,
3348
+) -> String {
3349
+ const DEFAULT_TIMEOUT_SECONDS: u64 = 300;
3350
+ const MAX_TIMEOUT_SECONDS: u64 = 600;
3351
+ const POLL_INTERVAL: std::time::Duration = std::time::Duration::from_millis(200);
3352
+ const PROGRESS_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
3353
+
3354
+ let task_id = match parse_task_id(arguments) {
3355
+ Ok(id) => id,
3356
+ Err(err) => return err,
3357
+ };
3358
+ let args: Value = serde_json::from_str(arguments).unwrap_or(Value::Null);
3359
+ let until_source = args
3360
+ .get("until")
3361
+ .and_then(Value::as_str)
3362
+ .filter(|pattern| !pattern.is_empty());
3363
+ let until = match until_source.map(regex::Regex::new).transpose() {
3364
+ Ok(re) => re,
3365
+ Err(err) => return format!("Error: invalid \"until\" pattern: {err}"),
3366
+ };
3367
+ let timeout_seconds = args
3368
+ .get("timeout_seconds")
3369
+ .and_then(Value::as_u64)
3370
+ .unwrap_or(DEFAULT_TIMEOUT_SECONDS)
3371
+ .clamp(1, MAX_TIMEOUT_SECONDS);
3372
+
3373
+ let started = tokio::time::Instant::now();
3374
+ let deadline = started + std::time::Duration::from_secs(timeout_seconds);
3375
+ let signal = owner.as_deref().map(abort_signal);
3376
+
3377
+ let mut gathered = String::new();
3378
+ let mut dropped = false;
3379
+ // The unterminated last line so far, so a match is not missed when the
3380
+ // line arrives in two chunks.
3381
+ let mut partial = String::new();
3382
+ let mut last_report = started;
3383
+ let mut reported_len = 0;
3384
+
3385
+ enum Stop {
3386
+ Exited(String),
3387
+ Matched,
3388
+ Timeout,
3389
+ Aborted,
3390
+ }
3391
+
3392
+ let (status, stop) = loop {
3393
+ let step_owner = owner.clone();
3394
+ let step =
3395
+ tokio::task::spawn_blocking(move || monitor_step(task_id, step_owner.as_deref())).await;
3396
+ let step = match step {
3397
+ Ok(Some(step)) => step,
3398
+ Ok(None) => return unknown_task(task_id),
3399
+ Err(err) => return format!("Error: monitor_command task failed: {err}"),
3400
+ };
3401
+
3402
+ dropped |= step.dropped;
3403
+ let mut matched = false;
3404
+ if !step.output.is_empty() {
3405
+ gathered.push_str(&step.output);
3406
+ dropped |= cap_tail(&mut gathered);
3407
+ if let Some(re) = &until {
3408
+ let scan = format!("{partial}{}", step.output);
3409
+ matched = scan.split('\n').any(|line| re.is_match(line));
3410
+ partial = if scan.ends_with('\n') {
3411
+ String::new()
3412
+ } else {
3413
+ scan.rsplit('\n').next().unwrap_or_default().to_string()
3414
+ };
3415
+ }
3416
+ }
3417
+
3418
+ if step.exit_code.is_some() {
3419
+ break (step.status.clone(), Stop::Exited(step.status));
3420
+ }
3421
+ if matched {
3422
+ break (step.status, Stop::Matched);
3423
+ }
3424
+ let now = tokio::time::Instant::now();
3425
+ if now >= deadline {
3426
+ break (step.status, Stop::Timeout);
3427
+ }
3428
+ if signal.as_ref().is_some_and(|signal| signal.is_aborted()) {
3429
+ break (step.status, Stop::Aborted);
3430
+ }
3431
+
3432
+ if now.duration_since(last_report) >= PROGRESS_INTERVAL && gathered.len() != reported_len {
3433
+ last_report = now;
3434
+ reported_len = gathered.len();
3435
+ let text = format!(
3436
+ "{} Watching for {}s.\n{gathered}",
3437
+ step.status,
3438
+ now.duration_since(started).as_secs()
3439
+ );
3440
+ report_progress(owner.as_deref(), tool_call_id, &text);
3441
+ }
3442
+
3443
+ match &signal {
3444
+ Some(signal) => {
3445
+ tokio::select! {
3446
+ _ = tokio::time::sleep(POLL_INTERVAL) => {}
3447
+ _ = signal.aborted() => {}
3448
+ }
3449
+ }
3450
+ None => tokio::time::sleep(POLL_INTERVAL).await,
3451
+ }
3452
+ };
3453
+
3454
+ let headline = match stop {
3455
+ Stop::Exited(status) => status,
3456
+ Stop::Matched => format!(
3457
+ "{} Stopped watching: a line matched \"{}\".",
3458
+ status,
3459
+ until_source.unwrap_or_default()
3460
+ ),
3461
+ Stop::Timeout => {
3462
+ format!("{status} Stopped watching after {timeout_seconds}s; the task keeps running.")
3463
+ }
3464
+ Stop::Aborted => format!(
3465
+ "{status} Stopped watching because the turn was cancelled; the task keeps running."
3466
+ ),
3467
+ };
3468
+ let note = dropped_note(dropped);
3469
+ if gathered.is_empty() {
3470
+ format!("{headline} No output while watching.{note}")
3471
+ } else {
3472
+ format!("{headline} Output while watching:{note}\n{gathered}")
3473
+ }
3474
+}
3475
+
3476
/// Stop and forget every background task `owner` started. Called when its
3477
/// session closes: nobody is left who could poll or kill them. Returns how
3478
/// many were still running.
3671
#[test]
3672
fn test_all_tools_count() {
3673
let tools = all_tools();
3421
- assert_eq!(tools.len(), 15);
3674
+ assert_eq!(tools.len(), 16);
3675
assert_eq!(tools[0].name, "read_file");
3676
assert_eq!(tools[1].name, "create_directory");
3677
assert_eq!(tools[2].name, "list_directory");
3687
assert_eq!(tools[12].name, "remember");
3688
assert_eq!(tools[13].name, "command_output");
3689
assert_eq!(tools[14].name, "kill_command");
3690
+ assert_eq!(tools[15].name, "monitor_command");
3691
}
3692
3693
#[test]
5028
exec_kill_command(&serde_json::json!({ "task_id": task_id }).to_string(), None);
5029
}
5030
5031
+ fn start_test_task(command: &str, owner: Option<&str>) -> u64 {
5032
+ let args = serde_json::json!({
5033
+ "command": command,
5034
+ "cwd": std::env::temp_dir(),
5035
+ "run_in_background": true
5036
+ })
5037
+ .to_string();
5038
+ background_task_id(&exec_run_command(&args, owner))
5039
+ }
5040
+
5041
+ #[cfg(unix)]
5042
+ const SLOW_TAIL: &str = "sleep 30";
5043
+ #[cfg(windows)]
5044
+ const SLOW_TAIL: &str = "ping -n 31 127.0.0.1 > nul";
5045
+
5046
+ fn kill_test_task(task_id: u64, owner: Option<&str>) {
5047
+ exec_kill_command(
5048
+ &serde_json::json!({ "task_id": task_id }).to_string(),
5049
+ owner,
5050
+ );
5051
+ }
5052
+
5053
+ #[tokio::test]
5054
+ async fn test_monitor_command_returns_on_exit_with_output() {
5055
+ let task_id = start_test_task("echo monitored", None);
5056
+ let args = serde_json::json!({ "task_id": task_id }).to_string();
5057
+ let result = exec_monitor_command(&args, None, None).await;
5058
+ assert!(result.contains("exited with code 0"), "got: {result}");
5059
+ assert!(result.contains("monitored"), "got: {result}");
5060
+ }
5061
+
5062
+ #[tokio::test]
5063
+ async fn test_monitor_command_stops_when_until_matches() {
5064
+ let task_id = start_test_task(&format!("echo build-ready && {SLOW_TAIL}"), None);
5065
+ let args =
5066
+ serde_json::json!({ "task_id": task_id, "until": "build-(ready|failed)" }).to_string();
5067
+ let started = std::time::Instant::now();
5068
+ let result = exec_monitor_command(&args, None, None).await;
5069
+ assert!(started.elapsed().as_secs() < 20, "waited for the task");
5070
+ assert!(result.contains("still running"), "got: {result}");
5071
+ assert!(result.contains("a line matched"), "got: {result}");
5072
+ assert!(result.contains("build-ready"), "got: {result}");
5073
+ kill_test_task(task_id, None);
5074
+ }
5075
+
5076
+ #[tokio::test]
5077
+ async fn test_monitor_command_times_out_on_a_quiet_task() {
5078
+ let task_id = start_test_task(SLOW_TAIL, None);
5079
+ let args = serde_json::json!({ "task_id": task_id, "timeout_seconds": 1 }).to_string();
5080
+ let result = exec_monitor_command(&args, None, None).await;
5081
+ assert!(result.contains("still running"), "got: {result}");
5082
+ assert!(result.contains("after 1s"), "got: {result}");
5083
+ kill_test_task(task_id, None);
5084
+ }
5085
+
5086
+ #[tokio::test]
5087
+ async fn test_monitor_command_ends_when_the_turn_is_aborted() {
5088
+ let owner = "monitor-abort-test";
5089
+ reset_abort(owner);
5090
+ let task_id = start_test_task(SLOW_TAIL, Some(owner));
5091
+ let args = serde_json::json!({ "task_id": task_id }).to_string();
5092
+ tokio::spawn(async move {
5093
+ tokio::time::sleep(std::time::Duration::from_millis(400)).await;
5094
+ abort_session_tools(owner);
5095
+ });
5096
+ let started = std::time::Instant::now();
5097
+ let result = exec_monitor_command(&args, Some(owner.to_string()), None).await;
5098
+ assert!(started.elapsed().as_secs() < 20, "ignored the abort");
5099
+ assert!(result.contains("turn was cancelled"), "got: {result}");
5100
+ kill_test_task(task_id, Some(owner));
5101
+ forget_abort(owner);
5102
+ }
5103
+
5104
+ #[tokio::test]
5105
+ async fn test_monitor_command_rejects_other_owners_and_bad_input() {
5106
+ let task_id = start_test_task(SLOW_TAIL, Some("monitor-owner-a"));
5107
+ let args = serde_json::json!({ "task_id": task_id }).to_string();
5108
+ let result = exec_monitor_command(&args, Some("monitor-owner-b".into()), None).await;
5109
+ assert!(result.contains("no background task"), "got: {result}");
5110
+ let bad = serde_json::json!({ "task_id": task_id, "until": "(" }).to_string();
5111
+ let result = exec_monitor_command(&bad, Some("monitor-owner-a".into()), None).await;
5112
+ assert!(result.contains("invalid \"until\""), "got: {result}");
5113
+ kill_test_task(task_id, Some("monitor-owner-a"));
5114
+ }
5115
+
5116
#[test]
5117
fn test_kill_command_stops_background_task() {
5118
#[cfg(unix)]