From 92594488daae85c6c015d6b626d90c17638683f7 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 20 Aug 2026 10:32:58 +0900 Subject: [PATCH 1/3] feat: stream workdir command output to consoles --- crates/protocol/src/lib.rs | 122 ++++- crates/protocol/src/typescript.rs | 10 +- crates/tools/src/bash.rs | 3 +- crates/tui/src/app.rs | 3 + crates/workdir/src/lib.rs | 14 + crates/workdir/src/local.rs | 515 ++++++++++++++++-- crates/workdir/src/operation.rs | 54 +- crates/worker-runtime/src/runtime.rs | 10 +- crates/worker/src/controller.rs | 112 +++- crates/worker/src/in_flight.rs | 144 ++++- crates/worker/src/internal_worker.rs | 5 +- crates/worker/tests/controller_test.rs | 108 +++- web/workspace/src/lib/generated/protocol.ts | 14 +- .../src/lib/workspace/console/model.test.ts | 117 ++++ .../src/lib/workspace/console/model.ts | 177 +++++- 15 files changed, 1332 insertions(+), 76 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index d44427cf..5d7932c9 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -547,6 +547,12 @@ pub enum Event { Status { status: WorkerStatus, }, + /// Bounded, provider-owned command telemetry for the live Console. This is + /// intentionally not a history entry and is reconstructed from + /// `Snapshot.in_flight.commands` after reconnect. + Command { + event: CommandEvent, + }, /// Reply to `Method::ListCompletions`. Delivered only to the /// requesting socket (not broadcast). `entries` is empty when no /// candidates match or when the requested kind has no resolver @@ -714,8 +720,71 @@ pub struct RewindSummary { pub tool_side_effect_warning: bool, } -/// Unfinished model output included in `Event::Snapshot` for clients that -/// attach while an LLM response is still streaming. +/// Live provider-owned command status. These values are operational Console +/// state only and are never appended to Worker history. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum CommandStatus { + Running, + Completed, + Failed, + TimedOut, + Cancelled, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(rename_all = "snake_case")] +pub enum CommandStream { + Stdout, + Stderr, +} + +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct CommandStreamSlice { + pub start_offset: u64, + pub end_offset: u64, + pub content: String, + pub truncated: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct CommandSnapshot { + pub command_id: String, + pub tool_call_id: Option, + pub status: CommandStatus, + pub stdout: CommandStreamSlice, + pub stderr: CommandStreamSlice, + pub exit_code: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum CommandEvent { + Started { + command_id: String, + tool_call_id: Option, + }, + Output { + command_id: String, + stream: CommandStream, + start_offset: u64, + end_offset: u64, + content: String, + }, + Terminal { + command_id: String, + status: CommandStatus, + exit_code: Option, + }, +} + +/// Unfinished model output and active command state included in +/// `Event::Snapshot` for clients that attach while work is still streaming. /// /// These blocks are presentation state only: they are reconstructed from the /// active Worker controller and must not be treated as committed assistant @@ -726,11 +795,13 @@ pub struct RewindSummary { pub struct InFlightSnapshot { #[serde(default, skip_serializing_if = "Vec::is_empty")] pub blocks: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub commands: Vec, } impl InFlightSnapshot { pub fn is_empty(&self) -> bool { - self.blocks.is_empty() + self.blocks.is_empty() && self.commands.is_empty() } } @@ -1375,6 +1446,19 @@ mod tests { state: InFlightToolCallState::StreamingArgs, }, ], + commands: vec![CommandSnapshot { + command_id: "command-1".into(), + tool_call_id: Some("call_1".into()), + status: CommandStatus::Running, + stdout: CommandStreamSlice { + start_offset: 4, + end_offset: 8, + content: "tail".into(), + truncated: true, + }, + stderr: CommandStreamSlice::default(), + exit_code: None, + }], }, internal_workers: Vec::new(), }; @@ -1444,6 +1528,38 @@ mod tests { )); } + #[test] + fn event_command_output_roundtrip_preserves_stream_and_offsets() { + let event = Event::Command { + event: CommandEvent::Output { + command_id: "command-1".into(), + stream: CommandStream::Stderr, + start_offset: 8, + end_offset: 12, + content: "warn".into(), + }, + }; + let json = serde_json::to_string(&event).unwrap(); + let parsed: serde_json::Value = serde_json::from_str(&json).unwrap(); + assert_eq!(parsed["event"], "command"); + assert_eq!(parsed["data"]["event"]["kind"], "output"); + assert_eq!(parsed["data"]["event"]["stream"], "stderr"); + assert_eq!(parsed["data"]["event"]["start_offset"], 8); + assert_eq!(parsed["data"]["event"]["end_offset"], 12); + assert!(matches!( + serde_json::from_str::(&json).unwrap(), + Event::Command { + event: CommandEvent::Output { + command_id, + stream: CommandStream::Stderr, + start_offset: 8, + end_offset: 12, + content, + } + } if command_id == "command-1" && content == "warn" + )); + } + #[test] fn event_snapshot_legacy_without_status_defaults_to_idle() { let json = r#"{"event":"snapshot","data":{"entries":[],"greeting":{"worker_name":"test","cwd":"/tmp","provider":"anthropic","model":"claude","scope_summary":"","tools":[]}}}"#; diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index 54d9cecf..53245d7b 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -3,8 +3,9 @@ use std::path::PathBuf; use ts_rs::{Config, TS}; use crate::{ - Alert, AlertLevel, AlertSource, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting, - InFlightBlock, InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, + Alert, AlertLevel, AlertSource, CommandEvent, CommandSnapshot, CommandStatus, CommandStream, + CommandStreamSlice, CompletionEntry, CompletionKind, ErrorCode, Event, Greeting, InFlightBlock, + InFlightSnapshot, InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot, InvokeKind, MemoryWorkerEvent, Method, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, TurnResult, WorkerEvent, WorkerStatus, @@ -47,6 +48,11 @@ pub fn generated_protocol_types() -> String { push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index 0b7d57bb..50fe8625 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -43,7 +43,7 @@ impl Tool for BashTool { async fn execute( &self, input_json: &str, - _ctx: llm_engine::tool::ToolExecutionContext, + ctx: llm_engine::tool::ToolExecutionContext, ) -> Result { let params: BashParams = serde_json::from_str(input_json) .map_err(|error| ToolError::InvalidArgument(format!("invalid Bash input: {error}")))?; @@ -58,6 +58,7 @@ impl Tool for BashTool { command: params.command, timeout_secs, output_limit: INLINE_BYTE_BUDGET, + tool_call_id: Some(ctx.call_id), }) .await .map_err(crate::ToolsError::from)?; diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index 473b7a91..32c532d7 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -1322,6 +1322,9 @@ impl App { self.rewind_refresh_fence = false; self.set_worker_status(status); } + // Command telemetry is an operational Web Console surface. The + // TUI continues to render the final Bash ToolResult from history. + Event::Command { .. } => {} Event::Completions { kind, entries } => { // Apply only if the popup is still on the same // (kind, prefix) the request was issued for; an diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index fdd05175..0c226c57 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -16,6 +16,7 @@ use std::sync::Arc; use async_trait::async_trait; use serde::{Deserialize, Serialize}; +use tokio::sync::broadcast; pub use delegation::{ AppliedWorkdirDelegation, ReadOnlyWorkdirSession, WorkdirDelegation, @@ -192,6 +193,19 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync { request: CommandOutputRequest, ) -> Result; async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError>; + + /// Subscribe to bounded provider-owned command telemetry. Implementations + /// that do not expose live command observation may keep the default. + fn subscribe_command_events(&self) -> Option> { + None + } + + /// Return the bounded current command state used to recover from a lagged + /// provider subscription without replaying command output into history. + fn command_snapshot(&self) -> Vec { + Vec::new() + } + /// Terminal, idempotent release of this Worker-bound operation session. async fn close(&self) -> Result<(), WorkdirError>; } diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index 51d3cc7a..c4eb28b4 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -14,21 +14,22 @@ use std::io::Write as _; use std::io::{Read as _, Seek as _, SeekFrom}; use std::path::{Path, PathBuf}; use std::process::Stdio; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::{Arc, Mutex as StdMutex}; use std::time::Duration; use async_trait::async_trait; use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope}; use sha2::{Digest, Sha256}; use tokio::process::Command; -use tokio::sync::{Mutex, Notify}; +use tokio::sync::{Mutex, Notify, broadcast, watch}; use tokio::task::JoinHandle; use crate::{ - CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest, - EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, - ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission, + CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, + CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, EditRequest, EditResult, + GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadRequest, + ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirError, WorkdirPath, WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle, WriteRequest, WriteResult, @@ -36,15 +37,146 @@ use crate::{ #[cfg(test)] use crate::{EntryKind, WriteOutcome}; +const COMMAND_EVENT_CHANNEL_CAPACITY: usize = 256; +const COMMAND_EVENT_CHUNK_BYTES: usize = 8 * 1024; +const COMMAND_SNAPSHOT_STREAM_BYTES: usize = 32 * 1024; + #[derive(Debug)] enum LocalCommand { Running { task: JoinHandle>, completion: Arc, + cancel: watch::Sender, }, Completed(CommandOutput), } +#[derive(Debug, Clone)] +struct CommandTelemetry { + inner: Arc, +} + +#[derive(Debug)] +struct CommandTelemetryInner { + snapshots: StdMutex>, + events: broadcast::Sender, +} + +impl CommandTelemetry { + fn new() -> Self { + let (events, _) = broadcast::channel(COMMAND_EVENT_CHANNEL_CAPACITY); + Self { + inner: Arc::new(CommandTelemetryInner { + snapshots: StdMutex::new(HashMap::new()), + events, + }), + } + } + + fn subscribe(&self) -> broadcast::Receiver { + self.inner.events.subscribe() + } + + fn snapshot(&self) -> Vec { + let mut snapshots = self + .inner + .snapshots + .lock() + .expect("command telemetry mutex poisoned") + .values() + .cloned() + .collect::>(); + snapshots.sort_by(|left, right| left.command_id.cmp(&right.command_id)); + snapshots + } + + fn started(&self, command_id: &str, tool_call_id: Option) { + self.inner + .snapshots + .lock() + .expect("command telemetry mutex poisoned") + .insert( + command_id.to_string(), + CommandSnapshot { + command_id: command_id.to_string(), + tool_call_id: tool_call_id.clone(), + status: CommandStatus::Running, + stdout: CommandStreamSlice::default(), + stderr: CommandStreamSlice::default(), + exit_code: None, + }, + ); + let _ = self.inner.events.send(CommandEvent::Started { + command_id: command_id.to_string(), + tool_call_id, + }); + } + + fn output(&self, command_id: &str, stream: CommandStream, start_offset: u64, bytes: &[u8]) { + if bytes.is_empty() { + return; + } + let end_offset = start_offset.saturating_add(bytes.len() as u64); + let content = String::from_utf8_lossy(bytes).into_owned(); + if let Some(snapshot) = self + .inner + .snapshots + .lock() + .expect("command telemetry mutex poisoned") + .get_mut(command_id) + { + let target = match stream { + CommandStream::Stdout => &mut snapshot.stdout, + CommandStream::Stderr => &mut snapshot.stderr, + }; + target.end_offset = end_offset; + target.content.push_str(&content); + if target.content.len() > COMMAND_SNAPSHOT_STREAM_BYTES { + let mut cut = target.content.len() - COMMAND_SNAPSHOT_STREAM_BYTES; + while cut < target.content.len() && !target.content.is_char_boundary(cut) { + cut += 1; + } + target.content.drain(..cut); + target.start_offset = end_offset.saturating_sub(target.content.len() as u64); + target.truncated = true; + } + } + let _ = self.inner.events.send(CommandEvent::Output { + command_id: command_id.to_string(), + stream, + start_offset, + end_offset, + content, + }); + } + + fn terminal(&self, command_id: &str, status: CommandStatus, exit_code: Option) { + if let Some(snapshot) = self + .inner + .snapshots + .lock() + .expect("command telemetry mutex poisoned") + .get_mut(command_id) + { + snapshot.status = status; + snapshot.exit_code = exit_code; + } + let _ = self.inner.events.send(CommandEvent::Terminal { + command_id: command_id.to_string(), + status, + exit_code, + }); + } + + fn remove(&self, command_id: &str) { + self.inner + .snapshots + .lock() + .expect("command telemetry mutex poisoned") + .remove(command_id); + } +} + #[derive(Debug)] struct ScopeAccess(Arc); @@ -69,6 +201,7 @@ struct LocalWorkdirSessionInner { close_lock: Mutex<()>, next_command_id: AtomicU64, commands: Mutex>, + command_telemetry: CommandTelemetry, } impl Drop for LocalWorkdirSessionInner { @@ -171,6 +304,7 @@ impl LocalWorkdirSession { close_lock: Mutex::new(()), next_command_id: AtomicU64::new(1), commands: Mutex::new(HashMap::new()), + command_telemetry: CommandTelemetry::new(), }), } } @@ -502,23 +636,35 @@ impl WorkdirSession for LocalWorkdirSession { async fn start_command(&self, request: CommandRequest) -> Result { self.ensure_capability(WorkdirSessionCapability::Command)?; + self.ensure_open()?; let id = self.inner.next_command_id.fetch_add(1, Ordering::Relaxed); let handle = CommandHandle(format!("command-{id}")); let cwd = self.inner.cwd.clone(); let completion = Arc::new(Notify::new()); let task_completion = Arc::clone(&completion); + let command_id = handle.0.clone(); + let telemetry = self.inner.command_telemetry.clone(); + let (cancel, cancel_rx) = watch::channel(false); let task = tokio::spawn(async move { - let output = run_command(cwd, request).await; + let output = run_command(cwd, request, command_id, telemetry, cancel_rx).await; task_completion.notify_one(); output }); let mut commands = self.inner.commands.lock().await; if let Err(error) = self.ensure_open() { + let _ = cancel.send(true); task.abort(); completion.notify_one(); return Err(error); } - commands.insert(handle.0.clone(), LocalCommand::Running { task, completion }); + commands.insert( + handle.0.clone(), + LocalCommand::Running { + task, + completion, + cancel, + }, + ); Ok(handle) } @@ -530,7 +676,13 @@ impl WorkdirSession for LocalWorkdirSession { .ok_or_else(|| WorkdirError::UnknownCommand(handle.0.clone()))?; Ok(match command { LocalCommand::Running { task, .. } if !task.is_finished() => CommandStatus::Running, - LocalCommand::Running { .. } => CommandStatus::Completed, + LocalCommand::Running { .. } => self + .inner + .command_telemetry + .snapshot() + .into_iter() + .find(|snapshot| snapshot.command_id == handle.0) + .map_or(CommandStatus::Completed, |snapshot| snapshot.status), LocalCommand::Completed(output) => output.status, }) } @@ -584,36 +736,75 @@ impl WorkdirSession for LocalWorkdirSession { if !self.inner.closed.load(Ordering::Acquire) { commands.insert(request.handle.0, LocalCommand::Completed(output)); } + } else { + self.inner.command_telemetry.remove(&request.handle.0); } Ok(page) } async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> { self.ensure_capability(WorkdirSessionCapability::Command)?; - let command = self - .inner - .commands - .lock() - .await - .remove(&handle.0) - .ok_or_else(|| WorkdirError::UnknownCommand(handle.0))?; - if let LocalCommand::Running { task, completion } = command { - task.abort(); - completion.notify_one(); + let cancel = { + let commands = self.inner.commands.lock().await; + let command = commands + .get(&handle.0) + .ok_or_else(|| WorkdirError::UnknownCommand(handle.0.clone()))?; + match command { + LocalCommand::Running { task, cancel, .. } if !task.is_finished() => { + Some(cancel.clone()) + } + _ => None, + } + }; + if let Some(cancel) = cancel { + let _ = cancel.send(true); } Ok(()) } + fn subscribe_command_events(&self) -> Option> { + self.inner + .capabilities + .supports(WorkdirSessionCapability::Command) + .then(|| self.inner.command_telemetry.subscribe()) + } + + fn command_snapshot(&self) -> Vec { + if self + .inner + .capabilities + .supports(WorkdirSessionCapability::Command) + { + self.inner.command_telemetry.snapshot() + } else { + Vec::new() + } + } + async fn close(&self) -> Result<(), WorkdirError> { let _close_guard = self.inner.close_lock.lock().await; if self.inner.closed.swap(true, Ordering::AcqRel) { return Ok(()); } - let mut commands = self.inner.commands.lock().await; - for (_, command) in commands.drain() { - if let LocalCommand::Running { task, completion } = command { - task.abort(); - completion.notify_one(); + let commands = { + let mut commands = self.inner.commands.lock().await; + commands + .drain() + .map(|(_, command)| command) + .collect::>() + }; + for command in commands { + match command { + LocalCommand::Running { + task, + completion, + cancel, + } => { + let _ = cancel.send(true); + let _ = task.await; + completion.notify_one(); + } + LocalCommand::Completed(_) => {} } } Ok(()) @@ -680,7 +871,13 @@ fn sanitize_error(error: WorkdirError, logical: &WorkdirPath) -> WorkdirError { } } -async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result { +async fn run_command( + cwd: PathBuf, + request: CommandRequest, + command_id: String, + telemetry: CommandTelemetry, + mut cancel: watch::Receiver, +) -> Result { let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?; let stderr = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?; let stdout_path = stdout.into_temp_path(); @@ -690,7 +887,8 @@ async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result Result { - let status = result.map_err(|error| WorkdirError::io(&cwd, error))?; - let (content, truncated) = - read_command_output_files(&stdout_path, &stderr_path, request.output_limit.max(1))?; - return Ok(CommandOutput { - status: CommandStatus::Completed, - exit_code: status.code(), - timed_out: false, - content, - next_cursor: None, - truncated, - }); - } - Err(_) => { - let _ = child.kill().await; - true + Ok(child) => child, + Err(error) => { + telemetry.terminal(&command_id, CommandStatus::Failed, None); + return Err(WorkdirError::io(&cwd, error)); } }; + let mut stdout_reader = + std::fs::File::open(&stdout_path).map_err(|error| WorkdirError::io(&stdout_path, error))?; + let mut stderr_reader = + std::fs::File::open(&stderr_path).map_err(|error| WorkdirError::io(&stderr_path, error))?; + let mut stdout_offset = 0; + let mut stderr_offset = 0; + let mut timeout = Box::pin(tokio::time::sleep(Duration::from_secs( + request.timeout_secs.max(1), + ))); + let mut interval = tokio::time::interval(Duration::from_millis(50)); + interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + interval.tick().await; + + let (status, exit_code) = loop { + tokio::select! { + exit = child.wait() => { + let exit = exit.map_err(|error| WorkdirError::io(&cwd, error))?; + break ( + if exit.success() { CommandStatus::Completed } else { CommandStatus::Failed }, + exit.code(), + ); + } + _ = &mut timeout => { + let _ = child.start_kill(); + let exit_code = child.wait().await.ok().and_then(|status| status.code()); + break (CommandStatus::TimedOut, exit_code); + } + changed = cancel.changed() => { + if changed.is_err() || *cancel.borrow() { + let _ = child.start_kill(); + let exit_code = child.wait().await.ok().and_then(|status| status.code()); + break (CommandStatus::Cancelled, exit_code); + } + } + _ = interval.tick() => { + publish_available_output( + &mut stdout_reader, + &mut stdout_offset, + &telemetry, + &command_id, + CommandStream::Stdout, + &stdout_path, + )?; + publish_available_output( + &mut stderr_reader, + &mut stderr_offset, + &telemetry, + &command_id, + CommandStream::Stderr, + &stderr_path, + )?; + } + } + }; + + publish_available_output( + &mut stdout_reader, + &mut stdout_offset, + &telemetry, + &command_id, + CommandStream::Stdout, + &stdout_path, + )?; + publish_available_output( + &mut stderr_reader, + &mut stderr_offset, + &telemetry, + &command_id, + CommandStream::Stderr, + &stderr_path, + )?; + telemetry.terminal(&command_id, status, exit_code); + let (content, truncated) = read_command_output_files(&stdout_path, &stderr_path, request.output_limit.max(1))?; Ok(CommandOutput { - status: CommandStatus::Failed, - exit_code: None, - timed_out, + status, + exit_code, + timed_out: status == CommandStatus::TimedOut, content, next_cursor: None, truncated, }) } +fn publish_available_output( + file: &mut std::fs::File, + offset: &mut u64, + telemetry: &CommandTelemetry, + command_id: &str, + stream: CommandStream, + path: &Path, +) -> Result<(), WorkdirError> { + file.seek(SeekFrom::Start(*offset)) + .map_err(|error| WorkdirError::io(path, error))?; + loop { + let mut buffer = vec![0; COMMAND_EVENT_CHUNK_BYTES]; + let read = file + .read(&mut buffer) + .map_err(|error| WorkdirError::io(path, error))?; + if read == 0 { + return Ok(()); + } + buffer.truncate(read); + telemetry.output(command_id, stream, *offset, &buffer); + *offset = offset.saturating_add(read as u64); + if read < COMMAND_EVENT_CHUNK_BYTES { + return Ok(()); + } + } +} + fn read_command_output_files( stdout_path: &Path, stderr_path: &Path, @@ -1024,6 +1303,7 @@ mod tests { command: "sleep 30".to_owned(), timeout_secs: 60, output_limit: 1024, + tool_call_id: None, }, ) .await @@ -1549,6 +1829,7 @@ mod tests { command: "pwd && printf provider-command".into(), timeout_secs: 5, output_limit: 4096, + tool_call_id: None, }, ) .await @@ -1583,6 +1864,7 @@ mod tests { command: "printf 'aéz'".into(), timeout_secs: 5, output_limit: 1024, + tool_call_id: None, }, ) .await @@ -1620,6 +1902,130 @@ mod tests { )); } + #[tokio::test] + async fn provider_streams_bounded_command_lifecycle_and_distinct_output() { + let dir = TempDir::new().unwrap(); + let workdir = make_fs(&dir); + let mut events = WorkdirSession::subscribe_command_events(&workdir) + .expect("local command observation must be available"); + let handle = WorkdirSession::start_command( + &workdir, + CommandRequest { + command: "printf ready; printf warning >&2; sleep 0.2; printf done".into(), + timeout_secs: 5, + output_limit: 1024, + tool_call_id: Some("tool-7".into()), + }, + ) + .await + .unwrap(); + + let mut stdout = String::new(); + let mut stderr = String::new(); + let mut terminal = None; + while terminal.is_none() { + let event = tokio::time::timeout(Duration::from_secs(2), events.recv()) + .await + .expect("command telemetry should not stall") + .unwrap(); + match event { + CommandEvent::Started { + command_id, + tool_call_id, + } => { + assert_eq!(command_id, handle.0); + assert_eq!(tool_call_id.as_deref(), Some("tool-7")); + } + CommandEvent::Output { + command_id, + stream, + content, + .. + } => { + assert_eq!(command_id, handle.0); + match stream { + CommandStream::Stdout => stdout.push_str(&content), + CommandStream::Stderr => stderr.push_str(&content), + } + } + CommandEvent::Terminal { + command_id, + status, + exit_code, + } => { + assert_eq!(command_id, handle.0); + terminal = Some((status, exit_code)); + } + } + } + assert_eq!(terminal, Some((CommandStatus::Completed, Some(0)))); + assert_eq!(stdout, "readydone"); + assert_eq!(stderr, "warning"); + let snapshot = WorkdirSession::command_snapshot(&workdir); + assert_eq!(snapshot.len(), 1); + assert_eq!(snapshot[0].status, CommandStatus::Completed); + assert_eq!(snapshot[0].stdout.content, "readydone"); + assert_eq!(snapshot[0].stderr.content, "warning"); + + let output = WorkdirSession::command_output( + &workdir, + CommandOutputRequest { + handle, + cursor: 0, + limit: 1024, + wait: true, + }, + ) + .await + .unwrap(); + assert_eq!(output.status, CommandStatus::Completed); + assert!(WorkdirSession::command_snapshot(&workdir).is_empty()); + } + + #[tokio::test] + async fn provider_distinguishes_timed_out_terminal_state() { + let dir = TempDir::new().unwrap(); + let workdir = make_fs(&dir); + let mut events = WorkdirSession::subscribe_command_events(&workdir).unwrap(); + let handle = WorkdirSession::start_command( + &workdir, + CommandRequest { + command: "sleep 30".into(), + timeout_secs: 1, + output_limit: 1024, + tool_call_id: None, + }, + ) + .await + .unwrap(); + let output = WorkdirSession::command_output( + &workdir, + CommandOutputRequest { + handle: handle.clone(), + cursor: 0, + limit: 1024, + wait: true, + }, + ) + .await + .unwrap(); + assert_eq!(output.status, CommandStatus::TimedOut); + assert!(output.timed_out); + + let mut terminal = None; + while let Ok(event) = events.try_recv() { + if let CommandEvent::Terminal { + command_id, + status, + exit_code, + } = event + { + terminal = Some((command_id, status, exit_code)); + } + } + assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None))); + } + #[tokio::test] async fn provider_cancels_active_command() { let dir = TempDir::new().unwrap(); @@ -1630,6 +2036,7 @@ mod tests { command: "sleep 30".into(), timeout_secs: 60, output_limit: 1024, + tool_call_id: None, }, ) .await @@ -1658,12 +2065,12 @@ mod tests { WorkdirSession::cancel_command(&workdir, handle.clone()) .await .unwrap(); - let waiter_error = tokio::time::timeout(Duration::from_secs(1), waiter) + let output = tokio::time::timeout(Duration::from_secs(1), waiter) .await .expect("cancel should wake command output waiters") .unwrap() - .unwrap_err(); - assert!(matches!(waiter_error, WorkdirError::UnknownCommand(_))); + .unwrap(); + assert_eq!(output.status, CommandStatus::Cancelled); assert!(matches!( WorkdirSession::command_status(&workdir, handle).await, Err(WorkdirError::UnknownCommand(_)) diff --git a/crates/workdir/src/operation.rs b/crates/workdir/src/operation.rs index e1e0d15e..54a1f282 100644 --- a/crates/workdir/src/operation.rs +++ b/crates/workdir/src/operation.rs @@ -9,6 +9,11 @@ pub struct CommandRequest { pub command: String, pub timeout_secs: u64, pub output_limit: usize, + /// Optional caller-owned correlation id. Bash supplies its tool-call id so + /// user-facing command telemetry can update the corresponding Console row + /// without exposing provider/session handles. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub tool_call_id: Option, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] @@ -24,8 +29,55 @@ pub struct CommandOutputRequest { pub enum CommandStatus { Running, Completed, - Cancelled, Failed, + TimedOut, + Cancelled, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum CommandStream { + Stdout, + Stderr, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +pub struct CommandStreamSlice { + pub start_offset: u64, + pub end_offset: u64, + pub content: String, + pub truncated: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct CommandSnapshot { + pub command_id: String, + pub tool_call_id: Option, + pub status: CommandStatus, + pub stdout: CommandStreamSlice, + pub stderr: CommandStreamSlice, + pub exit_code: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum CommandEvent { + Started { + command_id: String, + tool_call_id: Option, + }, + Output { + command_id: String, + stream: CommandStream, + start_offset: u64, + end_offset: u64, + content: String, + }, + Terminal { + command_id: String, + status: CommandStatus, + exit_code: Option, + }, } #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index b97c5b58..525ea88c 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -1430,7 +1430,10 @@ impl Runtime { context_tokens: 0, }, status: protocol::WorkerStatus::Idle, - in_flight: protocol::InFlightSnapshot { blocks: Vec::new() }, + in_flight: protocol::InFlightSnapshot { + blocks: Vec::new(), + commands: Vec::new(), + }, internal_workers: Vec::new(), }) } @@ -3765,7 +3768,10 @@ mod tests { context_tokens: 64, }, status: protocol::WorkerStatus::Running, - in_flight: protocol::InFlightSnapshot { blocks: Vec::new() }, + in_flight: protocol::InFlightSnapshot { + blocks: Vec::new(), + commands: Vec::new(), + }, internal_workers: Vec::new(), }, ); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 4db07122..7b5ad021 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -28,8 +28,14 @@ use crate::worker::{ WorkerRunResult, }; use protocol::{ - AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, - TurnResult, WorkerStatus, + AlertLevel, AlertSource, CommandEvent as ProtocolCommandEvent, + CommandSnapshot as ProtocolCommandSnapshot, CommandStatus as ProtocolCommandStatus, + CommandStream as ProtocolCommandStream, CommandStreamSlice as ProtocolCommandStreamSlice, + ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, TurnResult, WorkerStatus, +}; +use workdir::{ + CommandEvent as WorkdirCommandEvent, CommandSnapshot as WorkdirCommandSnapshot, + CommandStatus as WorkdirCommandStatus, CommandStream as WorkdirCommandStream, WorkdirSession, }; // --------------------------------------------------------------------------- @@ -424,6 +430,9 @@ impl WorkerController { Some(method_tx.downgrade()), ) .await?; + if let Some(session) = fs_for_view.as_ref() { + wire_workdir_command_events(session, &in_flight); + } // Intake role Workers self-terminate only after a successful // TicketIntakeReady turn has fully settled back to Idle. The request @@ -498,6 +507,105 @@ impl WorkerController { } } +pub(crate) fn wire_workdir_command_events( + session: &Arc, + in_flight: &InFlightEvents, +) { + in_flight.replace_command_snapshot( + session + .command_snapshot() + .into_iter() + .map(protocol_command_snapshot) + .collect(), + ); + let Some(mut events) = session.subscribe_command_events() else { + return; + }; + let in_flight = in_flight.clone(); + tokio::spawn(async move { + loop { + match events.recv().await { + Ok(event) => in_flight.publish_command_event(protocol_command_event(event)), + Err(broadcast::error::RecvError::Lagged(_)) => { + // Never retain stale command output after a provider-local + // observer lag. The next chunk reconstructs a bounded tail + // with its absolute offset and marks the gap truncated. + in_flight.replace_command_snapshot(Vec::new()); + } + Err(broadcast::error::RecvError::Closed) => break, + } + } + }); +} + +fn protocol_command_snapshot(snapshot: WorkdirCommandSnapshot) -> ProtocolCommandSnapshot { + ProtocolCommandSnapshot { + command_id: snapshot.command_id, + tool_call_id: snapshot.tool_call_id, + status: protocol_command_status(snapshot.status), + stdout: ProtocolCommandStreamSlice { + start_offset: snapshot.stdout.start_offset, + end_offset: snapshot.stdout.end_offset, + content: snapshot.stdout.content, + truncated: snapshot.stdout.truncated, + }, + stderr: ProtocolCommandStreamSlice { + start_offset: snapshot.stderr.start_offset, + end_offset: snapshot.stderr.end_offset, + content: snapshot.stderr.content, + truncated: snapshot.stderr.truncated, + }, + exit_code: snapshot.exit_code, + } +} + +fn protocol_command_event(event: WorkdirCommandEvent) -> ProtocolCommandEvent { + match event { + WorkdirCommandEvent::Started { + command_id, + tool_call_id, + } => ProtocolCommandEvent::Started { + command_id, + tool_call_id, + }, + WorkdirCommandEvent::Output { + command_id, + stream, + start_offset, + end_offset, + content, + } => ProtocolCommandEvent::Output { + command_id, + stream: match stream { + WorkdirCommandStream::Stdout => ProtocolCommandStream::Stdout, + WorkdirCommandStream::Stderr => ProtocolCommandStream::Stderr, + }, + start_offset, + end_offset, + content, + }, + WorkdirCommandEvent::Terminal { + command_id, + status, + exit_code, + } => ProtocolCommandEvent::Terminal { + command_id, + status: protocol_command_status(status), + exit_code, + }, + } +} + +fn protocol_command_status(status: WorkdirCommandStatus) -> ProtocolCommandStatus { + match status { + WorkdirCommandStatus::Running => ProtocolCommandStatus::Running, + WorkdirCommandStatus::Completed => ProtocolCommandStatus::Completed, + WorkdirCommandStatus::Failed => ProtocolCommandStatus::Failed, + WorkdirCommandStatus::TimedOut => ProtocolCommandStatus::TimedOut, + WorkdirCommandStatus::Cancelled => ProtocolCommandStatus::Cancelled, + } +} + /// Wire the per-event broadcast bridges on the Worker's Engine. Each callback /// re-publishes a worker-level signal as a `protocol::Event` on `event_tx` /// so subscribers (TUI, socket clients) get a single typed stream. diff --git a/crates/worker/src/in_flight.rs b/crates/worker/src/in_flight.rs index 6a08d881..cbc819df 100644 --- a/crates/worker/src/in_flight.rs +++ b/crates/worker/src/in_flight.rs @@ -1,9 +1,14 @@ use std::sync::{Arc, Mutex, MutexGuard}; -use protocol::{Event, InFlightBlock, InFlightSnapshot, InFlightToolCallState}; +use protocol::{ + CommandEvent, CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, Event, + InFlightBlock, InFlightSnapshot, InFlightToolCallState, +}; use session_store::{LoggedContentPart, LoggedItem}; use tokio::sync::broadcast; +const COMMAND_SNAPSHOT_STREAM_BYTES: usize = 32 * 1024; + #[derive(Debug, Clone, Copy, PartialEq, Eq)] pub struct InFlightBlockId(u64); @@ -17,6 +22,7 @@ pub struct InFlightEvents { pub(crate) struct InFlightInner { next_block_id: u64, blocks: Vec, + commands: Vec, } #[derive(Debug, Clone)] @@ -46,6 +52,7 @@ impl InFlightEvents { inner: Arc::new(Mutex::new(InFlightInner { next_block_id: 1, blocks: Vec::new(), + commands: Vec::new(), })), event_tx, } @@ -201,6 +208,15 @@ impl InFlightEvents { f() } + pub(crate) fn publish_command_event(&self, event: CommandEvent) { + self.lock().apply_command_event(&event); + let _ = self.event_tx.send(Event::Command { event }); + } + + pub(crate) fn replace_command_snapshot(&self, commands: Vec) { + self.lock().commands = commands; + } + pub(crate) fn clear(&self) { let mut inner = self.lock(); inner.clear(); @@ -224,6 +240,82 @@ impl InFlightInner { .find(|block| block.block_id() == block_id) } + fn apply_command_event(&mut self, event: &CommandEvent) { + match event { + CommandEvent::Started { + command_id, + tool_call_id, + } => { + self.commands + .retain(|command| command.command_id != *command_id); + self.commands.push(CommandSnapshot { + command_id: command_id.clone(), + tool_call_id: tool_call_id.clone(), + status: CommandStatus::Running, + stdout: CommandStreamSlice::default(), + stderr: CommandStreamSlice::default(), + exit_code: None, + }); + } + CommandEvent::Output { + command_id, + stream, + start_offset, + end_offset, + content, + } => { + let command = match self + .commands + .iter_mut() + .find(|command| command.command_id == *command_id) + { + Some(command) => command, + None => { + self.commands.push(CommandSnapshot { + command_id: command_id.clone(), + tool_call_id: None, + status: CommandStatus::Running, + stdout: CommandStreamSlice::default(), + stderr: CommandStreamSlice::default(), + exit_code: None, + }); + self.commands.last_mut().expect("command was inserted") + } + }; + let target = match stream { + CommandStream::Stdout => &mut command.stdout, + CommandStream::Stderr => &mut command.stderr, + }; + if target.end_offset != *start_offset { + target.content.clear(); + target.start_offset = *start_offset; + target.truncated = *start_offset > 0; + } + target.content.push_str(content); + target.end_offset = *end_offset; + if target.content.len() > COMMAND_SNAPSHOT_STREAM_BYTES { + let mut cut = target.content.len() - COMMAND_SNAPSHOT_STREAM_BYTES; + while cut < target.content.len() && !target.content.is_char_boundary(cut) { + cut += 1; + } + target.content.drain(..cut); + target.start_offset = target + .end_offset + .saturating_sub(target.content.len() as u64); + target.truncated = true; + } + } + CommandEvent::Terminal { command_id, .. } => { + // Terminal state is delivered as a live protocol event. It is + // no longer in-flight snapshot state, and removing it here + // also prevents queued output from an aborted turn from + // surviving the subsequent terminal event after `clear()`. + self.commands + .retain(|command| command.command_id != *command_id); + } + } + } + fn clear_for_committed_item(&mut self, item: &LoggedItem) { match item { LoggedItem::Message { role, content } @@ -273,14 +365,16 @@ impl InFlightInner { .iter() .filter_map(TrackedBlock::to_snapshot_block) .collect(), + commands: self.commands.clone(), } } fn clear(&mut self) -> bool { - if self.blocks.is_empty() { + if self.blocks.is_empty() && self.commands.is_empty() { false } else { self.blocks.clear(); + self.commands.clear(); true } } @@ -583,6 +677,52 @@ mod tests { ); } + #[test] + fn command_events_are_bounded_and_recoverable_from_snapshot() { + let (event_tx, _) = broadcast::channel(16); + let mut rx = event_tx.subscribe(); + let in_flight = InFlightEvents::new(event_tx); + in_flight.publish_command_event(CommandEvent::Started { + command_id: "command-1".into(), + tool_call_id: Some("tool-1".into()), + }); + in_flight.publish_command_event(CommandEvent::Output { + command_id: "command-1".into(), + stream: CommandStream::Stdout, + start_offset: 0, + end_offset: 5, + content: "ready".into(), + }); + + let guard = in_flight.snapshot_guard(); + let snapshot = snapshot_from_guard(&guard); + assert_eq!(snapshot.commands.len(), 1); + assert_eq!(snapshot.commands[0].tool_call_id.as_deref(), Some("tool-1")); + assert_eq!(snapshot.commands[0].stdout.content, "ready"); + assert_eq!(snapshot.commands[0].status, CommandStatus::Running); + drop(guard); + assert!(matches!( + rx.try_recv().unwrap(), + Event::Command { + event: CommandEvent::Started { .. } + } + )); + assert!(matches!( + rx.try_recv().unwrap(), + Event::Command { + event: CommandEvent::Output { .. } + } + )); + + in_flight.publish_command_event(CommandEvent::Terminal { + command_id: "command-1".into(), + status: CommandStatus::TimedOut, + exit_code: None, + }); + let guard = in_flight.snapshot_guard(); + assert!(snapshot_from_guard(&guard).commands.is_empty()); + } + #[test] fn clear_discards_uncommitted_blocks_without_protocol_event() { let (event_tx, _) = broadcast::channel(16); diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index 0a21d0d2..5577da78 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -17,7 +17,7 @@ use session_store::{LogEntry, SegmentId, SessionId, Store, StoreError, TraceEntr use tokio::sync::broadcast; use uuid::Uuid; -use crate::controller::wire_event_bridges_on_engine; +use crate::controller::{wire_event_bridges_on_engine, wire_workdir_command_events}; use crate::feature::FeatureRegistryBuilder; use crate::in_flight::{InFlightEvents, snapshot_from_guard}; use crate::ipc::alerter::Alerter; @@ -555,6 +555,9 @@ pub(crate) async fn prepare_internal_worker_session( spawn_internal_log_event_bridge(sink.clone(), event_tx.clone()); let alerter = Alerter::new(event_tx.clone()); let in_flight = InFlightEvents::new(event_tx.clone()); + if let Some(session) = worker.workdir_session() { + wire_workdir_command_events(session, &in_flight); + } let actor_in_flight = in_flight.clone(); worker.attach_alerter(alerter.clone()); worker.attach_event_tx(event_tx.clone()); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index bbe43ba5..41391433 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -12,8 +12,8 @@ use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use session_store::{CombinedStore, FsWorkerStore}; use session_store::{FsStore, LogEntry}; use workdir::{ - CommandRequest, LocalWorkdirSession, Workdir, WorkdirError, WorkdirSessionCapabilities, - WorkdirSessionHandle, + CommandOutputRequest, CommandRequest, LocalWorkdirSession, Workdir, WorkdirError, + WorkdirSessionCapabilities, WorkdirSessionHandle, }; use worker::{ @@ -232,6 +232,7 @@ async fn shutdown_closes_bound_workdir_session() { command: "sleep 30".to_owned(), timeout_secs: 60, output_limit: 1024, + tool_call_id: None, }) .await .unwrap(); @@ -253,6 +254,108 @@ async fn shutdown_closes_bound_workdir_session() { )); } +#[tokio::test] +async fn controller_projects_workdir_command_events_and_snapshot_state() { + let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await; + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( + Workdir::new("controller-command-observation-workdir"), + pwd.clone(), + pwd, + worker.scope().clone(), + WorkdirSessionCapabilities::ALL, + )); + worker.bind_workdir_session(Some(Arc::clone(&session))); + let handle = spawn_controller(worker).await; + let mut events = handle.subscribe(); + + let command = session + .start_command(CommandRequest { + command: "printf ready; sleep 0.3; printf done".to_owned(), + timeout_secs: 5, + output_limit: 1024, + tool_call_id: Some("tool-command-1".into()), + }) + .await + .unwrap(); + + let mut saw_started = false; + let mut saw_output = false; + while !saw_output { + let event = tokio::time::timeout(std::time::Duration::from_secs(2), events.recv()) + .await + .expect("command event should arrive") + .unwrap(); + match event { + Event::Command { + event: + protocol::CommandEvent::Started { + command_id, + tool_call_id, + }, + } => { + assert_eq!(command_id, command.0); + assert_eq!(tool_call_id.as_deref(), Some("tool-command-1")); + saw_started = true; + } + Event::Command { + event: + protocol::CommandEvent::Output { + command_id, + stream: protocol::CommandStream::Stdout, + content, + .. + }, + } if command_id == command.0 && content.contains("ready") => saw_output = true, + _ => {} + } + } + assert!(saw_started); + + let Event::Snapshot { in_flight, .. } = handle.snapshot_event() else { + panic!("worker snapshot expected"); + }; + assert_eq!(in_flight.commands.len(), 1); + assert_eq!(in_flight.commands[0].command_id, command.0); + assert_eq!(in_flight.commands[0].stdout.content, "ready"); + assert_eq!( + in_flight.commands[0].status, + protocol::CommandStatus::Running + ); + + let saw_terminal = drain_until(&mut events, std::time::Duration::from_secs(2), |event| { + matches!( + event, + Event::Command { + event: protocol::CommandEvent::Terminal { + command_id, + status: protocol::CommandStatus::Completed, + exit_code: Some(0), + } + } if command_id == &command.0 + ) + }) + .await; + assert!(saw_terminal, "completed command event should arrive"); + + let output = session + .command_output(CommandOutputRequest { + handle: command, + cursor: 0, + limit: 1024, + wait: true, + }) + .await + .unwrap(); + assert_eq!(output.status, workdir::CommandStatus::Completed); + let (entries, _) = handle.sink.subscribe_with_snapshot(); + let durable_history = serde_json::to_string(&entries).unwrap(); + assert!( + !durable_history.contains("ready") && !durable_history.contains("done"), + "operational command chunks must not be appended to Worker history: {durable_history}" + ); + handle.send(Method::Shutdown).await.unwrap(); +} + #[tokio::test] async fn controller_startup_failure_closes_bound_workdir_session() { let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await; @@ -279,6 +382,7 @@ async fn controller_startup_failure_closes_bound_workdir_session() { command: "printf unreachable".to_owned(), timeout_secs: 5, output_limit: 1024, + tool_call_id: None, }) .await, Err(WorkdirError::Unavailable(_)) diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 471f4dab..6a85bd7e 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -22,6 +22,16 @@ export type Permission = "read" | "write"; export type InFlightToolCallState = "pending" | "streaming_args" | "done"; +export type CommandStatus = "running" | "completed" | "failed" | "timed_out" | "cancelled"; + +export type CommandStream = "stdout" | "stderr"; + +export type CommandStreamSlice = { start_offset: number, end_offset: number, content: string, truncated: boolean, }; + +export type CommandSnapshot = { command_id: string, tool_call_id: string | null, status: CommandStatus, stdout: CommandStreamSlice, stderr: CommandStreamSlice, exit_code: number | null, }; + +export type CommandEvent = { "kind": "started", command_id: string, tool_call_id: string | null, } | { "kind": "output", command_id: string, stream: CommandStream, start_offset: number, end_offset: number, content: string, } | { "kind": "terminal", command_id: string, status: CommandStatus, exit_code: number | null, }; + export type ScopeRule = { /** * Target path. Must be absolute by the time a `Scope` is built from @@ -51,7 +61,7 @@ export type RewindSummary = { truncated_to_entries: number, discarded_entries: n export type InFlightBlock = { "kind": "text", text: string, finished?: boolean, } | { "kind": "thinking", text: string, finished?: boolean, } | { "kind": "tool_call", id: string, name: string, args: string, state?: InFlightToolCallState, }; -export type InFlightSnapshot = { blocks?: Array, }; +export type InFlightSnapshot = { blocks?: Array, commands?: Array, }; export type InternalWorkerKind = "sub_worker"; @@ -178,4 +188,4 @@ in_flight?: InFlightSnapshot, * Parent-owned Internal Worker sessions visible to this client. * Service-private Internal Workers are deliberately excluded. */ -internal_workers?: Array, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, revision: number, event: Event, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array, } } | { "event": "rewind_applied", "data": { entries: Array, input: Array, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" }; +internal_workers?: Array, } } | { "event": "internal_worker", "data": { worker: InternalWorkerRef, revision: number, event: Event, } } | { "event": "segment_rotated", "data": { entry: unknown, } } | { "event": "status", "data": { status: WorkerStatus, } } | { "event": "command", "data": { event: CommandEvent, } } | { "event": "completions", "data": { kind: CompletionKind, entries: Array, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array, } } | { "event": "rewind_applied", "data": { entries: Array, input: Array, summary: RewindSummary, } } | { "event": "workers_listed", "data": { workers: unknown, } } | { "event": "worker_restored", "data": { result: unknown, } } | { "event": "peer_registered", "data": { result: unknown, } } | { "event": "alert", "data": Alert } | { "event": "memory_worker", "data": MemoryWorkerEvent } | { "event": "compact_start" } | { "event": "compact_done", "data": { new_segment_id: string, } } | { "event": "compact_failed", "data": { error: string, } } | { "event": "shutdown" }; diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index 041fdbfb..b10c087c 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -334,6 +334,123 @@ Deno.test("projectConsole groups tool call lifecycle into one Call block", () => ); }); +Deno.test("projectConsole streams distinct Bash stdout and stderr through terminal status", () => { + const projection = projectConsole([ + { + eventId: "command-tool", + event: { + event: "tool_call_done", + data: { + id: "bash-stream", + name: "Bash", + arguments: JSON.stringify({ command: "long-command" }), + }, + } satisfies Event, + }, + { + eventId: "command-started", + event: { + event: "command", + data: { + event: { + kind: "started", + command_id: "command-1", + tool_call_id: "bash-stream", + }, + }, + } satisfies Event, + }, + { + eventId: "command-stdout", + event: { + event: "command", + data: { + event: { + kind: "output", + command_id: "command-1", + stream: "stdout", + start_offset: 0, + end_offset: 6, + content: "ready\n", + }, + }, + } satisfies Event, + }, + { + eventId: "command-stderr", + event: { + event: "command", + data: { + event: { + kind: "output", + command_id: "command-1", + stream: "stderr", + start_offset: 0, + end_offset: 5, + content: "warn\n", + }, + }, + } satisfies Event, + }, + { + eventId: "command-terminal", + event: { + event: "command", + data: { + event: { + kind: "terminal", + command_id: "command-1", + status: "failed", + exit_code: 7, + }, + }, + } satisfies Event, + }, + ]); + + const [line] = projection.lines.filter((line) => line.kind === "tool"); + assert(line.body.includes("Bash — failed (exit 7)"), line.body); + assert(line.body.includes("stdout:\nready\n"), line.body); + assert(line.body.includes("stderr:\nwarn\n"), line.body); + assertEquals(line.streaming, false); + assertEquals(line.error, true); +}); + +Deno.test("snapshot restores bounded in-flight Bash command output", () => { + const snapshot = snapshotEvent("/repo"); + if (snapshot.event !== "snapshot") throw new Error("snapshot fixture expected"); + snapshot.data.status = "running"; + snapshot.data.in_flight = { + blocks: [{ + kind: "tool_call", + id: "bash-snapshot", + name: "Bash", + args: JSON.stringify({ command: "slow" }), + state: "done", + }], + commands: [{ + command_id: "command-2", + tool_call_id: "bash-snapshot", + status: "running", + stdout: { + start_offset: 1024, + end_offset: 1031, + content: "tail\n", + truncated: true, + }, + stderr: { start_offset: 0, end_offset: 0, content: "", truncated: false }, + exit_code: null, + }], + }; + + const projection = projectConsole([{ eventId: "snapshot-command", event: snapshot }]); + const [line] = projection.lines.filter((line) => line.kind === "tool"); + assert(line.body.includes("Bash — running…"), line.body); + assert(line.body.includes("[stdout tail; earlier output omitted]"), line.body); + assert(line.body.includes("stdout:\ntail\n"), line.body); + assertEquals(line.streaming, true); +}); + Deno.test("projectConsole caps default tool request and result previews", () => { const projection = projectConsole([ { diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 2e632fff..15dcbaba 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -1,5 +1,8 @@ import type { Alert, + CommandEvent, + CommandSnapshot, + CommandStreamSlice, Event as ProtocolEvent, InFlightBlock, InFlightToolCallState, @@ -42,6 +45,7 @@ type ToolCallView = { output?: string | null; isError?: boolean; cwd?: string | null; + command?: CommandSnapshot; }; export type ConsoleDiffLine = { @@ -239,6 +243,126 @@ function appendSnapshotInFlightLines( }); } +const COMMAND_STREAM_DISPLAY_BYTES = 32 * 1024; + +function appendSnapshotCommands( + projection: ConsoleProjection, + commands: CommandSnapshot[], + eventId: string, +): void { + commands.forEach((command) => upsertCommandSnapshot(projection, eventId, command)); +} + +function upsertCommandSnapshot( + projection: ConsoleProjection, + eventId: string, + command: CommandSnapshot, +): void { + const toolCallId = command.tool_call_id ?? `command:${command.command_id}`; + const existingIndex = findToolCallLineIndex(projection, toolCallId); + const existing = existingIndex >= 0 + ? projection.lines[existingIndex].toolCall + : undefined; + upsertToolCall(projection, eventId, toolCallId, { + name: existing?.name ?? "Bash", + state: existing?.state ?? "running", + command, + }); +} + +function applyCommandEvent( + projection: ConsoleProjection, + eventId: string, + event: CommandEvent, +): void { + if (event.kind === "started") { + upsertCommandSnapshot(projection, eventId, { + command_id: event.command_id, + tool_call_id: event.tool_call_id, + status: "running", + stdout: emptyCommandStream(), + stderr: emptyCommandStream(), + exit_code: null, + }); + return; + } + + const index = projection.lines.findIndex((line) => + line.toolCall?.command?.command_id === event.command_id + ); + if (index < 0) { + if (event.kind === "output") { + const stream = commandStreamFromEvent(event); + upsertCommandSnapshot(projection, eventId, { + command_id: event.command_id, + tool_call_id: null, + status: "running", + stdout: event.stream === "stdout" ? stream : emptyCommandStream(), + stderr: event.stream === "stderr" ? stream : emptyCommandStream(), + exit_code: null, + }); + } + return; + } + + const existing = projection.lines[index].toolCall!.command!; + if (event.kind === "terminal") { + upsertCommandSnapshot(projection, eventId, { + ...existing, + status: event.status, + exit_code: event.exit_code, + }); + return; + } + const updatedStream = appendCommandStream( + event.stream === "stdout" ? existing.stdout : existing.stderr, + event.start_offset, + event.end_offset, + event.content, + ); + upsertCommandSnapshot(projection, eventId, { + ...existing, + stdout: event.stream === "stdout" ? updatedStream : existing.stdout, + stderr: event.stream === "stderr" ? updatedStream : existing.stderr, + }); +} + +function emptyCommandStream(): CommandStreamSlice { + return { start_offset: 0, end_offset: 0, content: "", truncated: false }; +} + +function commandStreamFromEvent( + event: Extract, +): CommandStreamSlice { + return appendCommandStream( + emptyCommandStream(), + event.start_offset, + event.end_offset, + event.content, + ); +} + +function appendCommandStream( + existing: CommandStreamSlice, + startOffset: number, + endOffset: number, + content: string, +): CommandStreamSlice { + if (endOffset <= existing.end_offset) return existing; + const contiguous = startOffset === existing.end_offset; + const combined = contiguous ? `${existing.content}${content}` : content; + const tail = combined.length > COMMAND_STREAM_DISPLAY_BYTES + ? combined.slice(-COMMAND_STREAM_DISPLAY_BYTES) + : combined; + return { + start_offset: endOffset - tail.length, + end_offset: endOffset, + content: tail, + truncated: existing.truncated || !contiguous || tail.length < combined.length || + startOffset > 0, + }; +} + function projectInternalWorkerSnapshot( snapshot: InternalWorkerSnapshot, eventId: string, @@ -256,6 +380,11 @@ function projectInternalWorkerSnapshot( `${eventId}:internal:${snapshot.worker.session_id}:in-flight`, cwd, ); + appendSnapshotCommands( + console, + snapshot.in_flight?.commands ?? [], + `${eventId}:internal:${snapshot.worker.session_id}:command`, + ); if (snapshot.error) { console.lines.push({ id: `${eventId}:internal:${snapshot.worker.session_id}:error`, @@ -403,6 +532,11 @@ export function applyProtocolEvent( `${envelope.eventId}:snapshot-in-flight`, next.cwd, ); + appendSnapshotCommands( + next, + event.data.in_flight?.commands ?? [], + `${envelope.eventId}:snapshot-command`, + ); next.internalWorkers = (event.data.internal_workers ?? []).map((worker) => projectInternalWorkerSnapshot(worker, envelope.eventId, next.cwd) ); @@ -436,6 +570,9 @@ export function applyProtocolEvent( case "status": next.status = event.data.status; break; + case "command": + applyCommandEvent(next, envelope.eventId, event.data.event); + break; case "segment_rotated": { const retainedErrors = next.lines.filter((line) => line.kind === "error"); const segment = snapshotProjectionFromEntries( @@ -786,6 +923,10 @@ function refreshedToolLine(item: ConsoleLine): ConsoleLine { if (!toolCall) { return item; } + const commandTerminal = toolCall.command !== undefined && + toolCall.command.status !== "running"; + const commandError = toolCall.command !== undefined && + ["failed", "timed_out", "cancelled"].includes(toolCall.command.status); return { ...item, title: item.title.startsWith("Call · Tool result") @@ -794,8 +935,8 @@ function refreshedToolLine(item: ConsoleLine): ConsoleLine { body: renderToolCall(toolCall), detail: toolCallDetail(toolCall), diff: toolCall.name === "Edit" ? editDiff(toolCall) : undefined, - streaming: !["done", "error"].includes(toolCall.state), - error: toolCall.state === "error", + streaming: !["done", "error"].includes(toolCall.state) && !commandTerminal, + error: toolCall.state === "error" || commandError, }; } @@ -1040,9 +1181,37 @@ function renderBashTool(toolCall: ToolCallView): string { const args = parsedArgs(toolCall); const command = stringField(args, "command"); return compactLines([ - `Bash — ${stateSuffix(toolCall.state)}`, + `Bash — ${commandStateSuffix(toolCall)}`, command ? `$ ${command}` : argsText(toolCall), - cappedDisplaySection(resultText(toolCall), 10), + ["done", "error"].includes(toolCall.state) + ? cappedDisplaySection(resultText(toolCall), 10) + : renderLiveCommandOutput(toolCall.command), + ]); +} + +function commandStateSuffix(toolCall: ToolCallView): string { + const command = toolCall.command; + if (!command) return stateSuffix(toolCall.state); + if (command.status === "completed") { + return command.exit_code === null + ? "completed" + : `completed (exit ${command.exit_code})`; + } + if (command.status === "failed") { + return command.exit_code === null ? "failed" : `failed (exit ${command.exit_code})`; + } + if (command.status === "timed_out") return "timed out"; + if (command.status === "cancelled") return "cancelled"; + return "running…"; +} + +function renderLiveCommandOutput(command?: CommandSnapshot): string | undefined { + if (!command) return undefined; + return compactLines([ + command.stdout.truncated ? "[stdout tail; earlier output omitted]" : undefined, + command.stdout.content ? `stdout:\n${command.stdout.content}` : undefined, + command.stderr.truncated ? "[stderr tail; earlier output omitted]" : undefined, + command.stderr.content ? `stderr:\n${command.stderr.content}` : undefined, ]); } From a82234a75ec6cee6839ee8e48fd6077039d384f6 Mon Sep 17 00:00:00 2001 From: Hare Date: Thu, 20 Aug 2026 10:33:12 +0900 Subject: [PATCH 2/3] docs: report subworker feature installation failure --- ...D01-internal-subworker-feature-installation.md | 15 +++++++++++++++ 1 file changed, 15 insertions(+) create mode 100644 docs/report/00001KZXWKD01-internal-subworker-feature-installation.md diff --git a/docs/report/00001KZXWKD01-internal-subworker-feature-installation.md b/docs/report/00001KZXWKD01-internal-subworker-feature-installation.md new file mode 100644 index 00000000..d2020762 --- /dev/null +++ b/docs/report/00001KZXWKD01-internal-subworker-feature-installation.md @@ -0,0 +1,15 @@ +# Internal SubWorker feature installation fails before analysis starts + +While implementing Ticket `00001KZXWKD01`, two read-only Internal SubWorkers were requested to investigate the backend and Web Console paths. Both `SubWorkerSpawn` operations failed before the child session started with: + +```text +install Internal Worker features: Worker feature installation failed: +builtin:worker-observation: required service requirement is not available: +builtin:worker.control +``` + +The requested `builtin:coder` child had read-only scope and did not need peer Worker observation for the delegated investigation. The failure prevented context splitting, so the parent Worker performed the investigation directly. No implementation or validation authority was lost. + +## Improvement direction + +Resolve the effective Internal SubWorker Profile so its installed feature set is satisfiable under the parent-provided services. Either install the required `worker.control` service before `worker-observation`, or avoid enabling `worker-observation` for a child that has no corresponding observation grant/service. Startup validation should identify the Profile feature that introduced the unsatisfied dependency and distinguish a configuration error from unavailable delegated authority. From cfb173c57063b13d3cdcd3714dc18aebb59b1f75 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 21 Aug 2026 03:49:45 +0900 Subject: [PATCH 3/3] fix: preserve command stream boundaries and lifecycle evidence --- crates/protocol/src/lib.rs | 14 ++ crates/workdir/src/local.rs | 205 ++++++++++++++++-- crates/workdir/src/operation.rs | 8 + crates/worker/src/controller.rs | 13 ++ crates/worker/src/in_flight.rs | 15 ++ crates/worker/tests/controller_test.rs | 2 + web/workspace/src/lib/generated/protocol.ts | 4 +- .../src/lib/workspace/console/model.test.ts | 14 ++ .../src/lib/workspace/console/model.ts | 29 +++ 9 files changed, 285 insertions(+), 19 deletions(-) diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 5d7932c9..9e3c06e0 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -756,6 +756,9 @@ pub struct CommandSnapshot { pub command_id: String, pub tool_call_id: Option, pub status: CommandStatus, + pub started_at_ms: u64, + pub observed_at_ms: u64, + pub last_output_at_ms: Option, pub stdout: CommandStreamSlice, pub stderr: CommandStreamSlice, pub exit_code: Option, @@ -768,6 +771,7 @@ pub enum CommandEvent { Started { command_id: String, tool_call_id: Option, + observed_at_ms: u64, }, Output { command_id: String, @@ -775,11 +779,15 @@ pub enum CommandEvent { start_offset: u64, end_offset: u64, content: String, + observed_at_ms: u64, }, Terminal { command_id: String, status: CommandStatus, exit_code: Option, + stdout_end_offset: u64, + stderr_end_offset: u64, + observed_at_ms: u64, }, } @@ -1450,6 +1458,9 @@ mod tests { command_id: "command-1".into(), tool_call_id: Some("call_1".into()), status: CommandStatus::Running, + started_at_ms: 100, + observed_at_ms: 120, + last_output_at_ms: Some(120), stdout: CommandStreamSlice { start_offset: 4, end_offset: 8, @@ -1537,6 +1548,7 @@ mod tests { start_offset: 8, end_offset: 12, content: "warn".into(), + observed_at_ms: 42, }, }; let json = serde_json::to_string(&event).unwrap(); @@ -1546,6 +1558,7 @@ mod tests { assert_eq!(parsed["data"]["event"]["stream"], "stderr"); assert_eq!(parsed["data"]["event"]["start_offset"], 8); assert_eq!(parsed["data"]["event"]["end_offset"], 12); + assert_eq!(parsed["data"]["event"]["observed_at_ms"], 42); assert!(matches!( serde_json::from_str::(&json).unwrap(), Event::Command { @@ -1555,6 +1568,7 @@ mod tests { start_offset: 8, end_offset: 12, content, + observed_at_ms: 42, } } if command_id == "command-1" && content == "warn" )); diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index c4eb28b4..71b67952 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -16,7 +16,7 @@ use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex as StdMutex}; -use std::time::Duration; +use std::time::{Duration, SystemTime, UNIX_EPOCH}; use async_trait::async_trait; use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope}; @@ -41,6 +41,15 @@ const COMMAND_EVENT_CHANNEL_CAPACITY: usize = 256; const COMMAND_EVENT_CHUNK_BYTES: usize = 8 * 1024; const COMMAND_SNAPSHOT_STREAM_BYTES: usize = 32 * 1024; +fn command_observed_at_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .try_into() + .unwrap_or(u64::MAX) +} + #[derive(Debug)] enum LocalCommand { Running { @@ -91,6 +100,7 @@ impl CommandTelemetry { } fn started(&self, command_id: &str, tool_call_id: Option) { + let observed_at_ms = command_observed_at_ms(); self.inner .snapshots .lock() @@ -101,6 +111,9 @@ impl CommandTelemetry { command_id: command_id.to_string(), tool_call_id: tool_call_id.clone(), status: CommandStatus::Running, + started_at_ms: observed_at_ms, + observed_at_ms, + last_output_at_ms: None, stdout: CommandStreamSlice::default(), stderr: CommandStreamSlice::default(), exit_code: None, @@ -109,6 +122,7 @@ impl CommandTelemetry { let _ = self.inner.events.send(CommandEvent::Started { command_id: command_id.to_string(), tool_call_id, + observed_at_ms, }); } @@ -118,6 +132,7 @@ impl CommandTelemetry { } let end_offset = start_offset.saturating_add(bytes.len() as u64); let content = String::from_utf8_lossy(bytes).into_owned(); + let observed_at_ms = command_observed_at_ms(); if let Some(snapshot) = self .inner .snapshots @@ -125,6 +140,8 @@ impl CommandTelemetry { .expect("command telemetry mutex poisoned") .get_mut(command_id) { + snapshot.observed_at_ms = observed_at_ms; + snapshot.last_output_at_ms = Some(observed_at_ms); let target = match stream { CommandStream::Stdout => &mut snapshot.stdout, CommandStream::Stderr => &mut snapshot.stderr, @@ -147,11 +164,13 @@ impl CommandTelemetry { start_offset, end_offset, content, + observed_at_ms, }); } fn terminal(&self, command_id: &str, status: CommandStatus, exit_code: Option) { - if let Some(snapshot) = self + let observed_at_ms = command_observed_at_ms(); + let (stdout_end_offset, stderr_end_offset) = if let Some(snapshot) = self .inner .snapshots .lock() @@ -160,11 +179,18 @@ impl CommandTelemetry { { snapshot.status = status; snapshot.exit_code = exit_code; - } + snapshot.observed_at_ms = observed_at_ms; + (snapshot.stdout.end_offset, snapshot.stderr.end_offset) + } else { + (0, 0) + }; let _ = self.inner.events.send(CommandEvent::Terminal { command_id: command_id.to_string(), status, exit_code, + stdout_end_offset, + stderr_end_offset, + observed_at_ms, }); } @@ -909,8 +935,8 @@ async fn run_command( std::fs::File::open(&stdout_path).map_err(|error| WorkdirError::io(&stdout_path, error))?; let mut stderr_reader = std::fs::File::open(&stderr_path).map_err(|error| WorkdirError::io(&stderr_path, error))?; - let mut stdout_offset = 0; - let mut stderr_offset = 0; + let mut stdout_decoder = CommandOutputDecoder::default(); + let mut stderr_decoder = CommandOutputDecoder::default(); let mut timeout = Box::pin(tokio::time::sleep(Duration::from_secs( request.timeout_secs.max(1), ))); @@ -942,19 +968,21 @@ async fn run_command( _ = interval.tick() => { publish_available_output( &mut stdout_reader, - &mut stdout_offset, + &mut stdout_decoder, &telemetry, &command_id, CommandStream::Stdout, &stdout_path, + false, )?; publish_available_output( &mut stderr_reader, - &mut stderr_offset, + &mut stderr_decoder, &telemetry, &command_id, CommandStream::Stderr, &stderr_path, + false, )?; } } @@ -962,19 +990,21 @@ async fn run_command( publish_available_output( &mut stdout_reader, - &mut stdout_offset, + &mut stdout_decoder, &telemetry, &command_id, CommandStream::Stdout, &stdout_path, + true, )?; publish_available_output( &mut stderr_reader, - &mut stderr_offset, + &mut stderr_decoder, &telemetry, &command_id, CommandStream::Stderr, &stderr_path, + true, )?; telemetry.terminal(&command_id, status, exit_code); @@ -990,15 +1020,23 @@ async fn run_command( }) } +#[derive(Debug, Default)] +struct CommandOutputDecoder { + read_offset: u64, + emitted_offset: u64, + pending: Vec, +} + fn publish_available_output( file: &mut std::fs::File, - offset: &mut u64, + decoder: &mut CommandOutputDecoder, telemetry: &CommandTelemetry, command_id: &str, stream: CommandStream, path: &Path, + flush: bool, ) -> Result<(), WorkdirError> { - file.seek(SeekFrom::Start(*offset)) + file.seek(SeekFrom::Start(decoder.read_offset)) .map_err(|error| WorkdirError::io(path, error))?; loop { let mut buffer = vec![0; COMMAND_EVENT_CHUNK_BYTES]; @@ -1006,17 +1044,67 @@ fn publish_available_output( .read(&mut buffer) .map_err(|error| WorkdirError::io(path, error))?; if read == 0 { + publish_decoded_output(decoder, telemetry, command_id, stream, flush); return Ok(()); } - buffer.truncate(read); - telemetry.output(command_id, stream, *offset, &buffer); - *offset = offset.saturating_add(read as u64); + decoder.pending.extend_from_slice(&buffer[..read]); + decoder.read_offset = decoder.read_offset.saturating_add(read as u64); + publish_decoded_output(decoder, telemetry, command_id, stream, false); if read < COMMAND_EVENT_CHUNK_BYTES { + if flush { + publish_decoded_output(decoder, telemetry, command_id, stream, true); + } return Ok(()); } } } +fn publish_decoded_output( + decoder: &mut CommandOutputDecoder, + telemetry: &CommandTelemetry, + command_id: &str, + stream: CommandStream, + flush: bool, +) { + let prefix_len = if flush { + decoder.pending.len() + } else { + stable_utf8_prefix_len(&decoder.pending) + }; + if prefix_len == 0 { + return; + } + telemetry.output( + command_id, + stream, + decoder.emitted_offset, + &decoder.pending[..prefix_len], + ); + decoder.emitted_offset = decoder.emitted_offset.saturating_add(prefix_len as u64); + decoder.pending.drain(..prefix_len); +} + +/// Return the byte prefix that can be decoded now without replacing a valid +/// UTF-8 scalar whose remaining bytes may arrive in a later file read. Definite +/// invalid sequences remain in the prefix and are rendered lossily, preserving +/// the existing arbitrary-byte output behavior. +fn stable_utf8_prefix_len(bytes: &[u8]) -> usize { + let mut inspected = 0; + while inspected < bytes.len() { + match std::str::from_utf8(&bytes[inspected..]) { + Ok(_) => return bytes.len(), + Err(error) => { + inspected += error.valid_up_to(); + match error.error_len() { + Some(invalid_len) => inspected += invalid_len, + None => return inspected, + } + } + } + } + inspected +} + fn read_command_output_files( stdout_path: &Path, stderr_path: &Path, @@ -1902,6 +1990,64 @@ mod tests { )); } + #[test] + fn command_output_decoder_preserves_utf8_split_across_file_reads() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("command.out"); + let mut first_write = vec![b'a'; COMMAND_EVENT_CHUNK_BYTES - 1]; + first_write.push(0xe2); + std::fs::write(&path, first_write).unwrap(); + + let telemetry = CommandTelemetry::new(); + let mut events = telemetry.subscribe(); + telemetry.started("command-utf8", None); + let mut decoder = CommandOutputDecoder::default(); + let mut reader = std::fs::File::open(&path).unwrap(); + publish_available_output( + &mut reader, + &mut decoder, + &telemetry, + "command-utf8", + CommandStream::Stdout, + &path, + false, + ) + .unwrap(); + assert_eq!(decoder.pending, vec![0xe2]); + + let mut writer = std::fs::OpenOptions::new() + .append(true) + .open(&path) + .unwrap(); + writer.write_all(&[0x82, 0xac]).unwrap(); + writer.flush().unwrap(); + publish_available_output( + &mut reader, + &mut decoder, + &telemetry, + "command-utf8", + CommandStream::Stdout, + &path, + false, + ) + .unwrap(); + + let output = std::iter::from_fn(|| events.try_recv().ok()) + .filter_map(|event| match event { + CommandEvent::Output { + stream: CommandStream::Stdout, + content, + .. + } => Some(content), + _ => None, + }) + .collect::(); + assert_eq!(output.len(), COMMAND_EVENT_CHUNK_BYTES - 1 + "€".len()); + assert!(output.ends_with('€')); + assert!(!output.contains('\u{fffd}')); + assert!(decoder.pending.is_empty()); + } + #[tokio::test] async fn provider_streams_bounded_command_lifecycle_and_distinct_output() { let dir = TempDir::new().unwrap(); @@ -1921,6 +2067,7 @@ mod tests { .unwrap(); let mut stdout = String::new(); + let mut stdout_chunks = 0; let mut stderr = String::new(); let mut terminal = None; while terminal.is_none() { @@ -1932,6 +2079,7 @@ mod tests { CommandEvent::Started { command_id, tool_call_id, + .. } => { assert_eq!(command_id, handle.0); assert_eq!(tool_call_id.as_deref(), Some("tool-7")); @@ -1944,7 +2092,10 @@ mod tests { } => { assert_eq!(command_id, handle.0); match stream { - CommandStream::Stdout => stdout.push_str(&content), + CommandStream::Stdout => { + stdout_chunks += 1; + stdout.push_str(&content); + } CommandStream::Stderr => stderr.push_str(&content), } } @@ -1952,13 +2103,32 @@ mod tests { command_id, status, exit_code, + stdout_end_offset, + stderr_end_offset, + observed_at_ms, } => { assert_eq!(command_id, handle.0); - terminal = Some((status, exit_code)); + terminal = Some(( + status, + exit_code, + stdout_end_offset, + stderr_end_offset, + observed_at_ms, + )); } } } - assert_eq!(terminal, Some((CommandStatus::Completed, Some(0)))); + let (status, exit_code, stdout_end_offset, stderr_end_offset, observed_at_ms) = + terminal.unwrap(); + assert_eq!(status, CommandStatus::Completed); + assert_eq!(exit_code, Some(0)); + assert_eq!(stdout_end_offset, "readydone".len() as u64); + assert_eq!(stderr_end_offset, "warning".len() as u64); + assert!(observed_at_ms > 0); + assert!( + stdout_chunks >= 2, + "long-running output should stream incrementally" + ); assert_eq!(stdout, "readydone"); assert_eq!(stderr, "warning"); let snapshot = WorkdirSession::command_snapshot(&workdir); @@ -2018,6 +2188,7 @@ mod tests { command_id, status, exit_code, + .. } = event { terminal = Some((command_id, status, exit_code)); diff --git a/crates/workdir/src/operation.rs b/crates/workdir/src/operation.rs index 54a1f282..47527ad8 100644 --- a/crates/workdir/src/operation.rs +++ b/crates/workdir/src/operation.rs @@ -54,6 +54,9 @@ pub struct CommandSnapshot { pub command_id: String, pub tool_call_id: Option, pub status: CommandStatus, + pub started_at_ms: u64, + pub observed_at_ms: u64, + pub last_output_at_ms: Option, pub stdout: CommandStreamSlice, pub stderr: CommandStreamSlice, pub exit_code: Option, @@ -65,6 +68,7 @@ pub enum CommandEvent { Started { command_id: String, tool_call_id: Option, + observed_at_ms: u64, }, Output { command_id: String, @@ -72,11 +76,15 @@ pub enum CommandEvent { start_offset: u64, end_offset: u64, content: String, + observed_at_ms: u64, }, Terminal { command_id: String, status: CommandStatus, exit_code: Option, + stdout_end_offset: u64, + stderr_end_offset: u64, + observed_at_ms: u64, }, } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 7b5ad021..8baeb1fc 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -543,6 +543,9 @@ fn protocol_command_snapshot(snapshot: WorkdirCommandSnapshot) -> ProtocolComman command_id: snapshot.command_id, tool_call_id: snapshot.tool_call_id, status: protocol_command_status(snapshot.status), + started_at_ms: snapshot.started_at_ms, + observed_at_ms: snapshot.observed_at_ms, + last_output_at_ms: snapshot.last_output_at_ms, stdout: ProtocolCommandStreamSlice { start_offset: snapshot.stdout.start_offset, end_offset: snapshot.stdout.end_offset, @@ -564,9 +567,11 @@ fn protocol_command_event(event: WorkdirCommandEvent) -> ProtocolCommandEvent { WorkdirCommandEvent::Started { command_id, tool_call_id, + observed_at_ms, } => ProtocolCommandEvent::Started { command_id, tool_call_id, + observed_at_ms, }, WorkdirCommandEvent::Output { command_id, @@ -574,6 +579,7 @@ fn protocol_command_event(event: WorkdirCommandEvent) -> ProtocolCommandEvent { start_offset, end_offset, content, + observed_at_ms, } => ProtocolCommandEvent::Output { command_id, stream: match stream { @@ -583,15 +589,22 @@ fn protocol_command_event(event: WorkdirCommandEvent) -> ProtocolCommandEvent { start_offset, end_offset, content, + observed_at_ms, }, WorkdirCommandEvent::Terminal { command_id, status, exit_code, + stdout_end_offset, + stderr_end_offset, + observed_at_ms, } => ProtocolCommandEvent::Terminal { command_id, status: protocol_command_status(status), exit_code, + stdout_end_offset, + stderr_end_offset, + observed_at_ms, }, } } diff --git a/crates/worker/src/in_flight.rs b/crates/worker/src/in_flight.rs index cbc819df..ecf56f75 100644 --- a/crates/worker/src/in_flight.rs +++ b/crates/worker/src/in_flight.rs @@ -245,6 +245,7 @@ impl InFlightInner { CommandEvent::Started { command_id, tool_call_id, + observed_at_ms, } => { self.commands .retain(|command| command.command_id != *command_id); @@ -252,6 +253,9 @@ impl InFlightInner { command_id: command_id.clone(), tool_call_id: tool_call_id.clone(), status: CommandStatus::Running, + started_at_ms: *observed_at_ms, + observed_at_ms: *observed_at_ms, + last_output_at_ms: None, stdout: CommandStreamSlice::default(), stderr: CommandStreamSlice::default(), exit_code: None, @@ -263,6 +267,7 @@ impl InFlightInner { start_offset, end_offset, content, + observed_at_ms, } => { let command = match self .commands @@ -275,6 +280,9 @@ impl InFlightInner { command_id: command_id.clone(), tool_call_id: None, status: CommandStatus::Running, + started_at_ms: *observed_at_ms, + observed_at_ms: *observed_at_ms, + last_output_at_ms: Some(*observed_at_ms), stdout: CommandStreamSlice::default(), stderr: CommandStreamSlice::default(), exit_code: None, @@ -282,6 +290,8 @@ impl InFlightInner { self.commands.last_mut().expect("command was inserted") } }; + command.observed_at_ms = *observed_at_ms; + command.last_output_at_ms = Some(*observed_at_ms); let target = match stream { CommandStream::Stdout => &mut command.stdout, CommandStream::Stderr => &mut command.stderr, @@ -685,6 +695,7 @@ mod tests { in_flight.publish_command_event(CommandEvent::Started { command_id: "command-1".into(), tool_call_id: Some("tool-1".into()), + observed_at_ms: 100, }); in_flight.publish_command_event(CommandEvent::Output { command_id: "command-1".into(), @@ -692,6 +703,7 @@ mod tests { start_offset: 0, end_offset: 5, content: "ready".into(), + observed_at_ms: 110, }); let guard = in_flight.snapshot_guard(); @@ -718,6 +730,9 @@ mod tests { command_id: "command-1".into(), status: CommandStatus::TimedOut, exit_code: None, + stdout_end_offset: 5, + stderr_end_offset: 0, + observed_at_ms: 200, }); let guard = in_flight.snapshot_guard(); assert!(snapshot_from_guard(&guard).commands.is_empty()); diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 41391433..a51a85f3 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -291,6 +291,7 @@ async fn controller_projects_workdir_command_events_and_snapshot_state() { protocol::CommandEvent::Started { command_id, tool_call_id, + .. }, } => { assert_eq!(command_id, command.0); @@ -330,6 +331,7 @@ async fn controller_projects_workdir_command_events_and_snapshot_state() { command_id, status: protocol::CommandStatus::Completed, exit_code: Some(0), + .. } } if command_id == &command.0 ) diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 6a85bd7e..c6308359 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -28,9 +28,9 @@ export type CommandStream = "stdout" | "stderr"; export type CommandStreamSlice = { start_offset: number, end_offset: number, content: string, truncated: boolean, }; -export type CommandSnapshot = { command_id: string, tool_call_id: string | null, status: CommandStatus, stdout: CommandStreamSlice, stderr: CommandStreamSlice, exit_code: number | null, }; +export type CommandSnapshot = { command_id: string, tool_call_id: string | null, status: CommandStatus, started_at_ms: number, observed_at_ms: number, last_output_at_ms: number | null, stdout: CommandStreamSlice, stderr: CommandStreamSlice, exit_code: number | null, }; -export type CommandEvent = { "kind": "started", command_id: string, tool_call_id: string | null, } | { "kind": "output", command_id: string, stream: CommandStream, start_offset: number, end_offset: number, content: string, } | { "kind": "terminal", command_id: string, status: CommandStatus, exit_code: number | null, }; +export type CommandEvent = { "kind": "started", command_id: string, tool_call_id: string | null, observed_at_ms: number, } | { "kind": "output", command_id: string, stream: CommandStream, start_offset: number, end_offset: number, content: string, observed_at_ms: number, } | { "kind": "terminal", command_id: string, status: CommandStatus, exit_code: number | null, stdout_end_offset: number, stderr_end_offset: number, observed_at_ms: number, }; export type ScopeRule = { /** diff --git a/web/workspace/src/lib/workspace/console/model.test.ts b/web/workspace/src/lib/workspace/console/model.test.ts index b10c087c..328d7231 100644 --- a/web/workspace/src/lib/workspace/console/model.test.ts +++ b/web/workspace/src/lib/workspace/console/model.test.ts @@ -356,6 +356,7 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin kind: "started", command_id: "command-1", tool_call_id: "bash-stream", + observed_at_ms: 1000, }, }, } satisfies Event, @@ -372,6 +373,7 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin start_offset: 0, end_offset: 6, content: "ready\n", + observed_at_ms: 1100, }, }, } satisfies Event, @@ -388,6 +390,7 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin start_offset: 0, end_offset: 5, content: "warn\n", + observed_at_ms: 1200, }, }, } satisfies Event, @@ -402,6 +405,9 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin command_id: "command-1", status: "failed", exit_code: 7, + stdout_end_offset: 6, + stderr_end_offset: 5, + observed_at_ms: 1300, }, }, } satisfies Event, @@ -410,6 +416,7 @@ Deno.test("projectConsole streams distinct Bash stdout and stderr through termin const [line] = projection.lines.filter((line) => line.kind === "tool"); assert(line.body.includes("Bash — failed (exit 7)"), line.body); + assert(line.body.includes("elapsed 300ms"), line.body); assert(line.body.includes("stdout:\nready\n"), line.body); assert(line.body.includes("stderr:\nwarn\n"), line.body); assertEquals(line.streaming, false); @@ -432,6 +439,9 @@ Deno.test("snapshot restores bounded in-flight Bash command output", () => { command_id: "command-2", tool_call_id: "bash-snapshot", status: "running", + started_at_ms: 1000, + observed_at_ms: 1250, + last_output_at_ms: 1200, stdout: { start_offset: 1024, end_offset: 1031, @@ -446,6 +456,10 @@ Deno.test("snapshot restores bounded in-flight Bash command output", () => { const projection = projectConsole([{ eventId: "snapshot-command", event: snapshot }]); const [line] = projection.lines.filter((line) => line.kind === "tool"); assert(line.body.includes("Bash — running…"), line.body); + assert( + line.body.includes("elapsed 250ms · last output at +200ms"), + line.body, + ); assert(line.body.includes("[stdout tail; earlier output omitted]"), line.body); assert(line.body.includes("stdout:\ntail\n"), line.body); assertEquals(line.streaming, true); diff --git a/web/workspace/src/lib/workspace/console/model.ts b/web/workspace/src/lib/workspace/console/model.ts index 15dcbaba..eac24b2c 100644 --- a/web/workspace/src/lib/workspace/console/model.ts +++ b/web/workspace/src/lib/workspace/console/model.ts @@ -280,6 +280,9 @@ function applyCommandEvent( command_id: event.command_id, tool_call_id: event.tool_call_id, status: "running", + started_at_ms: event.observed_at_ms, + observed_at_ms: event.observed_at_ms, + last_output_at_ms: null, stdout: emptyCommandStream(), stderr: emptyCommandStream(), exit_code: null, @@ -297,6 +300,9 @@ function applyCommandEvent( command_id: event.command_id, tool_call_id: null, status: "running", + started_at_ms: event.observed_at_ms, + observed_at_ms: event.observed_at_ms, + last_output_at_ms: event.observed_at_ms, stdout: event.stream === "stdout" ? stream : emptyCommandStream(), stderr: event.stream === "stderr" ? stream : emptyCommandStream(), exit_code: null, @@ -311,6 +317,7 @@ function applyCommandEvent( ...existing, status: event.status, exit_code: event.exit_code, + observed_at_ms: event.observed_at_ms, }); return; } @@ -322,6 +329,8 @@ function applyCommandEvent( ); upsertCommandSnapshot(projection, eventId, { ...existing, + observed_at_ms: event.observed_at_ms, + last_output_at_ms: event.observed_at_ms, stdout: event.stream === "stdout" ? updatedStream : existing.stdout, stderr: event.stream === "stderr" ? updatedStream : existing.stderr, }); @@ -1183,6 +1192,7 @@ function renderBashTool(toolCall: ToolCallView): string { return compactLines([ `Bash — ${commandStateSuffix(toolCall)}`, command ? `$ ${command}` : argsText(toolCall), + commandTiming(toolCall.command), ["done", "error"].includes(toolCall.state) ? cappedDisplaySection(resultText(toolCall), 10) : renderLiveCommandOutput(toolCall.command), @@ -1205,6 +1215,25 @@ function commandStateSuffix(toolCall: ToolCallView): string { return "running…"; } +function commandTiming(command?: CommandSnapshot): string | undefined { + if (!command) return undefined; + const elapsed = Math.max(0, command.observed_at_ms - command.started_at_ms); + if (command.status !== "running") return `elapsed ${durationLabel(elapsed)}`; + if (command.last_output_at_ms === null) { + return `elapsed ${durationLabel(elapsed)} · awaiting first output`; + } + const lastOutputElapsed = Math.max( + 0, + command.last_output_at_ms - command.started_at_ms, + ); + return `elapsed ${durationLabel(elapsed)} · last output at +${durationLabel(lastOutputElapsed)}`; +} + +function durationLabel(milliseconds: number): string { + if (milliseconds < 1000) return `${milliseconds}ms`; + return `${(milliseconds / 1000).toFixed(milliseconds < 10_000 ? 1 : 0)}s`; +} + function renderLiveCommandOutput(command?: CommandSnapshot): string | undefined { if (!command) return undefined; return compactLines([