1use std::collections::HashMap;
2use std::fs::File;
3use std::process::Stdio;
4use std::sync::{Arc, Mutex};
5use std::time::Duration;
6
7use command_group::{AsyncCommandGroup, AsyncGroupChild};
8use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
9use tokio::sync::mpsc;
10use tokio_util::sync::CancellationToken;
11
12use crate::error::RuntimeError;
13use crate::task_registry::{TaskKind, TaskRegistry, TaskStatus};
14use crate::tool::{BoxFut, Tier, Tool, ToolArgs, ToolCtx, ToolResult};
15use crate::value::Value;
16
17const DEFAULT_SPAWN_TIMEOUT_MS: u64 = 1_800_000;
18const MAX_SPAWN_TIMEOUT_MS: u64 = 86_400_000;
19const DEFAULT_MAX_OUTPUT_BYTES: u64 = 10_485_760;
20const RING_BUFFER_BYTES: usize = 65_536;
21const IO_DRAIN_TIMEOUT: Duration = Duration::from_secs(2);
22const DEFAULT_OUTPUT_LIMIT: usize = 32_000;
23
24#[derive(Debug, Clone, Hash, PartialEq, Eq)]
25pub struct BgHandle {
26 session_id: String,
27 local_id: u64,
28}
29
30impl BgHandle {
31 #[allow(clippy::inherent_to_string)]
32 pub fn to_string(&self) -> String {
33 format!("bg_{}_{}", self.session_id, self.local_id)
34 }
35
36 pub fn parse(s: &str) -> Option<Self> {
37 let rest = s.strip_prefix("bg_")?;
38 let idx = rest.rfind('_')?;
39 let session_id = rest[..idx].to_string();
40 let local_id = rest[idx + 1..].parse().ok()?;
41 Some(Self {
42 session_id,
43 local_id,
44 })
45 }
46}
47
48#[derive(Debug, Clone)]
49pub enum BgStatus {
50 Running {
51 pid: u32,
52 started_at: i64,
53 },
54 Exited {
55 exit_code: i32,
56 started_at: i64,
57 ended_at: i64,
58 },
59 TimedOut {
60 started_at: i64,
61 ended_at: i64,
62 },
63 Killed {
64 started_at: i64,
65 ended_at: i64,
66 },
67 Failed {
68 error: String,
69 started_at: i64,
70 ended_at: i64,
71 },
72}
73
74impl BgStatus {
75 fn kind(&self) -> &'static str {
76 match self {
77 Self::Running { .. } => "running",
78 Self::Exited { .. } => "exited",
79 Self::TimedOut { .. } => "timed_out",
80 Self::Killed { .. } => "killed",
81 Self::Failed { .. } => "failed",
82 }
83 }
84
85 fn exit_code(&self) -> Option<i32> {
86 match self {
87 Self::Exited { exit_code, .. } => Some(*exit_code),
88 _ => None,
89 }
90 }
91
92 fn started_at(&self) -> i64 {
93 match self {
94 Self::Running { started_at, .. }
95 | Self::Exited { started_at, .. }
96 | Self::TimedOut { started_at, .. }
97 | Self::Killed { started_at, .. }
98 | Self::Failed { started_at, .. } => *started_at,
99 }
100 }
101
102 fn ended_at(&self) -> Option<i64> {
103 match self {
104 Self::Exited { ended_at, .. }
105 | Self::TimedOut { ended_at, .. }
106 | Self::Killed { ended_at, .. }
107 | Self::Failed { ended_at, .. } => Some(*ended_at),
108 _ => None,
109 }
110 }
111
112 fn is_finished(&self) -> bool {
113 !matches!(self, Self::Running { .. })
114 }
115
116 fn error(&self) -> Option<&str> {
117 match self {
118 Self::Failed { error, .. } => Some(error),
119 _ => None,
120 }
121 }
122}
123
124#[derive(Debug, Default)]
125pub struct BgOutput {
126 pub combined: Vec<u8>,
127 pub total_bytes: u64,
128 pub truncated: bool,
129 buffer_start: usize,
130}
131
132fn framed_output(kind: StreamKind, data: &[u8]) -> Vec<u8> {
133 let prefix: &[u8] = match kind {
134 StreamKind::Stdout => b"[out] ",
135 StreamKind::Stderr => b"[err] ",
136 };
137 let mut frame = Vec::with_capacity(prefix.len() + data.len() + 1);
138 frame.extend_from_slice(prefix);
139 frame.extend_from_slice(data);
140 if !data.ends_with(b"\n") {
141 frame.push(b'\n');
142 }
143 frame
144}
145
146impl BgOutput {
147 fn push(&mut self, kind: StreamKind, data: &[u8], max: u64) -> Vec<u8> {
148 let mut new_total = self.total_bytes + data.len() as u64;
149 let mut to_write = data;
150 if new_total > max {
151 let allowed = max.saturating_sub(self.total_bytes) as usize;
152 to_write = &data[..allowed.min(data.len())];
153 new_total = max;
154 self.truncated = true;
155 }
156 let frame = if to_write.is_empty() {
157 Vec::new()
158 } else {
159 framed_output(kind, to_write)
160 };
161 self.combined.extend_from_slice(&frame);
162 self.total_bytes = new_total;
163 let max_ring = RING_BUFFER_BYTES;
164 if self.combined.len() > max_ring {
165 let drop = self.combined.len() - max_ring;
166 self.combined.drain(..drop);
167 self.buffer_start += drop;
168 }
169 frame
170 }
171
172 fn read_from(&self, cursor: usize, limit: usize) -> (Vec<u8>, usize, usize, bool, bool) {
173 let end = self.buffer_start + self.combined.len();
174 let start = cursor.max(self.buffer_start).min(end);
175 let local_cursor = start - self.buffer_start;
176 let (chunk, local_next, eof) = page_bytes(&self.combined, local_cursor, limit);
177 (
178 chunk,
179 start,
180 self.buffer_start + local_next,
181 eof,
182 cursor < self.buffer_start,
183 )
184 }
185}
186
187fn page_bytes(data: &[u8], cursor: usize, limit: usize) -> (Vec<u8>, usize, bool) {
188 if cursor >= data.len() {
189 return (Vec::new(), data.len(), true);
190 }
191 let remaining = &data[cursor..];
192 let mut take = remaining.len().min(limit);
193 if take < remaining.len()
194 && let Err(error) = std::str::from_utf8(&remaining[..take])
195 && error.error_len().is_none()
196 && error.valid_up_to() > 0
197 {
198 take = error.valid_up_to();
199 }
200 let chunk = remaining[..take].to_vec();
201 let next = cursor + take;
202 let eof = next >= data.len();
203 (chunk, next, eof)
204}
205
206#[derive(Clone, Copy)]
207enum StreamKind {
208 Stdout,
209 Stderr,
210}
211
212pub(crate) enum BgControl {
213 Kill,
214}
215
216pub struct BgEntry {
217 pub session_id: String,
218 pub(crate) control_tx: mpsc::Sender<BgControl>,
219 pub status: Arc<Mutex<BgStatus>>,
220 pub output: Arc<Mutex<BgOutput>>,
221 pub log_path: std::path::PathBuf,
222 pub task_id: Option<crate::task_registry::TaskId>,
223}
224
225impl crate::watch::Watchable for BgEntry {
226 fn watch_output(
227 self: std::sync::Arc<Self>,
228 pattern: String,
229 cancel: tokio_util::sync::CancellationToken,
230 ) -> std::pin::Pin<Box<dyn std::future::Future<Output = crate::watch::WatchResult> + Send>>
231 {
232 let output = self.output.clone();
233 let status = self.status.clone();
234 Box::pin(async move {
235 loop {
236 tokio::select! {
237 _ = cancel.cancelled() => return crate::watch::WatchResult::Cancelled,
238 _ = tokio::time::sleep(std::time::Duration::from_millis(100)) => {
239 let text = {
240 let out = output.lock().unwrap();
241 String::from_utf8_lossy(&out.combined).into_owned()
242 };
243 if let Some(pos) = text.find(&pattern) {
244 return crate::watch::WatchResult::Matched {
245 row: None,
246 col: Some(pos as u16),
247 text: pattern,
248 };
249 }
250 let st = status.lock().unwrap().clone();
251 if !matches!(st, BgStatus::Running { .. }) {
252 return crate::watch::WatchResult::SourceExited;
253 }
254 }
255 }
256 }
257 })
258 }
259}
260
261#[derive(Default)]
262pub struct BgRegistry {
263 entries: Mutex<HashMap<String, Arc<BgEntry>>>,
264 task_registry: Option<TaskRegistry>,
265}
266
267impl BgRegistry {
268 pub fn new() -> Self {
269 Self::default()
270 }
271
272 pub fn with_task_registry(mut self, tr: TaskRegistry) -> Self {
273 self.task_registry = Some(tr);
274 self
275 }
276
277 pub fn kill_all(&self) {
278 let entries = self.entries.lock().unwrap();
279 for (_, entry) in entries.iter() {
280 let _ = entry.control_tx.try_send(BgControl::Kill);
281 }
282 }
283
284 pub fn spawn(
285 self: &Arc<Self>,
286 cmd: String,
287 timeout_ms: Option<u64>,
288 max_output_bytes: u64,
289 ctx: &ToolCtx,
290 ) -> Result<Value, RuntimeError> {
291 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".to_string());
292 let local_id = uuid::Uuid::now_v7().as_u64_pair().0;
293 let handle = BgHandle {
294 session_id: session_id.clone(),
295 local_id,
296 };
297 let handle_str = handle.to_string();
298
299 let dir = ctx.session_dir.clone().ok_or_else(|| {
300 RuntimeError::ToolFailed("bash.spawn: session_dir not available".into())
301 })?;
302 std::fs::create_dir_all(&dir).map_err(|e| {
303 RuntimeError::ToolFailed(format!("bash.spawn: create session_dir: {e}"))
304 })?;
305 let log_path = dir.join(format!("bg_{}.log", handle_str));
306 let log_file = open_log_file(&log_path)
307 .map_err(|error| RuntimeError::ToolFailed(format!("bash.spawn: {error}")))?;
308
309 let timeout = match timeout_ms {
310 Some(0) => None,
311 Some(ms) => Some(Duration::from_millis(ms.min(MAX_SPAWN_TIMEOUT_MS))),
312 None => Some(Duration::from_millis(DEFAULT_SPAWN_TIMEOUT_MS)),
313 };
314
315 let (control_tx, control_rx) = mpsc::channel::<BgControl>(8);
316 let status = Arc::new(Mutex::new(BgStatus::Running {
317 pid: 0,
318 started_at: now_ms(),
319 }));
320 let output = Arc::new(Mutex::new(BgOutput::default()));
321 let cancel = ctx.cancel.clone();
322 let task_cancel = cancel.child_token();
323
324 let task_id = self.task_registry.as_ref().map(|tr| {
325 tr.register(
326 TaskKind::Bash,
327 cmd.clone(),
328 handle_str.clone(),
329 session_id.clone(),
330 task_cancel.clone(),
331 )
332 });
333
334 let entry = Arc::new(BgEntry {
335 session_id: session_id.clone(),
336 control_tx,
337 status: status.clone(),
338 output: output.clone(),
339 log_path: log_path.clone(),
340 task_id: task_id.clone(),
341 });
342 {
343 let mut entries = self.entries.lock().unwrap();
344 entries.insert(handle_str.clone(), entry.clone());
345 }
346
347 let registry = Arc::clone(self);
348 let handle_str_for_task = handle_str.clone();
349 let status_for_task = status.clone();
350 let log_path_for_return = log_path.clone();
351 let stream_tx = ctx.stream_tx.clone();
352 let handle_for_task = handle_str.clone();
353 let task_registry = self.task_registry.clone();
354 let task_id_for_spawn = task_id.clone();
355 let flow_run_id = ctx.flow_run_id.as_ref().map(|r| r.0.to_string());
356 tokio::spawn(async move {
357 run_bg_process(
358 handle_str_for_task,
359 cmd,
360 timeout,
361 max_output_bytes,
362 log_file,
363 status_for_task,
364 output,
365 control_rx,
366 task_cancel,
367 registry,
368 stream_tx,
369 handle_for_task,
370 task_registry,
371 task_id_for_spawn,
372 flow_run_id,
373 )
374 .await;
375 });
376
377 let pid = {
378 let s = status.lock().unwrap();
379 if let BgStatus::Running { pid, .. } = &*s {
380 *pid
381 } else {
382 0
383 }
384 };
385
386 Ok(Value::Struct(vec![
387 ("handle".into(), Value::Str(handle_str)),
388 ("status".into(), Value::Str("running".into())),
389 ("pid".into(), Value::Int(pid as i64)),
390 (
391 "log_path".into(),
392 Value::Str(log_path_for_return.to_string_lossy().into_owned()),
393 ),
394 ]))
395 }
396
397 pub fn lookup(&self, handle_str: &str, session_id: &str) -> Result<Arc<BgEntry>, RuntimeError> {
398 let handle = BgHandle::parse(handle_str).ok_or_else(|| {
399 RuntimeError::ToolFailed(format!("bash: invalid handle `{handle_str}`"))
400 })?;
401 if handle.session_id != session_id {
402 return Err(RuntimeError::ToolFailed(format!(
403 "bash: handle `{handle_str}` does not belong to session `{session_id}`"
404 )));
405 }
406 let entries = self.entries.lock().unwrap();
407 entries.get(handle_str).cloned().ok_or_else(|| {
408 RuntimeError::ToolFailed(format!("bash: handle `{handle_str}` not found"))
409 })
410 }
411
412 pub fn status(&self, handle_str: &str, session_id: &str) -> Result<Value, RuntimeError> {
413 let entry = self.lookup(handle_str, session_id)?;
414 let st = entry.status.lock().unwrap().clone();
415 let out = entry.output.lock().unwrap();
416 let mut fields = vec![
417 ("handle".into(), Value::Str(handle_str.into())),
418 ("status".into(), Value::Str(st.kind().into())),
419 ("started_at".into(), Value::Int(st.started_at())),
420 (
421 "log_path".into(),
422 Value::Str(entry.log_path.to_string_lossy().into_owned()),
423 ),
424 ];
425 if let Some(ec) = st.exit_code() {
426 fields.push(("exit_code".into(), Value::Int(ec as i64)));
427 }
428 if let Some(ended) = st.ended_at() {
429 fields.push(("ended_at".into(), Value::Int(ended)));
430 }
431 if let Some(error) = st.error() {
432 fields.push(("error".into(), Value::Str(error.into())));
433 }
434 fields.push(("bytes_total".into(), Value::Int(out.total_bytes as i64)));
435 fields.push(("output_truncated".into(), Value::Bool(out.truncated)));
436 Ok(Value::Struct(fields))
437 }
438
439 pub fn output(
440 &self,
441 handle_str: &str,
442 session_id: &str,
443 session_dir: Option<&std::path::Path>,
444 cursor: usize,
445 limit: usize,
446 ) -> Result<Value, RuntimeError> {
447 if let Ok(entry) = self.lookup(handle_str, session_id) {
448 let st = entry.status.lock().unwrap().clone();
449 let out = entry.output.lock().unwrap();
450 let (chunk, actual_cursor, next, eof, fell_behind) = out.read_from(cursor, limit);
451 return Ok(Value::Struct(vec![
452 ("handle".into(), Value::Str(handle_str.into())),
453 ("status".into(), Value::Str(st.kind().into())),
454 (
455 "chunk".into(),
456 Value::Str(String::from_utf8_lossy(&chunk).into_owned()),
457 ),
458 ("cursor".into(), Value::Int(actual_cursor as i64)),
459 ("next_cursor".into(), Value::Int(next as i64)),
460 (
461 "continuation".into(),
462 Value::Struct(vec![
463 ("type".into(), Value::Str("ByteCursor".into())),
464 ("next_byte".into(), Value::Int(next as i64)),
465 ("has_more".into(), Value::Bool(!eof)),
466 ]),
467 ),
468 ("eof".into(), Value::Bool(eof)),
469 (
470 "truncated".into(),
471 Value::Bool(out.truncated || fell_behind),
472 ),
473 ("live".into(), Value::Bool(true)),
474 ]));
475 }
476
477 let Some(dir) = session_dir else {
478 return Err(RuntimeError::ToolFailed(format!(
479 "bash: handle `{handle_str}` not found"
480 )));
481 };
482 let log_path = dir.join(format!("bg_{handle_str}.log"));
483 let data = std::fs::read(&log_path).map_err(|_| {
484 RuntimeError::ToolFailed(format!("bash: handle `{handle_str}` not found"))
485 })?;
486 if cursor >= data.len() {
487 return Ok(Value::Struct(vec![
488 ("handle".into(), Value::Str(handle_str.into())),
489 ("status".into(), Value::Str("exited".into())),
490 ("chunk".into(), Value::Str(String::new())),
491 ("cursor".into(), Value::Int(data.len() as i64)),
492 ("next_cursor".into(), Value::Int(data.len() as i64)),
493 (
494 "continuation".into(),
495 Value::Struct(vec![
496 ("type".into(), Value::Str("ByteCursor".into())),
497 ("next_byte".into(), Value::Int(data.len() as i64)),
498 ("has_more".into(), Value::Bool(false)),
499 ]),
500 ),
501 ("eof".into(), Value::Bool(true)),
502 ("truncated".into(), Value::Bool(false)),
503 ("live".into(), Value::Bool(false)),
504 ]));
505 }
506 let (chunk, next, eof) = page_bytes(&data, cursor, limit);
507 Ok(Value::Struct(vec![
508 ("handle".into(), Value::Str(handle_str.into())),
509 ("status".into(), Value::Str("exited".into())),
510 (
511 "chunk".into(),
512 Value::Str(String::from_utf8_lossy(&chunk).into_owned()),
513 ),
514 ("cursor".into(), Value::Int(cursor as i64)),
515 ("next_cursor".into(), Value::Int(next as i64)),
516 (
517 "continuation".into(),
518 Value::Struct(vec![
519 ("type".into(), Value::Str("ByteCursor".into())),
520 ("next_byte".into(), Value::Int(next as i64)),
521 ("has_more".into(), Value::Bool(!eof)),
522 ]),
523 ),
524 ("eof".into(), Value::Bool(eof)),
525 ("truncated".into(), Value::Bool(false)),
526 ("live".into(), Value::Bool(false)),
527 ]))
528 }
529
530 pub fn output_for_llm(
531 &self,
532 handle_str: &str,
533 session_id: &str,
534 session_dir: Option<&std::path::Path>,
535 cursor: usize,
536 limit: usize,
537 output_store: Option<&crate::tools::tool_output::OutputStore>,
538 ) -> Result<Value, RuntimeError> {
539 let persisted = || {
540 let dir = session_dir.ok_or_else(|| {
541 RuntimeError::ToolFailed(
542 "bash.output: complete persisted output is unavailable; output is truncated and cannot be continued".into(),
543 )
544 })?;
545 let log_path = dir.join(format!("bg_{handle_str}.log"));
546 std::fs::read_to_string(log_path).map_err(|_| {
547 RuntimeError::ToolFailed(
548 "bash.output: complete persisted output is unavailable; output is truncated and cannot be continued".into(),
549 )
550 })
551 };
552 let full = if let Ok(entry) = self.lookup(handle_str, session_id) {
553 let out = entry.output.lock().unwrap();
554 if out.buffer_start == 0 && !out.truncated {
555 String::from_utf8(out.combined.clone()).map_err(|_| {
556 RuntimeError::ToolFailed(
557 "bash.output: current output is not valid UTF-8; output is truncated and cannot be continued".into(),
558 )
559 })?
560 } else {
561 drop(out);
562 persisted()?
563 }
564 } else {
565 persisted()?
566 };
567 if full.len() <= limit {
568 return self.output(handle_str, session_id, session_dir, cursor, limit);
569 }
570 let store = output_store.ok_or_else(|| {
571 RuntimeError::ToolFailed(
572 "bash.output: session output store unavailable; oversized output cannot be continued".into(),
573 )
574 })?;
575 let output_id = store.register(handle_str, &full).ok_or_else(|| {
576 RuntimeError::ToolFailed(
577 "bash.output: complete output could not be persisted; output is truncated and cannot be continued".into(),
578 )
579 })?;
580 let offset = cursor.min(full.len());
581 if !full.is_char_boundary(offset) {
582 return Err(RuntimeError::ToolFailed(
583 "bash.output: cursor is not a valid UTF-8 byte boundary".into(),
584 ));
585 }
586 let (chunk, next, eof) = page_bytes(full.as_bytes(), offset, limit);
587 Ok(Value::Struct(vec![
588 (
589 "content".into(),
590 Value::Str(String::from_utf8(chunk).map_err(|_| {
591 RuntimeError::ToolFailed(
592 "bash.output: persisted output is not valid UTF-8".into(),
593 )
594 })?),
595 ),
596 ("output_id".into(), Value::Str(output_id)),
597 ("total_bytes".into(), Value::Int(full.len() as i64)),
598 (
599 "next".into(),
600 Value::Struct(vec![
601 ("mode".into(), Value::Str("bytes".into())),
602 ("offset".into(), Value::Int(next as i64)),
603 ("has_more".into(), Value::Bool(!eof)),
604 ]),
605 ),
606 ]))
607 }
608
609 pub fn kill(&self, handle_str: &str, session_id: &str) -> Result<Value, RuntimeError> {
610 let entry = self.lookup(handle_str, session_id)?;
611 let _ = entry.control_tx.try_send(BgControl::Kill);
612 let st = entry.status.lock().unwrap().clone();
613 Ok(Value::Struct(vec![
614 ("handle".into(), Value::Str(handle_str.into())),
615 ("status".into(), Value::Str(st.kind().into())),
616 ]))
617 }
618
619 fn remove(&self, handle_str: &str) {
620 self.entries.lock().unwrap().remove(handle_str);
621 }
622
623 #[doc(hidden)]
624 pub fn clear_for_test(&self) {
625 self.entries.lock().unwrap().clear();
626 }
627
628 pub fn list(
629 &self,
630 session_id: &str,
631 session_dir: Option<&std::path::Path>,
632 all: bool,
633 ) -> Value {
634 let entries = self.entries.lock().unwrap();
635 let mut live_handles: std::collections::HashSet<String> = std::collections::HashSet::new();
636 let mut items: Vec<Value> = entries
637 .iter()
638 .filter(|(_, e)| e.session_id == session_id)
639 .map(|(handle, entry)| {
640 live_handles.insert(handle.clone());
641 let st = entry.status.lock().unwrap().clone();
642 let out = entry.output.lock().unwrap();
643 let mut fields = vec![
644 ("handle".into(), Value::Str(handle.clone())),
645 ("status".into(), Value::Str(st.kind().into())),
646 ("started_at".into(), Value::Int(st.started_at())),
647 ("live".into(), Value::Bool(true)),
648 ];
649 if let Some(ec) = st.exit_code() {
650 fields.push(("exit_code".into(), Value::Int(ec as i64)));
651 }
652 fields.push(("bytes_total".into(), Value::Int(out.total_bytes as i64)));
653 Value::Struct(fields)
654 })
655 .collect();
656
657 if all {
658 if let Some(dir) = session_dir {
659 if let Ok(rd) = std::fs::read_dir(dir) {
660 for entry in rd.flatten() {
661 let name = entry.file_name();
662 let name = name.to_string_lossy();
663 let Some(rest) = name
664 .strip_prefix("bg_")
665 .and_then(|s| s.strip_suffix(".log"))
666 else {
667 continue;
668 };
669 let handle: String = rest.to_string();
670 if live_handles.contains(&handle) {
671 continue;
672 }
673 let Ok(meta) = entry.metadata() else {
674 continue;
675 };
676 let modified = meta
677 .modified()
678 .ok()
679 .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
680 .map(|d| d.as_millis() as i64)
681 .unwrap_or(0);
682 items.push(Value::Struct(vec![
683 ("handle".into(), Value::Str(handle)),
684 ("status".into(), Value::Str("exited".into())),
685 ("started_at".into(), Value::Int(modified)),
686 ("live".into(), Value::Bool(false)),
687 ("bytes_total".into(), Value::Int(meta.len() as i64)),
688 ]));
689 }
690 }
691 }
692 }
693
694 Value::List(items)
695 }
696}
697
698impl Drop for BgRegistry {
699 fn drop(&mut self) {
700 self.kill_all();
701 }
702}
703
704#[allow(clippy::too_many_arguments)]
705async fn run_bg_process(
706 handle_str: String,
707 cmd: String,
708 timeout: Option<Duration>,
709 max_output_bytes: u64,
710 log_file: File,
711 status: Arc<Mutex<BgStatus>>,
712 output: Arc<Mutex<BgOutput>>,
713 mut control_rx: mpsc::Receiver<BgControl>,
714 cancel: CancellationToken,
715 registry: Arc<BgRegistry>,
716 stream_tx: Option<tokio::sync::broadcast::Sender<crate::stream::StreamFrame>>,
717 handle_for_stream: String,
718 task_registry: Option<TaskRegistry>,
719 task_id: Option<crate::task_registry::TaskId>,
720 flow_run_id: Option<String>,
721) {
722 let started_at = now_ms();
723 let mut command = tokio::process::Command::new("sh");
724 command
725 .arg("-c")
726 .arg(&cmd)
727 .stdin(Stdio::null())
728 .stdout(Stdio::piped())
729 .stderr(Stdio::piped());
730 let mut child: AsyncGroupChild = match command.group().kill_on_drop(true).spawn() {
731 Ok(c) => c,
732 Err(e) => {
733 *status.lock().unwrap() = BgStatus::Failed {
734 error: format!("spawn: {e}"),
735 started_at,
736 ended_at: now_ms(),
737 };
738 registry.remove(&handle_str);
739 return;
740 }
741 };
742 let pid = child.id();
743 *status.lock().unwrap() = BgStatus::Running {
744 pid: pid.unwrap_or(0),
745 started_at,
746 };
747
748 let stdout = child.inner().stdout.take();
749 let stderr = child.inner().stderr.take();
750 let (log_tx, log_rx) = mpsc::unbounded_channel::<Vec<u8>>();
751 let log_writer = tokio::spawn(write_log(log_file, log_rx));
752
753 let stdout_reader = stdout.map(|s| {
754 let ctx = ReadStreamCtx {
755 output: output.clone(),
756 log_tx: log_tx.clone(),
757 kind: StreamKind::Stdout,
758 max_output_bytes,
759 stream_tx: stream_tx.clone(),
760 handle: handle_for_stream.clone(),
761 flow_run_id: flow_run_id.clone(),
762 };
763 tokio::spawn(read_stream(BufReader::new(s), ctx))
764 });
765 let stderr_reader = stderr.map(|s| {
766 let ctx = ReadStreamCtx {
767 output: output.clone(),
768 log_tx: log_tx.clone(),
769 kind: StreamKind::Stderr,
770 max_output_bytes,
771 stream_tx: stream_tx.clone(),
772 handle: handle_for_stream.clone(),
773 flow_run_id: flow_run_id.clone(),
774 };
775 tokio::spawn(read_stream(BufReader::new(s), ctx))
776 });
777
778 let exit_reason = tokio::select! {
779 biased;
780 _ = cancel.cancelled() => ExitReason::Cancelled,
781 ctrl = control_rx.recv() => {
782 match ctrl {
783 Some(BgControl::Kill) => ExitReason::Kill,
784 None => ExitReason::Natural,
785 }
786 }
787 _ = async {
788 if let Some(t) = timeout {
789 tokio::time::sleep(t).await;
790 } else {
791 std::future::pending::<()>().await;
792 }
793 } => ExitReason::Timeout,
794 s = child.wait() => ExitReason::Exited(s),
795 };
796
797 let ended_at = now_ms();
798 let mut final_status = match &exit_reason {
799 ExitReason::Exited(Ok(s)) => BgStatus::Exited {
800 exit_code: s.code().unwrap_or(-1),
801 started_at,
802 ended_at,
803 },
804 ExitReason::Timeout => {
805 let _ = child.start_kill();
806 let _ = tokio::time::timeout(Duration::from_millis(500), child.wait()).await;
807 BgStatus::TimedOut {
808 started_at,
809 ended_at,
810 }
811 }
812 ExitReason::Kill => {
813 let _ = child.start_kill();
814 let _ = tokio::time::timeout(Duration::from_millis(500), child.wait()).await;
815 BgStatus::Killed {
816 started_at,
817 ended_at,
818 }
819 }
820 ExitReason::Cancelled => {
821 let _ = child.start_kill();
822 let _ = tokio::time::timeout(Duration::from_millis(500), child.wait()).await;
823 BgStatus::Killed {
824 started_at,
825 ended_at,
826 }
827 }
828 ExitReason::Exited(Err(_)) => {
829 let _ = child.start_kill();
830 BgStatus::Failed {
831 error: "wait failed".into(),
832 started_at,
833 ended_at,
834 }
835 }
836 ExitReason::Natural => {
837 let s = child.wait().await;
838 BgStatus::Exited {
839 exit_code: s.ok().and_then(|s| s.code()).unwrap_or(-1),
840 started_at,
841 ended_at: now_ms(),
842 }
843 }
844 };
845
846 if let Some(mut r) = stdout_reader {
847 if tokio::time::timeout(IO_DRAIN_TIMEOUT, &mut r)
848 .await
849 .is_err()
850 {
851 r.abort();
852 let _ = r.await;
853 }
854 }
855 if let Some(mut r) = stderr_reader {
856 if tokio::time::timeout(IO_DRAIN_TIMEOUT, &mut r)
857 .await
858 .is_err()
859 {
860 r.abort();
861 let _ = r.await;
862 }
863 }
864 drop(log_tx);
865 let mut log_writer = log_writer;
866 let log_result = match tokio::time::timeout(IO_DRAIN_TIMEOUT, &mut log_writer).await {
867 Ok(Ok(result)) => result,
868 Ok(Err(join_error)) => Err(format!("log writer task failed: {join_error}")),
869 Err(_) => {
870 log_writer.abort();
871 let _ = log_writer.await;
872 Err("log writer timed out".into())
873 }
874 };
875 if let Err(error) = log_result {
876 final_status = BgStatus::Failed {
877 error,
878 started_at,
879 ended_at: now_ms(),
880 };
881 }
882
883 let exit_code = match &final_status {
884 BgStatus::Exited { exit_code, .. } => Some(*exit_code),
885 _ => None,
886 };
887 *status.lock().unwrap() = final_status.clone();
888
889 if let Some(tx) = &stream_tx {
890 let _ = tx.send(crate::stream::StreamFrame::BashExited {
891 handle: handle_for_stream,
892 exit_code,
893 error: final_status.error().map(str::to_owned),
894 run_id: flow_run_id,
895 });
896 }
897
898 if let (Some(tr), Some(tid)) = (task_registry, task_id) {
899 let ts = match &final_status {
900 BgStatus::Exited { exit_code, .. } if *exit_code == 0 => TaskStatus::Ok,
901 BgStatus::Killed { .. } | BgStatus::TimedOut { .. } => TaskStatus::Killed,
902 _ => TaskStatus::Err,
903 };
904 tr.finish(&tid, ts);
905 }
906}
907
908fn open_log_file(log_path: &std::path::Path) -> Result<File, String> {
909 File::options()
910 .write(true)
911 .create_new(true)
912 .open(log_path)
913 .map_err(|e| format!("open log: {e}"))
914}
915
916async fn write_log(file: File, mut log_rx: mpsc::UnboundedReceiver<Vec<u8>>) -> Result<(), String> {
917 let mut file = tokio::fs::File::from_std(file);
918 while let Some(frame) = log_rx.recv().await {
919 file.write_all(&frame)
920 .await
921 .map_err(|e| format!("write log: {e}"))?;
922 }
923 file.flush().await.map_err(|e| format!("flush log: {e}"))?;
924 Ok(())
925}
926
927struct ReadStreamCtx {
928 output: Arc<Mutex<BgOutput>>,
929 log_tx: mpsc::UnboundedSender<Vec<u8>>,
930 kind: StreamKind,
931 max_output_bytes: u64,
932 stream_tx: Option<tokio::sync::broadcast::Sender<crate::stream::StreamFrame>>,
933 handle: String,
934 flow_run_id: Option<String>,
935}
936
937async fn read_stream<R: tokio::io::AsyncBufRead + Unpin>(mut reader: R, ctx: ReadStreamCtx) {
938 let kind_str = match ctx.kind {
939 StreamKind::Stdout => "stdout",
940 StreamKind::Stderr => "stderr",
941 };
942 let mut buf = String::new();
943 loop {
944 buf.clear();
945 match reader.read_line(&mut buf).await {
946 Ok(0) => break,
947 Ok(_) => {
948 let data = buf.as_bytes();
949 {
950 let mut out = ctx.output.lock().unwrap();
951 let _ = ctx.log_tx.send(framed_output(ctx.kind, data));
952 let _ = out.push(ctx.kind, data, ctx.max_output_bytes);
953 }
954 if let Some(tx) = &ctx.stream_tx {
955 let _ = tx.send(crate::stream::StreamFrame::BashChunk {
956 handle: ctx.handle.clone(),
957 kind: kind_str.to_string(),
958 line: buf.clone(),
959 run_id: ctx.flow_run_id.clone(),
960 });
961 }
962 }
963 Err(_) => break,
964 }
965 }
966}
967
968enum ExitReason {
969 Exited(std::io::Result<std::process::ExitStatus>),
970 Timeout,
971 Kill,
972 Cancelled,
973 Natural,
974}
975
976fn now_ms() -> i64 {
977 chrono::Utc::now().timestamp_millis()
978}
979
980pub struct BashSpawn;
981
982impl Tool for BashSpawn {
983 fn name(&self) -> &str {
984 "bash.spawn"
985 }
986
987 fn tier(&self) -> Tier {
988 Tier::Four
989 }
990
991 fn description(&self) -> Option<&str> {
992 Some(
993 "Run a shell command via `sh -c`.\n\n\
994block=false (default): command runs in background, returns immediately with a\n\
995handle. The command keeps running — use bash.output to read its output later,\n\
996bash.status to check if it finished, bash.kill to stop it. Use this for:\n\
997- long-running commands (servers, watchers)\n\
998- commands where you need to check output incrementally\n\
999- when you want to do other things while the command runs\n\n\
1000block=true: waits for the command to finish, then returns stdout/stderr/exit_code.\n\
1001Use block_timeout_ms to set a max wait (default 30s). Use this for:\n\
1002- short commands where you need the result immediately (ls, git status, echo)\n\
1003- commands that finish quickly\n\n\
1004Do NOT use `sleep` in your command to wait — use block=true with block_timeout_ms\n\
1005instead, or use the sleep tool to pause the workflow.",
1006 )
1007 }
1008
1009 fn input_schema(&self) -> serde_json::Value {
1010 serde_json::json!({
1011 "type": "object",
1012 "properties": {
1013 "cmd": {"type": "string", "description": "Shell command line."},
1014 "block": {"type": "boolean", "default": false, "description": "If true, wait for process to exit before returning."},
1015 "block_timeout_ms": {"type": "integer", "description": "Only with block=true. Max wait. 0 = no timeout. Default 30000."},
1016 "timeout_ms": {"type": "integer", "description": "Process kill timeout in ms. Default 1800000 (30min). 0 = no timeout."},
1017 "max_output_bytes": {"type": "integer", "description": "Max combined output bytes. Default 10485760 (10MB)."}
1018 },
1019 "required": ["cmd"]
1020 })
1021 }
1022
1023 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
1024 Box::pin(async move {
1025 let cmd = extract_string(&args, "cmd", 0)?;
1026 let block = args
1027 .named("block")
1028 .and_then(|v| {
1029 if let Value::Bool(b) = v {
1030 Some(*b)
1031 } else {
1032 None
1033 }
1034 })
1035 .unwrap_or(false);
1036 let block_timeout_ms = extract_optional_int(&args, "block_timeout_ms")
1037 .map(|v| v as u64)
1038 .unwrap_or(30_000);
1039 let timeout_ms = extract_optional_int(&args, "timeout_ms").map(|v| v as u64);
1040 let max_output = extract_optional_int(&args, "max_output_bytes")
1041 .map(|v| v as u64)
1042 .unwrap_or(DEFAULT_MAX_OUTPUT_BYTES);
1043 let registry = ctx.bg_registry.clone().ok_or_else(|| {
1044 RuntimeError::ToolFailed("bash.spawn: registry not available".into())
1045 })?;
1046 let handle_str = registry.spawn(cmd, timeout_ms, max_output, ctx)?;
1047
1048 if !block {
1049 return Ok(handle_str);
1050 }
1051
1052 let handle_s = handle_str
1053 .field("handle")
1054 .and_then(|v| {
1055 if let Value::Str(s) = v {
1056 Some(s.clone())
1057 } else {
1058 None
1059 }
1060 })
1061 .ok_or_else(|| {
1062 RuntimeError::ToolFailed("bash.spawn: missing handle field".into())
1063 })?;
1064 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".into());
1065
1066 let deadline = if block_timeout_ms == 0 {
1067 None
1068 } else {
1069 Some(tokio::time::Instant::now() + Duration::from_millis(block_timeout_ms))
1070 };
1071 loop {
1072 let entry = registry.lookup(&handle_s, &session_id)?;
1073 let finished = {
1074 let st = entry.status.lock().unwrap();
1075 st.is_finished()
1076 };
1077 if finished {
1078 break;
1079 }
1080 if let Some(d) = deadline {
1081 if tokio::time::Instant::now() >= d {
1082 break;
1083 }
1084 }
1085 tokio::time::sleep(Duration::from_millis(50)).await;
1086 }
1087
1088 let entry = registry.lookup(&handle_s, &session_id)?;
1089 let st = entry.status.lock().unwrap().clone();
1090 let out = entry.output.lock().unwrap();
1091 let combined = String::from_utf8_lossy(&out.combined).into_owned();
1092 let log_path = entry.log_path.to_string_lossy().into_owned();
1093 Ok(Value::Struct(vec![
1094 ("handle".into(), Value::Str(handle_s)),
1095 ("status".into(), Value::Str(st.kind().into())),
1096 (
1097 "exit_code".into(),
1098 st.exit_code()
1099 .map(|c| Value::Int(c as i64))
1100 .unwrap_or(Value::Unit),
1101 ),
1102 (
1103 "error".into(),
1104 st.error()
1105 .map(|e| Value::Str(e.into()))
1106 .unwrap_or(Value::Unit),
1107 ),
1108 ("output".into(), Value::Str(combined)),
1109 ("bytes_total".into(), Value::Int(out.total_bytes as i64)),
1110 ("log_path".into(), Value::Str(log_path)),
1111 ]))
1112 })
1113 }
1114}
1115
1116pub struct BashStatus;
1117
1118impl Tool for BashStatus {
1119 fn name(&self) -> &str {
1120 "bash.status"
1121 }
1122
1123 fn tier(&self) -> Tier {
1124 Tier::Four
1125 }
1126
1127 fn description(&self) -> Option<&str> {
1128 Some("Check the status of a background bash process.")
1129 }
1130
1131 fn input_schema(&self) -> serde_json::Value {
1132 serde_json::json!({
1133 "type": "object",
1134 "properties": {"handle": {"type": "string"}},
1135 "required": ["handle"]
1136 })
1137 }
1138
1139 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
1140 Box::pin(async move {
1141 let handle = extract_string(&args, "handle", 0)?;
1142 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".to_string());
1143 let registry = ctx.bg_registry.clone().ok_or_else(|| {
1144 RuntimeError::ToolFailed("bash.status: registry not available".into())
1145 })?;
1146 registry.status(&handle, &session_id)
1147 })
1148 }
1149}
1150
1151pub struct BashOutput;
1152
1153impl Tool for BashOutput {
1154 fn name(&self) -> &str {
1155 "bash.output"
1156 }
1157
1158 fn tier(&self) -> Tier {
1159 Tier::Four
1160 }
1161
1162 fn description(&self) -> Option<&str> {
1163 Some("Read output from a background bash process by byte cursor.")
1164 }
1165
1166 fn input_schema(&self) -> serde_json::Value {
1167 serde_json::json!({
1168 "type": "object",
1169 "properties": {
1170 "handle": {"type": "string"},
1171 "cursor": {"type": "integer", "description": "Byte offset to start reading. Default 0."},
1172 "limit_bytes": {"type": "integer", "description": "Max bytes to return. Default 32000."}
1173 },
1174 "required": ["handle"]
1175 })
1176 }
1177
1178 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
1179 Box::pin(async move {
1180 let handle = extract_string(&args, "handle", 0)?;
1181 let cursor = extract_optional_int(&args, "cursor").unwrap_or(0).max(0) as usize;
1182 let limit = (extract_optional_int(&args, "limit_bytes")
1183 .unwrap_or(DEFAULT_OUTPUT_LIMIT as i64)
1184 .max(1) as usize)
1185 .min(ctx.tool_output_budget.max_bytes);
1186 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".to_string());
1187 let registry = ctx.bg_registry.clone().ok_or_else(|| {
1188 RuntimeError::ToolFailed("bash.output: registry not available".into())
1189 })?;
1190 registry.output_for_llm(
1191 &handle,
1192 &session_id,
1193 ctx.session_dir.as_deref(),
1194 cursor,
1195 limit,
1196 ctx.output_store.as_deref(),
1197 )
1198 })
1199 }
1200}
1201
1202pub struct BashKill;
1203
1204impl Tool for BashKill {
1205 fn name(&self) -> &str {
1206 "bash.kill"
1207 }
1208
1209 fn tier(&self) -> Tier {
1210 Tier::Four
1211 }
1212
1213 fn description(&self) -> Option<&str> {
1214 Some(
1215 "Kill a background bash process. signal=term (default) sends SIGTERM, signal=kill sends SIGKILL.",
1216 )
1217 }
1218
1219 fn input_schema(&self) -> serde_json::Value {
1220 serde_json::json!({
1221 "type": "object",
1222 "properties": {
1223 "handle": {"type": "string"},
1224 "signal": {"type": "string", "enum": ["term", "kill"], "description": "Default term."}
1225 },
1226 "required": ["handle"]
1227 })
1228 }
1229
1230 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
1231 Box::pin(async move {
1232 let handle = extract_string(&args, "handle", 0)?;
1233 let _ = extract_string(&args, "signal", 1);
1234 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".to_string());
1235 let registry = ctx.bg_registry.clone().ok_or_else(|| {
1236 RuntimeError::ToolFailed("bash.kill: registry not available".into())
1237 })?;
1238 registry.kill(&handle, &session_id)
1239 })
1240 }
1241}
1242
1243pub struct BashList;
1244
1245impl Tool for BashList {
1246 fn name(&self) -> &str {
1247 "bash.list"
1248 }
1249
1250 fn tier(&self) -> Tier {
1251 Tier::Four
1252 }
1253
1254 fn description(&self) -> Option<&str> {
1255 Some(
1256 "List background bash processes for the current session. Default: only live processes. Pass all=true to include historical processes whose log files persist in session_dir (status=exited, live=false).",
1257 )
1258 }
1259
1260 fn input_schema(&self) -> serde_json::Value {
1261 serde_json::json!({
1262 "type": "object",
1263 "properties": {
1264 "all": {"type": "boolean", "description": "Include historical processes (default false)."}
1265 }
1266 })
1267 }
1268
1269 fn call<'a>(&'a self, args: ToolArgs, ctx: &'a ToolCtx) -> BoxFut<'a, ToolResult> {
1270 Box::pin(async move {
1271 let all = extract_optional_bool(&args, "all").unwrap_or(false);
1272 let session_id = ctx.session_id.clone().unwrap_or_else(|| "anon".to_string());
1273 let registry = ctx.bg_registry.clone().ok_or_else(|| {
1274 RuntimeError::ToolFailed("bash.list: registry not available".into())
1275 })?;
1276 Ok(registry.list(&session_id, ctx.session_dir.as_deref(), all))
1277 })
1278 }
1279}
1280
1281fn extract_string(args: &ToolArgs, name: &str, pos: usize) -> Result<String, RuntimeError> {
1282 let value = match args.named(name) {
1283 Some(v) => v,
1284 None => args.positional(pos)?,
1285 };
1286 match value {
1287 Value::Str(s) => Ok(s.clone()),
1288 other => Err(RuntimeError::TypeMismatch {
1289 expected: "string".into(),
1290 actual: other.kind_name().into(),
1291 }),
1292 }
1293}
1294
1295fn extract_optional_int(args: &ToolArgs, name: &str) -> Option<i64> {
1296 match args.named(name)? {
1297 Value::Int(n) => Some(*n),
1298 _ => None,
1299 }
1300}
1301
1302fn extract_optional_bool(args: &ToolArgs, name: &str) -> Option<bool> {
1303 match args.named(name)? {
1304 Value::Bool(b) => Some(*b),
1305 _ => None,
1306 }
1307}
1308
1309#[cfg(test)]
1310mod tests {
1311 use super::*;
1312 use crate::tool::{ToolArgs, ToolCtx};
1313 use crate::value::Value;
1314 use std::sync::Arc;
1315 use tempfile::TempDir;
1316
1317 fn ctx_with_registry(registry: Arc<BgRegistry>, dir: &std::path::Path) -> ToolCtx {
1318 let mut ctx = ToolCtx::new();
1319 ctx.bg_registry = Some(registry);
1320 ctx.session_dir = Some(dir.to_path_buf());
1321 ctx.session_id = Some("test-session".to_string());
1322 ctx
1323 }
1324
1325 #[test]
1326 fn pushed_frame_matches_live_output_after_budget_truncation() {
1327 let mut output = BgOutput::default();
1328 let first = output.push(StreamKind::Stdout, b"first\n", 6);
1329 let second = output.push(StreamKind::Stderr, b"second\n", 6);
1330
1331 assert_eq!(first, b"[out] first\n");
1332 assert!(second.is_empty());
1333 assert_eq!(output.combined, first);
1334 assert_eq!(output.total_bytes, 6);
1335 assert!(output.truncated);
1336 }
1337
1338 #[test]
1339 fn ring_buffer_cursors_remain_absolute_after_eviction() {
1340 let first_data = vec![b'a'; 40_000];
1341 let second_data = vec![b'b'; 40_000];
1342 let mut full = Vec::new();
1343 full.extend_from_slice(b"[out] ");
1344 full.extend_from_slice(&first_data);
1345 full.push(b'\n');
1346 full.extend_from_slice(b"[out] ");
1347 full.extend_from_slice(&second_data);
1348 full.push(b'\n');
1349
1350 let mut output = BgOutput::default();
1351 output.push(StreamKind::Stdout, &first_data, u64::MAX);
1352 let (_, _, cursor, _, fell_behind) = output.read_from(0, 32_000);
1353 assert_eq!(cursor, 32_000);
1354 assert!(!fell_behind);
1355
1356 output.push(StreamKind::Stdout, &second_data, u64::MAX);
1357 assert!(output.buffer_start > 0);
1358 let (chunk, actual_cursor, next, _, fell_behind) = output.read_from(cursor, 32_000);
1359 assert_eq!(actual_cursor, cursor);
1360 assert_eq!(chunk, full[cursor..next]);
1361 assert!(!fell_behind);
1362
1363 let (chunk, actual_cursor, next, _, fell_behind) = output.read_from(0, 32);
1364 assert_eq!(actual_cursor, output.buffer_start);
1365 assert_eq!(chunk, full[actual_cursor..next]);
1366 assert!(fell_behind);
1367
1368 let (chunk, actual_cursor, next, eof, fell_behind) = output.read_from(usize::MAX, 32);
1369 assert!(chunk.is_empty());
1370 assert_eq!(actual_cursor, full.len());
1371 assert_eq!(next, full.len());
1372 assert!(eof);
1373 assert!(!fell_behind);
1374 }
1375
1376 #[test]
1377 fn page_bytes_preserves_utf8_across_pages() {
1378 let data = "你好世界".as_bytes();
1379 let (first, next, eof) = page_bytes(data, 0, 4);
1380 assert_eq!(first, "你".as_bytes());
1381 assert_eq!(next, 3);
1382 assert!(!eof);
1383 let (second, next, eof) = page_bytes(data, next, 4);
1384 assert_eq!(second, "好".as_bytes());
1385 assert_eq!(next, 6);
1386 assert!(!eof);
1387 }
1388
1389 #[test]
1390 fn persisted_output_pages_utf8_and_returns_continuation() {
1391 let registry = BgRegistry::new();
1392 let dir = TempDir::new().unwrap();
1393 let handle = "missing";
1394 std::fs::write(dir.path().join("bg_missing.log"), "你好").unwrap();
1395
1396 let first = registry
1397 .output(handle, "test-session", Some(dir.path()), 0, 4)
1398 .unwrap();
1399 let Value::Struct(fields) = first else {
1400 panic!("expected output fields");
1401 };
1402 assert!(matches!(
1403 fields.iter().find(|(name, _)| name == "chunk"),
1404 Some((_, Value::Str(chunk))) if chunk == "你"
1405 ));
1406 assert!(matches!(
1407 fields.iter().find(|(name, _)| name == "next_cursor"),
1408 Some((_, Value::Int(3)))
1409 ));
1410
1411 let second = registry
1412 .output(handle, "test-session", Some(dir.path()), 3, 4)
1413 .unwrap();
1414 let Value::Struct(fields) = second else {
1415 panic!("expected output fields");
1416 };
1417 assert!(matches!(
1418 fields.iter().find(|(name, _)| name == "chunk"),
1419 Some((_, Value::Str(chunk))) if chunk == "好"
1420 ));
1421 let Value::Struct(cursor) = fields
1422 .iter()
1423 .find_map(|(name, value)| (name == "continuation").then_some(value))
1424 .unwrap()
1425 else {
1426 panic!("expected continuation");
1427 };
1428 assert!(matches!(
1429 cursor.iter().find(|(name, _)| name == "next_byte"),
1430 Some((_, Value::Int(6)))
1431 ));
1432 assert!(matches!(
1433 cursor.iter().find(|(name, _)| name == "has_more"),
1434 Some((_, Value::Bool(false)))
1435 ));
1436
1437 let beyond_eof = registry
1438 .output(handle, "test-session", Some(dir.path()), 99, 4)
1439 .unwrap();
1440 let Value::Struct(fields) = beyond_eof else {
1441 panic!("expected output fields");
1442 };
1443 assert!(matches!(
1444 fields.iter().find(|(name, _)| name == "cursor"),
1445 Some((_, Value::Int(6)))
1446 ));
1447 assert!(matches!(
1448 fields.iter().find(|(name, _)| name == "next_cursor"),
1449 Some((_, Value::Int(6)))
1450 ));
1451 let Value::Struct(cursor) = fields
1452 .iter()
1453 .find_map(|(name, value)| (name == "continuation").then_some(value))
1454 .unwrap()
1455 else {
1456 panic!("expected continuation");
1457 };
1458 assert!(matches!(
1459 cursor.iter().find(|(name, _)| name == "next_byte"),
1460 Some((_, Value::Int(6)))
1461 ));
1462 }
1463
1464 #[test]
1465 fn oversized_output_registers_and_reassembles_through_output_store() {
1466 let registry = BgRegistry::new();
1467 let dir = TempDir::new().unwrap();
1468 let full = format!("{}{}", "前缀🚀".repeat(104_857), "前缀");
1469 assert_eq!(full.len(), 1_048_576);
1470 std::fs::write(dir.path().join("bg_missing.log"), full.as_bytes()).unwrap();
1471 let store = crate::tools::tool_output::OutputStore::at(dir.path());
1472
1473 let value = registry
1474 .output_for_llm(
1475 "missing",
1476 "test-session",
1477 Some(dir.path()),
1478 0,
1479 1024,
1480 Some(&store),
1481 )
1482 .unwrap();
1483 let Value::Struct(fields) = value else {
1484 panic!("expected output fields");
1485 };
1486 let output_id = fields
1487 .iter()
1488 .find_map(|(name, value)| (name == "output_id").then_some(value))
1489 .and_then(|value| match value {
1490 Value::Str(id) => Some(id.clone()),
1491 _ => None,
1492 })
1493 .unwrap();
1494 let initial_content = fields
1495 .iter()
1496 .find_map(|(name, value)| (name == "content").then_some(value))
1497 .and_then(|value| match value {
1498 Value::Str(content) => Some(content.clone()),
1499 _ => None,
1500 })
1501 .unwrap();
1502 let total_bytes = fields
1503 .iter()
1504 .find_map(|(name, value)| (name == "total_bytes").then_some(value))
1505 .and_then(|value| match value {
1506 Value::Int(total_bytes) => Some(*total_bytes as usize),
1507 _ => None,
1508 })
1509 .unwrap();
1510 let Value::Struct(next_fields) = fields
1511 .iter()
1512 .find_map(|(name, value)| (name == "next").then_some(value))
1513 .unwrap()
1514 else {
1515 panic!("expected next fields");
1516 };
1517 let next_offset = next_fields
1518 .iter()
1519 .find_map(|(name, value)| (name == "offset").then_some(value))
1520 .and_then(|value| match value {
1521 Value::Int(offset) => Some(*offset as usize),
1522 _ => None,
1523 })
1524 .unwrap();
1525 let has_more = next_fields
1526 .iter()
1527 .find_map(|(name, value)| (name == "has_more").then_some(value))
1528 .and_then(|value| match value {
1529 Value::Bool(has_more) => Some(*has_more),
1530 _ => None,
1531 })
1532 .unwrap();
1533 assert!(output_id.starts_with("out_"));
1534 assert_eq!(total_bytes, 1_048_576);
1535 assert_eq!(next_offset, initial_content.len());
1536 assert!(next_offset <= 1_024);
1537 assert!(full.is_char_boundary(next_offset));
1538 assert!(has_more);
1539 assert!(!fields.iter().any(|(name, _)| name == "continuation"));
1540
1541 let budget = crate::tools::tool_output::ToolOutputBudget {
1542 max_lines: usize::MAX,
1543 max_bytes: 1024,
1544 max_line_bytes: usize::MAX,
1545 };
1546 let mut offset = next_offset;
1547 let mut assembled = initial_content;
1548 assert!(has_more);
1549 loop {
1550 let page = store.read_bytes(&output_id, offset, 1024, budget).unwrap();
1551 assembled.push_str(&page.content);
1552 if !page.has_more {
1553 break;
1554 }
1555 offset = page.next_offset;
1556 }
1557 assert_eq!(assembled.len(), total_bytes);
1558 assert_eq!(assembled, full);
1559 }
1560
1561 #[test]
1562 fn page_bytes_always_advances_for_incomplete_utf8() {
1563 let data = [0xf0, 0x9f, 0x9a, 0x80];
1564 let (chunk, next, eof) = page_bytes(&data, 0, 1);
1565 assert_eq!(chunk, vec![0xf0]);
1566 assert_eq!(next, 1);
1567 assert!(!eof);
1568 }
1569
1570 #[test]
1571 fn failed_status_preserves_error_reason() {
1572 let status = BgStatus::Failed {
1573 error: "open log: permission denied".into(),
1574 started_at: 1,
1575 ended_at: 2,
1576 };
1577 assert_eq!(status.kind(), "failed");
1578 assert_eq!(status.error(), Some("open log: permission denied"));
1579 assert_eq!(status.exit_code(), None);
1580 assert!(status.is_finished());
1581 }
1582
1583 #[test]
1584 fn handle_parse_roundtrip() {
1585 let h = BgHandle {
1586 session_id: "abc".into(),
1587 local_id: 42,
1588 };
1589 let s = h.to_string();
1590 assert_eq!(s, "bg_abc_42");
1591 let back = BgHandle::parse(&s).unwrap();
1592 assert_eq!(back, h);
1593 }
1594
1595 #[test]
1596 fn handle_parse_rejects_bad_format() {
1597 assert!(BgHandle::parse("not_bg").is_none());
1598 assert!(BgHandle::parse("bg_nosuffix").is_none());
1599 assert!(BgHandle::parse("bg_x_notnum").is_none());
1600 }
1601
1602 #[test]
1603 fn log_file_reports_open_failure_before_spawn() {
1604 let dir = TempDir::new().unwrap();
1605 let log_path = dir.path().join("log");
1606 std::fs::create_dir(&log_path).unwrap();
1607
1608 let error = open_log_file(&log_path).unwrap_err();
1609 assert!(error.contains("open log"));
1610 }
1611
1612 #[tokio::test]
1613 async fn spawn_returns_immediately_with_running_status() {
1614 let registry = Arc::new(BgRegistry::new());
1615 let dir = TempDir::new().unwrap();
1616 let ctx = ctx_with_registry(registry.clone(), dir.path());
1617 let args = ToolArgs {
1618 positional: vec![Value::Str("echo hello".into())],
1619 named: vec![],
1620 };
1621 let v = BashSpawn.call(args, &ctx).await.unwrap();
1622 let Value::Struct(fields) = v else {
1623 panic!("expected struct")
1624 };
1625 let handle = fields
1626 .iter()
1627 .find(|(k, _)| k == "handle")
1628 .and_then(|(_, v)| {
1629 if let Value::Str(s) = v {
1630 Some(s.clone())
1631 } else {
1632 None
1633 }
1634 })
1635 .unwrap();
1636 assert!(handle.starts_with("bg_"));
1637 let status_val = fields.iter().find(|(k, _)| k == "status").unwrap();
1638 assert!(matches!(&status_val.1, Value::Str(s) if s == "running"));
1639 }
1640
1641 #[tokio::test]
1642 async fn spawn_then_status_reaches_exited() {
1643 let registry = Arc::new(BgRegistry::new());
1644 let dir = TempDir::new().unwrap();
1645 let ctx = ctx_with_registry(registry.clone(), dir.path());
1646 let spawn_args = ToolArgs {
1647 positional: vec![Value::Str("echo hello".into())],
1648 named: vec![],
1649 };
1650 let v = BashSpawn.call(spawn_args, &ctx).await.unwrap();
1651 let Value::Struct(fields) = v else { panic!() };
1652 let handle = fields
1653 .iter()
1654 .find(|(k, _)| k == "handle")
1655 .and_then(|(_, v)| {
1656 if let Value::Str(s) = v {
1657 Some(s.clone())
1658 } else {
1659 None
1660 }
1661 })
1662 .unwrap();
1663
1664 for _ in 0..50 {
1665 tokio::time::sleep(Duration::from_millis(50)).await;
1666 let status_args = ToolArgs {
1667 positional: vec![Value::Str(handle.clone())],
1668 named: vec![],
1669 };
1670 let s = BashStatus.call(status_args, &ctx).await.unwrap();
1671 if let Value::Struct(sf) = s {
1672 let kind = sf.iter().find(|(k, _)| k == "status").unwrap();
1673 if matches!(&kind.1, Value::Str(s) if s == "exited") {
1674 let ec = sf.iter().find(|(k, _)| k == "exit_code").unwrap();
1675 assert!(matches!(ec.1, Value::Int(0)));
1676 return;
1677 }
1678 }
1679 }
1680 panic!("process did not exit in time");
1681 }
1682
1683 #[tokio::test]
1684 async fn spawn_output_captures_stdout() {
1685 let registry = Arc::new(BgRegistry::new());
1686 let dir = TempDir::new().unwrap();
1687 let ctx = ctx_with_registry(registry.clone(), dir.path());
1688 let spawn_args = ToolArgs {
1689 positional: vec![Value::Str("echo line1; echo line2".into())],
1690 named: vec![],
1691 };
1692 let v = BashSpawn.call(spawn_args, &ctx).await.unwrap();
1693 let Value::Struct(fields) = v else { panic!() };
1694 let handle = fields
1695 .iter()
1696 .find(|(k, _)| k == "handle")
1697 .and_then(|(_, v)| {
1698 if let Value::Str(s) = v {
1699 Some(s.clone())
1700 } else {
1701 None
1702 }
1703 })
1704 .unwrap();
1705 let log_path = fields
1706 .iter()
1707 .find(|(k, _)| k == "log_path")
1708 .and_then(|(_, v)| {
1709 if let Value::Str(s) = v {
1710 Some(s.clone())
1711 } else {
1712 None
1713 }
1714 })
1715 .unwrap();
1716
1717 tokio::time::sleep(Duration::from_millis(300)).await;
1718
1719 let out_args = ToolArgs {
1720 positional: vec![Value::Str(handle.clone())],
1721 named: vec![],
1722 };
1723 let o = BashOutput.call(out_args, &ctx).await.unwrap();
1724 let Value::Struct(of) = o else { panic!() };
1725 let chunk = of.iter().find(|(k, _)| k == "chunk").unwrap();
1726 if let Value::Str(s) = &chunk.1 {
1727 assert!(s.contains("line1"), "chunk should contain line1: {s}");
1728 assert!(s.contains("line2"), "chunk should contain line2: {s}");
1729 assert_eq!(std::fs::read_to_string(log_path).unwrap(), *s);
1730 } else {
1731 panic!("chunk not str");
1732 }
1733 }
1734
1735 #[tokio::test]
1736 async fn kill_terminates_long_running_process() {
1737 let registry = Arc::new(BgRegistry::new());
1738 let dir = TempDir::new().unwrap();
1739 let ctx = ctx_with_registry(registry.clone(), dir.path());
1740 let spawn_args = ToolArgs {
1741 positional: vec![Value::Str("sleep 100".into())],
1742 named: vec![],
1743 };
1744 let v = BashSpawn.call(spawn_args, &ctx).await.unwrap();
1745 let Value::Struct(fields) = v else { panic!() };
1746 let handle = fields
1747 .iter()
1748 .find(|(k, _)| k == "handle")
1749 .and_then(|(_, v)| {
1750 if let Value::Str(s) = v {
1751 Some(s.clone())
1752 } else {
1753 None
1754 }
1755 })
1756 .unwrap();
1757
1758 let kill_args = ToolArgs {
1759 positional: vec![Value::Str(handle.clone())],
1760 named: vec![],
1761 };
1762 BashKill.call(kill_args, &ctx).await.unwrap();
1763
1764 for _ in 0..50 {
1765 tokio::time::sleep(Duration::from_millis(50)).await;
1766 let status_args = ToolArgs {
1767 positional: vec![Value::Str(handle.clone())],
1768 named: vec![],
1769 };
1770 let s = BashStatus.call(status_args, &ctx).await.unwrap();
1771 if let Value::Struct(sf) = s {
1772 let kind = sf.iter().find(|(k, _)| k == "status").unwrap();
1773 if matches!(&kind.1, Value::Str(s) if s == "killed") {
1774 return;
1775 }
1776 }
1777 }
1778 panic!("process not killed in time");
1779 }
1780
1781 #[tokio::test]
1782 async fn cross_session_access_rejected() {
1783 let registry = Arc::new(BgRegistry::new());
1784 let dir = TempDir::new().unwrap();
1785 let ctx_a = ctx_with_registry(registry.clone(), dir.path());
1786
1787 let spawn_args = ToolArgs {
1788 positional: vec![Value::Str("sleep 10".into())],
1789 named: vec![],
1790 };
1791 let v = BashSpawn.call(spawn_args, &ctx_a).await.unwrap();
1792 let Value::Struct(fields) = v else { panic!() };
1793 let handle = fields
1794 .iter()
1795 .find(|(k, _)| k == "handle")
1796 .and_then(|(_, v)| {
1797 if let Value::Str(s) = v {
1798 Some(s.clone())
1799 } else {
1800 None
1801 }
1802 })
1803 .unwrap();
1804
1805 let mut ctx_b = ToolCtx::new();
1806 ctx_b.bg_registry = Some(registry.clone());
1807 ctx_b.session_dir = Some(dir.path().to_path_buf());
1808 ctx_b.session_id = Some("other-session".to_string());
1809 let status_args = ToolArgs {
1810 positional: vec![Value::Str(handle)],
1811 named: vec![],
1812 };
1813 let err = BashStatus.call(status_args, &ctx_b).await.err().unwrap();
1814 assert!(format!("{err}").contains("does not belong to session"));
1815 }
1816
1817 #[tokio::test]
1818 async fn read_stream_keeps_complete_log_after_memory_budget_is_exhausted() {
1819 let output = Arc::new(Mutex::new(BgOutput::default()));
1820 let (log_tx, mut log_rx) = mpsc::unbounded_channel();
1821 let (mut writer, reader) = tokio::io::duplex(1024);
1822 let input = b"first line\nsecond line\n";
1823 let write_task = tokio::spawn(async move {
1824 use tokio::io::AsyncWriteExt;
1825 writer.write_all(input).await.unwrap();
1826 });
1827
1828 read_stream(
1829 BufReader::new(reader),
1830 ReadStreamCtx {
1831 output: Arc::clone(&output),
1832 log_tx,
1833 kind: StreamKind::Stdout,
1834 max_output_bytes: 5,
1835 stream_tx: None,
1836 handle: "test".into(),
1837 flow_run_id: None,
1838 },
1839 )
1840 .await;
1841 write_task.await.unwrap();
1842
1843 let frames: Vec<Vec<u8>> = std::iter::from_fn(|| log_rx.try_recv().ok()).collect();
1844 assert_eq!(frames.concat(), b"[out] first line\n[out] second line\n");
1845 assert!(output.lock().unwrap().truncated);
1846 }
1847}