docs: report subworker feature installation failure

Cargo.toml

@@ -109,7 +109,7 @@ serde = "1.0"
 serde_json = "1.0"
 serde_yaml = "0.9.34"
 tar = "0.4"
-rusqlite = { version = "0.37", features = ["bundled"] }
+rusqlite = { version = "0.37", features = ["backup", "bundled"] }
 ring = "0.17.14"
 sha2 = "0.11"
 tempfile = "3.27"

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<String>,
+    pub status: CommandStatus,
+    pub stdout: CommandStreamSlice,
+    pub stderr: CommandStreamSlice,
+    pub exit_code: Option<i32>,
+}
+
+#[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<String>,
+    },
+    Output {
+        command_id: String,
+        stream: CommandStream,
+        start_offset: u64,
+        end_offset: u64,
+        content: String,
+    },
+    Terminal {
+        command_id: String,
+        status: CommandStatus,
+        exit_code: Option<i32>,
+    },
+}
+
+/// 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<InFlightBlock>,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub commands: Vec<CommandSnapshot>,
 }
 
 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::<Event>(&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":[]}}}"#;

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::<ErrorCode>(&cfg, &mut output);
     push_decl::<Permission>(&cfg, &mut output);
     push_decl::<InFlightToolCallState>(&cfg, &mut output);
+    push_decl::<CommandStatus>(&cfg, &mut output);
+    push_decl::<CommandStream>(&cfg, &mut output);
+    push_decl::<CommandStreamSlice>(&cfg, &mut output);
+    push_decl::<CommandSnapshot>(&cfg, &mut output);
+    push_decl::<CommandEvent>(&cfg, &mut output);
     push_decl::<ScopeRule>(&cfg, &mut output);
     push_decl::<CompletionEntry>(&cfg, &mut output);
     push_decl::<RewindTargetId>(&cfg, &mut output);

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<ToolOutput, ToolError> {
         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)?;

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

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<CommandOutput, WorkdirError>;
     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<broadcast::Receiver<CommandEvent>> {
+        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<CommandSnapshot> {
+        Vec::new()
+    }
+
     /// Terminal, idempotent release of this Worker-bound operation session.
     async fn close(&self) -> Result<(), WorkdirError>;
 }

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<Result<CommandOutput, WorkdirError>>,
         completion: Arc<Notify>,
+        cancel: watch::Sender<bool>,
     },
     Completed(CommandOutput),
 }
 
+#[derive(Debug, Clone)]
+struct CommandTelemetry {
+    inner: Arc<CommandTelemetryInner>,
+}
+
+#[derive(Debug)]
+struct CommandTelemetryInner {
+    snapshots: StdMutex<HashMap<String, CommandSnapshot>>,
+    events: broadcast::Sender<CommandEvent>,
+}
+
+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<CommandEvent> {
+        self.inner.events.subscribe()
+    }
+
+    fn snapshot(&self) -> Vec<CommandSnapshot> {
+        let mut snapshots = self
+            .inner
+            .snapshots
+            .lock()
+            .expect("command telemetry mutex poisoned")
+            .values()
+            .cloned()
+            .collect::<Vec<_>>();
+        snapshots.sort_by(|left, right| left.command_id.cmp(&right.command_id));
+        snapshots
+    }
+
+    fn started(&self, command_id: &str, tool_call_id: Option<String>) {
+        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<i32>) {
+        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<Scope>);
 
@@ -69,6 +201,7 @@ struct LocalWorkdirSessionInner {
     close_lock: Mutex<()>,
     next_command_id: AtomicU64,
     commands: Mutex<HashMap<String, LocalCommand>>,
+    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<CommandHandle, WorkdirError> {
         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<broadcast::Receiver<CommandEvent>> {
+        self.inner
+            .capabilities
+            .supports(WorkdirSessionCapability::Command)
+            .then(|| self.inner.command_telemetry.subscribe())
+    }
+
+    fn command_snapshot(&self) -> Vec<CommandSnapshot> {
+        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::<Vec<_>>()
+        };
+        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<CommandOutput, WorkdirError> {
+async fn run_command(
+    cwd: PathBuf,
+    request: CommandRequest,
+    command_id: String,
+    telemetry: CommandTelemetry,
+    mut cancel: watch::Receiver<bool>,
+) -> Result<CommandOutput, WorkdirError> {
     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<CommandOut
     let stderr_file = std::fs::File::create(&stderr_path)
         .map_err(|error| WorkdirError::io(&stderr_path, error))?;
 
-    let mut child = Command::new("bash")
+    telemetry.started(&command_id, request.tool_call_id.clone());
+    let mut child = match Command::new("bash")
         .arg("-c")
         .arg(&request.command)
         .current_dir(&cwd)
@@ -699,45 +897,126 @@ async fn run_command(cwd: PathBuf, request: CommandRequest) -> Result<CommandOut
         .stderr(Stdio::from(stderr_file))
         .kill_on_drop(true)
         .spawn()
-        .map_err(|error| WorkdirError::io(&cwd, error))?;
-
-    let timed_out = match tokio::time::timeout(
-        Duration::from_secs(request.timeout_secs.max(1)),
-        child.wait(),
-    )
-    .await
     {
-        Ok(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(_))

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<String>,
 }
 
 #[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<String>,
+    pub status: CommandStatus,
+    pub stdout: CommandStreamSlice,
+    pub stderr: CommandStreamSlice,
+    pub exit_code: Option<i32>,
+}
+
+#[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<String>,
+    },
+    Output {
+        command_id: String,
+        stream: CommandStream,
+        start_offset: u64,
+        end_offset: u64,
+        content: String,
+    },
+    Terminal {
+        command_id: String,
+        status: CommandStatus,
+        exit_code: Option<i32>,
+    },
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]

crates/worker-runtime/src/fs_store.rs

@@ -2,7 +2,9 @@ use crate::catalog::{CreateWorkerRequest, WorkingDirectoryStatus};
 use crate::config_bundle::ConfigBundle;
 use crate::diagnostics::{DiagnosticSeverity, RuntimeDiagnostic};
 use crate::error::RuntimeError;
-use crate::identity::{WorkerId, WorkerRef};
+use crate::identity::{
+    LegacyWorkerIdentityMapping, WorkerId, WorkerRef, legacy_worker_identity_mapping_digest,
+};
 use crate::management::{RuntimeBackendKind, RuntimeStatus};
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
@@ -11,10 +13,11 @@ use std::io::{BufReader, Write};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicU64, Ordering};
 
-const SCHEMA_VERSION: u32 = 2;
+const SCHEMA_VERSION: u32 = 3;
 const RUNTIME_FILE: &str = "runtime.json";
 const WORKERS_DIR: &str = "workers";
 const WORKER_FILE: &str = "worker.json";
+const WORKER_METADATA_FILE: &str = "metadata.json";
 const LEGACY_OBSERVATIONS_FILE: &str = "observations.jsonl";
 
 static NEXT_TMP_SEQUENCE: AtomicU64 = AtomicU64::new(1);
@@ -52,6 +55,18 @@ pub struct FsRuntimeStore {
 }
 
 impl FsRuntimeStore {
+    pub fn migration_plan(
+        options: &FsRuntimeStoreOptions,
+    ) -> Result<FsRuntimeStoreMigrationPlan, RuntimeError> {
+        plan_runtime_store_migration(&options.root, &options.runtime_id).map(|(plan, _)| plan)
+    }
+
+    pub fn migrate(
+        options: &FsRuntimeStoreOptions,
+    ) -> Result<FsRuntimeStoreMigrationPlan, RuntimeError> {
+        migrate_runtime_store(&options.root, &options.runtime_id)
+    }
+
     pub fn root(&self) -> &Path {
         &self.root
     }
@@ -92,7 +107,7 @@ impl FsRuntimeStore {
         }
 
         if existed {
-            migrate_v1_worker_identity(&root, runtime_id)?;
+            migrate_runtime_store(&root, runtime_id)?;
         }
         let store = Self { root };
         let state = if existed {
@@ -286,11 +301,42 @@ fn runtime_store_corrupt(path: &Path, message: String) -> RuntimeError {
     }
 }
 
-fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), RuntimeError> {
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct FsRuntimeStoreMigrationPlan {
+    pub current_schema_version: u32,
+    pub target_schema_version: u32,
+    pub migration_required: bool,
+    pub worker_count: usize,
+    pub migrated_worker_aggregate_count: usize,
+    pub migrated_diagnostic_worker_ref_count: usize,
+    pub cleared_diagnostic_worker_ref_count: usize,
+    pub mapping_digest: String,
+    pub mappings: Vec<LegacyWorkerIdentityMapping>,
+    pub excluded_ephemeral_paths: Vec<String>,
+}
+
+#[derive(Clone, Debug)]
+struct PlannedRuntimeWorkerMigration {
+    worker_id: WorkerId,
+    source_dir: PathBuf,
+    workspace_id: Option<String>,
+    legacy_mapping: Option<LegacyWorkerIdentityMapping>,
+}
+
+fn plan_runtime_store_migration(
+    root: &Path,
+    runtime_id: &str,
+) -> Result<
+    (
+        FsRuntimeStoreMigrationPlan,
+        Vec<PlannedRuntimeWorkerMigration>,
+    ),
+    RuntimeError,
+> {
     let runtime_path = root.join(RUNTIME_FILE);
     let bytes =
         fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?;
-    let mut document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
+    let document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
         runtime_store_corrupt(
             &runtime_path,
             format!("decode Runtime state {}: {error}", runtime_path.display()),
@@ -305,108 +351,803 @@ fn migrate_v1_worker_identity(root: &Path, runtime_id: &str) -> Result<(), Runti
                 "Runtime state is missing schema_version".to_string(),
             )
         })?;
-    if schema_version == u64::from(SCHEMA_VERSION) {
-        return Ok(());
-    }
-    if schema_version != 1 {
-        return Err(runtime_store_corrupt(
+    let current_schema_version = u32::try_from(schema_version).map_err(|_| {
+        runtime_store_corrupt(
             &runtime_path,
+            format!("Runtime store schema version {schema_version} is out of range"),
+        )
+    })?;
+    let staging = migration_sibling(root, "schema-v3-staging")?;
+    let backup = migration_sibling(root, "pre-schema-v3-backup")?;
+    if staging.exists() || backup.exists() {
+        return Err(runtime_store_corrupt(
+            root,
             format!(
-                "unsupported Runtime store schema version {schema_version}; expected 1 or {SCHEMA_VERSION}"
+                "unfinished Runtime migration artifact exists (staging={}, backup={})",
+                staging.display(),
+                backup.display()
             ),
         ));
     }
+    if current_schema_version == SCHEMA_VERSION {
+        let plan = FsRuntimeStoreMigrationPlan {
+            current_schema_version,
+            target_schema_version: SCHEMA_VERSION,
+            migration_required: false,
+            worker_count: 0,
+            migrated_worker_aggregate_count: 0,
+            migrated_diagnostic_worker_ref_count: 0,
+            cleared_diagnostic_worker_ref_count: 0,
+            mapping_digest: legacy_worker_identity_mapping_digest(&[]),
+            mappings: Vec::new(),
+            excluded_ephemeral_paths: Vec::new(),
+        };
+        return Ok((plan, Vec::new()));
+    }
+    if !matches!(current_schema_version, 1 | 2) {
+        return Err(runtime_store_corrupt(
+            &runtime_path,
+            format!(
+                "unsupported Runtime store schema version {schema_version}; expected 1, 2, or {SCHEMA_VERSION}"
+            ),
+        ));
+    }
+    let excluded_ephemeral_paths = runtime_tree_exclusions(root)?;
 
-    let legacy_ids = document
-        .get("workers")
-        .and_then(serde_json::Value::as_array)
-        .ok_or_else(|| {
-            runtime_store_corrupt(
-                &runtime_path,
-                "Runtime state workers must be an array".to_string(),
-            )
-        })?
-        .iter()
-        .map(|value| {
-            value.as_u64().ok_or_else(|| {
-                runtime_store_corrupt(
-                    &runtime_path,
-                    "legacy Worker id must be unsigned".to_string(),
-                )
-            })
-        })
-        .collect::<Result<Vec<_>, _>>()?;
-
-    let mut migrated_ids = Vec::with_capacity(legacy_ids.len());
-    for legacy_id in legacy_ids {
-        let legacy_dir = root.join("workers").join(legacy_id.to_string());
-        let legacy_snapshot_path = legacy_dir.join(WORKER_FILE);
-        let bytes = fs::read(&legacy_snapshot_path)
-            .map_err(|error| runtime_io_error("read", &legacy_snapshot_path, error))?;
-        let mut snapshot: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
-            runtime_store_corrupt(
-                &runtime_path,
-                format!(
-                    "decode Worker snapshot {}: {error}",
-                    legacy_snapshot_path.display()
-                ),
-            )
+    let workers_dir = root.join(WORKERS_DIR);
+    let mut entries = fs::read_dir(&workers_dir)
+        .map_err(|error| runtime_io_error("read workers", &workers_dir, error))?
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|error| runtime_io_error("read workers", &workers_dir, error))?;
+    entries.sort_by_key(|entry| entry.file_name());
+    let mut planned = Vec::with_capacity(entries.len());
+    let mut target_ids = std::collections::BTreeSet::new();
+    for entry in entries {
+        let source_dir = entry.path();
+        if !source_dir.is_dir() {
+            return Err(runtime_store_corrupt(
+                &source_dir,
+                "workers directory contains a non-directory entry".to_string(),
+            ));
+        }
+        let name = entry.file_name();
+        let name = name.to_str().ok_or_else(|| {
+            runtime_store_corrupt(&source_dir, "Worker directory is not UTF-8".to_string())
         })?;
-        let workspace_id = snapshot
-            .get("workspace_id")
-            .and_then(serde_json::Value::as_str)
-            .unwrap_or("local")
-            .to_string();
-        let worker_id = WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_id);
-        let worker_id_text = worker_id.to_string();
-        snapshot["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
-        snapshot["worker_id"] = serde_json::Value::String(worker_id_text.clone());
-        snapshot["worker_ref"]["worker_id"] = serde_json::Value::String(worker_id_text.clone());
-        let request = snapshot
-            .get_mut("request")
-            .and_then(serde_json::Value::as_object_mut)
-            .ok_or_else(|| {
+        let snapshot_path = source_dir.join(WORKER_FILE);
+        let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker snapshot")?;
+        let (worker_id, workspace_id, legacy_mapping) = if current_schema_version == 1 {
+            let legacy_worker_id = name.parse::<u64>().map_err(|_| {
                 runtime_store_corrupt(
-                    &runtime_path,
-                    "Worker snapshot request must be an object".to_string(),
+                    &source_dir,
+                    format!("legacy Worker directory name must be numeric, found {name}"),
                 )
             })?;
-        let fingerprint = request
-            .remove("idempotency_fingerprint")
-            .and_then(|value| value.as_str().map(ToOwned::to_owned))
-            .unwrap_or_else(|| format!("legacy:{workspace_id}:{runtime_id}:{legacy_id}"));
-        request.remove("idempotency_key");
-        request.insert(
+            let workspace_id = snapshot
+                .get("workspace_id")
+                .and_then(serde_json::Value::as_str)
+                .filter(|workspace_id| !workspace_id.is_empty())
+                .ok_or_else(|| {
+                    runtime_store_corrupt(
+                        &snapshot_path,
+                        "legacy Worker snapshot is missing workspace_id; unscoped Workers require an explicit migration disposition"
+                            .to_string(),
+                    )
+                })?
+                .to_string();
+            let worker_id =
+                WorkerId::from_legacy_binding(&workspace_id, runtime_id, legacy_worker_id);
+            let mapping = LegacyWorkerIdentityMapping {
+                workspace_id: workspace_id.clone(),
+                runtime_id: runtime_id.to_string(),
+                legacy_worker_id,
+                worker_id,
+            };
+            (worker_id, Some(workspace_id), Some(mapping))
+        } else {
+            let worker_id = name.parse::<WorkerId>().map_err(|_| {
+                runtime_store_corrupt(
+                    &source_dir,
+                    format!("schema-v2 Worker directory name must be a UUIDv7, found {name}"),
+                )
+            })?;
+            (worker_id, None, None)
+        };
+        if !target_ids.insert(worker_id) {
+            return Err(runtime_store_corrupt(
+                &snapshot_path,
+                format!("Worker identity maps to duplicate target {worker_id}"),
+            ));
+        }
+        let target_dir = workers_dir.join(worker_id.to_string());
+        if target_dir.exists() && target_dir != source_dir {
+            return Err(runtime_store_corrupt(
+                &target_dir,
+                format!("target Worker directory {worker_id} already exists"),
+            ));
+        }
+        planned.push(PlannedRuntimeWorkerMigration {
+            worker_id,
+            source_dir,
+            workspace_id,
+            legacy_mapping,
+        });
+    }
+    let mappings = planned
+        .iter()
+        .filter_map(|worker| worker.legacy_mapping.clone())
+        .collect::<Vec<_>>();
+    let mut migrated_worker_aggregate_count = 0;
+    for worker in &mut planned {
+        let snapshot_path = worker.source_dir.join(WORKER_FILE);
+        let snapshot: serde_json::Value = read_json(&snapshot_path, "read Worker snapshot")?;
+        let migrated = migrate_worker_document(
+            snapshot,
+            current_schema_version,
+            worker.legacy_mapping.as_ref(),
+            &snapshot_path,
+        )?;
+        let snapshot = validate_migrated_worker_document(&migrated, &snapshot_path)?;
+        if snapshot.worker_id != worker.worker_id {
+            return Err(runtime_store_corrupt(
+                &snapshot_path,
+                format!(
+                    "Worker snapshot id {} does not match directory identity {}",
+                    snapshot.worker_id, worker.worker_id
+                ),
+            ));
+        }
+        worker.workspace_id = worker
+            .workspace_id
+            .clone()
+            .or(snapshot.workspace_id)
+            .or_else(|| {
+                snapshot
+                    .request
+                    .workspace_api
+                    .map(|workspace_api| workspace_api.workspace_id)
+            });
+
+        let metadata_path = worker.source_dir.join(WORKER_METADATA_FILE);
+        if metadata_path.is_file() {
+            let metadata: serde_json::Value =
+                read_json(&metadata_path, "read Worker aggregate metadata")?;
+            let (_, migrated) =
+                migrate_worker_aggregate_document(metadata, worker, runtime_id, &metadata_path)?;
+            migrated_worker_aggregate_count += usize::from(migrated);
+        }
+    }
+    let (_, diagnostic_refs) =
+        migrate_runtime_document(document, current_schema_version, &mappings, &runtime_path)?;
+    let plan = FsRuntimeStoreMigrationPlan {
+        current_schema_version,
+        target_schema_version: SCHEMA_VERSION,
+        migration_required: true,
+        worker_count: planned.len(),
+        migrated_worker_aggregate_count,
+        migrated_diagnostic_worker_ref_count: diagnostic_refs.migrated,
+        cleared_diagnostic_worker_ref_count: diagnostic_refs.cleared,
+        mapping_digest: legacy_worker_identity_mapping_digest(&mappings),
+        mappings,
+        excluded_ephemeral_paths,
+    };
+    Ok((plan, planned))
+}
+
+#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
+struct DiagnosticWorkerRefMigrationCounts {
+    migrated: usize,
+    cleared: usize,
+}
+
+fn migrate_v1_worker_document(
+    mut snapshot: serde_json::Value,
+    mapping: &LegacyWorkerIdentityMapping,
+    snapshot_path: &Path,
+) -> Result<serde_json::Value, RuntimeError> {
+    let worker_id_text = mapping.worker_id.to_string();
+    let snapshot_object = snapshot.as_object_mut().ok_or_else(|| {
+        runtime_store_corrupt(
+            snapshot_path,
+            "Worker snapshot must be an object".to_string(),
+        )
+    })?;
+    snapshot_object.insert(
+        "schema_version".to_string(),
+        serde_json::Value::from(SCHEMA_VERSION),
+    );
+    snapshot_object.insert(
+        "worker_id".to_string(),
+        serde_json::Value::String(worker_id_text.clone()),
+    );
+    snapshot_object
+        .get_mut("worker_ref")
+        .and_then(serde_json::Value::as_object_mut)
+        .ok_or_else(|| {
+            runtime_store_corrupt(
+                snapshot_path,
+                "Worker snapshot worker_ref must be an object".to_string(),
+            )
+        })?
+        .insert(
             "worker_id".to_string(),
             serde_json::Value::String(worker_id_text.clone()),
         );
-        request.insert(
-            "create_fingerprint".to_string(),
-            serde_json::Value::String(fingerprint),
-        );
+    let request = snapshot_object
+        .get_mut("request")
+        .and_then(serde_json::Value::as_object_mut)
+        .ok_or_else(|| {
+            runtime_store_corrupt(
+                snapshot_path,
+                "Worker snapshot request must be an object".to_string(),
+            )
+        })?;
+    let fingerprint = request
+        .remove("idempotency_fingerprint")
+        .and_then(|value| value.as_str().map(ToOwned::to_owned))
+        .unwrap_or_else(|| {
+            format!(
+                "legacy:{}:{}:{}",
+                mapping.workspace_id, mapping.runtime_id, mapping.legacy_worker_id
+            )
+        });
+    request.remove("idempotency_key");
+    request.insert(
+        "worker_id".to_string(),
+        serde_json::Value::String(worker_id_text),
+    );
+    request.insert(
+        "create_fingerprint".to_string(),
+        serde_json::Value::String(fingerprint),
+    );
+    Ok(snapshot)
+}
 
+fn migrate_worker_document(
+    mut document: serde_json::Value,
+    source_schema_version: u32,
+    mapping: Option<&LegacyWorkerIdentityMapping>,
+    snapshot_path: &Path,
+) -> Result<serde_json::Value, RuntimeError> {
+    if source_schema_version == 1 {
+        return migrate_v1_worker_document(
+            document,
+            mapping.ok_or_else(|| {
+                runtime_store_corrupt(
+                    snapshot_path,
+                    "schema-v1 Worker migration is missing its identity mapping".to_string(),
+                )
+            })?,
+            snapshot_path,
+        );
+    }
+    let object = document.as_object_mut().ok_or_else(|| {
+        runtime_store_corrupt(
+            snapshot_path,
+            "Worker snapshot must be an object".to_string(),
+        )
+    })?;
+    object.insert(
+        "schema_version".to_string(),
+        serde_json::Value::from(SCHEMA_VERSION),
+    );
+    Ok(document)
+}
+
+fn validate_migrated_worker_document(
+    document: &serde_json::Value,
+    snapshot_path: &Path,
+) -> Result<WorkerSnapshot, RuntimeError> {
+    let snapshot: WorkerSnapshot = serde_json::from_value(document.clone()).map_err(|error| {
+        runtime_store_corrupt(
+            snapshot_path,
+            format!("decode migrated Worker snapshot: {error}"),
+        )
+    })?;
+    snapshot.validate(snapshot_path)?;
+    Ok(snapshot)
+}
+
+fn runtime_worker_name(worker_id: WorkerId) -> String {
+    format!("worker-runtime-{worker_id}")
+}
+
+fn migrate_worker_aggregate_document(
+    mut document: serde_json::Value,
+    worker: &PlannedRuntimeWorkerMigration,
+    runtime_id: &str,
+    metadata_path: &Path,
+) -> Result<(serde_json::Value, bool), RuntimeError> {
+    let expected_name = runtime_worker_name(worker.worker_id);
+    let metadata = document.as_object_mut().ok_or_else(|| {
+        runtime_store_corrupt(
+            metadata_path,
+            "Worker aggregate metadata must be an object".to_string(),
+        )
+    })?;
+    let actual_name = metadata
+        .get("worker_name")
+        .and_then(serde_json::Value::as_str)
+        .ok_or_else(|| {
+            runtime_store_corrupt(
+                metadata_path,
+                "Worker aggregate metadata is missing worker_name".to_string(),
+            )
+        })?
+        .to_string();
+    let migrated = actual_name != expected_name;
+    if migrated {
+        let legacy_worker_id = actual_name
+            .strip_prefix("worker-runtime-")
+            .and_then(|worker_id| worker_id.parse::<u64>().ok())
+            .ok_or_else(|| {
+                runtime_store_corrupt(
+                    metadata_path,
+                    format!(
+                        "Worker aggregate identity {actual_name} is neither the expected UUID identity nor a legacy numeric identity"
+                    ),
+                )
+            })?;
+        let workspace_id = worker.workspace_id.as_deref().ok_or_else(|| {
+            runtime_store_corrupt(
+                metadata_path,
+                "legacy Worker aggregate identity has no Workspace binding".to_string(),
+            )
+        })?;
+        let mapped = WorkerId::from_legacy_binding(workspace_id, runtime_id, legacy_worker_id);
+        if mapped != worker.worker_id {
+            return Err(runtime_store_corrupt(
+                metadata_path,
+                format!(
+                    "legacy Worker aggregate identity {actual_name} maps to {mapped}, expected {}",
+                    worker.worker_id
+                ),
+            ));
+        }
+    }
+
+    if let Some(snapshot) = metadata
+        .get_mut("resolved_manifest_snapshot")
+        .filter(|snapshot| !snapshot.is_null())
+    {
+        let manifest: manifest::WorkerManifest =
+            serde_json::from_value(snapshot.clone()).map_err(|error| {
+                runtime_store_corrupt(
+                    metadata_path,
+                    format!("decode Worker aggregate resolved manifest snapshot: {error}"),
+                )
+            })?;
+        if manifest.worker.name != actual_name {
+            return Err(runtime_store_corrupt(
+                metadata_path,
+                format!(
+                    "Worker aggregate manifest identity {} does not match metadata identity {actual_name}",
+                    manifest.worker.name
+                ),
+            ));
+        }
+        snapshot
+            .as_object_mut()
+            .and_then(|manifest| manifest.get_mut("worker"))
+            .and_then(serde_json::Value::as_object_mut)
+            .ok_or_else(|| {
+                runtime_store_corrupt(
+                    metadata_path,
+                    "Worker aggregate resolved manifest is missing worker metadata".to_string(),
+                )
+            })?
+            .insert(
+                "name".to_string(),
+                serde_json::Value::String(expected_name.clone()),
+            );
+    }
+    metadata.insert(
+        "worker_name".to_string(),
+        serde_json::Value::String(expected_name.clone()),
+    );
+
+    let metadata: session_store::WorkerMetadata = serde_json::from_value(document.clone())
+        .map_err(|error| {
+            runtime_store_corrupt(
+                metadata_path,
+                format!("decode migrated Worker aggregate metadata: {error}"),
+            )
+        })?;
+    if metadata.worker_name != expected_name {
+        return Err(runtime_store_corrupt(
+            metadata_path,
+            "migrated Worker aggregate identity does not match its Worker UUID".to_string(),
+        ));
+    }
+    if let Some(snapshot) = metadata.resolved_manifest_snapshot {
+        let manifest: manifest::WorkerManifest =
+            serde_json::from_value(snapshot).map_err(|error| {
+                runtime_store_corrupt(
+                    metadata_path,
+                    format!("decode migrated Worker aggregate resolved manifest: {error}"),
+                )
+            })?;
+        if manifest.worker.name != expected_name {
+            return Err(runtime_store_corrupt(
+                metadata_path,
+                "migrated Worker aggregate manifest identity does not match its Worker UUID"
+                    .to_string(),
+            ));
+        }
+    }
+    Ok((document, migrated))
+}
+
+fn migrate_runtime_document(
+    mut document: serde_json::Value,
+    source_schema_version: u32,
+    mappings: &[LegacyWorkerIdentityMapping],
+    runtime_path: &Path,
+) -> Result<(serde_json::Value, DiagnosticWorkerRefMigrationCounts), RuntimeError> {
+    let mapped_worker_ids = mappings
+        .iter()
+        .map(|mapping| (mapping.legacy_worker_id, mapping.worker_id))
+        .collect::<BTreeMap<_, _>>();
+    let mut counts = DiagnosticWorkerRefMigrationCounts::default();
+    if source_schema_version == 1
+        && let Some(diagnostics) = document.get_mut("diagnostics")
+    {
+        let diagnostics = diagnostics.as_array_mut().ok_or_else(|| {
+            runtime_store_corrupt(
+                runtime_path,
+                "Runtime snapshot diagnostics must be an array".to_string(),
+            )
+        })?;
+        for (index, diagnostic) in diagnostics.iter_mut().enumerate() {
+            let diagnostic = diagnostic.as_object_mut().ok_or_else(|| {
+                runtime_store_corrupt(
+                    runtime_path,
+                    format!("Runtime diagnostic {index} must be an object"),
+                )
+            })?;
+            let Some(worker_ref) = diagnostic.get("worker_ref") else {
+                continue;
+            };
+            if worker_ref.is_null() {
+                continue;
+            }
+            let legacy_worker_id = worker_ref
+                .as_object()
+                .and_then(|worker_ref| worker_ref.get("worker_id"))
+                .and_then(serde_json::Value::as_u64)
+                .ok_or_else(|| {
+                    runtime_store_corrupt(
+                        runtime_path,
+                        format!(
+                            "Runtime diagnostic {index} worker_ref.worker_id must be an unsigned legacy Worker id"
+                        ),
+                    )
+                })?;
+            if let Some(worker_id) = mapped_worker_ids.get(&legacy_worker_id) {
+                diagnostic
+                    .get_mut("worker_ref")
+                    .and_then(serde_json::Value::as_object_mut)
+                    .expect("validated diagnostic Worker reference")
+                    .insert(
+                        "worker_id".to_string(),
+                        serde_json::Value::String(worker_id.to_string()),
+                    );
+                counts.migrated += 1;
+            } else {
+                // The diagnostic remains useful historical evidence, but a deleted
+                // legacy Worker has no Workspace binding from which a stable UUID
+                // can be reconstructed.
+                diagnostic.remove("worker_ref");
+                counts.cleared += 1;
+            }
+        }
+    }
+
+    let object = document.as_object_mut().ok_or_else(|| {
+        runtime_store_corrupt(
+            runtime_path,
+            "Runtime snapshot must be an object".to_string(),
+        )
+    })?;
+    object.insert(
+        "schema_version".to_string(),
+        serde_json::Value::from(SCHEMA_VERSION),
+    );
+    object.remove("workers");
+    object.remove("next_worker_sequence");
+    let snapshot: RuntimeSnapshot = serde_json::from_value(document.clone()).map_err(|error| {
+        runtime_store_corrupt(
+            runtime_path,
+            format!("decode migrated Runtime snapshot: {error}"),
+        )
+    })?;
+    snapshot.validate(runtime_path)?;
+    Ok((document, counts))
+}
+
+fn migration_sibling(root: &Path, suffix: &str) -> Result<PathBuf, RuntimeError> {
+    let parent = root.parent().ok_or_else(|| {
+        runtime_store_corrupt(
+            root,
+            "Runtime store root has no parent directory".to_string(),
+        )
+    })?;
+    let name = root
+        .file_name()
+        .and_then(|name| name.to_str())
+        .ok_or_else(|| {
+            runtime_store_corrupt(root, "Runtime store root name is not UTF-8".to_string())
+        })?;
+    Ok(parent.join(format!(".{name}.{suffix}")))
+}
+
+fn runtime_ephemeral_socket(root: &Path, path: &Path) -> Result<bool, RuntimeError> {
+    let relative = path.strip_prefix(root).map_err(|_| {
+        runtime_store_corrupt(
+            path,
+            format!("Runtime migration path escaped root {}", root.display()),
+        )
+    })?;
+    let components = relative
+        .components()
+        .map(|component| component.as_os_str().to_string_lossy().into_owned())
+        .collect::<Vec<_>>();
+    let known_path = components.len() == 5
+        && components[0] == WORKERS_DIR
+        && (components[1].parse::<u64>().is_ok() || WorkerId::parse(&components[1]).is_some())
+        && components[2] == "runs"
+        && components[3].parse::<u64>().is_ok()
+        && components[4] == "worker.sock";
+    if !known_path {
+        return Ok(false);
+    }
+    #[cfg(unix)]
+    {
+        match std::os::unix::net::UnixStream::connect(path) {
+            Ok(_) => Err(runtime_store_corrupt(
+                path,
+                "Runtime migration found an active Worker socket; stop the legacy Runtime and Worker before migrating"
+                    .to_string(),
+            )),
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
+                ) => Ok(true),
+            Err(error) => Err(runtime_store_corrupt(
+                path,
+                format!("Runtime migration could not verify Worker socket liveness: {error}"),
+            )),
+        }
+    }
+    #[cfg(not(unix))]
+    Ok(false)
+}
+
+fn collect_runtime_tree_exclusions(
+    root: &Path,
+    source: &Path,
+    excluded: &mut Vec<String>,
+) -> Result<(), RuntimeError> {
+    let entries = fs::read_dir(source)
+        .map_err(|error| runtime_io_error("read migration source", source, error))?;
+    for entry in entries {
+        let entry =
+            entry.map_err(|error| runtime_io_error("read migration source", source, error))?;
+        let source_path = entry.path();
+        let file_type = entry
+            .file_type()
+            .map_err(|error| runtime_io_error("inspect migration source", &source_path, error))?;
+        if file_type.is_dir() {
+            collect_runtime_tree_exclusions(root, &source_path, excluded)?;
+        } else if file_type.is_file() {
+        } else if runtime_ephemeral_socket(root, &source_path)? {
+            excluded.push(
+                source_path
+                    .strip_prefix(root)
+                    .expect("validated Runtime migration path")
+                    .to_string_lossy()
+                    .into_owned(),
+            );
+        } else {
+            return Err(runtime_store_corrupt(
+                &source_path,
+                "Runtime migration refuses unknown symlinks and special files".to_string(),
+            ));
+        }
+    }
+    Ok(())
+}
+
+fn runtime_tree_exclusions(root: &Path) -> Result<Vec<String>, RuntimeError> {
+    let mut excluded = Vec::new();
+    collect_runtime_tree_exclusions(root, root, &mut excluded)?;
+    excluded.sort();
+    Ok(excluded)
+}
+
+fn copy_runtime_tree(root: &Path, source: &Path, target: &Path) -> Result<(), RuntimeError> {
+    fs::create_dir(target)
+        .map_err(|error| runtime_io_error("create migration staging", target, error))?;
+    let mut entries = fs::read_dir(source)
+        .map_err(|error| runtime_io_error("read migration source", source, error))?
+        .collect::<Result<Vec<_>, _>>()
+        .map_err(|error| runtime_io_error("read migration source", source, error))?;
+    entries.sort_by_key(|entry| entry.file_name());
+    for entry in entries {
+        let source_path = entry.path();
+        let target_path = target.join(entry.file_name());
+        let file_type = entry
+            .file_type()
+            .map_err(|error| runtime_io_error("inspect migration source", &source_path, error))?;
+        if file_type.is_dir() {
+            copy_runtime_tree(root, &source_path, &target_path)?;
+        } else if file_type.is_file() {
+            fs::copy(&source_path, &target_path)
+                .map_err(|error| runtime_io_error("copy migration source", &source_path, error))?;
+        } else if runtime_ephemeral_socket(root, &source_path)? {
+            continue;
+        } else {
+            return Err(runtime_store_corrupt(
+                &source_path,
+                "Runtime migration refuses unknown symlinks and special files".to_string(),
+            ));
+        }
+    }
+    Ok(())
+}
+
+fn migrate_runtime_store(
+    root: &Path,
+    runtime_id: &str,
+) -> Result<FsRuntimeStoreMigrationPlan, RuntimeError> {
+    let (plan, _) = plan_runtime_store_migration(root, runtime_id)?;
+    if !plan.migration_required {
+        return Ok(plan);
+    }
+    let staging = migration_sibling(root, "schema-v3-staging")?;
+    let backup = migration_sibling(root, "pre-schema-v3-backup")?;
+    if staging.exists() || backup.exists() {
+        return Err(runtime_store_corrupt(
+            root,
+            format!(
+                "unfinished Runtime migration artifact exists (staging={}, backup={}); recover or remove it before retrying",
+                staging.display(),
+                backup.display()
+            ),
+        ));
+    }
+    if let Err(error) = copy_runtime_tree(root, root, &staging) {
+        let _ = fs::remove_dir_all(&staging);
+        return Err(error);
+    }
+    let staged_plan = match migrate_runtime_store_in_place(&staging, runtime_id) {
+        Ok(plan) => plan,
+        Err(error) => {
+            let _ = fs::remove_dir_all(&staging);
+            return Err(error);
+        }
+    };
+    let staged_store = FsRuntimeStore {
+        root: staging.clone(),
+    };
+    if let Err(error) = staged_store.load_runtime_state() {
+        let _ = fs::remove_dir_all(&staging);
+        return Err(error);
+    }
+    fs::rename(root, &backup)
+        .map_err(|error| runtime_io_error("backup runtime store", root, error))?;
+    if let Err(error) = fs::rename(&staging, root) {
+        let rollback = fs::rename(&backup, root);
+        return match rollback {
+            Ok(()) => Err(runtime_io_error(
+                "activate migrated runtime store",
+                &staging,
+                error,
+            )),
+            Err(rollback_error) => Err(runtime_store_corrupt(
+                root,
+                format!(
+                    "activate migrated Runtime store failed: {error}; rollback failed: {rollback_error}; backup remains at {}",
+                    backup.display()
+                ),
+            )),
+        };
+    }
+    fs::remove_dir_all(&backup)
+        .map_err(|error| runtime_io_error("remove runtime migration backup", &backup, error))?;
+    debug_assert_eq!(plan.mapping_digest, staged_plan.mapping_digest);
+    Ok(plan)
+}
+
+fn migrate_runtime_store_in_place(
+    root: &Path,
+    runtime_id: &str,
+) -> Result<FsRuntimeStoreMigrationPlan, RuntimeError> {
+    let (plan, planned_workers) = plan_runtime_store_migration(root, runtime_id)?;
+    if !plan.migration_required {
+        return Ok(plan);
+    }
+    let runtime_path = root.join(RUNTIME_FILE);
+    let bytes =
+        fs::read(&runtime_path).map_err(|error| runtime_io_error("read", &runtime_path, error))?;
+    let document: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
+        runtime_store_corrupt(
+            &runtime_path,
+            format!("decode Runtime state {}: {error}", runtime_path.display()),
+        )
+    })?;
+
+    for planned_worker in &planned_workers {
+        let source_dir = &planned_worker.source_dir;
+        let source_snapshot_path = source_dir.join(WORKER_FILE);
+        let bytes = fs::read(&source_snapshot_path)
+            .map_err(|error| runtime_io_error("read", &source_snapshot_path, error))?;
+        let snapshot: serde_json::Value = serde_json::from_slice(&bytes).map_err(|error| {
+            runtime_store_corrupt(
+                &source_snapshot_path,
+                format!(
+                    "decode Worker snapshot {}: {error}",
+                    source_snapshot_path.display()
+                ),
+            )
+        })?;
+        let snapshot = migrate_worker_document(
+            snapshot,
+            plan.current_schema_version,
+            planned_worker.legacy_mapping.as_ref(),
+            &source_snapshot_path,
+        )?;
+        let metadata_path = source_dir.join(WORKER_METADATA_FILE);
+        let metadata = if metadata_path.is_file() {
+            let metadata: serde_json::Value =
+                read_json(&metadata_path, "read Worker aggregate metadata")?;
+            Some(
+                migrate_worker_aggregate_document(
+                    metadata,
+                    planned_worker,
+                    runtime_id,
+                    &metadata_path,
+                )?
+                .0,
+            )
+        } else {
+            None
+        };
+
+        let worker_id_text = planned_worker.worker_id.to_string();
         let migrated_dir = root.join("workers").join(&worker_id_text);
-        fs::rename(&legacy_dir, &migrated_dir)
-            .map_err(|error| runtime_io_error("rename", &legacy_dir, error))?;
+        if source_dir != &migrated_dir {
+            fs::rename(source_dir, &migrated_dir)
+                .map_err(|error| runtime_io_error("rename", source_dir, error))?;
+        }
         let migrated_snapshot_path = migrated_dir.join(WORKER_FILE);
         atomic_write_json(
             &migrated_snapshot_path,
             &snapshot,
             "migrate Worker identity",
         )?;
-        migrated_ids.push(serde_json::Value::String(worker_id_text));
+        if let Some(metadata) = metadata {
+            atomic_write_json(
+                &migrated_dir.join(WORKER_METADATA_FILE),
+                &metadata,
+                "migrate Worker aggregate identity",
+            )?;
+        }
     }
 
-    document["schema_version"] = serde_json::Value::from(SCHEMA_VERSION);
-    document["workers"] = serde_json::Value::Array(migrated_ids);
-    if let Some(object) = document.as_object_mut() {
-        object.remove("next_worker_sequence");
-    }
+    let (document, _) = migrate_runtime_document(
+        document,
+        plan.current_schema_version,
+        &plan.mappings,
+        &runtime_path,
+    )?;
     atomic_write_json(
         &runtime_path,
         &document,
         "migrate Runtime Worker identities",
-    )
+    )?;
+    Ok(plan)
 }
 
 #[derive(Clone, Debug, Serialize, Deserialize)]

crates/worker-runtime/src/identity.rs

@@ -1,4 +1,5 @@
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
+use sha2::{Digest, Sha256};
 use std::{fmt, str::FromStr};
 use uuid::{Uuid, Version};
 
@@ -27,8 +28,6 @@ impl WorkerId {
     }
 
     pub fn from_legacy_binding(workspace_id: &str, runtime_id: &str, value: u64) -> Self {
-        use sha2::{Digest, Sha256};
-
         let mut hasher = Sha256::new();
         hasher.update(b"yoi.workspace-worker-id.v1\0");
         hasher.update(workspace_id.as_bytes());
@@ -101,6 +100,45 @@ impl fmt::Display for WorkerIdParseError {
 
 impl std::error::Error for WorkerIdParseError {}
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+pub struct LegacyWorkerIdentityMapping {
+    pub workspace_id: String,
+    pub runtime_id: String,
+    pub legacy_worker_id: u64,
+    pub worker_id: WorkerId,
+}
+
+pub fn legacy_worker_identity_mapping_digest(mappings: &[LegacyWorkerIdentityMapping]) -> String {
+    let mut mappings = mappings.to_vec();
+    mappings.sort_by(|left, right| {
+        (
+            left.workspace_id.as_str(),
+            left.runtime_id.as_str(),
+            left.legacy_worker_id,
+            left.worker_id,
+        )
+            .cmp(&(
+                right.workspace_id.as_str(),
+                right.runtime_id.as_str(),
+                right.legacy_worker_id,
+                right.worker_id,
+            ))
+    });
+    let mut hasher = Sha256::new();
+    hasher.update(b"yoi.workspace-worker-migration-plan.v1\0");
+    for mapping in mappings {
+        hasher.update(mapping.workspace_id.as_bytes());
+        hasher.update([0]);
+        hasher.update(mapping.runtime_id.as_bytes());
+        hasher.update([0]);
+        hasher.update(mapping.legacy_worker_id.to_be_bytes());
+        hasher.update(mapping.worker_id.to_string().as_bytes());
+        hasher.update([b'\n']);
+    }
+    let digest = hasher.finalize();
+    digest.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
 /// Runtime-local authority reference for Worker operations. The contained id is
 /// nevertheless the Workspace-owned stable identity; the Runtime does not mint it.
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]

crates/worker-runtime/src/main.rs

@@ -18,7 +18,7 @@ use worker_runtime::auth::{
     RuntimeHttpAuthConfig, RuntimeIdentityMaterial, TrustedServerKey, decode_public_key,
 };
 use worker_runtime::error::RuntimeError;
-use worker_runtime::fs_store::FsRuntimeStoreOptions;
+use worker_runtime::fs_store::{FsRuntimeStore, FsRuntimeStoreOptions};
 use worker_runtime::http_server::{
     RuntimeHttpServerConfig, RuntimeHttpServerError, RuntimeHttpStoreSelection,
 };
@@ -44,6 +44,9 @@ fn main() -> ExitCode {
 
 fn run() -> Result<(), ProcessError> {
     let args = env::args().skip(1).collect::<Vec<_>>();
+    if matches!(args.first().map(String::as_str), Some("migrate")) {
+        return run_migration_command(args);
+    }
     if matches!(
         args.first().map(String::as_str),
         Some("identity" | "trust-server")
@@ -78,6 +81,76 @@ fn run() -> Result<(), ProcessError> {
     Ok(())
 }
 
+fn run_migration_command(mut args: Vec<String>) -> Result<(), ProcessError> {
+    args.remove(0);
+    let dry_run_index = args
+        .iter()
+        .position(|argument| argument == "--dry-run")
+        .ok_or_else(|| ProcessError::usage("migrate currently requires --dry-run".to_string()))?;
+    args.remove(dry_run_index);
+    let explicit_runtime_id =
+        if let Some(index) = args.iter().position(|argument| argument == "--runtime-id") {
+            if index + 1 >= args.len() {
+                return Err(ProcessError::usage(
+                    "--runtime-id requires a value".to_string(),
+                ));
+            }
+            let runtime_id = args.remove(index + 1);
+            args.remove(index);
+            Some(runtime_id)
+        } else {
+            None
+        };
+    let config = parse_args(args)?.ok_or_else(|| {
+        ProcessError::usage("migrate requires a Runtime store configuration".to_string())
+    })?;
+    let root = match &config.http.store {
+        RuntimeHttpStoreSelection::Fs { root } => root.clone(),
+        RuntimeHttpStoreSelection::Memory => {
+            return Err(ProcessError::usage(
+                "migration dry-run requires the fs Runtime store".to_string(),
+            ));
+        }
+        _ => {
+            return Err(ProcessError::usage(
+                "unsupported Runtime catalog store selection".to_string(),
+            ));
+        }
+    };
+    let persisted_runtime_id = read_runtime_auth_file(&runtime_auth_path(&config))?
+        .identity
+        .map(|identity| identity.identity_id);
+    let runtime_id = match (persisted_runtime_id, explicit_runtime_id) {
+        (Some(persisted), Some(explicit)) if persisted != explicit => {
+            return Err(ProcessError::usage(format!(
+                "--runtime-id {explicit} does not match persisted Runtime identity {persisted}"
+            )));
+        }
+        (Some(persisted), _) => persisted,
+        (None, Some(explicit)) if !explicit.is_empty() => explicit,
+        (None, Some(_)) => {
+            return Err(ProcessError::usage(
+                "--runtime-id must not be empty".to_string(),
+            ));
+        }
+        (None, None) => {
+            return Err(ProcessError::usage(
+                "migration dry-run requires a persisted Runtime identity or explicit --runtime-id"
+                    .to_string(),
+            ));
+        }
+    };
+    let mut options = FsRuntimeStoreOptions::new(root).with_runtime_id(runtime_id);
+    options.display_name = config.http.display_name.clone();
+    let plan = FsRuntimeStore::migration_plan(&options).map_err(ProcessError::Runtime)?;
+    println!(
+        "{}",
+        serde_json::to_string_pretty(&plan)
+            .map_err(|error| ProcessError::Auth(format!("encode migration plan: {error}")))?
+    );
+    Ok(())
+}
+
 fn build_runtime(config: &ProcessConfig) -> Result<Runtime, ProcessError> {
     let fs_paths = config.resolved_fs_paths();
     let runtime_store_dir = match &config.http.store {
@@ -791,6 +864,7 @@ fn run_trust_server_command(mut args: VecDeque<String>) -> Result<(), ProcessErr
 
 fn usage() -> &'static str {
     r#"Usage: yoi-runtime [OPTIONS]
+       yoi-runtime migrate --dry-run [--runtime-id <ID>] [OPTIONS]
 
 Starts a worker-backed Runtime REST command API for a trusted backend/proxy.
 Browsers must not connect to this Runtime process directly.
@@ -911,6 +985,87 @@ mod tests {
         assert_eq!(paths.workdir_target, PathBuf::from("/tmp/yoi-workdirs"));
     }
 
+    #[test]
+    fn migration_dry_run_accepts_real_v1_document_without_workers_field() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        std::fs::create_dir_all(root.join("workers")).unwrap();
+        std::fs::write(
+            root.join("runtime.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "schema_version": 1,
+                "display_name": "local",
+                "backend": "fs_store",
+                "status": "running",
+                "next_diagnostic_id": 1,
+                "config_bundles": {},
+                "workspace_owners": {},
+                "assignments": [],
+                "execution": [],
+                "diagnostics": []
+            }))
+            .unwrap(),
+        )
+        .unwrap();
+        let before = std::fs::read(root.join("runtime.json")).unwrap();
+        run_migration_command(vec![
+            "migrate".to_string(),
+            "--dry-run".to_string(),
+            "--runtime-id".to_string(),
+            "local".to_string(),
+            "--store".to_string(),
+            "fs".to_string(),
+            "--fs-root".to_string(),
+            temp.path().display().to_string(),
+            "--fs-runtime-dir".to_string(),
+            root.display().to_string(),
+        ])
+        .unwrap();
+        assert_eq!(std::fs::read(root.join("runtime.json")).unwrap(), before);
+    }
+
+    #[test]
+    fn migration_dry_run_rejects_v1_document_that_cannot_decode_as_v3() {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("runtime");
+        std::fs::create_dir_all(root.join("workers")).unwrap();
+        std::fs::write(
+            root.join("runtime.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "schema_version": 1,
+                "display_name": "local",
+                "backend": "fs_store",
+                "status": 3,
+                "next_diagnostic_id": 1,
+                "config_bundles": {},
+                "workspace_owners": {},
+                "diagnostics": []
+            }))
+            .unwrap(),
+        )
+        .unwrap();
+        let before = std::fs::read(root.join("runtime.json")).unwrap();
+        let error = run_migration_command(vec![
+            "migrate".to_string(),
+            "--dry-run".to_string(),
+            "--runtime-id".to_string(),
+            "local".to_string(),
+            "--store".to_string(),
+            "fs".to_string(),
+            "--fs-root".to_string(),
+            temp.path().display().to_string(),
+            "--fs-runtime-dir".to_string(),
+            root.display().to_string(),
+        ])
+        .unwrap_err();
+        assert!(
+            error
+                .to_string()
+                .contains("decode migrated Runtime snapshot")
+        );
+        assert_eq!(std::fs::read(root.join("runtime.json")).unwrap(), before);
+    }
+
     #[test]
     fn no_store_disables_runtime_catalog_persistence() {
         let config = parse_args(["--no-store"]).unwrap().unwrap();

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(),
             },
         );
@@ -4181,37 +4187,196 @@ mod tests {
             serde_json::to_vec_pretty(&worker_json).unwrap(),
         )
         .unwrap();
+        let legacy_worker_name = "worker-runtime-7";
+        let legacy_manifest = manifest::WorkerManifest::from_toml(&format!(
+            r#"
+                [worker]
+                name = "{legacy_worker_name}"
+
+                [model]
+                scheme = "anthropic"
+                model_id = "test-model"
+
+                [engine]
+
+                [[scope.allow]]
+                target = "/tmp"
+                permission = "write"
+            "#,
+        ))
+        .unwrap();
+        std::fs::write(
+            legacy_dir.join("metadata.json"),
+            serde_json::to_vec_pretty(&serde_json::json!({
+                "worker_name": legacy_worker_name,
+                "workspace_id": "workspace-a",
+                "resolved_manifest_snapshot": legacy_manifest
+            }))
+            .unwrap(),
+        )
+        .unwrap();
 
         let runtime_path = root.join("runtime.json");
         let mut runtime_json: serde_json::Value =
             serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap();
         runtime_json["schema_version"] = serde_json::json!(1);
-        runtime_json["workers"] = serde_json::json!([7]);
+        runtime_json["workers"] = serde_json::json!({"legacy": "ignored"});
         runtime_json["next_worker_sequence"] = serde_json::json!(8);
+        runtime_json["next_diagnostic_id"] = serde_json::json!(3);
+        runtime_json["diagnostics"] = serde_json::json!([
+            {
+                "id": 1,
+                "severity": "warning",
+                "code": "mapped_legacy_worker",
+                "message": "mapped diagnostic",
+                "worker_ref": {"worker_id": 7}
+            },
+            {
+                "id": 2,
+                "severity": "warning",
+                "code": "deleted_legacy_worker",
+                "message": "unmapped diagnostic",
+                "worker_ref": {"worker_id": 6}
+            }
+        ]);
         std::fs::write(
             &runtime_path,
             serde_json::to_vec_pretty(&runtime_json).unwrap(),
         )
         .unwrap();
+        #[cfg(unix)]
+        {
+            let run_dir = legacy_dir.join("runs").join("6");
+            std::fs::create_dir_all(&run_dir).unwrap();
+            let socket =
+                std::os::unix::net::UnixListener::bind(run_dir.join("worker.sock")).unwrap();
+            drop(socket);
+        }
 
-        let restored = Runtime::with_fs_store(crate::fs_store::FsRuntimeStoreOptions {
+        let runtime_options = crate::fs_store::FsRuntimeStoreOptions {
             root: root.clone(),
             runtime_id: runtime_id.to_string(),
             display_name: None,
-        })
-        .unwrap();
+        };
+        let runtime_before_dry_run = std::fs::read(&runtime_path).unwrap();
+        let plan = crate::fs_store::FsRuntimeStore::migration_plan(&runtime_options).unwrap();
+        assert!(plan.migration_required);
+        assert_eq!(plan.worker_count, 1);
+        assert_eq!(plan.migrated_worker_aggregate_count, 1);
+        assert_eq!(plan.migrated_diagnostic_worker_ref_count, 1);
+        assert_eq!(plan.cleared_diagnostic_worker_ref_count, 1);
+        assert_eq!(plan.mappings[0].legacy_worker_id, 7);
+        #[cfg(unix)]
+        assert_eq!(
+            plan.excluded_ephemeral_paths,
+            vec!["workers/7/runs/6/worker.sock"]
+        );
+        assert_eq!(
+            std::fs::read(&runtime_path).unwrap(),
+            runtime_before_dry_run
+        );
+        assert!(legacy_dir.exists());
+
+        let restored = Runtime::with_fs_store(runtime_options.clone()).unwrap();
         let expected = WorkerId::from_legacy_binding("workspace-a", runtime_id, 7);
         let detail = restored.worker_detail(&WorkerRef::new(expected)).unwrap();
         assert_eq!(detail.worker_id, expected);
         assert_eq!(detail.worker_ref.worker_id, expected);
-        assert!(root.join("workers").join(expected.to_string()).exists());
+        let expected_worker_dir = root.join("workers").join(expected.to_string());
+        assert!(expected_worker_dir.exists());
+        #[cfg(unix)]
+        assert!(!expected_worker_dir.join("runs/6/worker.sock").exists());
         assert!(!legacy_dir.exists());
         let migrated_runtime: serde_json::Value =
-            serde_json::from_slice(&std::fs::read(runtime_path).unwrap()).unwrap();
-        assert_eq!(migrated_runtime["schema_version"], serde_json::json!(2));
+            serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap();
+        assert_eq!(migrated_runtime["schema_version"], serde_json::json!(3));
+        assert!(migrated_runtime.get("workers").is_none());
         assert!(migrated_runtime.get("next_worker_sequence").is_none());
+        assert_eq!(
+            migrated_runtime["diagnostics"][0]["worker_ref"]["worker_id"],
+            serde_json::json!(expected.to_string())
+        );
+        assert!(
+            migrated_runtime["diagnostics"][1]
+                .get("worker_ref")
+                .is_none()
+        );
+        let diagnostics = restored.diagnostics().unwrap();
+        assert_eq!(
+            diagnostics
+                .iter()
+                .find(|diagnostic| diagnostic.code == "mapped_legacy_worker")
+                .and_then(|diagnostic| diagnostic.worker_ref.as_ref()),
+            Some(&WorkerRef::new(expected))
+        );
+        assert!(
+            diagnostics
+                .iter()
+                .find(|diagnostic| diagnostic.code == "deleted_legacy_worker")
+                .is_some_and(|diagnostic| diagnostic.worker_ref.is_none())
+        );
+        let metadata_path = expected_worker_dir.join("metadata.json");
+        let mut migrated_metadata: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&metadata_path).unwrap()).unwrap();
+        let expected_worker_name = format!("worker-runtime-{expected}");
+        assert_eq!(
+            migrated_metadata["worker_name"],
+            serde_json::json!(expected_worker_name)
+        );
+        assert_eq!(
+            migrated_metadata["resolved_manifest_snapshot"]["worker"]["name"],
+            serde_json::json!(expected_worker_name)
+        );
 
         drop(restored);
+
+        let mut schema_v2_runtime: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&runtime_path).unwrap()).unwrap();
+        schema_v2_runtime["schema_version"] = serde_json::json!(2);
+        std::fs::write(
+            &runtime_path,
+            serde_json::to_vec_pretty(&schema_v2_runtime).unwrap(),
+        )
+        .unwrap();
+        let migrated_worker_path = expected_worker_dir.join("worker.json");
+        let mut schema_v2_worker: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(&migrated_worker_path).unwrap()).unwrap();
+        schema_v2_worker["schema_version"] = serde_json::json!(2);
+        std::fs::write(
+            &migrated_worker_path,
+            serde_json::to_vec_pretty(&schema_v2_worker).unwrap(),
+        )
+        .unwrap();
+        migrated_metadata["worker_name"] = serde_json::json!(legacy_worker_name);
+        migrated_metadata["resolved_manifest_snapshot"]["worker"]["name"] =
+            serde_json::json!(legacy_worker_name);
+        std::fs::write(
+            &metadata_path,
+            serde_json::to_vec_pretty(&migrated_metadata).unwrap(),
+        )
+        .unwrap();
+
+        let recovery_plan =
+            crate::fs_store::FsRuntimeStore::migration_plan(&runtime_options).unwrap();
+        assert_eq!(recovery_plan.current_schema_version, 2);
+        assert_eq!(recovery_plan.target_schema_version, 3);
+        assert!(recovery_plan.migration_required);
+        assert_eq!(recovery_plan.worker_count, 1);
+        assert_eq!(recovery_plan.migrated_worker_aggregate_count, 1);
+        assert!(recovery_plan.mappings.is_empty());
+
+        let recovered = Runtime::with_fs_store(runtime_options).unwrap();
+        let recovered_metadata: serde_json::Value =
+            serde_json::from_slice(&std::fs::read(metadata_path).unwrap()).unwrap();
+        assert_eq!(
+            recovered_metadata["worker_name"],
+            serde_json::json!(expected_worker_name)
+        );
+        assert_eq!(
+            recovered_metadata["resolved_manifest_snapshot"]["worker"]["name"],
+            serde_json::json!(expected_worker_name)
+        );
+        drop(recovered);
         let _ = std::fs::remove_dir_all(root);
     }
 

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<dyn WorkdirSession>,
+    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.

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<TrackedBlock>,
+    commands: Vec<CommandSnapshot>,
 }
 
 #[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<CommandSnapshot>) {
+        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);

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,10 @@ 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());
     worker.attach_in_flight_events(in_flight.clone());
@@ -587,6 +591,7 @@ pub(crate) async fn prepare_internal_worker_session(
         while let Some(command) = command_rx.recv().await {
             match command {
                 InternalWorkerSessionCommand::Run(input) => {
+                    actor_in_flight.clear();
                     let cancel_sender = worker.engine_mut().cancel_sender();
                     let mut run = std::pin::pin!(worker.run_text(&input));
                     loop {
@@ -599,6 +604,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                         Some(error.to_string()),
                                     ),
                                 };
+                                actor_in_flight.clear();
                                 status.store(turn_status.encode(), std::sync::atomic::Ordering::Release);
                                 if let Some(message) = error {
                                     *last_error.lock().unwrap() = Some(message.clone());
@@ -622,6 +628,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                     Some(InternalWorkerSessionCommand::Stop(done)) => {
                                         let _ = cancel_sender.send(()).await;
                                         let _ = (&mut run).await;
+                                        actor_in_flight.clear();
                                         status.store(InternalWorkerSessionStatus::Stopped.encode(), std::sync::atomic::Ordering::Release);
                                         let _ = event_tx.send(Event::Status { status: WorkerStatus::Paused });
                                         let _ = event_tx.send(Event::Shutdown);
@@ -634,6 +641,7 @@ pub(crate) async fn prepare_internal_worker_session(
                                     }
                                     None => {
                                         let _ = cancel_sender.send(()).await;
+                                        actor_in_flight.clear();
                                         return;
                                     }
                                 }
@@ -642,6 +650,7 @@ pub(crate) async fn prepare_internal_worker_session(
                     }
                 }
                 InternalWorkerSessionCommand::Stop(done) => {
+                    actor_in_flight.clear();
                     status.store(
                         InternalWorkerSessionStatus::Stopped.encode(),
                         std::sync::atomic::Ordering::Release,
@@ -656,6 +665,7 @@ pub(crate) async fn prepare_internal_worker_session(
                 }
             }
         }
+        actor_in_flight.clear();
     });
 
     Ok(handle)
@@ -1068,6 +1078,10 @@ permission = "write"
             handle.wait_until_idle().await,
             InternalWorkerSessionStatus::Idle
         );
+        handle
+            .in_flight
+            .tool_call_start("stale-call".to_string(), "Read".to_string());
+        assert_eq!(handle.protocol_snapshot().in_flight.blocks.len(), 1);
         let entries_after_first = handle.entries().len();
         assert!(entries_after_first >= 4);
         handle.send("follow-up").await.expect("send follow-up turn");
@@ -1075,6 +1089,7 @@ permission = "write"
             handle.wait_until_idle().await,
             InternalWorkerSessionStatus::Idle
         );
+        assert!(handle.protocol_snapshot().in_flight.blocks.is_empty());
         assert_eq!(calls.load(Ordering::SeqCst), 2);
         assert!(handle.entries().len() > entries_after_first);
 

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(_))

crates/workspace-server/src/main.rs

@@ -23,6 +23,7 @@ enum Command {
     ConfigDiff(WorkspacePathOptions),
     Identity(Vec<String>),
     TrustRuntime(Vec<String>),
+    MigrateDryRun { database: Option<PathBuf> },
     Skills(SkillsCommand),
     Help,
 }
@@ -85,6 +86,17 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
         Command::ConfigDiff(options) => run_config_diff(options),
         Command::Identity(args) => run_identity_command(args),
         Command::TrustRuntime(args) => run_trust_runtime_command(args),
+        Command::MigrateDryRun { database } => {
+            let database = database.unwrap_or_else(ServerConfig::default_server_database_path);
+            let plan = SqliteWorkspaceStore::migration_plan(&database).map_err(|error| {
+                CliError(format!(
+                    "migration dry-run failed for {}: {error}",
+                    database.display()
+                ))
+            })?;
+            println!("{}", serde_json::to_string_pretty(&plan)?);
+            Ok(())
+        }
         Command::Skills(command) => run_skills(command),
         Command::Help => Ok(()),
     }
@@ -107,6 +119,7 @@ fn parse_command(args: &[String]) -> Result<Command, CliError> {
         "config" => parse_config_command(rest),
         "identity" => Ok(Command::Identity(rest.to_vec())),
         "trust-runtime" => Ok(Command::TrustRuntime(rest.to_vec())),
+        "migrate" => parse_migrate_command(rest),
         "skills" => parse_skills_command(rest),
         "serve" => {
             if rest.iter().any(|arg| arg == "--help" || arg == "-h") {
@@ -120,7 +133,7 @@ fn parse_command(args: &[String]) -> Result<Command, CliError> {
             Ok(Command::Help)
         }
         other => Err(CliError(format!(
-            "unknown command `{other}`; expected `init`, `config`, `identity`, `trust-runtime`, `skills`, or `serve`"
+            "unknown command `{other}`; expected `init`, `config`, `identity`, `trust-runtime`, `migrate`, `skills`, or `serve`"
         ))),
     }
 }
@@ -718,6 +731,32 @@ fn parse_config_command(args: &[String]) -> Result<Command, CliError> {
     }
 }
 
+fn parse_migrate_command(args: &[String]) -> Result<Command, CliError> {
+    let mut dry_run = false;
+    let mut database = None;
+    let mut index = 0;
+    while index < args.len() {
+        match args[index].as_str() {
+            "--dry-run" => dry_run = true,
+            "--database" => {
+                index += 1;
+                database =
+                    Some(PathBuf::from(args.get(index).ok_or_else(|| {
+                        CliError("--database requires a path".to_string())
+                    })?));
+            }
+            value => {
+                return Err(CliError(format!("unknown migrate option: {value}")));
+            }
+        }
+        index += 1;
+    }
+    if !dry_run {
+        return Err(CliError("migrate currently requires --dry-run".to_string()));
+    }
+    Ok(Command::MigrateDryRun { database })
+}
+
 fn parse_skills_command(args: &[String]) -> Result<Command, CliError> {
     let Some((subcommand, rest)) = args.split_first() else {
         print_skills_help();
@@ -875,7 +914,8 @@ fn parse_listen(value: &str) -> Result<SocketAddr, CliError> {
 
 fn print_help() {
     println!(
-        "yoi-server\n\nUsage:\n  yoi-server init [OPTIONS]\n  yoi-server config <COMMAND> [OPTIONS]\n  yoi-server identity init --server-id <SERVER_ID> [--replace]\n  yoi-server identity show [--json]\n  yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n  yoi-server trust-runtime list [--json] [--include-revoked]\n  yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n  yoi-server skills <COMMAND> [OPTIONS]\n  yoi-server serve [OPTIONS]\n\nOptions:\n  -h, --help    Print help"
+        "yoi-server\n\nUsage:\n  yoi-server init [OPTIONS]\n  yoi-server config <COMMAND> [OPTIONS]\n  yoi-server identity init --server-id <SERVER_ID> [--replace]\n  yoi-server identity show [--json]\n  yoi-server trust-runtime add --runtime-id <RUNTIME_ID> --base-url <URL> --public-key <KEY> [--display-name <NAME>] [--replace]\n  yoi-server trust-runtime list [--json] [--include-revoked]\n  yoi-server trust-runtime revoke --runtime-id <RUNTIME_ID>\n  yoi-server skills <COMMAND> [OPTIONS]\n  yoi-server migrate --dry-run [--database <PATH>]
+  yoi-server serve [OPTIONS]\n\nOptions:\n  -h, --help    Print help"
     );
 }
 
@@ -899,7 +939,8 @@ fn print_skills_help() {
 
 fn print_serve_help() {
     println!(
-        "yoi-server serve\n\nUsage:\n  yoi-server serve [OPTIONS]\n\nDescription:\n  Serves the Workspace recorded in the Yoi server DB. Workspace records are stored in the XDG/Yoi data directory, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n      --listen <ADDR>     Listen address (default 127.0.0.1:8787)\n  -h, --help              Print help"
+        "yoi-server serve\n\nUsage:\n  yoi-server migrate --dry-run [--database <PATH>]
+  yoi-server serve [OPTIONS]\n\nDescription:\n  Serves the Workspace recorded in the Yoi server DB. Workspace records are stored in the XDG/Yoi data directory, and runtime sources are loaded from XDG runtimes.toml.\n\nOptions:\n      --listen <ADDR>     Listen address (default 127.0.0.1:8787)\n  -h, --help              Print help"
     );
 }
 
@@ -939,6 +980,22 @@ mod tests {
         assert_eq!(name, "debug-rust");
     }
 
+    #[test]
+    fn parse_migrate_requires_dry_run_and_accepts_database_path() {
+        let error = parse_migrate_command(&[]).unwrap_err();
+        assert_eq!(error.to_string(), "migrate currently requires --dry-run");
+        let command = parse_migrate_command(&[
+            "--dry-run".to_string(),
+            "--database".to_string(),
+            "/tmp/server.db".to_string(),
+        ])
+        .unwrap();
+        let Command::MigrateDryRun { database } = command else {
+            panic!("expected migration dry-run command");
+        };
+        assert_eq!(database, Some(PathBuf::from("/tmp/server.db")));
+    }
+
     #[test]
     fn parse_serve_accepts_listen_only() {
         let args = vec!["--listen".to_string(), "127.0.0.1:0".to_string()];

crates/workspace-server/src/retention.rs

@@ -166,6 +166,25 @@ pub enum WorkerRetentionError {
     Invalid(String),
 }
 
+pub(crate) fn repair_worker_diagnostics_archive_table(conn: &Connection) -> crate::Result<bool> {
+    let existed: bool = conn.query_row(
+        "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='worker_diagnostics_archives')",
+        [],
+        |row| row.get(0),
+    )?;
+    if !existed {
+        conn.execute_batch(
+            "CREATE TABLE worker_diagnostics_archives (
+              operation_id TEXT PRIMARY KEY, workspace_id TEXT NOT NULL, runtime_id TEXT NOT NULL,
+              worker_id TEXT NOT NULL, policy_id TEXT NOT NULL, policy_revision INTEGER NOT NULL,
+              committed_at TEXT NOT NULL, expires_at TEXT NOT NULL,
+              FOREIGN KEY(operation_id) REFERENCES worker_removal_operations(operation_id),
+              FOREIGN KEY(workspace_id) REFERENCES workspaces(workspace_id) ON DELETE CASCADE);",
+        )?;
+    }
+    Ok(!existed)
+}
+
 pub(crate) fn create_worker_retention_tables(conn: &Connection) -> crate::Result<()> {
     conn.execute_batch(r#"
       CREATE TABLE workspace_worker_retention_policy_revisions (

crates/workspace-server/src/store.rs

@@ -4,11 +4,15 @@ use std::time::Duration;
 
 use async_trait::async_trait;
 use flow::{CompiledFlowDefinition, FlowSourceKind, compile_flow_source};
-use rusqlite::{Connection, OptionalExtension, TransactionBehavior, params};
+use rusqlite::{
+    Connection, OpenFlags, OptionalExtension, TransactionBehavior, backup::Backup, params,
+};
 use serde::{Deserialize, Serialize};
 use uuid::Uuid;
 
-use worker_runtime::identity::{RuntimeWorkerRef, WorkerId};
+use worker_runtime::identity::{
+    LegacyWorkerIdentityMapping, RuntimeWorkerRef, WorkerId, legacy_worker_identity_mapping_digest,
+};
 
 use crate::{Error, Result};
 
@@ -204,7 +208,7 @@ const MIGRATIONS: &[Migration] = &[
     Migration {
         version: 37,
         name: "promote Workspace Worker UUIDv7 identity",
-        apply: promote_workspace_worker_uuid_identity,
+        apply: apply_workspace_worker_uuid_identity_migration,
     },
     Migration {
         version: 38,
@@ -219,6 +223,17 @@ struct Migration {
     apply: fn(&Connection) -> Result<()>,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+pub struct WorkspaceStoreMigrationPlan {
+    pub current_schema_version: i64,
+    pub target_schema_version: i64,
+    pub migration_required: bool,
+    pub worker_count: usize,
+    pub mapping_digest: String,
+    pub mappings: Vec<LegacyWorkerIdentityMapping>,
+    pub repairs: Vec<String>,
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct WorkspaceRecord {
     pub workspace_id: String,
@@ -976,6 +991,66 @@ pub struct SqliteWorkspaceStore {
 }
 
 impl SqliteWorkspaceStore {
+    pub fn migration_plan(path: impl AsRef<Path>) -> Result<WorkspaceStoreMigrationPlan> {
+        let path = path.as_ref();
+        let source = Connection::open_with_flags(
+            path,
+            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_URI,
+        )?;
+        let current_schema_version = current_schema_version(&source)?;
+        let target_schema_version = MIGRATIONS
+            .last()
+            .map(|migration| i64::from(migration.version))
+            .unwrap_or(current_schema_version);
+        let mut repairs = Vec::new();
+        if current_schema_version < 37 && !table_exists(&source, "worker_diagnostics_archives")? {
+            repairs.push("create missing worker_diagnostics_archives table".to_string());
+        }
+
+        let mut candidate = Connection::open_in_memory()?;
+        {
+            let backup = Backup::new(&source, &mut candidate)?;
+            backup.run_to_completion(5, Duration::from_millis(10), None)?;
+        }
+        configure_sqlite(&candidate)?;
+        let mappings = if current_schema_version < 37 {
+            apply_migrations_through(&candidate, 36)?;
+            let tx = candidate.unchecked_transaction()?;
+            crate::retention::repair_worker_diagnostics_archive_table(&tx)?;
+            let mappings = promote_workspace_worker_uuid_identity(&tx)?;
+            tx.execute(
+                "INSERT INTO __yoi_schema_migrations (version, name) VALUES (37, ?1)",
+                ["promote Workspace Worker UUIDv7 identity"],
+            )?;
+            tx.commit()?;
+            mappings
+        } else {
+            Vec::new()
+        };
+        apply_migrations_through(&candidate, i64::MAX)?;
+        ticket::migrate_sqlite_ticket_schema(&candidate)?;
+        merge_request::migrate(&candidate).map_err(|error| Error::Store(error.to_string()))?;
+        validate_workspace_repository_references(&candidate)?;
+        let foreign_key_failures: i64 =
+            candidate.query_row("SELECT COUNT(*) FROM pragma_foreign_key_check", [], |row| {
+                row.get(0)
+            })?;
+        if foreign_key_failures != 0 {
+            return Err(Error::Store(format!(
+                "migration dry-run found {foreign_key_failures} foreign key violation(s)"
+            )));
+        }
+        Ok(WorkspaceStoreMigrationPlan {
+            current_schema_version,
+            target_schema_version,
+            migration_required: current_schema_version < target_schema_version,
+            worker_count: mappings.len(),
+            mapping_digest: legacy_worker_identity_mapping_digest(&mappings),
+            mappings,
+            repairs,
+        })
+    }
+
     pub fn open(path: impl AsRef<Path>) -> Result<Self> {
         let conn = Connection::open(path)?;
         Self::from_connection(conn)
@@ -5310,7 +5385,13 @@ fn collect_legacy_text_worker_bindings(
     Ok(())
 }
 
-fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> {
+fn apply_workspace_worker_uuid_identity_migration(conn: &Connection) -> Result<()> {
+    promote_workspace_worker_uuid_identity(conn).map(|_| ())
+}
+
+fn promote_workspace_worker_uuid_identity(
+    conn: &Connection,
+) -> Result<Vec<LegacyWorkerIdentityMapping>> {
     conn.execute_batch(
         r#"
         PRAGMA defer_foreign_keys = ON;
@@ -5367,7 +5448,9 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> {
         collect_legacy_text_worker_bindings(conn, table, &mut legacy_workers)?;
     }
 
-    for (workspace_id, runtime_id, runtime_worker_id) in legacy_workers {
+    let mut mappings = Vec::with_capacity(legacy_workers.len());
+    for (workspace_id, runtime_id, runtime_worker_id) in &legacy_workers {
+        let worker_id = WorkerId::from_legacy_binding(workspace_id, runtime_id, *runtime_worker_id);
         conn.execute(
             "INSERT INTO worker_identity_v37(\
                 workspace_id, runtime_id, runtime_worker_id, worker_id\
@@ -5376,10 +5459,15 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> {
                 workspace_id,
                 runtime_id,
                 runtime_worker_id,
-                WorkerId::from_legacy_binding(&workspace_id, &runtime_id, runtime_worker_id)
-                    .to_string()
+                worker_id.to_string()
             ],
         )?;
+        mappings.push(LegacyWorkerIdentityMapping {
+            workspace_id: workspace_id.clone(),
+            runtime_id: runtime_id.clone(),
+            legacy_worker_id: *runtime_worker_id,
+            worker_id,
+        });
     }
 
     conn.execute_batch(
@@ -5581,7 +5669,7 @@ fn promote_workspace_worker_uuid_identity(conn: &Connection) -> Result<()> {
         DROP TABLE worker_identity_v37;
         "#,
     )?;
-    Ok(())
+    Ok(mappings)
 }
 
 fn allocate_resource_human_key(
@@ -5852,6 +5940,9 @@ pub(crate) fn apply_migrations_through(conn: &Connection, through_version: i64)
         i64::from(migration.version) > current && i64::from(migration.version) <= through_version
     }) {
         let tx = conn.unchecked_transaction()?;
+        if migration.version == 37 {
+            crate::retention::repair_worker_diagnostics_archive_table(&tx)?;
+        }
         (migration.apply)(&tx)?;
         tx.execute(
             "INSERT INTO __yoi_schema_migrations (version, name) VALUES (?1, ?2)",
@@ -6369,6 +6460,48 @@ mod tests {
             .unwrap();
     }
 
+    #[test]
+    fn migration_dry_run_repairs_missing_diagnostics_archive_without_mutating_source() {
+        let temp = tempfile::tempdir().unwrap();
+        let path = temp.path().join("server.db");
+        {
+            let conn = Connection::open(&path).unwrap();
+            configure_sqlite(&conn).unwrap();
+            apply_migrations_through(&conn, 36).unwrap();
+            conn.execute_batch(
+                "DROP TABLE worker_diagnostics_archives;
+                 INSERT INTO workspaces(workspace_id, display_name, state, created_at, updated_at)
+                 VALUES ('workspace-a', 'Workspace A', 'active', '1', '1');
+                 INSERT INTO worker_registry(
+                    workspace_id, runtime_id, runtime_worker_id, display_name,
+                    retention_state, created_at, updated_at
+                 ) VALUES ('workspace-a', 'runtime-a', 7, 'Worker 7', 'normal', '1', '1');",
+            )
+            .unwrap();
+        }
+        let before = std::fs::read(&path).unwrap();
+        let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
+        assert_eq!(plan.current_schema_version, 36);
+        assert_eq!(plan.target_schema_version, 38);
+        assert!(plan.migration_required);
+        assert_eq!(plan.worker_count, 1);
+        assert_eq!(plan.mappings[0].legacy_worker_id, 7);
+        assert_eq!(
+            plan.repairs,
+            vec!["create missing worker_diagnostics_archives table"]
+        );
+        assert_eq!(std::fs::read(&path).unwrap(), before);
+
+        let store = SqliteWorkspaceStore::open(&path).unwrap();
+        store
+            .with_conn(|conn| {
+                assert!(table_exists(conn, "worker_diagnostics_archives")?);
+                assert_eq!(current_schema_version(conn)?, 38);
+                Ok(())
+            })
+            .unwrap();
+    }
+
     #[test]
     fn v38_backfills_workspace_scoped_objective_and_worker_human_keys() {
         let conn = Connection::open_in_memory().unwrap();

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.

docs/report/2026-08-19-subworker-spawn-missing-worker-control.md

@@ -0,0 +1,23 @@
+# SubWorker spawn failed because the inherited profile required an unavailable control service
+
+Date: 2026-08-19
+
+## Observed behavior
+
+While splitting a migration investigation into read-only Runtime and Server analysis, both `SubWorkerSpawn` calls failed before the child session started:
+
+```text
+install Internal Worker features: Worker feature installation failed:
+builtin:worker-observation: required service requirement is not available:
+builtin:worker.control
+```
+
+The requested children used `builtin:coder` with read-only scopes. No child was created and no delegated work ran.
+
+## Impact
+
+A parent Worker with the SubWorker tools available cannot necessarily spawn a catalog profile whose transitive features require `builtin:worker.control`. The failure occurs at profile feature installation rather than being rejected when the profile is selected or omitted from the available SubWorker profile choices. The parent must continue the investigation without context splitting.
+
+## Expected behavior
+
+The SubWorker spawn layer should either install the parent-owned `worker.control` service before resolving dependent child features, provide a SubWorker-compatible profile projection that does not require unavailable Workspace Worker control, or reject the profile choice up front with an actionable capability diagnostic. A read-only delegated scope must remain read-only; satisfying the service dependency must not widen filesystem or Workspace authority.

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<InFlightBlock>, };
+export type InFlightSnapshot = { blocks?: Array<InFlightBlock>, commands?: Array<CommandSnapshot>, };
 
 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<InternalWorkerSnapshot>, } } | { "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<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, 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<InternalWorkerSnapshot>, } } | { "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<CompletionEntry>, } } | { "event": "rewind_targets", "data": { head_entries: number, targets: Array<RewindTarget>, } } | { "event": "rewind_applied", "data": { entries: Array<unknown>, input: Array<Segment>, 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" };

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([
     {
@@ -1314,9 +1431,36 @@ Deno.test("parent snapshot authoritatively replaces Internal Worker projections"
       kind: "sub_worker",
     },
     revision: 4,
-    entries: [],
+    entries: [{
+      kind: "assistant_item",
+      ts: 1,
+      item: {
+        kind: "tool_call",
+        call_id: "committed-call",
+        name: "Read",
+        arguments: JSON.stringify({ file_path: "/repo/a.md" }),
+      },
+    }, {
+      kind: "tool_result",
+      ts: 2,
+      item: {
+        kind: "tool_result",
+        call_id: "committed-call",
+        summary: "read file",
+        content: "content",
+        is_error: false,
+      },
+    }],
     status: "idle",
-    in_flight: { blocks: [] },
+    in_flight: {
+      blocks: [{
+        kind: "tool_call",
+        id: "committed-call",
+        name: "Read",
+        args: JSON.stringify({ file_path: "/repo/a.md" }),
+        state: "done",
+      }],
+    },
     internal_workers: [],
   }];
   const projector = createConsoleProjector();
@@ -1340,6 +1484,10 @@ Deno.test("parent snapshot authoritatively replaces Internal Worker projections"
   assertEquals(projection.internalWorkers.map((worker) => worker.worker.session_id), [
     "replacement",
   ]);
+  const childLines = projection.internalWorkers[0].console.lines;
+  assertEquals(childLines.length, 1);
+  assertEquals(new Set(childLines.map((line) => line.id)).size, 1);
+  assertEquals(childLines[0].kind, "tool");
 });
 
 Deno.test("snapshot restores TaskStore state from system history", () => {

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 = {
@@ -224,6 +228,141 @@ function projectVisibleConsole(
   };
 }
 
+function appendSnapshotInFlightLines(
+  projection: ConsoleProjection,
+  blocks: InFlightBlock[],
+  eventId: string,
+  cwd: string | null,
+): void {
+  const lineIds = new Set(projection.lines.map((line) => line.id));
+  blocks.forEach((block, index) => {
+    const pending = inFlightLine(`${eventId}:${index}`, block, cwd);
+    if (lineIds.has(pending.id)) return;
+    projection.lines.push(pending);
+    lineIds.add(pending.id);
+  });
+}
+
+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<CommandEvent, { kind: "output" }>,
+): 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,
@@ -235,15 +374,17 @@ function projectInternalWorkerSnapshot(
     cwd,
   );
   console.status = snapshot.status;
-  for (const block of snapshot.in_flight?.blocks ?? []) {
-    console.lines.push(
-      inFlightLine(
-        `${eventId}:internal:${snapshot.worker.session_id}:in-flight`,
-        block,
-        cwd,
-      ),
-    );
-  }
+  appendSnapshotInFlightLines(
+    console,
+    snapshot.in_flight?.blocks ?? [],
+    `${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`,
@@ -385,9 +526,17 @@ export function applyProtocolEvent(
       next.lines = snapshot.lines;
       next.tasks = snapshot.tasks;
       next.taskNextId = snapshot.taskNextId;
-      for (const block of event.data.in_flight?.blocks ?? []) {
-        next.lines.push(inFlightLine(envelope.eventId, block, next.cwd));
-      }
+      appendSnapshotInFlightLines(
+        next,
+        event.data.in_flight?.blocks ?? [],
+        `${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)
       );
@@ -421,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(
@@ -771,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")
@@ -779,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,
   };
 }
 
@@ -1025,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,
   ]);
 }