feat(command): stream command logs from CAS-backed storage

Persist command stdout/stderr through scratch logs and finalized CAS refs, expose byte-offset tailing through the API, and render separate streaming panels in the web run view.

Resolve command output blob refs for execution-time consumers such as edge routing and retros, and make Docker streaming timeout/cancel drain output before returning.

apps/fabro-web/app/lib/queries.ts

@@ -6,6 +6,8 @@ import type {
   PaginatedRunList,
   PaginatedRunStageList,
   PaginatedStageTurnList,
+  CommandLogResponse,
+  CommandOutputStream,
   RunBilling,
   RunProjection,
   ServerSettings,
@@ -156,6 +158,18 @@ export function useRunEventsList(id: string | undefined, enabled = true) {
   );
 }
 
+export function fetchRunCommandLog(
+  id: string,
+  stageId: string,
+  stream: CommandOutputStream,
+  offset: number,
+  limit?: number,
+) {
+  return apiFetcher<CommandLogResponse>(
+    queryKeys.runs.stageLog(id, stageId, stream, offset, limit),
+  );
+}
+
 export function useWorkflows() {
   return useSWR<PaginatedWorkflowListResponse | null>(
     queryKeys.workflows.list(),

apps/fabro-web/app/lib/query-keys.test.ts

@@ -8,6 +8,9 @@ describe("queryKeys", () => {
     expect(queryKeys.auth.me()).toBe("/api/v1/auth/me");
     expect(queryKeys.runs.files("run 1")).toBe("/api/v1/runs/run%201/files");
     expect(queryKeys.runs.graph("run-1", "TB")).toBe("/api/v1/runs/run-1/graph?direction=TB");
+    expect(queryKeys.runs.stageLog("run 1", "build step@2", "stderr", 12, 34)).toBe(
+      "/api/v1/runs/run%201/stages/build%20step%402/logs/stderr?offset=12&limit=34",
+    );
   });
 
   test("event-mapped keys match query hook resources", () => {

apps/fabro-web/app/lib/query-keys.ts

@@ -46,6 +46,17 @@ export const queryKeys = {
       withQuery(`/api/v1/runs/${pathSegment(id)}/events`, { limit }),
     stageTurns: (id: string, stageId: string) =>
       `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/turns`,
+    stageLog: (
+      id: string,
+      stageId: string,
+      stream: "stdout" | "stderr",
+      offset = 0,
+      limit = 65_536,
+    ) =>
+      withQuery(
+        `/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/logs/${stream}`,
+        { offset, limit },
+      ),
     preview: (id: string) => `/api/v1/runs/${pathSegment(id)}/preview`,
     cancel: (id: string) => `/api/v1/runs/${pathSegment(id)}/cancel`,
     archive: (id: string) => `/api/v1/runs/${pathSegment(id)}/archive`,

apps/fabro-web/app/routes/run-stages.tsx

@@ -1,4 +1,4 @@
-import { useEffect, useMemo, useState } from "react";
+import { useEffect, useMemo, useRef, useState } from "react";
 import { useParams } from "react-router";
 import { Marked } from "marked";
 
@@ -40,7 +40,7 @@ import type { Stage } from "../components/stage-sidebar";
 import { EmptyState } from "../components/state";
 import { CopyButton } from "../components/ui";
 import { formatDurationSecs } from "../lib/format";
-import { useRunEventsList, useRunStageTurns, useRunStages } from "../lib/queries";
+import { fetchRunCommandLog, useRunEventsList, useRunStageTurns, useRunStages } from "../lib/queries";
 import { mapRunStagesToSidebarStages } from "../lib/stage-sidebar";
 import type { StageTurn as ApiStageTurn, PaginatedStageTurnList, PaginatedEventList } from "@qltysh/fabro-api-client";
 
@@ -50,10 +50,11 @@ type TurnType =
   | { kind: "system"; content: string }
   | { kind: "assistant"; content: string }
   | { kind: "tool"; tools: ToolUse[] }
-  | { kind: "command"; script: string; language: string; stdout?: string; stderr?: string; exitCode?: number | null; durationMs?: number; timedOut?: boolean; running: boolean };
+  | { kind: "command"; stageId: string; script: string; language: string; stdout?: string; stderr?: string; exitCode?: number | null; durationMs?: number; timedOut?: boolean; running: boolean };
 
 interface RawEvent {
   node_id?: string;
+  stage_id?: string;
   event: string;
   properties?: Record<string, unknown>;
   text?: string;
@@ -70,7 +71,7 @@ function turnsFromEvents(events: RawEvent[], stageId: string): TurnType[] {
   // Collect tool pairs: started → completed
   const pendingTools = new Map<string, { toolName: string; input: string }>();
   // Track pending command for pairing started → completed
-  let pendingCommand: { script: string; language: string } | undefined;
+  let pendingCommand: { stageId: string; script: string; language: string } | undefined;
 
   for (const e of stageEvents) {
     const props = e.properties ?? {};
@@ -111,6 +112,7 @@ function turnsFromEvents(events: RawEvent[], stageId: string): TurnType[] {
       }
       case "command.started": {
         pendingCommand = {
+          stageId: e.stage_id ?? `${stageId}@1`,
           script: props.script as string ?? "",
           language: props.language as string ?? "shell",
         };
@@ -119,6 +121,7 @@ function turnsFromEvents(events: RawEvent[], stageId: string): TurnType[] {
       case "command.completed": {
         turns.push({
           kind: "command",
+          stageId: pendingCommand?.stageId ?? e.stage_id ?? `${stageId}@1`,
           script: pendingCommand?.script ?? "",
           language: pendingCommand?.language ?? "shell",
           stdout: props.stdout as string ?? "",
@@ -138,6 +141,7 @@ function turnsFromEvents(events: RawEvent[], stageId: string): TurnType[] {
   if (pendingCommand) {
     turns.push({
       kind: "command",
+      stageId: pendingCommand.stageId,
       script: pendingCommand.script,
       language: pendingCommand.language,
       running: true,
@@ -238,6 +242,9 @@ function StatusPill({
 }
 
 const COLLAPSE_AFTER_LINES = 20;
+const LOG_POLL_INTERVAL_MS = 1000;
+const LOG_FETCH_LIMIT_BYTES = 65_536;
+const LOG_MEMORY_CAP_BYTES = 5 * 1024 * 1024;
 
 function StreamLabel({ label }: { label: string }) {
   return (
@@ -247,18 +254,144 @@ function StreamLabel({ label }: { label: string }) {
   );
 }
 
+interface CommandLogState {
+  text: string;
+  eof: boolean;
+  loading: boolean;
+  error: boolean;
+  truncated: boolean;
+  casRef: string | null;
+  liveStreaming: boolean;
+  totalBytes: number;
+}
+
+function decodeBase64Bytes(value: string): Uint8Array {
+  if (!value) return new Uint8Array();
+  const binary = atob(value);
+  const bytes = new Uint8Array(binary.length);
+  for (let i = 0; i < binary.length; i += 1) {
+    bytes[i] = binary.charCodeAt(i);
+  }
+  return bytes;
+}
+
+function trimTextToBytes(text: string, maxBytes: number) {
+  const encoded = new TextEncoder().encode(text);
+  if (encoded.byteLength <= maxBytes) {
+    return { text, truncated: false };
+  }
+  const start = encoded.byteLength - maxBytes;
+  const trimmed = new TextDecoder().decode(encoded.slice(start));
+  return { text: trimmed.replace(/^\uFFFD/, ""), truncated: true };
+}
+
+function useCommandLog(
+  runId: string | undefined,
+  stageId: string | undefined,
+  stream: "stdout" | "stderr",
+  running: boolean,
+): CommandLogState {
+  const [state, setState] = useState<CommandLogState>({
+    text: "",
+    eof: false,
+    loading: true,
+    error: false,
+    truncated: false,
+    casRef: null,
+    liveStreaming: false,
+    totalBytes: 0,
+  });
+  const offsetRef = useRef(0);
+  const finalPollDoneRef = useRef(false);
+  const decoderRef = useRef(new TextDecoder());
+
+  useEffect(() => {
+    offsetRef.current = 0;
+    finalPollDoneRef.current = false;
+    decoderRef.current = new TextDecoder();
+    setState({
+      text: "",
+      eof: false,
+      loading: true,
+      error: false,
+      truncated: false,
+      casRef: null,
+      liveStreaming: false,
+      totalBytes: 0,
+    });
+  }, [runId, stageId, stream]);
+
+  useEffect(() => {
+    if (!runId || !stageId) return;
+    let cancelled = false;
+    let timer: ReturnType<typeof setTimeout> | undefined;
+
+    async function poll() {
+      try {
+        const chunk = await fetchRunCommandLog(
+          runId,
+          stageId,
+          stream,
+          offsetRef.current,
+          LOG_FETCH_LIMIT_BYTES,
+        );
+        if (cancelled) return;
+
+        offsetRef.current = chunk.next_offset;
+        const bytes = decodeBase64Bytes(chunk.bytes_base64);
+        const decoded = decoderRef.current.decode(bytes, { stream: !chunk.eof });
+        finalPollDoneRef.current = chunk.eof;
+        setState((current) => {
+          const next = trimTextToBytes(current.text + decoded, LOG_MEMORY_CAP_BYTES);
+          return {
+            text: next.text,
+            eof: chunk.eof,
+            loading: false,
+            error: false,
+            truncated: current.truncated || next.truncated,
+            casRef: chunk.cas_ref,
+            liveStreaming: chunk.live_streaming,
+            totalBytes: chunk.total_bytes,
+          };
+        });
+      } catch {
+        if (!cancelled) {
+          setState((current) => ({ ...current, loading: false, error: true }));
+        }
+      }
+
+      if (!cancelled && (running || !finalPollDoneRef.current)) {
+        timer = setTimeout(poll, LOG_POLL_INTERVAL_MS);
+      }
+    }
+
+    void poll();
+    return () => {
+      cancelled = true;
+      if (timer) clearTimeout(timer);
+    };
+  }, [runId, running, stageId, stream]);
+
+  return state;
+}
+
 function OutputStream({
   label,
-  content,
+  state,
   tone = "normal",
+  forceExpanded = false,
 }: {
   label: string;
-  content: string;
+  state: CommandLogState;
   tone?: "normal" | "error";
+  forceExpanded?: boolean;
 }) {
+  const content = state.text;
   const lines = content.split("\n");
   const isLong = lines.length > COLLAPSE_AFTER_LINES;
-  const [expanded, setExpanded] = useState(false);
+  const [expanded, setExpanded] = useState(forceExpanded);
+  const scrollRef = useRef<HTMLPreElement>(null);
+  const followTailRef = useRef(true);
   const visible = isLong && !expanded
     ? lines.slice(-COLLAPSE_AFTER_LINES).join("\n")
     : content;
@@ -267,13 +400,44 @@ function OutputStream({
     tone === "error"
       ? "whitespace-pre-wrap font-mono text-sm leading-relaxed text-coral sm:text-xs"
       : "whitespace-pre-wrap font-mono text-sm leading-relaxed text-fg-3 sm:text-xs";
+  const status = state.error
+    ? "Failed to load"
+    : state.loading
+      ? "Waiting"
+      : content.length > 0
+        ? state.eof
+          ? state.casRef
+            ? "Stored"
+            : "Complete"
+          : state.liveStreaming
+            ? "Streaming"
+            : "Running"
+        : state.eof
+          ? "No output"
+          : "Waiting";
+
+  useEffect(() => {
+    if (!forceExpanded) return;
+    setExpanded(true);
+  }, [forceExpanded]);
+
+  useEffect(() => {
+    const el = scrollRef.current;
+    if (el && followTailRef.current) {
+      el.scrollTop = el.scrollHeight;
+    }
+  }, [visible]);
 
   return (
     <div>
       <div className="mb-1 flex items-center gap-2">
         <StreamLabel label={label} />
+        <span className="text-[11px] text-fg-muted">{status}</span>
+        {state.truncated ? (
+          <span className="text-[11px] text-amber">Last 5 MiB</span>
+        ) : null}
         <CopyButton
-          value={content}
+          value={visible}
           label={`Copy ${label}`}
           className="-my-1"
         />
@@ -287,13 +451,36 @@ function OutputStream({
           Show {hiddenLines} earlier lines
         </button>
       ) : null}
-      <pre className={preClass}>{visible}</pre>
+      {content.length === 0 ? (
+        <div className="font-mono text-sm text-fg-muted sm:text-xs">
+          {state.error ? "Unable to fetch this stream." : "No bytes received yet."}
+        </div>
+      ) : (
+        <pre
+          ref={scrollRef}
+          onScroll={(event) => {
+            const el = event.currentTarget;
+            followTailRef.current = el.scrollHeight - el.scrollTop - el.clientHeight < 24;
+          }}
+          className={`${preClass} max-h-96 overflow-auto`}
+        >
+          {visible}
+        </pre>
+      )}
     </div>
   );
 }
 
-function CommandBlock({ turn }: { turn: Extract<TurnType, { kind: "command" }> }) {
+function CommandBlock({
+  runId,
+  turn,
+}: {
+  runId: string | undefined;
+  turn: Extract<TurnType, { kind: "command" }>;
+}) {
   const failed = !turn.running && turn.exitCode !== 0;
+  const stdout = useCommandLog(runId, turn.stageId, "stdout", turn.running);
+  const stderr = useCommandLog(runId, turn.stageId, "stderr", turn.running);
   const borderColor = turn.running ? "border-teal-500/20" : failed ? "border-coral/15" : "border-mint/15";
   const bgColor = turn.running ? "bg-teal-500/5" : failed ? "bg-coral/5" : "bg-mint/5";
 
@@ -339,19 +526,19 @@ function CommandBlock({ turn }: { turn: Extract<TurnType, { kind: "command" }> }
         </div>
       )}
 
-      {/* stdout */}
-      {turn.stdout && (
-        <div className="border-t border-line px-3 py-2.5">
-          <OutputStream label="stdout" content={turn.stdout} />
+      <div className="grid border-t border-line md:grid-cols-2">
+        <div className="border-line px-3 py-2.5 md:border-r">
+          <OutputStream label="stdout" state={stdout} />
         </div>
-      )}
-
-      {/* stderr */}
-      {turn.stderr && (
-        <div className="border-t border-line px-3 py-2.5">
-          <OutputStream label="stderr" content={turn.stderr} tone="error" />
+        <div className="border-t border-line px-3 py-2.5 md:border-t-0">
+          <OutputStream
+            label="stderr"
+            state={stderr}
+            tone="error"
+            forceExpanded={failed || stderr.text.length > 0}
+          />
         </div>
-      )}
+      </div>
     </div>
   );
 }
@@ -446,7 +633,7 @@ export default function RunStages() {
             case "tool":
               return <ToolBlock key={`turn-${i}`} tools={turn.tools} />;
             case "command":
-              return <CommandBlock key={`turn-${i}`} turn={turn} />;
+              return <CommandBlock key={`turn-${i}`} runId={id} turn={turn} />;
           }
         })}
       </div>

docs/public/agents/outputs.mdx

@@ -132,18 +132,18 @@ For the **CLI backend**, Fabro takes a different approach: it runs `git diff --n
 
 ## Artifact offloading
 
-When a stage produces a large context value -- an LLM response, command output, or any context update -- Fabro automatically offloads it into a global content-addressed blob store instead of leaving the full value inline in durable context.
+When a stage produces a large context value -- an LLM response or any context update -- Fabro automatically offloads it into a global content-addressed blob store instead of leaving the full value inline in durable context. Command stdout and stderr are streamed to stage log files while the command runs, then finalized into durable blob refs after completion.
 
 ### How offloading works
 
-After each node completes, Fabro checks every context update. If the serialized JSON of a value exceeds **100KB**, it is stored once by SHA-256 hash and replaced with a durable blob ref:
+After each node completes, Fabro checks every context update. If the serialized JSON of a value exceeds **100KB**, it is stored once by SHA-256 hash and replaced with a durable blob ref. Command stdout and stderr are always stored this way after completion, even when they are small or empty:
 
 ```
 response.plan  -->  blob://sha256/2cf24dba5fb0...
 command.output -->  blob://sha256/a4f3c1d9c2e1...
 ```
 
-Values under 100KB remain inline.
+Non-command values under 100KB remain inline.
 
 Checkpoints, checkpoint-completed events, forks, and resumes persist these `blob://` refs, not host-specific file paths.
 

docs/public/api-reference/fabro-api.yaml

@@ -1768,6 +1768,44 @@ paths:
               schema:
                 $ref: "#/components/schemas/ErrorResponse"
 
+  /api/v1/runs/{id}/stages/{stageId}/logs/{stream}:
+    get:
+      operationId: getRunStageCommandLog
+      tags: [Run Internals]
+      summary: Tail Command Log
+      description: Returns a byte-offset slice of a command stage stdout or stderr log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
+      parameters:
+        - $ref: "#/components/parameters/RunId"
+        - $ref: "#/components/parameters/StageId"
+        - $ref: "#/components/parameters/CommandLogStream"
+        - $ref: "#/components/parameters/CommandLogOffset"
+        - $ref: "#/components/parameters/CommandLogLimit"
+      responses:
+        "200":
+          description: Command log bytes.
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/CommandLogResponse"
+        "400":
+          description: Invalid stage, stream, offset, or limit.
+          headers:
+            x-request-id:
+              $ref: "#/components/headers/XRequestId"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+        "404":
+          description: Run or stage not found.
+          headers:
+            x-request-id:
+              $ref: "#/components/headers/XRequestId"
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/ErrorResponse"
+
   /api/v1/runs/{id}/questions:
     get:
       operationId: listRunQuestions
@@ -2891,6 +2929,38 @@ components:
         type: string
       example: code@2
 
+    CommandLogStream:
+      name: stream
+      in: path
+      required: true
+      description: Command output stream to read.
+      schema:
+        $ref: "#/components/schemas/CommandOutputStream"
+      example: stdout
+
+    CommandLogOffset:
+      name: offset
+      in: query
+      required: false
+      description: Byte offset to start reading from. Defaults to `0`.
+      schema:
+        type: integer
+        minimum: 0
+        default: 0
+      example: 65536
+
+    CommandLogLimit:
+      name: limit
+      in: query
+      required: false
+      description: Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
+      schema:
+        type: integer
+        minimum: 1
+        maximum: 1048576
+        default: 65536
+      example: 65536
+
     BlobId:
       name: blobId
       in: path
@@ -4814,6 +4884,62 @@ components:
           description: Blob identifier.
           example: 550e8400-e29b-41d4-a716-446655440000
 
+    CommandOutputStream:
+      description: Command output stream name.
+      type: string
+      enum:
+        - stdout
+        - stderr
+
+    CommandLogResponse:
+      description: Byte-offset command log slice.
+      type: object
+      required:
+        - stream
+        - offset
+        - next_offset
+        - total_bytes
+        - bytes_base64
+        - eof
+        - cas_ref
+        - live_streaming
+      properties:
+        stream:
+          $ref: "#/components/schemas/CommandOutputStream"
+        offset:
+          type: integer
+          minimum: 0
+          description: Actual byte offset used for this slice.
+          example: 0
+        next_offset:
+          type: integer
+          minimum: 0
+          description: Byte offset for the next tail request.
+          example: 4096
+        total_bytes:
+          type: integer
+          minimum: 0
+          description: Total bytes currently available for the stream.
+          example: 8192
+        bytes_base64:
+          type: string
+          description: Base64-encoded raw log bytes.
+          example: aGVsbG8K
+        eof:
+          type: boolean
+          description: Whether the stream is finalized.
+          example: false
+        cas_ref:
+          oneOf:
+            - type: string
+              pattern: '^blob://sha256/[0-9a-f]{64}$'
+            - type: "null"
+          description: Final CAS reference once the command has completed.
+        live_streaming:
+          type: boolean
+          description: Whether the sandbox provided live output while the command was running.
+          example: true
+
     ArtifactEntry:
       description: A single artifact filename.
       type: object
@@ -4979,6 +5105,16 @@ components:
           type: ["string", "null"]
         stderr:
           type: ["string", "null"]
+        stdout_bytes:
+          type: ["integer", "null"]
+          minimum: 0
+        stderr_bytes:
+          type: ["integer", "null"]
+          minimum: 0
+        streams_separated:
+          type: ["boolean", "null"]
+        live_streaming:
+          type: ["boolean", "null"]
 
     InterviewOption:
       description: Option stored with an interview question in the event log.

docs/public/execution/context.mdx

@@ -44,8 +44,8 @@ Agents can also emit arbitrary context updates by including a JSON object with a
 
 | Key | Value |
 |---|---|
-| `command.output` | The command's stdout |
-| `command.stderr` | The command's stderr |
+| `command.output` | The command's stdout. Durable context stores this as a `blob://sha256/...` ref after the command completes; downstream prompts resolve it back to text. |
+| `command.stderr` | The command's stderr. Durable context stores this as a `blob://sha256/...` ref after the command completes; downstream prompts resolve it back to text. |
 
 ### Human gates
 
@@ -196,10 +196,11 @@ Internal keys (prefixed with `internal.`, `current`, `graph.`, `thread.`, `respo
 
 ## Artifact offloading
 
-When a stage produces a large output (over 100KB of serialized JSON), Fabro stores the serialized bytes in a global content-addressed blob store and replaces the context value with a durable blob ref:
+When a stage produces a large output (over 100KB of serialized JSON), Fabro stores the serialized bytes in a global content-addressed blob store and replaces the context value with a durable blob ref. Command stdout and stderr are always finalized into blob refs after command completion, even when they are small or empty:
 
 ```
 response.plan → blob://sha256/2cf24dba5fb0...
+command.output → blob://sha256/a4f3c1d9c2e1...
 ```
 
 Checkpoints and checkpoint-completed events persist these `blob://` refs, not host-specific file paths.

lib/crates/fabro-agent/src/lib.rs

@@ -40,8 +40,9 @@ pub use memory::discover_memory;
 pub use profiles::{AnthropicProfile, EnvContext, GeminiProfile, OpenAiProfile};
 pub use read_before_write_sandbox::ReadBeforeWriteSandbox;
 pub use sandbox::{
-    DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, WorktreeEvent,
-    WorktreeEventCallback, WorktreeOptions, WorktreeSandbox, format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
+    SandboxEvent, SandboxEventCallback, WorktreeEvent, WorktreeEventCallback, WorktreeOptions,
+    WorktreeSandbox, format_lines_numbered, shell_quote,
 };
 pub use session::Session;
 pub use skills::Skill;

lib/crates/fabro-agent/src/sandbox.rs

@@ -2,7 +2,7 @@
 // Re-export the delegate_sandbox! macro at crate root so existing
 // `crate::delegate_sandbox!` invocations continue to work.
 pub use fabro_sandbox::{
-    DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, WorktreeEvent,
-    WorktreeEventCallback, WorktreeOptions, WorktreeSandbox, delegate_sandbox,
-    format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
+    SandboxEvent, SandboxEventCallback, WorktreeEvent, WorktreeEventCallback, WorktreeOptions,
+    WorktreeSandbox, delegate_sandbox, format_lines_numbered, shell_quote,
 };

lib/crates/fabro-api/build.rs

@@ -315,6 +315,11 @@ fn main() {
         ("NodeStatusRecord", "fabro_types::NodeStatusRecord", &[]),
         ("StageOutcome", "fabro_types::StageOutcome", &[]),
         ("StageState", "fabro_types::StageState", &[]),
+        (
+            "CommandOutputStream",
+            "fabro_types::CommandOutputStream",
+            &[],
+        ),
         ("NodeState", "fabro_types::NodeState", &[]),
         ("SecretMetadata", "fabro_types::SecretMetadata", &[]),
         ("InterviewOption", "fabro_types::InterviewOption", &[]),

lib/crates/fabro-api/src/lib.rs

@@ -29,11 +29,11 @@ pub mod types {
         BlockedReason, FailureReason, RunControlAction, RunStatus, SuccessReason, TerminalStatus,
     };
     pub use fabro_types::{
-        ActorKind, ActorRef, BilledTokenCounts, DiffStats, DirtyStatus, EventEnvelope, GitContext,
-        InterviewOption, InterviewQuestionRecord, NodeState, NodeStatusRecord,
-        PendingInterviewRecord, PreRunPushOutcome, QuestionType, RepositoryReference, RunEvent,
-        RunProjection, RunSummary, SecretMetadata, SecretType, ServerSettings, StageOutcome,
-        StageState, WorkflowSettings,
+        ActorKind, ActorRef, BilledTokenCounts, CommandOutputStream, DiffStats, DirtyStatus,
+        EventEnvelope, GitContext, InterviewOption, InterviewQuestionRecord, NodeState,
+        NodeStatusRecord, PendingInterviewRecord, PreRunPushOutcome, QuestionType,
+        RepositoryReference, RunEvent, RunProjection, RunSummary, SecretMetadata, SecretType,
+        ServerSettings, StageOutcome, StageState, WorkflowSettings,
     };
 
     pub use crate::generated::types::*;

lib/crates/fabro-api/tests/command_output_stream_round_trip.rs

@@ -0,0 +1,44 @@
+use std::any::{TypeId, type_name};
+
+use fabro_api::types::CommandOutputStream as ApiCommandOutputStream;
+use fabro_types::CommandOutputStream;
+use serde_json::json;
+
+#[test]
+fn command_output_stream_reuses_canonical_type() {
+    assert_same_type::<ApiCommandOutputStream, CommandOutputStream>();
+}
+
+#[test]
+fn command_output_stream_serializes_as_stream_names() {
+    assert_eq!(
+        serde_json::to_value(CommandOutputStream::Stdout).unwrap(),
+        json!("stdout")
+    );
+    assert_eq!(
+        serde_json::to_value(CommandOutputStream::Stderr).unwrap(),
+        json!("stderr")
+    );
+}
+
+#[test]
+fn command_output_stream_deserializes_representative_values() {
+    assert_eq!(
+        serde_json::from_value::<ApiCommandOutputStream>(json!("stdout")).unwrap(),
+        CommandOutputStream::Stdout
+    );
+    assert_eq!(
+        serde_json::from_value::<ApiCommandOutputStream>(json!("stderr")).unwrap(),
+        CommandOutputStream::Stderr
+    );
+}
+
+fn assert_same_type<T: 'static, U: 'static>() {
+    assert_eq!(
+        TypeId::of::<T>(),
+        TypeId::of::<U>(),
+        "{} should be the same type as {}",
+        type_name::<T>(),
+        type_name::<U>()
+    );
+}

lib/crates/fabro-core/src/executor.rs

@@ -390,7 +390,11 @@ impl<G: Graph + 'static> Executor<G> {
         }
 
         // Normal edge selection
-        if let Some(selection) = graph.select_edge(node, outcome, &state.context) {
+        let routing_context = self
+            .handler
+            .context_for_edge_selection(&state.context, graph)
+            .await?;
+        if let Some(selection) = graph.select_edge(node, outcome, &routing_context) {
             let target = selection.edge.target().to_string();
             let is_restart = selection.edge.is_loop_restart();
 

lib/crates/fabro-core/src/handler.rs

@@ -15,6 +15,10 @@ pub trait NodeHandler<G: Graph>: Send + Sync {
         graph: &G,
     ) -> Result<Outcome<G::Meta>>;
 
+    async fn context_for_edge_selection(&self, context: &Context, _graph: &G) -> Result<Context> {
+        Ok(context.clone())
+    }
+
     fn retry_policy(&self, _node: &G::Node, _graph: &G) -> RetryPolicy {
         RetryPolicy::none()
     }

lib/crates/fabro-retro/src/retro_agent.rs

@@ -1,4 +1,6 @@
+use std::future::Future;
 use std::path::{Path, PathBuf};
+use std::pin::Pin;
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
@@ -11,6 +13,7 @@ use fabro_llm::client::Client;
 use fabro_llm::provider::Provider;
 use fabro_llm::types::ToolDefinition;
 use fabro_store::{EventEnvelope, RunProjection, SerializableProjection};
+use fabro_types::{RunBlobId, parse_blob_ref};
 use tokio::task::JoinHandle;
 
 use crate::retro::{RetroNarrative, SmoothnessRating};
@@ -120,6 +123,12 @@ pub struct RetroAgentResult {
     pub response:  String,
 }
 
+pub type RetroBlobReader = Arc<
+    dyn Fn(RunBlobId) -> Pin<Box<dyn Future<Output = anyhow::Result<Option<Vec<u8>>>> + Send>>
+        + Send
+        + Sync,
+>;
+
 #[must_use]
 pub fn build_retro_prompt(retro_data_dir: &str) -> String {
     format!(
@@ -140,6 +149,7 @@ pub async fn run_retro_agent(
     state: &RunProjection,
     events: &[EventEnvelope],
     run_dir: &Path,
+    blob_reader: Option<RetroBlobReader>,
     llm_client: &Client,
     provider: Provider,
     model: &str,
@@ -147,7 +157,15 @@ pub async fn run_retro_agent(
 ) -> anyhow::Result<RetroAgentResult> {
     // Upload data files into sandbox (needed for Daytona; no-op effect for local
     // since the agent can also read from the original paths via tools).
-    upload_data_files(sandbox, state, events, run_dir, RETRO_DATA_DIR).await?;
+    upload_data_files(
+        sandbox,
+        state,
+        events,
+        run_dir,
+        RETRO_DATA_DIR,
+        blob_reader.as_ref(),
+    )
+    .await?;
 
     // Build provider profile with the submit_retro tool
     let captured: Arc<Mutex<Option<RetroNarrative>>> = Arc::new(Mutex::new(None));
@@ -300,6 +318,7 @@ async fn upload_data_files(
     events: &[EventEnvelope],
     _run_dir: &Path,
     target_dir: &str,
+    blob_reader: Option<&RetroBlobReader>,
 ) -> anyhow::Result<()> {
     let progress_content = (!events.is_empty()).then(|| {
         let mut buf = String::new();
@@ -406,14 +425,14 @@ async fn upload_data_files(
             sandbox,
             target_dir,
             &base.join("stdout.log"),
-            node.stdout.clone(),
+            resolve_text_file_content(node.stdout.clone(), blob_reader).await?,
         )
         .await?;
         upload_file(
             sandbox,
             target_dir,
             &base.join("stderr.log"),
-            node.stderr.clone(),
+            resolve_text_file_content(node.stderr.clone(), blob_reader).await?,
         )
         .await?;
     }
@@ -421,6 +440,28 @@ async fn upload_data_files(
     Ok(())
 }
 
+async fn resolve_text_file_content(
+    content: Option<String>,
+    blob_reader: Option<&RetroBlobReader>,
+) -> anyhow::Result<Option<String>> {
+    let Some(content) = content else {
+        return Ok(None);
+    };
+    let Some(blob_id) = parse_blob_ref(&content) else {
+        return Ok(Some(content));
+    };
+    let Some(blob_reader) = blob_reader else {
+        return Ok(Some(content));
+    };
+
+    let bytes = blob_reader(blob_id)
+        .await?
+        .ok_or_else(|| anyhow::anyhow!("text blob missing: {blob_id}"))?;
+    let text = serde_json::from_slice::<String>(&bytes)
+        .map_err(|err| anyhow::anyhow!("text blob was not a JSON string: {err}"))?;
+    Ok(Some(text))
+}
+
 async fn upload_file(
     sandbox: &Arc<dyn Sandbox>,
     target_dir: &str,
@@ -573,11 +614,22 @@ mod tests {
             parallel_results:  Some(serde_json::json!([{ "stage": "fanout@1" }])),
             stdout:            Some("stdout".to_string()),
             stderr:            Some("stderr".to_string()),
+            stdout_bytes:      None,
+            stderr_bytes:      None,
+            streams_separated: None,
+            live_streaming:    None,
         });
 
-        upload_data_files(&sandbox, &state, &[], output_dir.path(), &target_dir_str)
-            .await
-            .expect("retro files should upload");
+        upload_data_files(
+            &sandbox,
+            &state,
+            &[],
+            output_dir.path(),
+            &target_dir_str,
+            None,
+        )
+        .await
+        .expect("retro files should upload");
 
         let run_json: serde_json::Value = serde_json::from_str(
             &fs::read_to_string(target_dir.join("run.json"))
@@ -622,4 +674,65 @@ mod tests {
             "progress file should be omitted when there are no events"
         );
     }
+
+    #[tokio::test]
+    async fn upload_data_files_resolves_command_stdout_stderr_blob_refs() {
+        let sandbox_root = tempfile::tempdir().expect("sandbox tempdir should exist");
+        let sandbox: Arc<dyn Sandbox> =
+            Arc::new(LocalSandbox::new(sandbox_root.path().to_path_buf()));
+        let output_dir = tempfile::tempdir().expect("retro tempdir should exist");
+        let target_dir = output_dir.path().join("retro");
+        let target_dir_str = target_dir.to_string_lossy().to_string();
+
+        let stdout_blob = serde_json::to_vec("resolved stdout").unwrap();
+        let stderr_blob = serde_json::to_vec("resolved stderr").unwrap();
+        let stdout_id = fabro_types::RunBlobId::new(&stdout_blob);
+        let stderr_id = fabro_types::RunBlobId::new(&stderr_blob);
+
+        let stage_id = StageId::new("build", 1);
+        let mut state = RunProjection::default();
+        state.set_node(stage_id, NodeState {
+            stdout: Some(fabro_types::format_blob_ref(&stdout_id)),
+            stderr: Some(fabro_types::format_blob_ref(&stderr_id)),
+            ..NodeState::default()
+        });
+
+        let reader: RetroBlobReader = Arc::new(move |blob_id| {
+            let stdout_blob = stdout_blob.clone();
+            let stderr_blob = stderr_blob.clone();
+            Box::pin(async move {
+                if blob_id == stdout_id {
+                    Ok(Some(stdout_blob))
+                } else if blob_id == stderr_id {
+                    Ok(Some(stderr_blob))
+                } else {
+                    Ok(None)
+                }
+            })
+        });
+
+        upload_data_files(
+            &sandbox,
+            &state,
+            &[],
+            output_dir.path(),
+            &target_dir_str,
+            Some(&reader),
+        )
+        .await
+        .expect("retro files should upload");
+
+        assert_eq!(
+            fs::read_to_string(target_dir.join("stages/build@1/stdout.log"))
+                .await
+                .expect("stdout file should exist"),
+            "resolved stdout"
+        );
+        assert_eq!(
+            fs::read_to_string(target_dir.join("stages/build@1/stderr.log"))
+                .await
+                .expect("stderr file should exist"),
+            "resolved stderr"
+        );
+    }
 }

lib/crates/fabro-sandbox/src/daytona/mod.rs

@@ -6,7 +6,7 @@ use std::time::Instant;
 use async_trait::async_trait;
 use daytona_sdk::api_types::SignedPortPreviewUrl;
 use fabro_github::GitHubCredentials;
-use fabro_types::RunId;
+use fabro_types::{CommandOutputStream, RunId};
 use rand::Rng;
 use tokio::sync::OnceCell;
 use tokio::{fs, time};
@@ -16,8 +16,8 @@ use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
 use crate::redact::redact_auth_url;
 use crate::sandbox::resolve_path;
 use crate::{
-    DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback,
-    format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
+    SandboxEvent, SandboxEventCallback, format_lines_numbered, shell_quote,
 };
 
 const WORKING_DIRECTORY: &str = "/home/daytona/workspace";
@@ -1149,6 +1149,39 @@ impl Sandbox for DaytonaSandbox {
         })
     }
 
+    async fn exec_command_streaming(
+        &self,
+        command: &str,
+        timeout_ms: u64,
+        working_dir: Option<&str>,
+        env_vars: Option<&HashMap<String, String>>,
+        cancel_token: Option<CancellationToken>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<ExecStreamingResult> {
+        let result = self
+            .exec_command(command, timeout_ms, working_dir, env_vars, cancel_token)
+            .await?;
+        if !result.stdout.is_empty() {
+            output_callback(
+                CommandOutputStream::Stdout,
+                result.stdout.as_bytes().to_vec(),
+            )
+            .await?;
+        }
+        if !result.stderr.is_empty() {
+            output_callback(
+                CommandOutputStream::Stderr,
+                result.stderr.as_bytes().to_vec(),
+            )
+            .await?;
+        }
+        Ok(ExecStreamingResult {
+            result,
+            streams_separated: false,
+            live_streaming: false,
+        })
+    }
+
     async fn grep(
         &self,
         pattern: &str,

lib/crates/fabro-sandbox/src/docker.rs

@@ -1,7 +1,8 @@
 use std::collections::HashMap;
 use std::fmt::Write as _;
 use std::io::Cursor;
-use std::time::Instant;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
 
 use async_trait::async_trait;
 use bollard::Docker;
@@ -15,7 +16,7 @@ use bollard::exec::{CreateExecOptions, StartExecResults};
 use bollard::image::CreateImageOptions;
 use bollard::models::HostConfig;
 use fabro_github::GitHubCredentials;
-use fabro_types::RunId;
+use fabro_types::{CommandOutputStream, RunId};
 use futures::StreamExt;
 use tokio::sync::OnceCell;
 use tokio::{fs, time};
@@ -25,8 +26,8 @@ use crate::clone_source::{self, CloneDecision, EmptyWorkspaceReason};
 use crate::redact::redact_auth_url;
 use crate::sandbox::resolve_path;
 use crate::{
-    DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback,
-    format_lines_numbered, shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
+    SandboxEvent, SandboxEventCallback, format_lines_numbered, shell_quote,
 };
 
 const WORKING_DIRECTORY: &str = "/workspace";
@@ -34,6 +35,7 @@ const GIT_CLONE_DEPTH: usize = 10;
 
 const MANAGED_LABEL: &str = "sh.fabro.managed";
 const RUN_ID_LABEL: &str = "sh.fabro.run_id";
+static EXEC_CONTROL_COUNTER: AtomicU64 = AtomicU64::new(1);
 
 #[derive(Clone, Debug, PartialEq)]
 pub struct DockerSandboxOptions {
@@ -226,6 +228,69 @@ impl DockerSandbox {
         Ok((stdout, stderr, exit_code))
     }
 
+    async fn docker_exec_streaming(
+        docker: Docker,
+        container_id: String,
+        cmd: Vec<String>,
+        working_dir: Option<String>,
+        env: Option<Vec<String>>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<(Vec<u8>, Vec<u8>, i32)> {
+        let exec_opts = CreateExecOptions {
+            cmd: Some(cmd),
+            attach_stdout: Some(true),
+            attach_stderr: Some(true),
+            working_dir,
+            env: env.map(|e| e.into_iter().collect()),
+            ..Default::default()
+        };
+
+        let exec_instance = docker
+            .create_exec(&container_id, exec_opts)
+            .await
+            .map_err(|e| crate::Error::context("Failed to create exec", e))?;
+
+        let start_result = docker
+            .start_exec(&exec_instance.id, None)
+            .await
+            .map_err(|e| crate::Error::context("Failed to start exec", e))?;
+
+        let mut stdout = Vec::new();
+        let mut stderr = Vec::new();
+
+        if let StartExecResults::Attached { mut output, .. } = start_result {
+            while let Some(chunk) = output.next().await {
+                match chunk {
+                    Ok(LogOutput::StdOut { message }) => {
+                        let bytes = message.to_vec();
+                        output_callback(CommandOutputStream::Stdout, bytes.clone()).await?;
+                        stdout.extend_from_slice(&bytes);
+                    }
+                    Ok(LogOutput::StdErr { message }) => {
+                        let bytes = message.to_vec();
+                        output_callback(CommandOutputStream::Stderr, bytes.clone()).await?;
+                        stderr.extend_from_slice(&bytes);
+                    }
+                    Ok(_) => {}
+                    Err(e) => {
+                        return Err(crate::Error::context("Error reading exec output", e));
+                    }
+                }
+            }
+        }
+
+        let inspect = docker
+            .inspect_exec(&exec_instance.id)
+            .await
+            .map_err(|e| crate::Error::context("Failed to inspect exec", e))?;
+
+        let exit_code = inspect
+            .exit_code
+            .and_then(|code| i32::try_from(code).ok())
+            .unwrap_or(-1);
+        Ok((stdout, stderr, exit_code))
+    }
+
     async fn docker_exec_shell(
         &self,
         command: &str,
@@ -282,6 +347,94 @@ impl DockerSandbox {
         }
     }
 
+    async fn docker_exec_shell_streaming(
+        &self,
+        command: &str,
+        timeout_ms: u64,
+        working_dir: Option<&str>,
+        env_vars: Option<&HashMap<String, String>>,
+        cancel_token: Option<CancellationToken>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<ExecStreamingResult> {
+        let start = Instant::now();
+        let effective_dir = working_dir.unwrap_or(WORKING_DIRECTORY).to_string();
+        let env: Option<Vec<String>> =
+            env_vars.map(|vars| vars.iter().map(|(k, v)| format!("{k}={v}")).collect());
+        let (stop_file, pid_file) = docker_exec_control_paths();
+        let controlled_command = docker_controlled_shell_command(command, &stop_file, &pid_file);
+        let cmd = vec![
+            "/bin/bash".to_string(),
+            "-lc".to_string(),
+            controlled_command,
+        ];
+
+        let timeout_duration = Duration::from_millis(timeout_ms);
+        let token = cancel_token.unwrap_or_default();
+
+        let container_id = self.container_id()?.to_string();
+        let mut output_task = tokio::spawn(Self::docker_exec_streaming(
+            self.docker.clone(),
+            container_id,
+            cmd,
+            Some(effective_dir.clone()),
+            env,
+            output_callback,
+        ));
+
+        let mut interrupted = false;
+        let output = tokio::select! {
+            joined = &mut output_task => {
+                joined
+                    .map_err(|e| crate::Error::context("Docker exec stream task failed", e))??
+            }
+            () = time::sleep(timeout_duration) => {
+                interrupted = true;
+                self.request_docker_exec_stop(&stop_file).await?;
+                output_task
+                    .await
+                    .map_err(|e| crate::Error::context("Docker exec stream task failed", e))??
+            }
+            () = token.cancelled() => {
+                interrupted = true;
+                self.request_docker_exec_stop(&stop_file).await?;
+                output_task
+                    .await
+                    .map_err(|e| crate::Error::context("Docker exec stream task failed", e))??
+            }
+        };
+
+        let (stdout, stderr, exit_code) = output;
+        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
+        Ok(ExecStreamingResult {
+            result:            ExecResult {
+                stdout: String::from_utf8_lossy(&stdout).into_owned(),
+                stderr: String::from_utf8_lossy(&stderr).into_owned(),
+                exit_code: if interrupted { -1 } else { exit_code },
+                timed_out: interrupted,
+                duration_ms,
+            },
+            streams_separated: true,
+            live_streaming:    true,
+        })
+    }
+
+    async fn request_docker_exec_stop(&self, stop_file: &str) -> crate::Result<()> {
+        let command = format!("touch {}", shell_quote(stop_file));
+        let (stdout, stderr, exit_code) = self
+            .docker_exec(
+                vec!["/bin/bash".to_string(), "-lc".to_string(), command.clone()],
+                Some("/"),
+                None,
+            )
+            .await?;
+        if exit_code != 0 {
+            return Err(crate::Error::message(format!(
+                "Failed to request Docker exec stop (exit {exit_code}): {stderr}{stdout}"
+            )));
+        }
+        Ok(())
+    }
+
     async fn ensure_image(&self) -> crate::Result<()> {
         if !self.config.auto_pull {
             return Ok(());
@@ -546,6 +699,48 @@ fn container_name(run_id: &RunId) -> String {
     format!("fabro-run-{run_id}")
 }
 
+fn docker_exec_control_paths() -> (String, String) {
+    let sequence = EXEC_CONTROL_COUNTER.fetch_add(1, Ordering::Relaxed);
+    let prefix = format!("/tmp/fabro-exec-{}-{sequence}", std::process::id());
+    (format!("{prefix}.stop"), format!("{prefix}.pid"))
+}
+
+fn docker_controlled_shell_command(command: &str, stop_file: &str, pid_file: &str) -> String {
+    format!(
+        "\
+stop_file={stop_file}; \
+pid_file={pid_file}; \
+user_command={command}; \
+rm -f \"$stop_file\" \"$pid_file\"; \
+( \
+  while [ ! -e \"$stop_file\" ]; do sleep 0.1; done; \
+  if [ -s \"$pid_file\" ]; then \
+    child=$(cat \"$pid_file\"); \
+    kill -TERM \"-$child\" 2>/dev/null || kill -TERM \"$child\" 2>/dev/null || true; \
+    sleep 0.2; \
+    kill -KILL \"-$child\" 2>/dev/null || kill -KILL \"$child\" 2>/dev/null || true; \
+  fi \
+) & watcher=$!; \
+if command -v setsid >/dev/null 2>&1; then \
+  setsid /bin/bash -lc \"$user_command\" & \
+else \
+  /bin/bash -lc \"$user_command\" & \
+fi; \
+child=$!; \
+echo \"$child\" > \"$pid_file\"; \
+wait \"$child\"; \
+status=$?; \
+kill \"$watcher\" 2>/dev/null || true; \
+wait \"$watcher\" 2>/dev/null || true; \
+rm -f \"$stop_file\" \"$pid_file\"; \
+exit \"$status\"\
+",
+        stop_file = shell_quote(stop_file),
+        pid_file = shell_quote(pid_file),
+        command = shell_quote(command),
+    )
+}
+
 fn git_clone_command(clone_url: &str, branch: Option<&str>) -> String {
     let mut command = "git -c maintenance.auto=0 -c gc.auto=0 clone".to_string();
     if let Some(branch) = branch {
@@ -961,6 +1156,27 @@ impl Sandbox for DockerSandbox {
             .await
     }
 
+    async fn exec_command_streaming(
+        &self,
+        command: &str,
+        timeout_ms: u64,
+        working_dir: Option<&str>,
+        env_vars: Option<&HashMap<String, String>>,
+        cancel_token: Option<CancellationToken>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<ExecStreamingResult> {
+        let dir = working_dir.map(Self::resolve_container_path);
+        self.docker_exec_shell_streaming(
+            command,
+            timeout_ms,
+            dir.as_deref(),
+            env_vars,
+            cancel_token,
+            output_callback,
+        )
+        .await
+    }
+
     async fn read_file(
         &self,
         path: &str,
@@ -1288,6 +1504,7 @@ mod tests {
         reason = "unit test reads an in-memory tar entry synchronously"
     )]
     use std::io::Read as _;
+    use std::sync::{Arc, Mutex};
 
     use super::*;
 
@@ -1347,6 +1564,87 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn streaming_timeout_terminates_docker_exec_before_returning() {
+        let image = "buildpack-deps:noble";
+        let Ok(docker) = Docker::connect_with_local_defaults() else {
+            return;
+        };
+        if docker.inspect_image(image).await.is_err() {
+            return;
+        }
+
+        let sandbox = DockerSandbox::new(
+            DockerSandboxOptions {
+                image: image.to_string(),
+                auto_pull: false,
+                skip_clone: true,
+                ..DockerSandboxOptions::default()
+            },
+            None,
+            None,
+            None,
+            None,
+        )
+        .expect("docker sandbox should construct");
+        sandbox
+            .initialize()
+            .await
+            .expect("docker sandbox should initialize");
+
+        let chunks = Arc::new(Mutex::new(Vec::new()));
+        let callback_chunks = Arc::clone(&chunks);
+        let callback: CommandOutputCallback = Arc::new(move |_stream, bytes| {
+            let callback_chunks = Arc::clone(&callback_chunks);
+            Box::pin(async move {
+                callback_chunks.lock().unwrap().extend(bytes);
+                Ok(())
+            })
+        });
+
+        let marker = "fabro_streaming_timeout_sentinel";
+        let result = sandbox
+            .exec_command_streaming(
+                &format!("trap '' HUP TERM; echo start; sleep 5 # {marker}"),
+                200,
+                None,
+                None,
+                None,
+                callback,
+            )
+            .await
+            .expect("streaming command should return a timeout result");
+
+        assert!(result.result.timed_out);
+        assert!(
+            String::from_utf8_lossy(&chunks.lock().unwrap()).contains("start"),
+            "stream should include output emitted before timeout"
+        );
+
+        let probe = sandbox
+            .exec_command(
+                "marker='fabro_streaming_timeout_''sentinel'; \
+                 ps -eo pid,args | awk -v marker=\"$marker\" \
+                 'index($0, marker) && $0 !~ /awk/ && $0 !~ /ps -eo/ { print }'",
+                1_000,
+                None,
+                None,
+                None,
+            )
+            .await
+            .expect("process probe should run");
+        sandbox
+            .cleanup()
+            .await
+            .expect("docker cleanup should succeed");
+
+        assert!(
+            !probe.stdout.contains(marker),
+            "timed-out docker exec should be terminated before returning, found: {}",
+            probe.stdout
+        );
+    }
+
     #[test]
     fn label_validation_rejects_unmanaged_container() {
         let labels = HashMap::new();

lib/crates/fabro-sandbox/src/lib.rs

@@ -36,9 +36,9 @@ pub use error::{Error, Result};
 pub use local::LocalSandbox;
 pub use read_guard::ReadBeforeWriteSandbox;
 pub use sandbox::{
-    DirEntry, ExecResult, GitRunInfo, GitSetupIntent, GrepOptions, Sandbox, SandboxEvent,
-    SandboxEventCallback, format_lines_numbered, git_push_via_exec, setup_git_via_exec,
-    shell_quote,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GitRunInfo, GitSetupIntent,
+    GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback, format_lines_numbered,
+    git_push_via_exec, setup_git_via_exec, shell_quote,
 };
 pub use sandbox_provider::SandboxProvider;
 pub use sandbox_record::SandboxRecord;

lib/crates/fabro-sandbox/src/local.rs

@@ -3,15 +3,16 @@ use std::time::Instant;
 
 use async_trait::async_trait;
 use fabro_static::EnvVars;
-use tokio::io::AsyncReadExt;
+use fabro_types::CommandOutputStream;
+use tokio::io::{AsyncRead, AsyncReadExt};
 use tokio::process::{Child, Command};
 use tokio::task::spawn_blocking;
 use tokio::{fs, time};
 use tokio_util::sync::CancellationToken;
 
 use crate::{
-    DirEntry, ExecResult, GrepOptions, Sandbox, SandboxEvent, SandboxEventCallback,
-    format_lines_numbered,
+    CommandOutputCallback, DirEntry, ExecResult, ExecStreamingResult, GrepOptions, Sandbox,
+    SandboxEvent, SandboxEventCallback, format_lines_numbered,
 };
 
 pub struct LocalSandbox {
@@ -323,6 +324,100 @@ impl Sandbox for LocalSandbox {
         })
     }
 
+    async fn exec_command_streaming(
+        &self,
+        command: &str,
+        timeout_ms: u64,
+        working_dir: Option<&str>,
+        env_vars: Option<&std::collections::HashMap<String, String>>,
+        cancel_token: Option<CancellationToken>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<ExecStreamingResult> {
+        let start = Instant::now();
+
+        let mut filtered_env: Vec<(String, String)> = process_env_vars()
+            .into_iter()
+            .filter(|(key, _)| !Self::should_filter_env_var(key))
+            .collect();
+
+        if let Some(extra) = env_vars {
+            for (k, v) in extra {
+                if !Self::should_filter_env_var(k) {
+                    filtered_env.push((k.clone(), v.clone()));
+                }
+            }
+        }
+
+        let effective_dir =
+            working_dir.map_or_else(|| self.working_directory.clone(), std::path::PathBuf::from);
+
+        let mut cmd = Command::new("/bin/bash");
+        cmd.arg("-c")
+            .arg(command)
+            .current_dir(&effective_dir)
+            .env_clear()
+            .envs(filtered_env)
+            .stdout(std::process::Stdio::piped())
+            .stderr(std::process::Stdio::piped());
+
+        #[cfg(unix)]
+        fabro_proc::pre_exec_setpgid(cmd.as_std_mut());
+
+        let mut child = cmd
+            .spawn()
+            .map_err(|e| crate::Error::context("Failed to spawn command", e))?;
+
+        let timeout_duration = std::time::Duration::from_millis(timeout_ms);
+        let token = cancel_token.unwrap_or_default();
+
+        let stdout_pipe = child.stdout.take();
+        let stderr_pipe = child.stderr.take();
+        let stdout_callback = output_callback.clone();
+        let stderr_callback = output_callback;
+        let stdout_task = tokio::spawn(async move {
+            drain_command_pipe(stdout_pipe, CommandOutputStream::Stdout, stdout_callback).await
+        });
+        let stderr_task = tokio::spawn(async move {
+            drain_command_pipe(stderr_pipe, CommandOutputStream::Stderr, stderr_callback).await
+        });
+
+        let (timed_out, exit_code) = tokio::select! {
+            status_result = child.wait() => {
+                let status = status_result
+                    .map_err(|e| crate::Error::context("Failed to wait for process", e))?;
+                (false, status.code().unwrap_or(-1))
+            }
+            () = time::sleep(timeout_duration) => {
+                sigterm_then_kill(&mut child).await;
+                (true, -1)
+            }
+            () = token.cancelled() => {
+                sigterm_then_kill(&mut child).await;
+                (true, -1)
+            }
+        };
+
+        let duration_ms = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
+        let stdout_bytes = stdout_task
+            .await
+            .map_err(|e| crate::Error::context("stdout stream task failed", e))??;
+        let stderr_bytes = stderr_task
+            .await
+            .map_err(|e| crate::Error::context("stderr stream task failed", e))??;
+
+        Ok(ExecStreamingResult {
+            result:            ExecResult {
+                stdout: String::from_utf8_lossy(&stdout_bytes).into_owned(),
+                stderr: String::from_utf8_lossy(&stderr_bytes).into_owned(),
+                exit_code,
+                timed_out,
+                duration_ms,
+            },
+            streams_separated: true,
+            live_streaming:    true,
+        })
+    }
+
     async fn grep(
         &self,
         pattern: &str,
@@ -583,6 +678,34 @@ async fn sigterm_then_kill(child: &mut Child) {
     }
 }
 
+async fn drain_command_pipe<R>(
+    mut reader: Option<R>,
+    stream: CommandOutputStream,
+    output_callback: CommandOutputCallback,
+) -> crate::Result<Vec<u8>>
+where
+    R: AsyncRead + Unpin,
+{
+    let mut output = Vec::new();
+    let Some(reader) = reader.as_mut() else {
+        return Ok(output);
+    };
+
+    let mut buf = [0_u8; 8192];
+    loop {
+        let read = reader
+            .read(&mut buf)
+            .await
+            .map_err(|e| crate::Error::context("Failed to read command output", e))?;
+        if read == 0 {
+            return Ok(output);
+        }
+        let chunk = buf[..read].to_vec();
+        output_callback(stream, chunk.clone()).await?;
+        output.extend_from_slice(&chunk);
+    }
+}
+
 #[cfg(test)]
 #[expect(
     clippy::disallowed_methods,

lib/crates/fabro-sandbox/src/sandbox.rs

@@ -1,10 +1,13 @@
 use std::collections::HashMap;
 use std::fmt::Write;
+use std::future::Future;
 use std::path::Path;
+use std::pin::Pin;
 use std::sync::Arc;
 use std::time::Duration;
 
 use async_trait::async_trait;
+use fabro_types::CommandOutputStream;
 use serde::{Deserialize, Serialize};
 use tokio::time;
 use tokio_util::sync::CancellationToken;
@@ -85,6 +88,27 @@ macro_rules! delegate_sandbox {
                     .await
             }
 
+            async fn exec_command_streaming(
+                &self,
+                command: &str,
+                timeout_ms: u64,
+                working_dir: Option<&str>,
+                env_vars: Option<&std::collections::HashMap<String, String>>,
+                cancel_token: Option<tokio_util::sync::CancellationToken>,
+                output_callback: $crate::CommandOutputCallback,
+            ) -> $crate::Result<$crate::ExecStreamingResult> {
+                self.$field
+                    .exec_command_streaming(
+                        command,
+                        timeout_ms,
+                        working_dir,
+                        env_vars,
+                        cancel_token,
+                        output_callback,
+                    )
+                    .await
+            }
+
             async fn glob(&self, pattern: &str, path: Option<&str>) -> $crate::Result<Vec<String>> {
                 self.$field.glob(pattern, path).await
             }
@@ -422,6 +446,19 @@ impl ExecResult {
     }
 }
 
+#[derive(Debug, Clone)]
+pub struct ExecStreamingResult {
+    pub result:            ExecResult,
+    pub streams_separated: bool,
+    pub live_streaming:    bool,
+}
+
+pub type CommandOutputCallback = Arc<
+    dyn Fn(CommandOutputStream, Vec<u8>) -> Pin<Box<dyn Future<Output = crate::Result<()>> + Send>>
+        + Send
+        + Sync,
+>;
+
 #[derive(Debug, Clone)]
 pub struct DirEntry {
     pub name:   String,
@@ -460,6 +497,38 @@ pub trait Sandbox: Send + Sync {
         env_vars: Option<&std::collections::HashMap<String, String>>,
         cancel_token: Option<CancellationToken>,
     ) -> crate::Result<ExecResult>;
+    async fn exec_command_streaming(
+        &self,
+        command: &str,
+        timeout_ms: u64,
+        working_dir: Option<&str>,
+        env_vars: Option<&std::collections::HashMap<String, String>>,
+        cancel_token: Option<CancellationToken>,
+        output_callback: CommandOutputCallback,
+    ) -> crate::Result<ExecStreamingResult> {
+        let result = self
+            .exec_command(command, timeout_ms, working_dir, env_vars, cancel_token)
+            .await?;
+        if !result.stdout.is_empty() {
+            output_callback(
+                CommandOutputStream::Stdout,
+                result.stdout.as_bytes().to_vec(),
+            )
+            .await?;
+        }
+        if !result.stderr.is_empty() {
+            output_callback(
+                CommandOutputStream::Stderr,
+                result.stderr.as_bytes().to_vec(),
+            )
+            .await?;
+        }
+        Ok(ExecStreamingResult {
+            result,
+            streams_separated: true,
+            live_streaming: false,
+        })
+    }
     async fn grep(
         &self,
         pattern: &str,

lib/crates/fabro-server/src/server.rs

@@ -73,14 +73,15 @@ use fabro_types::settings::server::{
 };
 use fabro_types::settings::{InterpString, RunNamespace};
 use fabro_types::{
-    ActorRef, EventBody, InterviewQuestionRecord, PullRequestRecord, QuestionType, RunBlobId,
-    RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance, RunServerProvenance,
-    RunSubjectProvenance, ServerSettings,
+    ActorRef, CommandOutputStream, EventBody, InterviewQuestionRecord, PullRequestRecord,
+    QuestionType, RunBlobId, RunClientProvenance, RunControlAction, RunEvent, RunId, RunProvenance,
+    RunServerProvenance, RunSubjectProvenance, ServerSettings, parse_blob_ref,
 };
 use fabro_util::error::{collect_causes, render_with_causes};
 use fabro_util::version::FABRO_VERSION;
 use fabro_vault::{Error as VaultError, SecretType, Vault};
 use fabro_workflow::artifact_upload::ArtifactSink;
+use fabro_workflow::command_log::{command_log_path, read_json_string_blob, read_log_slice};
 use fabro_workflow::event::{self as workflow_event, Emitter};
 use fabro_workflow::handler::HandlerRegistry;
 use fabro_workflow::pipeline::Persisted;
@@ -123,8 +124,8 @@ use crate::run_selector::{ResolveRunError, resolve_run_by_selector};
 use crate::server_secrets::{LlmClientResult, ServerSecrets};
 use crate::spawn_env::{apply_render_graph_env, apply_worker_env};
 use crate::worker_token::{
-    AuthorizeRunBlob, AuthorizeRunScoped, AuthorizeStageArtifact, WorkerTokenKeys,
-    issue_worker_token,
+    AuthorizeCommandLog, AuthorizeRunBlob, AuthorizeRunScoped, AuthorizeStageArtifact,
+    WorkerTokenKeys, issue_worker_token,
 };
 use crate::{
     canonical_host, demo, diagnostics, run_manifest, security_headers, static_files, web_auth,
@@ -1111,6 +1112,10 @@ fn demo_routes() -> Router<Arc<AppState>> {
         .route("/runs/{id}/attach", get(demo::run_events_stub))
         .route("/runs/{id}/blobs", post(not_implemented))
         .route("/runs/{id}/blobs/{blobId}", get(not_implemented))
+        .route(
+            "/runs/{id}/stages/{stageId}/logs/{stream}",
+            get(not_implemented),
+        )
         .route("/runs/{id}/checkpoint", get(demo::checkpoint_stub))
         .route("/runs/{id}/cancel", post(demo::cancel_stub))
         .route("/runs/{id}/start", post(demo::start_run_stub))
@@ -1208,6 +1213,10 @@ fn real_routes() -> Router<Arc<AppState>> {
         .route("/runs/{id}/attach", get(attach_run_events))
         .route("/runs/{id}/blobs", post(write_run_blob))
         .route("/runs/{id}/blobs/{blobId}", get(read_run_blob))
+        .route(
+            "/runs/{id}/stages/{stageId}/logs/{stream}",
+            get(get_run_stage_command_log),
+        )
         .route("/runs/{id}/checkpoint", get(get_checkpoint))
         .route("/runs/{id}/cancel", post(cancel_run))
         .route("/runs/{id}/start", post(start_run))
@@ -3034,6 +3043,30 @@ struct DeleteRunQuery {
     force: bool,
 }
 
+fn default_command_log_limit() -> u64 {
+    65_536
+}
+
+#[derive(Debug, serde::Deserialize)]
+struct CommandLogQuery {
+    #[serde(default)]
+    offset: u64,
+    #[serde(default = "default_command_log_limit")]
+    limit:  u64,
+}
+
+#[derive(Debug, serde::Serialize)]
+struct CommandLogResponseBody {
+    stream:         &'static str,
+    offset:         u64,
+    next_offset:    u64,
+    total_bytes:    u64,
+    bytes_base64:   String,
+    eof:            bool,
+    cas_ref:        Option<String>,
+    live_streaming: bool,
+}
+
 async fn resolve_run(
     _auth: AuthenticatedService,
     State(state): State<Arc<AppState>>,
@@ -5301,6 +5334,144 @@ async fn get_run_logs(
     }
 }
 
+async fn get_run_stage_command_log(
+    AuthorizeCommandLog(id, stage_id, stream): AuthorizeCommandLog,
+    State(state): State<Arc<AppState>>,
+    Query(query): Query<CommandLogQuery>,
+) -> Response {
+    const MAX_COMMAND_LOG_LIMIT: u64 = 1_048_576;
+
+    if query.limit == 0 {
+        return ApiError::bad_request("limit must be greater than 0").into_response();
+    }
+    let limit = query.limit.min(MAX_COMMAND_LOG_LIMIT);
+    let Ok(run_store) = state.store.open_run_reader(&id).await else {
+        return ApiError::not_found("Run not found.").into_response();
+    };
+    let run_state = match run_store.state().await {
+        Ok(run_state) => run_state,
+        Err(err) => {
+            return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
+                .into_response();
+        }
+    };
+    let Some(node) = run_state.node(&stage_id) else {
+        return ApiError::not_found("Stage not found.").into_response();
+    };
+
+    let stream_value = match stream {
+        CommandOutputStream::Stdout => node.stdout.as_deref(),
+        CommandOutputStream::Stderr => node.stderr.as_deref(),
+    };
+    let cas_ref = stream_value
+        .filter(|value| parse_blob_ref(value).is_some())
+        .map(str::to_string);
+    let live_streaming = node
+        .live_streaming
+        .unwrap_or_else(|| cas_ref.is_none() && node.status.is_none());
+    let run_dir = Storage::new(state.server_storage_dir())
+        .run_scratch(&id)
+        .root()
+        .to_path_buf();
+    let scratch_path = command_log_path(&run_dir, &stage_id, stream);
+
+    match read_log_slice(&scratch_path, query.offset, limit).await {
+        Ok((bytes, total_bytes)) => {
+            let offset = query.offset.min(total_bytes);
+            return Json(CommandLogResponseBody {
+                stream: stream.as_str(),
+                offset,
+                next_offset: offset + u64::try_from(bytes.len()).unwrap_or(u64::MAX),
+                total_bytes,
+                bytes_base64: BASE64_STANDARD.encode(bytes),
+                eof: cas_ref.is_some(),
+                cas_ref,
+                live_streaming,
+            })
+            .into_response();
+        }
+        Err(err) if err.kind() == ErrorKind::NotFound => {}
+        Err(err) => {
+            return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
+                .into_response();
+        }
+    }
+
+    if let Some(cas_ref) = cas_ref {
+        let text = match read_json_string_blob(&run_store.clone().into(), &cas_ref).await {
+            Ok(Some(text)) => text,
+            Ok(None) => String::new(),
+            Err(err) => {
+                return ApiError::new(StatusCode::INTERNAL_SERVER_ERROR, err.to_string())
+                    .into_response();
+            }
+        };
+        return command_log_text_response(
+            stream,
+            query.offset,
+            limit,
+            text.as_bytes(),
+            true,
+            Some(cas_ref),
+            live_streaming,
+        );
+    }
+
+    if let Some(inline_text) = stream_value {
+        return command_log_text_response(
+            stream,
+            query.offset,
+            limit,
+            inline_text.as_bytes(),
+            true,
+            None,
+            live_streaming,
+        );
+    }
+
+    let eof = node.status.is_some();
+    Json(CommandLogResponseBody {
+        stream: stream.as_str(),
+        offset: 0,
+        next_offset: 0,
+        total_bytes: 0,
+        bytes_base64: String::new(),
+        eof,
+        cas_ref: None,
+        live_streaming,
+    })
+    .into_response()
+}
+
+fn command_log_text_response(
+    stream: CommandOutputStream,
+    requested_offset: u64,
+    limit: u64,
+    bytes: &[u8],
+    eof: bool,
+    cas_ref: Option<String>,
+    live_streaming: bool,
+) -> Response {
+    let total_bytes = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
+    let offset = requested_offset.min(total_bytes);
+    let start = usize::try_from(offset).unwrap_or(bytes.len());
+    let end = start
+        .saturating_add(usize::try_from(limit).unwrap_or(usize::MAX))
+        .min(bytes.len());
+    let body = &bytes[start..end];
+    Json(CommandLogResponseBody {
+        stream: stream.as_str(),
+        offset,
+        next_offset: offset + u64::try_from(body.len()).unwrap_or(u64::MAX),
+        total_bytes,
+        bytes_base64: BASE64_STANDARD.encode(body),
+        eof,
+        cas_ref,
+        live_streaming,
+    })
+    .into_response()
+}
+
 #[expect(
     clippy::disallowed_types,
     reason = "Pull-request API validates public github.com URLs; these raw URLs are not credential-bearing log output."
@@ -10774,6 +10945,157 @@ slug = "fabro"
         assert_status!(response, StatusCode::NOT_FOUND).await;
     }
 
+    #[tokio::test]
+    async fn get_run_stage_command_log_returns_scratch_slice() {
+        let state = create_app_state_with_isolated_storage();
+        let app = build_router(Arc::clone(&state), AuthMode::Disabled);
+        let run_id = RunId::new();
+        let stage_id = StageId::new("script_node", 1);
+        create_durable_run_with_events(&state, run_id, &[
+            workflow_event::Event::RunSubmitted {
+                definition_blob: None,
+            },
+            workflow_event::Event::StageStarted {
+                node_id:      "script_node".to_string(),
+                name:         "Script".to_string(),
+                index:        1,
+                handler_type: "command".to_string(),
+                attempt:      1,
+                max_attempts: 1,
+            },
+            workflow_event::Event::CommandStarted {
+                node_id:    "script_node".to_string(),
+                script:     "echo hello world".to_string(),
+                command:    "echo hello world".to_string(),
+                language:   "shell".to_string(),
+                timeout_ms: None,
+            },
+        ])
+        .await;
+        let run_dir = Storage::new(state.server_storage_dir())
+            .run_scratch(&run_id)
+            .root()
+            .to_path_buf();
+        let log_path = command_log_path(&run_dir, &stage_id, CommandOutputStream::Stdout);
+        tokio::fs::create_dir_all(log_path.parent().unwrap())
+            .await
+            .unwrap();
+        tokio::fs::write(&log_path, b"hello world").await.unwrap();
+
+        let req = Request::builder()
+            .method("GET")
+            .uri(api(&format!(
+                "/runs/{run_id}/stages/{stage_id}/logs/stdout?offset=6&limit=5"
+            )))
+            .body(Body::empty())
+            .unwrap();
+
+        let response = app.oneshot(req).await.unwrap();
+        let body = response_json!(response, StatusCode::OK).await;
+        let bytes = BASE64_STANDARD
+            .decode(body["bytes_base64"].as_str().unwrap())
+            .unwrap();
+
+        assert_eq!(body["stream"], "stdout");
+        assert_eq!(body["offset"], 6);
+        assert_eq!(body["next_offset"], 11);
+        assert_eq!(body["total_bytes"], 11);
+        assert_eq!(bytes, b"world");
+        assert_eq!(body["eof"], false);
+        assert_eq!(body["cas_ref"], serde_json::Value::Null);
+        assert_eq!(body["live_streaming"], true);
+    }
+
+    #[tokio::test]
+    async fn get_run_stage_command_log_returns_cas_slice() {
+        let state = create_app_state_with_isolated_storage();
+        let app = build_router(Arc::clone(&state), AuthMode::Disabled);
+        let run_id = RunId::new();
+        let run_store = state.store.create_run(&run_id).await.unwrap();
+        let stdout_blob = run_store
+            .write_blob(&serde_json::to_vec("hello world").unwrap())
+            .await
+            .unwrap();
+        let stderr_blob = run_store
+            .write_blob(&serde_json::to_vec("").unwrap())
+            .await
+            .unwrap();
+        let stdout_ref = format!("blob://sha256/{stdout_blob}");
+        let stderr_ref = format!("blob://sha256/{stderr_blob}");
+        for event in [
+            workflow_event::Event::RunSubmitted {
+                definition_blob: None,
+            },
+            workflow_event::Event::StageStarted {
+                node_id:      "script_node".to_string(),
+                name:         "Script".to_string(),
+                index:        1,
+                handler_type: "command".to_string(),
+                attempt:      1,
+                max_attempts: 1,
+            },
+            workflow_event::Event::CommandCompleted {
+                node_id:           "script_node".to_string(),
+                stdout:            stdout_ref.clone(),
+                stderr:            stderr_ref,
+                exit_code:         Some(0),
+                duration_ms:       5,
+                timed_out:         false,
+                stdout_bytes:      11,
+                stderr_bytes:      0,
+                streams_separated: true,
+                live_streaming:    false,
+            },
+        ] {
+            workflow_event::append_event(&run_store, &run_id, &event)
+                .await
+                .unwrap();
+        }
+
+        let req = Request::builder()
+            .method("GET")
+            .uri(api(&format!(
+                "/runs/{run_id}/stages/script_node@1/logs/stdout?offset=6&limit=5"
+            )))
+            .body(Body::empty())
+            .unwrap();
+
+        let response = app.oneshot(req).await.unwrap();
+        let body = response_json!(response, StatusCode::OK).await;
+        let bytes = BASE64_STANDARD
+            .decode(body["bytes_base64"].as_str().unwrap())
+            .unwrap();
+
+        assert_eq!(body["stream"], "stdout");
+        assert_eq!(body["offset"], 6);
+        assert_eq!(body["next_offset"], 11);
+        assert_eq!(body["total_bytes"], 11);
+        assert_eq!(bytes, b"world");
+        assert_eq!(body["eof"], true);
+        assert_eq!(body["cas_ref"], stdout_ref);
+        assert_eq!(body["live_streaming"], false);
+    }
+
+    #[tokio::test]
+    async fn get_run_stage_command_log_returns_not_found_for_missing_stage() {
+        let state = create_app_state_with_isolated_storage();
+        let app = build_router(Arc::clone(&state), AuthMode::Disabled);
+        let run_id = RunId::new();
+        create_durable_run_with_events(&state, run_id, &[workflow_event::Event::RunSubmitted {
+            definition_blob: None,
+        }])
+        .await;
+
+        let req = Request::builder()
+            .method("GET")
+            .uri(api(&format!("/runs/{run_id}/stages/missing@1/logs/stdout")))
+            .body(Body::empty())
+            .unwrap();
+
+        let response = app.oneshot(req).await.unwrap();
+        assert_status!(response, StatusCode::NOT_FOUND).await;
+    }
+
     #[tokio::test]
     async fn get_run_pull_request_returns_live_detail_from_github() {
         let github = MockServer::start();
@@ -12132,6 +12454,78 @@ slug = "fabro"
         assert_status!(response, StatusCode::UNAUTHORIZED).await;
     }
 
+    #[tokio::test]
+    async fn worker_token_controls_command_log_route() {
+        let (state, app) = jwt_auth_app();
+        let user_jwt = issue_test_user_jwt();
+        let run_id = create_run_with_bearer(&app, &user_jwt).await;
+        let worker_token = issue_test_worker_token(&run_id);
+        let other_run_id = create_run_with_bearer(&app, &user_jwt).await;
+        let mismatched_worker_token = issue_test_worker_token(&other_run_id);
+        let run_store = state.store.open_run(&run_id).await.unwrap();
+        workflow_event::append_event(
+            &run_store,
+            &run_id,
+            &workflow_event::Event::CommandStarted {
+                node_id:    "code".to_string(),
+                script:     "echo hello".to_string(),
+                command:    "echo hello".to_string(),
+                language:   "shell".to_string(),
+                timeout_ms: None,
+            },
+        )
+        .await
+        .unwrap();
+
+        let response = app
+            .clone()
+            .oneshot(bearer_request(
+                Method::GET,
+                &format!("/runs/{run_id}/stages/code@1/logs/stdout"),
+                &worker_token,
+                Body::empty(),
+            ))
+            .await
+            .unwrap();
+        assert_status!(response, StatusCode::OK).await;
+
+        let response = app
+            .clone()
+            .oneshot(bearer_request(
+                Method::GET,
+                &format!("/runs/{run_id}/stages/code@1/logs/stdout"),
+                &user_jwt,
+                Body::empty(),
+            ))
+            .await
+            .unwrap();
+        assert_status!(response, StatusCode::OK).await;
+
+        let response = app
+            .clone()
+            .oneshot(bearer_request(
+                Method::GET,
+                &format!("/runs/{run_id}/stages/code@1/logs/stdout"),
+                &mismatched_worker_token,
+                Body::empty(),
+            ))
+            .await
+            .unwrap();
+        assert_status!(response, StatusCode::FORBIDDEN).await;
+
+        let response = app
+            .oneshot(
+                Request::builder()
+                    .method(Method::GET)
+                    .uri(api(&format!("/runs/{run_id}/stages/code@1/logs/stdout")))
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        assert_status!(response, StatusCode::UNAUTHORIZED).await;
+    }
+
     #[tokio::test]
     async fn worker_token_is_rejected_on_user_only_routes() {
         let (_state, app) = jwt_auth_app();

lib/crates/fabro-server/src/worker_token.rs

@@ -5,7 +5,7 @@ use axum::extract::{FromRequestParts, Path};
 use axum::http::StatusCode;
 use axum::http::request::Parts;
 use axum::response::{IntoResponse, Response};
-use fabro_types::{RunBlobId, RunId, StageId};
+use fabro_types::{CommandOutputStream, RunBlobId, RunId, StageId};
 use jsonwebtoken::{Algorithm, DecodingKey, EncodingKey, Header, Validation};
 use tracing::{debug, warn};
 use uuid::Uuid;
@@ -197,6 +197,34 @@ impl FromRequestParts<Arc<AppState>> for AuthorizeStageArtifact {
     }
 }
 
+pub(crate) struct AuthorizeCommandLog(
+    pub(crate) RunId,
+    pub(crate) StageId,
+    pub(crate) CommandOutputStream,
+);
+
+impl FromRequestParts<Arc<AppState>> for AuthorizeCommandLog {
+    type Rejection = Response;
+
+    async fn from_request_parts(
+        parts: &mut Parts,
+        state: &Arc<AppState>,
+    ) -> Result<Self, Self::Rejection> {
+        let Path((id, stage_id, stream)): Path<(String, String, String)> =
+            Path::from_request_parts(parts, state)
+                .await
+                .map_err(IntoResponse::into_response)?;
+        let run_id = parse_run_id_path(&id)?;
+        let stage_id = parse_stage_id_path(&stage_id)?;
+        let stream = stream
+            .parse::<CommandOutputStream>()
+            .map_err(|_| ApiError::bad_request("Invalid command log stream.").into_response())?;
+        authorize_run_scoped(parts, state.as_ref(), &run_id)
+            .map_err(IntoResponse::into_response)?;
+        Ok(Self(run_id, stage_id, stream))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::{Arc, Mutex as StdMutex};

lib/crates/fabro-store/src/artifact_store.rs

@@ -216,6 +216,15 @@ fn encode_path_segment(segment: &str) -> String {
     utf8_percent_encode(segment, ARTIFACT_SEGMENT_ENCODE_SET).to_string()
 }
 
+#[must_use]
+pub fn stage_storage_segment(node: &StageId) -> String {
+    format!(
+        "{}@{:04}",
+        encode_path_segment(node.node_id()),
+        node.visit()
+    )
+}
+
 fn decode_path_segment(kind: &str, value: &str) -> Result<String> {
     percent_decode_str(value)
         .decode_utf8()

lib/crates/fabro-store/src/lib.rs

@@ -10,7 +10,7 @@ mod serializable_projection;
 mod slate;
 mod types;
 
-pub use artifact_store::{ArtifactStore, NodeArtifact};
+pub use artifact_store::{ArtifactStore, NodeArtifact, stage_storage_segment};
 pub use error::{Error, Result};
 pub use fabro_types::{
     EventEnvelope, NodeState, PendingInterviewRecord, RunBlobId, RunProjection, RunSummary, StageId,

lib/crates/fabro-store/src/run_state.rs

@@ -342,6 +342,10 @@ impl RunProjectionReducer for RunProjection {
                 let node = self.node_mut(node_id, visit);
                 node.stdout = Some(props.stdout.clone());
                 node.stderr = Some(props.stderr.clone());
+                node.stdout_bytes = Some(props.stdout_bytes);
+                node.stderr_bytes = Some(props.stderr_bytes);
+                node.streams_separated = Some(props.streams_separated);
+                node.live_streaming = Some(props.live_streaming);
                 node.script_timing = Some(serde_json::to_value(props).map_err(|err| {
                     Error::InvalidEvent(format!("invalid command.completed payload: {err}"))
                 })?);

lib/crates/fabro-store/tests/serializable_projection.rs

@@ -96,6 +96,10 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() {
         parallel_results:  Some(json!([{ "stage": "fanout@1" }])),
         stdout:            Some("stdout".to_string()),
         stderr:            Some("stderr".to_string()),
+        stdout_bytes:      None,
+        stderr_bytes:      None,
+        streams_separated: None,
+        live_streaming:    None,
     });
 
     let serialized = serde_json::to_value(SerializableProjection(&projection))

lib/crates/fabro-types/src/command_output.rs

@@ -0,0 +1,44 @@
+use std::path::PathBuf;
+
+use serde::{Deserialize, Serialize};
+use strum::{Display, EnumString, IntoStaticStr};
+
+#[derive(
+    Debug,
+    Clone,
+    Copy,
+    PartialEq,
+    Eq,
+    Hash,
+    Serialize,
+    Deserialize,
+    Display,
+    EnumString,
+    IntoStaticStr,
+)]
+#[serde(rename_all = "snake_case")]
+#[strum(serialize_all = "snake_case")]
+pub enum CommandOutputStream {
+    Stdout,
+    Stderr,
+}
+
+impl CommandOutputStream {
+    #[must_use]
+    pub fn as_str(self) -> &'static str {
+        self.into()
+    }
+
+    #[must_use]
+    pub fn log_filename(self) -> &'static str {
+        match self {
+            Self::Stdout => "stdout.log",
+            Self::Stderr => "stderr.log",
+        }
+    }
+
+    #[must_use]
+    pub fn command_log_relative_path(self) -> PathBuf {
+        PathBuf::from("command").join(self.log_filename())
+    }
+}

lib/crates/fabro-types/src/lib.rs

@@ -5,6 +5,7 @@ pub mod auth;
 pub mod billing;
 pub mod blob_ref;
 pub mod checkpoint;
+pub mod command_output;
 pub mod conclusion;
 pub mod dense;
 pub mod diff;
@@ -42,6 +43,7 @@ pub use blob_ref::{
     format_blob_ref, parse_blob_ref, parse_legacy_blob_file_ref, parse_managed_blob_file_ref,
 };
 pub use checkpoint::Checkpoint;
+pub use command_output::CommandOutputStream;
 pub use conclusion::{Conclusion, StageSummary};
 pub use dense::{ServerSettings, UserSettings, WorkflowSettings};
 pub use diff::DiffStats;

lib/crates/fabro-types/src/run_event/misc.rs

@@ -201,12 +201,20 @@ pub struct CommandStartedProps {
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct CommandCompletedProps {
-    pub stdout:      String,
-    pub stderr:      String,
+    pub stdout:            String,
+    pub stderr:            String,
     #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub exit_code:   Option<i32>,
-    pub duration_ms: u64,
-    pub timed_out:   bool,
+    pub exit_code:         Option<i32>,
+    pub duration_ms:       u64,
+    pub timed_out:         bool,
+    #[serde(default)]
+    pub stdout_bytes:      u64,
+    #[serde(default)]
+    pub stderr_bytes:      u64,
+    #[serde(default)]
+    pub streams_separated: bool,
+    #[serde(default)]
+    pub live_streaming:    bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

lib/crates/fabro-types/src/run_projection.rs

@@ -49,6 +49,14 @@ pub struct NodeState {
     pub parallel_results:  Option<serde_json::Value>,
     pub stdout:            Option<String>,
     pub stderr:            Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stdout_bytes:      Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub stderr_bytes:      Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub streams_separated: Option<bool>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub live_streaming:    Option<bool>,
 }
 
 impl RunProjection {

lib/crates/fabro-workflow/src/artifact.rs

@@ -120,6 +120,19 @@ pub async fn resolve_context_for_execution(
     Ok(resolved)
 }
 
+pub async fn resolve_context_for_edge_selection(
+    context: &Context,
+    run_store: &RunStoreHandle,
+) -> Result<Context> {
+    let mut values = context.snapshot();
+    for key in [context::keys::COMMAND_OUTPUT, context::keys::COMMAND_STDERR] {
+        if let Some(Value::String(current)) = values.get_mut(key) {
+            *current = resolve_text_or_blob_ref_str(current, run_store).await?;
+        }
+    }
+    Ok(Context::from_values(values))
+}
+
 pub async fn resolve_outcomes_for_execution(
     node_outcomes: &HashMap<String, Outcome>,
     run_store: &RunStoreHandle,
@@ -144,6 +157,29 @@ pub async fn resolved_context_snapshot(
     Ok(values)
 }
 
+pub async fn resolve_text_or_blob_ref(value: &Value, run_store: &RunStoreHandle) -> Result<String> {
+    match value.as_str() {
+        Some(current) => resolve_text_or_blob_ref_str(current, run_store).await,
+        None => Ok(value.to_string()),
+    }
+}
+
+pub async fn resolve_text_or_blob_ref_str(
+    current: &str,
+    run_store: &RunStoreHandle,
+) -> Result<String> {
+    let Some(blob_id) = parse_blob_ref(current) else {
+        return Ok(current.to_string());
+    };
+    let bytes = run_store
+        .read_blob(&blob_id)
+        .await
+        .map_err(|e| Error::engine(format!("text blob read failed: {e}")))?
+        .ok_or_else(|| Error::engine(format!("text blob missing: {blob_id}")))?;
+    serde_json::from_slice::<String>(&bytes)
+        .map_err(|e| Error::engine(format!("text blob was not a JSON string: {e}")))
+}
+
 /// Sync artifact files to a remote sandbox.
 ///
 /// For each `file://` pointer in `updates`, checks whether the file is accessible
@@ -225,14 +261,15 @@ fn resolve_execution_values<'a>(
     run_dir: &'a Path,
 ) -> BoxFuture<'a, Result<()>> {
     Box::pin(async move {
-        for value in values.values_mut() {
-            resolve_execution_value(value, run_store, env, run_dir).await?;
+        for (key, value) in values.iter_mut() {
+            resolve_execution_value(Some(key.as_str()), value, run_store, env, run_dir).await?;
         }
         Ok(())
     })
 }
 
 fn resolve_execution_value<'a>(
+    key: Option<&'a str>,
     value: &'a mut Value,
     run_store: &'a RunStoreHandle,
     env: &'a dyn Sandbox,
@@ -241,7 +278,12 @@ fn resolve_execution_value<'a>(
     Box::pin(async move {
         match value {
             Value::String(current) => {
-                if let Some(blob_id) =
+                if matches!(
+                    key,
+                    Some(context::keys::COMMAND_OUTPUT | context::keys::COMMAND_STDERR)
+                ) {
+                    *current = resolve_text_or_blob_ref_str(current, run_store).await?;
+                } else if let Some(blob_id) =
                     parse_blob_ref(current).or_else(|| parse_legacy_blob_file_ref(current))
                 {
                     *current = materialize_blob_ref(&blob_id, run_store, env, run_dir).await?;
@@ -253,12 +295,12 @@ fn resolve_execution_value<'a>(
             }
             Value::Array(items) => {
                 for item in items {
-                    resolve_execution_value(item, run_store, env, run_dir).await?;
+                    resolve_execution_value(None, item, run_store, env, run_dir).await?;
                 }
             }
             Value::Object(map) => {
                 for item in map.values_mut() {
-                    resolve_execution_value(item, run_store, env, run_dir).await?;
+                    resolve_execution_value(None, item, run_store, env, run_dir).await?;
                 }
             }
             Value::Null | Value::Bool(_) | Value::Number(_) => {}

lib/crates/fabro-workflow/src/command_log.rs

@@ -0,0 +1,198 @@
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+
+use fabro_config::RunScratch;
+use fabro_store::stage_storage_segment;
+use fabro_types::{CommandOutputStream, StageId, format_blob_ref};
+use serde_json::Value;
+use tokio::fs::{self, File, OpenOptions};
+use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
+use tokio::sync::Mutex;
+
+use crate::error::{Error, Result};
+use crate::runtime_store::RunStoreHandle;
+
+#[derive(Debug, Clone)]
+pub struct FinalizedCommandLogs {
+    pub stdout_ref:   String,
+    pub stderr_ref:   String,
+    pub stdout_bytes: u64,
+    pub stderr_bytes: u64,
+    pub stdout_text:  String,
+    pub stderr_text:  String,
+}
+
+pub struct CommandLogRecorder {
+    stdout:       Mutex<File>,
+    stderr:       Mutex<File>,
+    stdout_bytes: AtomicU64,
+    stderr_bytes: AtomicU64,
+    stdout_path:  PathBuf,
+    stderr_path:  PathBuf,
+}
+
+impl CommandLogRecorder {
+    pub async fn create(run_dir: &Path, stage_id: &StageId) -> Result<Arc<Self>> {
+        let stdout_path = command_log_path(run_dir, stage_id, CommandOutputStream::Stdout);
+        let stderr_path = command_log_path(run_dir, stage_id, CommandOutputStream::Stderr);
+        if let Some(parent) = stdout_path.parent() {
+            fs::create_dir_all(parent).await.map_err(|err| {
+                Error::Io(format!(
+                    "creating command log directory {}: {err}",
+                    parent.display()
+                ))
+            })?;
+        }
+        let stdout = open_truncated(&stdout_path).await?;
+        let stderr = open_truncated(&stderr_path).await?;
+        Ok(Arc::new(Self {
+            stdout: Mutex::new(stdout),
+            stderr: Mutex::new(stderr),
+            stdout_bytes: AtomicU64::new(0),
+            stderr_bytes: AtomicU64::new(0),
+            stdout_path,
+            stderr_path,
+        }))
+    }
+
+    pub async fn append(&self, stream: CommandOutputStream, bytes: &[u8]) -> Result<()> {
+        if bytes.is_empty() {
+            return Ok(());
+        }
+        let mut file = match stream {
+            CommandOutputStream::Stdout => self.stdout.lock().await,
+            CommandOutputStream::Stderr => self.stderr.lock().await,
+        };
+        file.write_all(bytes)
+            .await
+            .map_err(|err| Error::Io(format!("writing command {stream} log failed: {err}")))?;
+        file.flush()
+            .await
+            .map_err(|err| Error::Io(format!("flushing command {stream} log failed: {err}")))?;
+        let len = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
+        match stream {
+            CommandOutputStream::Stdout => {
+                self.stdout_bytes.fetch_add(len, Ordering::Relaxed);
+            }
+            CommandOutputStream::Stderr => {
+                self.stderr_bytes.fetch_add(len, Ordering::Relaxed);
+            }
+        }
+        Ok(())
+    }
+
+    pub async fn finalize(&self, run_store: &RunStoreHandle) -> Result<FinalizedCommandLogs> {
+        self.flush_all().await?;
+        let stdout_text = read_lossy_text(&self.stdout_path).await?;
+        let stderr_text = read_lossy_text(&self.stderr_path).await?;
+        let stdout_bytes = self.stdout_bytes();
+        let stderr_bytes = self.stderr_bytes();
+        let stdout_ref = write_json_string_blob(run_store, &stdout_text).await?;
+        let stderr_ref = write_json_string_blob(run_store, &stderr_text).await?;
+        Ok(FinalizedCommandLogs {
+            stdout_ref,
+            stderr_ref,
+            stdout_bytes,
+            stderr_bytes,
+            stdout_text,
+            stderr_text,
+        })
+    }
+
+    pub fn stdout_bytes(&self) -> u64 {
+        self.stdout_bytes.load(Ordering::Relaxed)
+    }
+
+    pub fn stderr_bytes(&self) -> u64 {
+        self.stderr_bytes.load(Ordering::Relaxed)
+    }
+
+    async fn flush_all(&self) -> Result<()> {
+        self.stdout
+            .lock()
+            .await
+            .flush()
+            .await
+            .map_err(|err| Error::Io(format!("flushing stdout command log failed: {err}")))?;
+        self.stderr
+            .lock()
+            .await
+            .flush()
+            .await
+            .map_err(|err| Error::Io(format!("flushing stderr command log failed: {err}")))?;
+        Ok(())
+    }
+}
+
+pub fn command_log_path(
+    run_dir: &Path,
+    stage_id: &StageId,
+    stream: CommandOutputStream,
+) -> PathBuf {
+    RunScratch::new(run_dir)
+        .runtime_dir()
+        .join("stages")
+        .join(stage_storage_segment(stage_id))
+        .join(stream.command_log_relative_path())
+}
+
+pub async fn read_log_slice(
+    path: &Path,
+    offset: u64,
+    limit: u64,
+) -> std::io::Result<(Vec<u8>, u64)> {
+    let mut file = fs::File::open(path).await?;
+    let total = file.metadata().await?.len();
+    let start = offset.min(total);
+    file.seek(std::io::SeekFrom::Start(start)).await?;
+    let take = limit.min(total.saturating_sub(start));
+    let mut buf = vec![0; usize::try_from(take).unwrap_or(usize::MAX)];
+    file.read_exact(&mut buf).await?;
+    Ok((buf, total))
+}
+
+pub async fn read_json_string_blob(
+    run_store: &RunStoreHandle,
+    blob_ref: &str,
+) -> Result<Option<String>> {
+    let Some(blob_id) = fabro_types::parse_blob_ref(blob_ref) else {
+        return Ok(None);
+    };
+    let bytes = run_store
+        .read_blob(&blob_id)
+        .await
+        .map_err(|err| Error::engine(format!("command log blob read failed: {err}")))?
+        .ok_or_else(|| Error::engine(format!("command log blob missing: {blob_id}")))?;
+    let text = serde_json::from_slice::<String>(&bytes)
+        .map_err(|err| Error::engine(format!("command log blob was not a JSON string: {err}")))?;
+    Ok(Some(text))
+}
+
+async fn open_truncated(path: &Path) -> Result<File> {
+    OpenOptions::new()
+        .create(true)
+        .write(true)
+        .truncate(true)
+        .open(path)
+        .await
+        .map_err(|err| Error::Io(format!("opening command log {}: {err}", path.display())))
+}
+
+async fn read_lossy_text(path: &Path) -> Result<String> {
+    let bytes = fs::read(path)
+        .await
+        .map_err(|err| Error::Io(format!("reading command log {}: {err}", path.display())))?;
+    Ok(String::from_utf8_lossy(&bytes).into_owned())
+}
+
+async fn write_json_string_blob(run_store: &RunStoreHandle, text: &str) -> Result<String> {
+    let value = Value::String(text.to_string());
+    let bytes = serde_json::to_vec(&value)
+        .map_err(|err| Error::engine(format!("command log JSON serialization failed: {err}")))?;
+    let blob_id = run_store
+        .write_blob(&bytes)
+        .await
+        .map_err(|err| Error::engine(format!("command log blob write failed: {err}")))?;
+    Ok(format_blob_ref(&blob_id))
+}

lib/crates/fabro-workflow/src/event.rs

@@ -499,13 +499,17 @@ pub enum Event {
         timeout_ms: Option<u64>,
     },
     CommandCompleted {
-        node_id:     String,
-        stdout:      String,
-        stderr:      String,
+        node_id:           String,
+        stdout:            String,
+        stderr:            String,
         #[serde(default, skip_serializing_if = "Option::is_none")]
-        exit_code:   Option<i32>,
-        duration_ms: u64,
-        timed_out:   bool,
+        exit_code:         Option<i32>,
+        duration_ms:       u64,
+        timed_out:         bool,
+        stdout_bytes:      u64,
+        stderr_bytes:      u64,
+        streams_separated: bool,
+        live_streaming:    bool,
     },
     AgentCliStarted {
         node_id:  String,
@@ -1111,11 +1115,18 @@ impl Event {
                 exit_code,
                 duration_ms,
                 timed_out,
+                stdout_bytes,
+                stderr_bytes,
                 ..
             } => {
                 debug!(
                     node_id,
-                    exit_code, duration_ms, timed_out, "Command completed"
+                    exit_code,
+                    duration_ms,
+                    timed_out,
+                    stdout_bytes,
+                    stderr_bytes,
+                    "Command completed"
                 );
             }
             Self::AgentCliStarted {
@@ -2487,13 +2498,21 @@ fn event_body_from_event(event: &Event) -> EventBody {
             exit_code,
             duration_ms,
             timed_out,
+            stdout_bytes,
+            stderr_bytes,
+            streams_separated,
+            live_streaming,
             ..
         } => EventBody::CommandCompleted(fabro_types::CommandCompletedProps {
-            stdout:      stdout.clone(),
-            stderr:      stderr.clone(),
-            exit_code:   *exit_code,
-            duration_ms: *duration_ms,
-            timed_out:   *timed_out,
+            stdout:            stdout.clone(),
+            stderr:            stderr.clone(),
+            exit_code:         *exit_code,
+            duration_ms:       *duration_ms,
+            timed_out:         *timed_out,
+            stdout_bytes:      *stdout_bytes,
+            stderr_bytes:      *stderr_bytes,
+            streams_separated: *streams_separated,
+            live_streaming:    *live_streaming,
         }),
         Event::AgentCliStarted {
             visit,
@@ -2697,6 +2716,11 @@ impl StageScope {
             parallel_branch_id: Some(parallel_branch_id),
         }
     }
+
+    #[must_use]
+    pub fn stage_id(&self) -> StageId {
+        StageId::new(self.node_id.clone(), self.visit)
+    }
 }
 
 #[must_use]

lib/crates/fabro-workflow/src/git.rs

@@ -509,12 +509,16 @@ mod tests {
         .await
         .unwrap();
         append_event(&run, &fixtures::RUN_1, &Event::CommandCompleted {
-            node_id:     "work".into(),
-            stdout:      "hi\n".into(),
-            stderr:      String::new(),
-            exit_code:   Some(0),
-            duration_ms: 10,
-            timed_out:   false,
+            node_id:           "work".into(),
+            stdout:            "hi\n".into(),
+            stderr:            String::new(),
+            exit_code:         Some(0),
+            duration_ms:       10,
+            timed_out:         false,
+            stdout_bytes:      3,
+            stderr_bytes:      0,
+            streams_separated: true,
+            live_streaming:    true,
         })
         .await
         .unwrap();

lib/crates/fabro-workflow/src/handler/command.rs

@@ -1,9 +1,11 @@
 use std::path::Path;
 
 use async_trait::async_trait;
+use fabro_agent::CommandOutputCallback;
 use fabro_graphviz::graph::{Graph, Node};
 
 use super::{EngineServices, Handler};
+use crate::command_log::CommandLogRecorder;
 use crate::context::{Context, keys};
 use crate::error::Error;
 use crate::event::{Event, StageScope};
@@ -57,7 +59,7 @@ impl Handler for CommandHandler {
         node: &Node,
         context: &Context,
         _graph: &Graph,
-        _run_dir: &Path,
+        run_dir: &Path,
         services: &EngineServices,
     ) -> Result<Outcome, Error> {
         let script = node
@@ -107,25 +109,53 @@ impl Handler for CommandHandler {
             Some(&services.env)
         };
         let cancel_token = services.run.sandbox_cancel_token();
+        let stage_id = stage_scope.stage_id();
+        let recorder = CommandLogRecorder::create(run_dir, &stage_id).await?;
+        let output_callback: CommandOutputCallback = {
+            let recorder = recorder.clone();
+            std::sync::Arc::new(move |stream, bytes| {
+                let recorder = recorder.clone();
+                Box::pin(async move {
+                    recorder
+                        .append(stream, &bytes)
+                        .await
+                        .map_err(|err| fabro_sandbox::Error::message(err.to_string()))
+                })
+            })
+        };
 
         let result = services
             .run
             .sandbox
-            .exec_command(&command, timeout_ms, None, env_vars, cancel_token.clone())
+            .exec_command_streaming(
+                &command,
+                timeout_ms,
+                None,
+                env_vars,
+                cancel_token.clone(),
+                output_callback,
+            )
             .await;
         if let Some(token) = cancel_token {
             token.cancel();
         }
-        let result = result.map_err(|e| Error::handler(format!("Failed to spawn script: {e}")))?;
+        let streaming =
+            result.map_err(|e| Error::handler(format!("Failed to spawn script: {e}")))?;
+        let result = streaming.result;
+        let finalized = recorder.finalize(&services.run.run_store).await?;
 
         services.run.emitter.emit_scoped(
             &Event::CommandCompleted {
-                node_id:     node.id.clone(),
-                stdout:      result.stdout.clone(),
-                stderr:      result.stderr.clone(),
-                exit_code:   (!result.timed_out).then_some(result.exit_code),
-                duration_ms: result.duration_ms,
-                timed_out:   result.timed_out,
+                node_id:           node.id.clone(),
+                stdout:            finalized.stdout_ref.clone(),
+                stderr:            finalized.stderr_ref.clone(),
+                exit_code:         (!result.timed_out).then_some(result.exit_code),
+                duration_ms:       result.duration_ms,
+                timed_out:         result.timed_out,
+                stdout_bytes:      finalized.stdout_bytes,
+                stderr_bytes:      finalized.stderr_bytes,
+                streams_separated: streaming.streams_separated,
+                live_streaming:    streaming.live_streaming,
             },
             &stage_scope,
         );
@@ -140,54 +170,120 @@ impl Handler for CommandHandler {
             let mut outcome = Outcome::success();
             outcome.context_updates.insert(
                 keys::COMMAND_OUTPUT.to_string(),
-                serde_json::json!(result.stdout),
+                serde_json::json!(finalized.stdout_ref),
             );
             outcome.context_updates.insert(
                 keys::COMMAND_STDERR.to_string(),
-                serde_json::json!(result.stderr),
+                serde_json::json!(finalized.stderr_ref),
             );
             outcome.notes = Some(format!("Script completed: {script}"));
             Ok(outcome)
         } else {
             let mut reason = format!("Script failed with exit code: {}", result.exit_code);
-            if !result.stdout.trim().is_empty() {
+            let stdout_tail = tail_bytes(&finalized.stdout_text, 4096);
+            let stderr_tail = tail_bytes(&finalized.stderr_text, 4096);
+            if !stdout_tail.trim().is_empty() {
                 reason.push_str("\n\n## stdout\n");
-                reason.push_str(&result.stdout);
+                reason.push_str(&stdout_tail);
             }
-            if !result.stderr.trim().is_empty() {
+            if !stderr_tail.trim().is_empty() {
                 reason.push_str("\n\n## stderr\n");
-                reason.push_str(&result.stderr);
+                reason.push_str(&stderr_tail);
             }
             let mut outcome = Outcome::fail_classify(reason);
             outcome.context_updates.insert(
                 keys::COMMAND_OUTPUT.to_string(),
-                serde_json::json!(result.stdout),
+                serde_json::json!(finalized.stdout_ref),
             );
             outcome.context_updates.insert(
                 keys::COMMAND_STDERR.to_string(),
-                serde_json::json!(result.stderr),
+                serde_json::json!(finalized.stderr_ref),
             );
             Ok(outcome)
         }
     }
 }
 
+fn tail_bytes(text: &str, max_bytes: usize) -> String {
+    if text.len() <= max_bytes {
+        return text.to_string();
+    }
+    let mut start = text.len() - max_bytes;
+    while !text.is_char_boundary(start) {
+        start += 1;
+    }
+    text[start..].to_string()
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
     use std::sync::atomic::AtomicBool;
     use std::time::Duration;
 
+    use bytes::Bytes;
     use fabro_graphviz::graph::AttrValue;
     use fabro_store::{Database, RunDatabase, StageId};
     use fabro_types::fixtures;
     use object_store::memory::InMemory;
+    use tokio::sync::Mutex;
 
     use super::*;
     use crate::outcome::StageOutcome;
+    use crate::runtime_store::{RunStoreBackend, RunStoreHandle};
+
+    #[derive(Default)]
+    struct MemoryRunStoreBackend {
+        blobs: Mutex<std::collections::HashMap<fabro_types::RunBlobId, Bytes>>,
+    }
+
+    #[async_trait::async_trait]
+    impl RunStoreBackend for MemoryRunStoreBackend {
+        async fn load_state(&self) -> anyhow::Result<fabro_store::RunProjection> {
+            Ok(fabro_store::RunProjection::default())
+        }
+
+        async fn list_events(&self) -> anyhow::Result<Vec<fabro_store::EventEnvelope>> {
+            Ok(Vec::new())
+        }
+
+        async fn append_run_event(&self, _event: &fabro_types::RunEvent) -> anyhow::Result<()> {
+            Ok(())
+        }
+
+        async fn write_blob(&self, data: &[u8]) -> anyhow::Result<fabro_types::RunBlobId> {
+            let blob_id = fabro_types::RunBlobId::new(data);
+            self.blobs
+                .lock()
+                .await
+                .insert(blob_id, Bytes::copy_from_slice(data));
+            Ok(blob_id)
+        }
+
+        async fn read_blob(&self, id: &fabro_types::RunBlobId) -> anyhow::Result<Option<Bytes>> {
+            Ok(self.blobs.lock().await.get(id).cloned())
+        }
+    }
 
     fn make_services() -> EngineServices {
-        EngineServices::test_default()
+        let mut services = EngineServices::test_default();
+        services.run = services.run.with_run_store(RunStoreHandle::new(Arc::new(
+            MemoryRunStoreBackend::default(),
+        )));
+        services
+    }
+
+    async fn command_text(services: &EngineServices, value: &serde_json::Value) -> String {
+        crate::artifact::resolve_text_or_blob_ref(value, &services.run.run_store)
+            .await
+            .unwrap()
+    }
+
+    async fn command_log_text(services: &EngineServices, value: &str) -> String {
+        crate::command_log::read_json_string_blob(&services.run.run_store, value)
+            .await
+            .unwrap()
+            .unwrap_or_else(|| value.to_string())
     }
 
     fn test_store() -> Arc<Database> {
@@ -224,8 +320,9 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Failed {
@@ -304,16 +401,21 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Succeeded);
         assert!(outcome.notes.as_deref().unwrap().contains("echo hello"));
         let command_output = outcome.context_updates.get(keys::COMMAND_OUTPUT).unwrap();
-        assert!(command_output.as_str().unwrap().contains("hello"));
+        assert!(
+            command_text(&services, command_output)
+                .await
+                .contains("hello")
+        );
         let command_stderr = outcome.context_updates.get(keys::COMMAND_STDERR).unwrap();
-        assert_eq!(command_stderr.as_str().unwrap(), "");
+        assert_eq!(command_text(&services, command_stderr).await, "");
     }
 
     #[tokio::test]
@@ -326,8 +428,9 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Failed {
@@ -442,9 +545,13 @@ mod tests {
         let snapshot = run_store.state().await.unwrap();
         let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap();
         let stdout = node_state.stdout.as_deref().unwrap();
-        assert_eq!(stdout.trim(), "hello");
+        assert_eq!(command_log_text(&services, stdout).await.trim(), "hello");
         let stderr = node_state.stderr.as_deref().unwrap();
-        assert_eq!(stderr, "");
+        assert_eq!(command_log_text(&services, stderr).await, "");
+        assert_eq!(node_state.stdout_bytes, Some(6));
+        assert_eq!(node_state.stderr_bytes, Some(0));
+        assert_eq!(node_state.streams_separated, Some(true));
+        assert_eq!(node_state.live_streaming, Some(true));
     }
 
     #[tokio::test]
@@ -469,7 +576,7 @@ mod tests {
         let snapshot = run_store.state().await.unwrap();
         let node_state = snapshot.node(&StageId::new("script_node", 1)).unwrap();
         let stderr = node_state.stderr.as_deref().unwrap();
-        assert_eq!(stderr.trim(), "oops");
+        assert_eq!(command_log_text(&services, stderr).await.trim(), "oops");
     }
 
     #[tokio::test]
@@ -599,16 +706,16 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Succeeded);
         let command_output = outcome.context_updates.get(keys::COMMAND_OUTPUT).unwrap();
         assert!(
-            command_output
-                .as_str()
-                .unwrap()
+            command_text(&services, command_output)
+                .await
                 .contains("hello from python")
         );
     }
@@ -681,13 +788,18 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Succeeded);
         let command_output = outcome.context_updates.get(keys::COMMAND_OUTPUT).unwrap();
-        assert!(command_output.as_str().unwrap().contains("legacy"));
+        assert!(
+            command_text(&services, command_output)
+                .await
+                .contains("legacy")
+        );
     }
 
     #[tokio::test]
@@ -702,14 +814,17 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Succeeded);
         let command_stderr = outcome.context_updates.get(keys::COMMAND_STDERR).unwrap();
         assert!(
-            command_stderr.as_str().unwrap().contains("err"),
+            command_text(&services, command_stderr)
+                .await
+                .contains("err"),
             "command.stderr should contain 'err', got: {:?}",
             command_stderr
         );
@@ -822,7 +937,7 @@ mod tests {
     }
 
     fn make_spy_services(sandbox: std::sync::Arc<SpySandbox>) -> EngineServices {
-        let mut services = EngineServices::test_default();
+        let mut services = make_services();
         services.run = services.run.with_sandbox(sandbox);
         services
     }
@@ -847,21 +962,16 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_spy_services(spy.clone());
         let outcome = handler
-            .execute(
-                &node,
-                &context,
-                &graph,
-                run_dir.path(),
-                &make_spy_services(spy.clone()),
-            )
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
 
         assert_eq!(outcome.status, StageOutcome::Succeeded);
         let command_output = outcome.context_updates.get(keys::COMMAND_OUTPUT).unwrap();
         assert_eq!(
-            command_output.as_str().unwrap(),
+            command_text(&services, command_output).await,
             "SANDBOX_MARKER\n",
             "CommandHandler must delegate to the sandbox, not spawn a host process"
         );
@@ -1017,8 +1127,9 @@ mod tests {
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
 
+        let services = make_services();
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Failed {
@@ -1065,9 +1176,10 @@ mod tests {
         let context = Context::new();
         let graph = Graph::new("test");
         let run_dir = tempfile::tempdir().unwrap();
+        let services = make_services();
 
         let outcome = handler
-            .execute(&node, &context, &graph, run_dir.path(), &make_services())
+            .execute(&node, &context, &graph, run_dir.path(), &services)
             .await
             .unwrap();
         assert_eq!(outcome.status, StageOutcome::Failed {
@@ -1078,7 +1190,9 @@ mod tests {
             .get(keys::COMMAND_OUTPUT)
             .expect("command.output should be set on failure");
         assert!(
-            command_output.as_str().unwrap().contains("build output"),
+            command_text(&services, command_output)
+                .await
+                .contains("build output"),
             "command.output should contain stdout, got: {command_output:?}"
         );
     }

lib/crates/fabro-workflow/src/lib.rs

@@ -114,6 +114,7 @@ pub fn extract_stage_durations_from_events(events: &[EventEnvelope]) -> HashMap<
 pub mod artifact;
 pub mod artifact_snapshot;
 pub mod artifact_upload;
+pub mod command_log;
 pub(crate) mod condition;
 pub mod context;
 pub mod devcontainer_bridge;

lib/crates/fabro-workflow/src/node_handler.rs

@@ -123,6 +123,23 @@ impl NodeHandler<WorkflowGraph> for WorkflowNodeHandler {
         }
     }
 
+    async fn context_for_edge_selection(
+        &self,
+        context: &Context,
+        _graph: &WorkflowGraph,
+    ) -> CoreResult<Context> {
+        artifact::resolve_context_for_edge_selection(context, &self.services.run.run_store)
+            .await
+            .map_err(|err| {
+                CoreError::handler(HandlerErrorDetail {
+                    message:   err.to_string(),
+                    retryable: true,
+                    category:  Some(FailureCategory::TransientInfra),
+                    signature: None,
+                })
+            })
+    }
+
     fn retry_policy(&self, node: &WorkflowNode, _graph: &WorkflowGraph) -> CoreRetryPolicy {
         let gv_node = node.inner();
         build_retry_policy(gv_node, &self.graph)

lib/crates/fabro-workflow/src/pipeline/execute/tests.rs

@@ -17,7 +17,7 @@ use fabro_hooks::HookSettings;
 use fabro_interview::AutoApproveInterviewer;
 use fabro_sandbox::SandboxSpec;
 use fabro_store::Database;
-use fabro_types::{RunId, WorkflowSettings, fixtures};
+use fabro_types::{RunId, WorkflowSettings, fixtures, format_blob_ref};
 use object_store::memory::InMemory;
 
 use super::*;
@@ -420,6 +420,29 @@ impl HandlerTrait for PanickingHandler {
     }
 }
 
+struct BlobCommandOutputHandler;
+
+#[async_trait]
+impl HandlerTrait for BlobCommandOutputHandler {
+    async fn execute(
+        &self,
+        _node: &Node,
+        _context: &Context,
+        _graph: &Graph,
+        _run_dir: &Path,
+        services: &crate::handler::EngineServices,
+    ) -> std::result::Result<Outcome, Error> {
+        let blob = serde_json::to_vec("routed-ok").unwrap();
+        let blob_id = services.run.run_store.write_blob(&blob).await.unwrap();
+        let mut outcome = Outcome::success();
+        outcome.context_updates.insert(
+            context::keys::COMMAND_OUTPUT.to_string(),
+            serde_json::json!(format_blob_ref(&blob_id)),
+        );
+        Ok(outcome)
+    }
+}
+
 struct FailOnceThenSucceedHandler {
     call_count: AtomicU32,
 }
@@ -655,6 +678,78 @@ async fn execute_conditional_routing_uses_unconditional_success_path() {
     assert!(!cp.completed_nodes.contains(&"path_a".to_string()));
 }
 
+#[tokio::test]
+async fn execute_conditional_routing_resolves_command_output_blob_refs() {
+    let dir = tempfile::tempdir().unwrap();
+    let mut g = Graph::new("command_output_route");
+
+    let mut start = Node::new("start");
+    start.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Mdiamond".to_string()),
+    );
+    g.nodes.insert("start".to_string(), start);
+
+    let mut commandish = Node::new("commandish");
+    commandish.attrs.insert(
+        "type".to_string(),
+        AttrValue::String("blob_command_output".to_string()),
+    );
+    commandish
+        .attrs
+        .insert("max_retries".to_string(), AttrValue::Integer(0));
+    g.nodes.insert("commandish".to_string(), commandish);
+
+    g.nodes.insert("matched".to_string(), Node::new("matched"));
+    g.nodes
+        .insert("fallback".to_string(), Node::new("fallback"));
+
+    let mut exit = Node::new("exit");
+    exit.attrs.insert(
+        "shape".to_string(),
+        AttrValue::String("Msquare".to_string()),
+    );
+    g.nodes.insert("exit".to_string(), exit);
+
+    g.edges.push(Edge::new("start", "commandish"));
+    let mut matched = Edge::new("commandish", "matched");
+    matched.attrs.insert(
+        "condition".to_string(),
+        AttrValue::String("command.output contains routed-ok".to_string()),
+    );
+    g.edges.push(matched);
+    g.edges.push(Edge::new("commandish", "fallback"));
+    g.edges.push(Edge::new("matched", "exit"));
+    g.edges.push(Edge::new("fallback", "exit"));
+
+    let mut registry = make_registry();
+    registry.register("blob_command_output", Box::new(BlobCommandOutputHandler));
+    let executed = execute_test_run_with_options(
+        test_run_options(dir.path(), "test-run"),
+        g,
+        Some(Arc::new(registry)),
+    )
+    .await;
+
+    let cp = executed
+        .engine
+        .run
+        .run_store
+        .state()
+        .await
+        .unwrap()
+        .checkpoint
+        .unwrap();
+    assert!(cp.completed_nodes.contains(&"matched".to_string()));
+    assert!(!cp.completed_nodes.contains(&"fallback".to_string()));
+    assert!(
+        cp.context_values[context::keys::COMMAND_OUTPUT]
+            .as_str()
+            .is_some_and(|value| value.starts_with("blob://sha256/")),
+        "durable checkpoint context should keep the command output blob ref"
+    );
+}
+
 #[tokio::test]
 async fn execute_persists_start_record_and_node_status() {
     let dir = tempfile::tempdir().unwrap();

lib/crates/fabro-workflow/src/pipeline/retro.rs

@@ -4,7 +4,7 @@ use fabro_agent::SessionEvent;
 use fabro_llm::client::Client;
 use fabro_retro::retro::{Retro, derive_retro};
 use fabro_retro::retro_agent::{
-    RETRO_DATA_DIR, build_retro_prompt, dry_run_narrative, run_retro_agent,
+    RETRO_DATA_DIR, RetroBlobReader, build_retro_prompt, dry_run_narrative, run_retro_agent,
 };
 
 use super::types::{Executed, RetroOptions, Retroed};
@@ -81,11 +81,22 @@ pub async fn run_retro(options: &RetroOptions, dry_run: bool) -> Option<Retro> {
                             });
                         }
                     });
+                let run_store = services.run_store.clone();
+                let blob_reader: RetroBlobReader = Arc::new(move |blob_id| {
+                    let run_store = run_store.clone();
+                    Box::pin(async move {
+                        Ok(run_store
+                            .read_blob(&blob_id)
+                            .await?
+                            .map(|bytes| bytes.to_vec()))
+                    })
+                });
                 run_retro_agent(
                     &services.sandbox,
                     &state,
                     &events,
                     &options.run_dir,
+                    Some(blob_reader),
                     &client,
                     services.provider,
                     &options.model,

lib/crates/fabro-workflow/src/run_dump.rs

@@ -179,21 +179,35 @@ impl RunDump {
     {
         let mut cache = HashMap::new();
         for entry in &mut self.entries {
-            if let RunDumpContents::Json(value) = &mut entry.contents {
-                let mut blob_ids = Vec::new();
-                collect_blob_refs_in_value(value, &mut blob_ids);
-                for blob_id in blob_ids {
-                    if cache.contains_key(&blob_id) {
-                        continue;
+            match &mut entry.contents {
+                RunDumpContents::Json(value) => {
+                    let mut blob_ids = Vec::new();
+                    collect_blob_refs_in_value(value, &mut blob_ids);
+                    for blob_id in blob_ids {
+                        if cache.contains_key(&blob_id) {
+                            continue;
+                        }
+                        let blob = read_blob(blob_id).await?.with_context(|| {
+                            format!("blob {blob_id:?} is missing from the store")
+                        })?;
+                        let hydrated: serde_json::Value = serde_json::from_slice(&blob)
+                            .with_context(|| format!("blob {blob_id:?} is not valid JSON"))?;
+                        cache.insert(blob_id, hydrated);
                     }
+                    replace_blob_refs_in_value(value, &cache)?;
+                }
+                RunDumpContents::Text(text) => {
+                    let Some(blob_id) = parse_blob_ref(text) else {
+                        continue;
+                    };
                     let blob = read_blob(blob_id)
                         .await?
                         .with_context(|| format!("blob {blob_id:?} is missing from the store"))?;
-                    let hydrated: serde_json::Value = serde_json::from_slice(&blob)
-                        .with_context(|| format!("blob {blob_id:?} is not valid JSON"))?;
-                    cache.insert(blob_id, hydrated);
+                    *text = serde_json::from_slice::<String>(&blob).with_context(|| {
+                        format!("blob {blob_id:?} is not a JSON string text log")
+                    })?;
                 }
-                replace_blob_refs_in_value(value, &cache)?;
+                RunDumpContents::Bytes(_) => {}
             }
         }
         Ok(())
@@ -524,6 +538,10 @@ mod tests {
             parallel_results:  Some(serde_json::json!([{ "stage": "fanout@1" }])),
             stdout:            Some("stdout".to_string()),
             stderr:            Some("stderr".to_string()),
+            stdout_bytes:      None,
+            stderr_bytes:      None,
+            streams_separated: None,
+            live_streaming:    None,
         });
 
         let dump = RunDump::from_projection(&projection);

lib/crates/fabro-workflow/tests/it/integration.rs

@@ -19,7 +19,7 @@
 use std::collections::VecDeque;
 use std::collections::hash_map::DefaultHasher;
 use std::hash::{Hash, Hasher};
-use std::path::Path;
+use std::path::{Path, PathBuf};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -32,7 +32,7 @@ use fabro_interview::{
 };
 use fabro_llm::provider::Provider;
 use fabro_store::{ArtifactStore, Database};
-use fabro_types::{RunEvent, RunId, StageId, WorkflowSettings};
+use fabro_types::{RunEvent, RunId, StageId, WorkflowSettings, parse_blob_ref};
 use fabro_validate::{Severity, validate, validate_or_raise};
 use fabro_workflow::context::Context;
 use fabro_workflow::error::{Error, FailureSignatureExt};
@@ -175,6 +175,79 @@ fn load_run_checkpoint(run_dir: &Path) -> Result<Checkpoint, Box<dyn std::error:
         .ok_or_else(|| "checkpoint should exist in run store".into())
 }
 
+fn run_store_dir_and_mode(run_dir: &Path) -> Result<(PathBuf, bool), Box<dyn std::error::Error>> {
+    let uses_shared_store = run_dir
+        .parent()
+        .and_then(Path::file_name)
+        .is_some_and(|name| name == "scratch");
+    let store_dir = if uses_shared_store {
+        let runs_dir = run_dir.parent().ok_or("run dir should have parent")?;
+        let storage_dir = runs_dir.parent().ok_or("runs dir should have parent")?;
+        storage_dir.join("store")
+    } else {
+        test_store_dir(run_dir)
+    };
+    Ok((store_dir, uses_shared_store))
+}
+
+#[expect(
+    clippy::disallowed_methods,
+    reason = "This helper spins up a dedicated current-thread runtime when called from inside an existing Tokio runtime."
+)]
+fn resolve_checkpoint_text(
+    run_dir: &Path,
+    value: &serde_json::Value,
+) -> Result<String, Box<dyn std::error::Error>> {
+    let Some(current) = value.as_str() else {
+        return Ok(value.to_string());
+    };
+    let Some(blob_id) = parse_blob_ref(current) else {
+        return Ok(current.to_string());
+    };
+
+    let run_dir = run_dir.to_path_buf();
+    let (store_dir, uses_shared_store) = run_store_dir_and_mode(&run_dir)?;
+    std::thread::spawn(
+        move || -> Result<_, Box<dyn std::error::Error + Send + Sync>> {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()?;
+            let object_store = Arc::new(LocalFileSystem::new_with_prefix(store_dir)?);
+            let store = Arc::new(Database::new(
+                object_store,
+                "",
+                Duration::from_millis(1),
+                None,
+            ));
+            let run_id = if uses_shared_store {
+                run_dir
+                    .file_name()
+                    .ok_or("run dir should have file name")?
+                    .to_string_lossy()
+                    .rsplit('-')
+                    .next()
+                    .ok_or("run dir should contain run id suffix")?
+                    .parse()?
+            } else {
+                runtime
+                    .block_on(store.list_runs(&fabro_store::ListRunsQuery::default()))?
+                    .into_iter()
+                    .next()
+                    .ok_or("test store should contain one run")?
+                    .run_id
+            };
+            let run = runtime.block_on(store.open_run_reader(&run_id))?;
+            let bytes = runtime
+                .block_on(run.read_blob(&blob_id))?
+                .ok_or("checkpoint blob should exist")?;
+            Ok(serde_json::from_slice::<String>(&bytes)?)
+        },
+    )
+    .join()
+    .map_err(|_| "checkpoint text resolver thread panicked")?
+    .map_err(|err| err.to_string().into())
+}
+
 fn save_checkpoint(path: &Path, checkpoint: &Checkpoint) {
     let serialized_checkpoint =
         serde_json::to_string_pretty(checkpoint).expect("checkpoint should serialize to JSON");
@@ -2330,12 +2403,8 @@ async fn tool_handler_e2e() {
         .context_values
         .get("command.output")
         .expect("command.output should exist");
-    assert!(
-        command_output
-            .as_str()
-            .unwrap()
-            .contains("hello-from-script")
-    );
+    let command_output = resolve_checkpoint_text(dir.path(), command_output).unwrap();
+    assert!(command_output.contains("hello-from-script"));
 }
 
 #[tokio::test]
@@ -2701,7 +2770,8 @@ async fn scenario_ship_a_feature() {
         .context_values
         .get("command.output")
         .expect("command.output");
-    assert!(command_output.as_str().unwrap().contains("PASS"));
+    let command_output = resolve_checkpoint_text(dir.path(), command_output).unwrap();
+    assert!(command_output.contains("PASS"));
     assert!(cp.completed_nodes.contains(&"plan".to_string()));
     assert!(cp.completed_nodes.contains(&"implement".to_string()));
     assert!(cp.completed_nodes.contains(&"test".to_string()));

lib/packages/fabro-api-client/src/.openapi-generator/FILES

@@ -43,6 +43,8 @@ models/check-run-status.ts
 models/check-run.ts
 models/close-run-pull-request-response.ts
 models/code-location.ts
+models/command-log-response.ts
+models/command-output-stream.ts
 models/completion-content-part.ts
 models/completion-message.ts
 models/completion-response.ts

lib/packages/fabro-api-client/src/api/run-internals-api.ts

@@ -26,6 +26,10 @@ import type { AppendEventResponse } from '../models';
 // @ts-ignore
 import type { ArtifactListResponse } from '../models';
 // @ts-ignore
+import type { CommandLogResponse } from '../models';
+// @ts-ignore
+import type { CommandOutputStream } from '../models';
+// @ts-ignore
 import type { ErrorResponse } from '../models';
 // @ts-ignore
 import type { PaginatedEventList } from '../models';
@@ -182,6 +186,64 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
                 options: localVarRequestOptions,
             };
         },
+        /**
+         * Returns a byte-offset slice of a command stage stdout or stderr log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
+         * @summary Tail Command Log
+         * @param {string} id Unique run identifier (ULID).
+         * @param {string} stageId Identifier of a stage within a run\'s workflow graph, serialized as `node_id@visit`.
+         * @param {CommandOutputStream} stream Command output stream to read.
+         * @param {number} [offset] Byte offset to start reading from. Defaults to `0`.
+         * @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
+         * @param {*} [options] Override http request option.
+         * @throws {RequiredError}
+         */
+        getRunStageCommandLog: async (id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
+            // verify required parameter 'id' is not null or undefined
+            assertParamExists('getRunStageCommandLog', 'id', id)
+            // verify required parameter 'stageId' is not null or undefined
+            assertParamExists('getRunStageCommandLog', 'stageId', stageId)
+            // verify required parameter 'stream' is not null or undefined
+            assertParamExists('getRunStageCommandLog', 'stream', stream)
+            const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/logs/{stream}`
+                .replace(`{${"id"}}`, encodeURIComponent(String(id)))
+                .replace(`{${"stageId"}}`, encodeURIComponent(String(stageId)))
+                .replace(`{${"stream"}}`, encodeURIComponent(String(stream)));
+            // use dummy base URL string because the URL constructor only accepts absolute URLs.
+            const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
+            let baseOptions;
+            if (configuration) {
+                baseOptions = configuration.baseOptions;
+            }
+
+            const localVarRequestOptions = { method: 'GET', ...baseOptions, ...options};
+            const localVarHeaderParameter = {} as any;
+            const localVarQueryParameter = {} as any;
+
+            // authentication SessionCookie required
+
+            // authentication BearerAuth required
+            // http bearer authentication required
+            await setBearerAuthToObject(localVarHeaderParameter, configuration)
+
+            if (offset !== undefined) {
+                localVarQueryParameter['offset'] = offset;
+            }
+
+            if (limit !== undefined) {
+                localVarQueryParameter['limit'] = limit;
+            }
+
+            localVarHeaderParameter['Accept'] = 'application/json';
+
+            setSearchParams(localVarUrlObj, localVarQueryParameter);
+            let headersFromBaseOptions = baseOptions && baseOptions.headers ? baseOptions.headers : {};
+            localVarRequestOptions.headers = {...localVarHeaderParameter, ...headersFromBaseOptions, ...options.headers};
+
+            return {
+                url: toPathString(localVarUrlObj),
+                options: localVarRequestOptions,
+            };
+        },
         /**
          * Returns the internal event-sourced run projection. This is not a stable public contract.
          * @summary Get Run State
@@ -784,6 +846,23 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
             const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.getRunLogs']?.[localVarOperationServerIndex]?.url;
             return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
         },
+        /**
+         * Returns a byte-offset slice of a command stage stdout or stderr log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
+         * @summary Tail Command Log
+         * @param {string} id Unique run identifier (ULID).
+         * @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
+         * @param {CommandOutputStream} stream Command output stream to read.
+         * @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
+         * @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
+         * @param {*} [options] Override http request option.
+         * @throws {RequiredError}
+         */
+        async getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<CommandLogResponse>> {
+            const localVarAxiosArgs = await localVarAxiosParamCreator.getRunStageCommandLog(id, stageId, stream, offset, limit, options);
+            const localVarOperationServerIndex = configuration?.serverIndex ?? 0;
+            const localVarOperationServerBasePath = operationServerMap['RunInternalsApi.getRunStageCommandLog']?.[localVarOperationServerIndex]?.url;
+            return (axios, basePath) => createRequestFunction(localVarAxiosArgs, globalAxios, BASE_PATH, configuration)(axios, localVarOperationServerBasePath || basePath);
+        },
         /**
          * Returns the internal event-sourced run projection. This is not a stable public contract.
          * @summary Get Run State
@@ -996,6 +1075,20 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
         getRunLogs(id: string, options?: RawAxiosRequestConfig): AxiosPromise<string> {
             return localVarFp.getRunLogs(id, options).then((request) => request(axios, basePath));
         },
+        /**
+         * Returns a byte-offset slice of a command stage stdout or stderr log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
+         * @summary Tail Command Log
+         * @param {string} id Unique run identifier (ULID).
+         * @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
+         * @param {CommandOutputStream} stream Command output stream to read.
+         * @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
+         * @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
+         * @param {*} [options] Override http request option.
+         * @throws {RequiredError}
+         */
+        getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise<CommandLogResponse> {
+            return localVarFp.getRunStageCommandLog(id, stageId, stream, offset, limit, options).then((request) => request(axios, basePath));
+        },
         /**
          * Returns the internal event-sourced run projection. This is not a stable public contract.
          * @summary Get Run State
@@ -1173,6 +1266,21 @@ export class RunInternalsApi extends BaseAPI {
         return RunInternalsApiFp(this.configuration).getRunLogs(id, options).then((request) => request(this.axios, this.basePath));
     }
 
+    /**
+     * Returns a byte-offset slice of a command stage stdout or stderr log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
+     * @summary Tail Command Log
+     * @param {string} id Unique run identifier (ULID).
+     * @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
+     * @param {CommandOutputStream} stream Command output stream to read.
+     * @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
+     * @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
+     * @param {*} [options] Override http request option.
+     * @throws {RequiredError}
+     */
+    public getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig) {
+        return RunInternalsApiFp(this.configuration).getRunStageCommandLog(id, stageId, stream, offset, limit, options).then((request) => request(this.axios, this.basePath));
+    }
+
     /**
      * Returns the internal event-sourced run projection. This is not a stable public contract.
      * @summary Get Run State

lib/packages/fabro-api-client/src/models/command-log-response.ts

@@ -0,0 +1,53 @@
+/* tslint:disable */
+/* eslint-disable */
+/**
+ * Fabro Run API
+ * HTTP API for managing Fabro workflow run executions.
+ *
+ * The version of the OpenAPI document: 0.1.0
+ * 
+ *
+ * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
+ * https://openapi-generator.tech
+ * Do not edit the class manually.
+ */
+
+
+// May contain unused imports in some cases
+// @ts-ignore
+import type { CommandOutputStream } from './command-output-stream';
+
+/**
+ * Byte-offset command log slice.
+ */
+export interface CommandLogResponse {
+    'stream': CommandOutputStream;
+    /**
+     * Actual byte offset used for this slice.
+     */
+    'offset': number;
+    /**
+     * Byte offset for the next tail request.
+     */
+    'next_offset': number;
+    /**
+     * Total bytes currently available for the stream.
+     */
+    'total_bytes': number;
+    /**
+     * Base64-encoded raw log bytes.
+     */
+    'bytes_base64': string;
+    /**
+     * Whether the stream is finalized.
+     */
+    'eof': boolean;
+    'cas_ref': string | null;
+    /**
+     * Whether the sandbox provided live output while the command was running.
+     */
+    'live_streaming': boolean;
+}
+
+
+

lib/packages/fabro-api-client/src/models/command-output-stream.ts

@@ -0,0 +1,29 @@
+/* tslint:disable */
+/* eslint-disable */
+/**
+ * Fabro Run API
+ * HTTP API for managing Fabro workflow run executions.
+ *
+ * The version of the OpenAPI document: 0.1.0
+ * 
+ *
+ * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
+ * https://openapi-generator.tech
+ * Do not edit the class manually.
+ */
+
+
+
+/**
+ * Command output stream name.
+ */
+
+export const CommandOutputStream = {
+    STDOUT: 'stdout',
+    STDERR: 'stderr'
+} as const;
+
+export type CommandOutputStream = typeof CommandOutputStream[keyof typeof CommandOutputStream];
+
+
+

lib/packages/fabro-api-client/src/models/index.ts

@@ -23,6 +23,8 @@ export * from './check-run';
 export * from './check-run-status';
 export * from './close-run-pull-request-response';
 export * from './code-location';
+export * from './command-log-response';
+export * from './command-output-stream';
 export * from './completion-content-part';
 export * from './completion-message';
 export * from './completion-response';

lib/packages/fabro-api-client/src/models/node-state.ts

@@ -5,7 +5,7 @@
  * HTTP API for managing Fabro workflow run executions.
  *
  * The version of the OpenAPI document: 0.1.0
- * 
+ *
  *
  * NOTE: This class is auto generated by OpenAPI Generator (https://openapi-generator.tech).
  * https://openapi-generator.tech
@@ -31,5 +31,8 @@ export interface NodeState {
     'parallel_results'?: any;
     'stdout'?: string | null;
     'stderr'?: string | null;
+    'stdout_bytes'?: number | null;
+    'stderr_bytes'?: number | null;
+    'streams_separated'?: boolean | null;
+    'live_streaming'?: boolean | null;
 }
-