From 8ac400df1ccd5ff53bbf047eb65def5f6f89ee4e Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Thu, 30 Apr 2026 11:05:33 -0400 Subject: [PATCH] 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 | 14 + apps/fabro-web/app/lib/query-keys.test.ts | 3 + apps/fabro-web/app/lib/query-keys.ts | 11 + apps/fabro-web/app/routes/run-stages.tsx | 231 +++++++++- docs/public/agents/outputs.mdx | 6 +- docs/public/api-reference/fabro-api.yaml | 136 ++++++ docs/public/execution/context.mdx | 7 +- lib/crates/fabro-agent/src/lib.rs | 5 +- lib/crates/fabro-agent/src/sandbox.rs | 6 +- lib/crates/fabro-api/build.rs | 5 + lib/crates/fabro-api/src/lib.rs | 10 +- .../tests/command_output_stream_round_trip.rs | 44 ++ lib/crates/fabro-core/src/executor.rs | 6 +- lib/crates/fabro-core/src/handler.rs | 4 + lib/crates/fabro-retro/src/retro_agent.rs | 125 +++++- lib/crates/fabro-sandbox/src/daytona/mod.rs | 39 +- lib/crates/fabro-sandbox/src/docker.rs | 306 ++++++++++++- lib/crates/fabro-sandbox/src/lib.rs | 6 +- lib/crates/fabro-sandbox/src/local.rs | 129 +++++- lib/crates/fabro-sandbox/src/sandbox.rs | 69 +++ lib/crates/fabro-server/src/server.rs | 404 +++++++++++++++++- lib/crates/fabro-server/src/worker_token.rs | 30 +- lib/crates/fabro-store/src/artifact_store.rs | 9 + lib/crates/fabro-store/src/lib.rs | 2 +- lib/crates/fabro-store/src/run_state.rs | 4 + .../tests/serializable_projection.rs | 4 + lib/crates/fabro-types/src/command_output.rs | 44 ++ lib/crates/fabro-types/src/lib.rs | 2 + lib/crates/fabro-types/src/run_event/misc.rs | 18 +- lib/crates/fabro-types/src/run_projection.rs | 8 + lib/crates/fabro-workflow/src/artifact.rs | 52 ++- lib/crates/fabro-workflow/src/command_log.rs | 198 +++++++++ lib/crates/fabro-workflow/src/event.rs | 48 ++- lib/crates/fabro-workflow/src/git.rs | 16 +- .../fabro-workflow/src/handler/command.rs | 206 +++++++-- lib/crates/fabro-workflow/src/lib.rs | 1 + lib/crates/fabro-workflow/src/node_handler.rs | 17 + .../src/pipeline/execute/tests.rs | 97 ++++- .../fabro-workflow/src/pipeline/retro.rs | 13 +- lib/crates/fabro-workflow/src/run_dump.rs | 38 +- .../fabro-workflow/tests/it/integration.rs | 88 +++- .../src/.openapi-generator/FILES | 2 + .../src/api/run-internals-api.ts | 108 +++++ .../src/models/command-log-response.ts | 53 +++ .../src/models/command-output-stream.ts | 29 ++ .../fabro-api-client/src/models/index.ts | 2 + .../fabro-api-client/src/models/node-state.ts | 7 +- 47 files changed, 2500 insertions(+), 162 deletions(-) create mode 100644 lib/crates/fabro-api/tests/command_output_stream_round_trip.rs create mode 100644 lib/crates/fabro-types/src/command_output.rs create mode 100644 lib/crates/fabro-workflow/src/command_log.rs create mode 100644 lib/packages/fabro-api-client/src/models/command-log-response.ts create mode 100644 lib/packages/fabro-api-client/src/models/command-output-stream.ts diff --git a/apps/fabro-web/app/lib/queries.ts b/apps/fabro-web/app/lib/queries.ts index a0a1925fb..35070247e 100644 --- a/apps/fabro-web/app/lib/queries.ts +++ b/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( + queryKeys.runs.stageLog(id, stageId, stream, offset, limit), + ); +} + export function useWorkflows() { return useSWR( queryKeys.workflows.list(), diff --git a/apps/fabro-web/app/lib/query-keys.test.ts b/apps/fabro-web/app/lib/query-keys.test.ts index 2b44658dd..8f036ac3e 100644 --- a/apps/fabro-web/app/lib/query-keys.test.ts +++ b/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", () => { diff --git a/apps/fabro-web/app/lib/query-keys.ts b/apps/fabro-web/app/lib/query-keys.ts index 11421bf5a..7d3e847e7 100644 --- a/apps/fabro-web/app/lib/query-keys.ts +++ b/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`, diff --git a/apps/fabro-web/app/routes/run-stages.tsx b/apps/fabro-web/app/routes/run-stages.tsx index bdd4bd1c0..88b4621fd 100644 --- a/apps/fabro-web/app/routes/run-stages.tsx +++ b/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; text?: string; @@ -70,7 +71,7 @@ function turnsFromEvents(events: RawEvent[], stageId: string): TurnType[] { // Collect tool pairs: started → completed const pendingTools = new Map(); // 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({ + 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 | 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(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 (
+ {status} + {state.truncated ? ( + Last 5 MiB + ) : null} @@ -287,13 +451,36 @@ function OutputStream({ Show {hiddenLines} earlier lines ) : null} - 
{visible}
+ {content.length === 0 ? ( +
+ {state.error ? "Unable to fetch this stream." : "No bytes received yet."} +
+ ) : ( + 
 {
+            const el = event.currentTarget;
+            followTailRef.current = el.scrollHeight - el.scrollTop - el.clientHeight < 24;
+          }}
+          className={`${preClass} max-h-96 overflow-auto`}
+        >
+          {visible}
+
+ )}
); } -function CommandBlock({ turn }: { turn: Extract }) { +function CommandBlock({ + runId, + turn, +}: { + runId: string | undefined; + turn: Extract; +}) { 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 }
)} - {/* stdout */} - {turn.stdout && ( -
- +
+
+
- )} - - {/* stderr */} - {turn.stderr && ( -
- +
+ 0} + />
- )} +
); } @@ -446,7 +633,7 @@ export default function RunStages() { case "tool": return ; case "command": - return ; + return ; } })}
diff --git a/docs/public/agents/outputs.mdx b/docs/public/agents/outputs.mdx index a15d9a9ff..5b6c6d39f 100644 --- a/docs/public/agents/outputs.mdx +++ b/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. diff --git a/docs/public/api-reference/fabro-api.yaml b/docs/public/api-reference/fabro-api.yaml index 3d704b27b..53ad92424 100644 --- a/docs/public/api-reference/fabro-api.yaml +++ b/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. diff --git a/docs/public/execution/context.mdx b/docs/public/execution/context.mdx index 8fa5ac2c4..87eaf81e4 100644 --- a/docs/public/execution/context.mdx +++ b/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. diff --git a/lib/crates/fabro-agent/src/lib.rs b/lib/crates/fabro-agent/src/lib.rs index ea9673e6e..b0fe24dab 100644 --- a/lib/crates/fabro-agent/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-agent/src/sandbox.rs b/lib/crates/fabro-agent/src/sandbox.rs index 6df954184..0a7335364 100644 --- a/lib/crates/fabro-agent/src/sandbox.rs +++ b/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, }; diff --git a/lib/crates/fabro-api/build.rs b/lib/crates/fabro-api/build.rs index 553ca33be..30e35bb3a 100644 --- a/lib/crates/fabro-api/build.rs +++ b/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", &[]), diff --git a/lib/crates/fabro-api/src/lib.rs b/lib/crates/fabro-api/src/lib.rs index 5818812c1..3211ca942 100644 --- a/lib/crates/fabro-api/src/lib.rs +++ b/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::*; diff --git a/lib/crates/fabro-api/tests/command_output_stream_round_trip.rs b/lib/crates/fabro-api/tests/command_output_stream_round_trip.rs new file mode 100644 index 000000000..9abd55162 --- /dev/null +++ b/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::(); +} + +#[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::(json!("stdout")).unwrap(), + CommandOutputStream::Stdout + ); + assert_eq!( + serde_json::from_value::(json!("stderr")).unwrap(), + CommandOutputStream::Stderr + ); +} + +fn assert_same_type() { + assert_eq!( + TypeId::of::(), + TypeId::of::(), + "{} should be the same type as {}", + type_name::(), + type_name::() + ); +} diff --git a/lib/crates/fabro-core/src/executor.rs b/lib/crates/fabro-core/src/executor.rs index 3c6987095..91b584f99 100644 --- a/lib/crates/fabro-core/src/executor.rs +++ b/lib/crates/fabro-core/src/executor.rs @@ -390,7 +390,11 @@ impl Executor { } // 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(); diff --git a/lib/crates/fabro-core/src/handler.rs b/lib/crates/fabro-core/src/handler.rs index e0a371adb..c530ce760 100644 --- a/lib/crates/fabro-core/src/handler.rs +++ b/lib/crates/fabro-core/src/handler.rs @@ -15,6 +15,10 @@ pub trait NodeHandler: Send + Sync { graph: &G, ) -> Result>; + async fn context_for_edge_selection(&self, context: &Context, _graph: &G) -> Result { + Ok(context.clone()) + } + fn retry_policy(&self, _node: &G::Node, _graph: &G) -> RetryPolicy { RetryPolicy::none() } diff --git a/lib/crates/fabro-retro/src/retro_agent.rs b/lib/crates/fabro-retro/src/retro_agent.rs index 6a66ada28..ece61c55c 100644 --- a/lib/crates/fabro-retro/src/retro_agent.rs +++ b/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>>> + 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, llm_client: &Client, provider: Provider, model: &str, @@ -147,7 +157,15 @@ pub async fn run_retro_agent( ) -> anyhow::Result { // 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>> = 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, + blob_reader: Option<&RetroBlobReader>, +) -> anyhow::Result> { + 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::(&bytes) + .map_err(|err| anyhow::anyhow!("text blob was not a JSON string: {err}"))?; + Ok(Some(text)) +} + async fn upload_file( sandbox: &Arc, 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 = + 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" + ); + } } diff --git a/lib/crates/fabro-sandbox/src/daytona/mod.rs b/lib/crates/fabro-sandbox/src/daytona/mod.rs index 3a51ac505..a93e057ff 100644 --- a/lib/crates/fabro-sandbox/src/daytona/mod.rs +++ b/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>, + cancel_token: Option, + output_callback: CommandOutputCallback, + ) -> crate::Result { + 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, diff --git a/lib/crates/fabro-sandbox/src/docker.rs b/lib/crates/fabro-sandbox/src/docker.rs index c707d99eb..e7b5ae57a 100644 --- a/lib/crates/fabro-sandbox/src/docker.rs +++ b/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, + working_dir: Option, + env: Option>, + output_callback: CommandOutputCallback, + ) -> crate::Result<(Vec, Vec, 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>, + cancel_token: Option, + output_callback: CommandOutputCallback, + ) -> crate::Result { + let start = Instant::now(); + let effective_dir = working_dir.unwrap_or(WORKING_DIRECTORY).to_string(); + let env: Option> = + 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>, + cancel_token: Option, + output_callback: CommandOutputCallback, + ) -> crate::Result { + 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(); diff --git a/lib/crates/fabro-sandbox/src/lib.rs b/lib/crates/fabro-sandbox/src/lib.rs index 3b4a77aab..1fc600583 100644 --- a/lib/crates/fabro-sandbox/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-sandbox/src/local.rs b/lib/crates/fabro-sandbox/src/local.rs index 2889faa43..f019e7765 100644 --- a/lib/crates/fabro-sandbox/src/local.rs +++ b/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>, + cancel_token: Option, + output_callback: CommandOutputCallback, + ) -> crate::Result { + 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( + mut reader: Option, + stream: CommandOutputStream, + output_callback: CommandOutputCallback, +) -> crate::Result> +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, diff --git a/lib/crates/fabro-sandbox/src/sandbox.rs b/lib/crates/fabro-sandbox/src/sandbox.rs index bf160f2ae..3afe17325 100644 --- a/lib/crates/fabro-sandbox/src/sandbox.rs +++ b/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>, + cancel_token: Option, + 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> { 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) -> Pin> + 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>, cancel_token: Option, ) -> crate::Result; + async fn exec_command_streaming( + &self, + command: &str, + timeout_ms: u64, + working_dir: Option<&str>, + env_vars: Option<&std::collections::HashMap>, + cancel_token: Option, + output_callback: CommandOutputCallback, + ) -> crate::Result { + 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, diff --git a/lib/crates/fabro-server/src/server.rs b/lib/crates/fabro-server/src/server.rs index 82fe3406a..f29293563 100644 --- a/lib/crates/fabro-server/src/server.rs +++ b/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> { .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> { .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, + live_streaming: bool, +} + async fn resolve_run( _auth: AuthenticatedService, State(state): State>, @@ -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>, + Query(query): Query, +) -> 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, + 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(); diff --git a/lib/crates/fabro-server/src/worker_token.rs b/lib/crates/fabro-server/src/worker_token.rs index 9178cf7dd..a5c94820e 100644 --- a/lib/crates/fabro-server/src/worker_token.rs +++ b/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> for AuthorizeStageArtifact { } } +pub(crate) struct AuthorizeCommandLog( + pub(crate) RunId, + pub(crate) StageId, + pub(crate) CommandOutputStream, +); + +impl FromRequestParts> for AuthorizeCommandLog { + type Rejection = Response; + + async fn from_request_parts( + parts: &mut Parts, + state: &Arc, + ) -> Result { + 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::() + .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}; diff --git a/lib/crates/fabro-store/src/artifact_store.rs b/lib/crates/fabro-store/src/artifact_store.rs index 3f3e72afb..d09bfcfdd 100644 --- a/lib/crates/fabro-store/src/artifact_store.rs +++ b/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 { percent_decode_str(value) .decode_utf8() diff --git a/lib/crates/fabro-store/src/lib.rs b/lib/crates/fabro-store/src/lib.rs index 050fd5ad3..a88f9c77e 100644 --- a/lib/crates/fabro-store/src/lib.rs +++ b/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, diff --git a/lib/crates/fabro-store/src/run_state.rs b/lib/crates/fabro-store/src/run_state.rs index 862291daa..5b0829c47 100644 --- a/lib/crates/fabro-store/src/run_state.rs +++ b/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}")) })?); diff --git a/lib/crates/fabro-store/tests/serializable_projection.rs b/lib/crates/fabro-store/tests/serializable_projection.rs index 5a8f276ae..f9120e09b 100644 --- a/lib/crates/fabro-store/tests/serializable_projection.rs +++ b/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)) diff --git a/lib/crates/fabro-types/src/command_output.rs b/lib/crates/fabro-types/src/command_output.rs new file mode 100644 index 000000000..26b38d830 --- /dev/null +++ b/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()) + } +} diff --git a/lib/crates/fabro-types/src/lib.rs b/lib/crates/fabro-types/src/lib.rs index 8f7a9e86b..440a71f03 100644 --- a/lib/crates/fabro-types/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-types/src/run_event/misc.rs b/lib/crates/fabro-types/src/run_event/misc.rs index 7a3c4d302..d86658107 100644 --- a/lib/crates/fabro-types/src/run_event/misc.rs +++ b/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, - pub duration_ms: u64, - pub timed_out: bool, + pub exit_code: Option, + 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)] diff --git a/lib/crates/fabro-types/src/run_projection.rs b/lib/crates/fabro-types/src/run_projection.rs index 177e68225..e0866a92d 100644 --- a/lib/crates/fabro-types/src/run_projection.rs +++ b/lib/crates/fabro-types/src/run_projection.rs @@ -49,6 +49,14 @@ pub struct NodeState { pub parallel_results: Option, pub stdout: Option, pub stderr: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stdout_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub stderr_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub streams_separated: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub live_streaming: Option, } impl RunProjection { diff --git a/lib/crates/fabro-workflow/src/artifact.rs b/lib/crates/fabro-workflow/src/artifact.rs index 42a27cbd3..160308120 100644 --- a/lib/crates/fabro-workflow/src/artifact.rs +++ b/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 { + 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, 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 { + 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 { + 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::(&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(_) => {} diff --git a/lib/crates/fabro-workflow/src/command_log.rs b/lib/crates/fabro-workflow/src/command_log.rs new file mode 100644 index 000000000..250a6335a --- /dev/null +++ b/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, + stderr: Mutex, + 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> { + 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 { + 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, 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> { + 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::(&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 { + 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 { + 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 { + 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)) +} diff --git a/lib/crates/fabro-workflow/src/event.rs b/lib/crates/fabro-workflow/src/event.rs index 036dbb5dc..6aa91ec1d 100644 --- a/lib/crates/fabro-workflow/src/event.rs +++ b/lib/crates/fabro-workflow/src/event.rs @@ -499,13 +499,17 @@ pub enum Event { timeout_ms: Option, }, 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, - duration_ms: u64, - timed_out: bool, + exit_code: Option, + 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] diff --git a/lib/crates/fabro-workflow/src/git.rs b/lib/crates/fabro-workflow/src/git.rs index cac291e0b..592dccfe1 100644 --- a/lib/crates/fabro-workflow/src/git.rs +++ b/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(); diff --git a/lib/crates/fabro-workflow/src/handler/command.rs b/lib/crates/fabro-workflow/src/handler/command.rs index 4148dd7a4..e944f98cf 100644 --- a/lib/crates/fabro-workflow/src/handler/command.rs +++ b/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 { 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>, + } + + #[async_trait::async_trait] + impl RunStoreBackend for MemoryRunStoreBackend { + async fn load_state(&self) -> anyhow::Result { + Ok(fabro_store::RunProjection::default()) + } + + async fn list_events(&self) -> anyhow::Result> { + 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 { + 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> { + 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 { @@ -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) -> 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:?}" ); } diff --git a/lib/crates/fabro-workflow/src/lib.rs b/lib/crates/fabro-workflow/src/lib.rs index 2de9a5972..6baba63ba 100644 --- a/lib/crates/fabro-workflow/src/lib.rs +++ b/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; diff --git a/lib/crates/fabro-workflow/src/node_handler.rs b/lib/crates/fabro-workflow/src/node_handler.rs index 653db56bf..417253727 100644 --- a/lib/crates/fabro-workflow/src/node_handler.rs +++ b/lib/crates/fabro-workflow/src/node_handler.rs @@ -123,6 +123,23 @@ impl NodeHandler for WorkflowNodeHandler { } } + async fn context_for_edge_selection( + &self, + context: &Context, + _graph: &WorkflowGraph, + ) -> CoreResult { + 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) diff --git a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs b/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs index f65e087fd..8377ca3ec 100644 --- a/lib/crates/fabro-workflow/src/pipeline/execute/tests.rs +++ b/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 { + 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(); diff --git a/lib/crates/fabro-workflow/src/pipeline/retro.rs b/lib/crates/fabro-workflow/src/pipeline/retro.rs index 5e6177aa0..b99e4fc27 100644 --- a/lib/crates/fabro-workflow/src/pipeline/retro.rs +++ b/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 { }); } }); + 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, diff --git a/lib/crates/fabro-workflow/src/run_dump.rs b/lib/crates/fabro-workflow/src/run_dump.rs index 8035dd65f..ccf4ceefc 100644 --- a/lib/crates/fabro-workflow/src/run_dump.rs +++ b/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::(&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); diff --git a/lib/crates/fabro-workflow/tests/it/integration.rs b/lib/crates/fabro-workflow/tests/it/integration.rs index 05e5b890c..aa7bb7513 100644 --- a/lib/crates/fabro-workflow/tests/it/integration.rs +++ b/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 Result<(PathBuf, bool), Box> { + 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> { + 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> { + 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::(&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())); diff --git a/lib/packages/fabro-api-client/src/.openapi-generator/FILES b/lib/packages/fabro-api-client/src/.openapi-generator/FILES index e6c3ad19a..7bd14c09c 100644 --- a/lib/packages/fabro-api-client/src/.openapi-generator/FILES +++ b/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 diff --git a/lib/packages/fabro-api-client/src/api/run-internals-api.ts b/lib/packages/fabro-api-client/src/api/run-internals-api.ts index 7496dc295..b8325989a 100644 --- a/lib/packages/fabro-api-client/src/api/run-internals-api.ts +++ b/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 => { + // 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\'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} + */ + async getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise> { + 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 { 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\'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(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise { + 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\'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} + */ + 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 diff --git a/lib/packages/fabro-api-client/src/models/command-log-response.ts b/lib/packages/fabro-api-client/src/models/command-log-response.ts new file mode 100644 index 000000000..b6ad7b365 --- /dev/null +++ b/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; +} + + + diff --git a/lib/packages/fabro-api-client/src/models/command-output-stream.ts b/lib/packages/fabro-api-client/src/models/command-output-stream.ts new file mode 100644 index 000000000..8fb234b4e --- /dev/null +++ b/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]; + + + diff --git a/lib/packages/fabro-api-client/src/models/index.ts b/lib/packages/fabro-api-client/src/models/index.ts index b3173a238..ce2e6a927 100644 --- a/lib/packages/fabro-api-client/src/models/index.ts +++ b/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'; diff --git a/lib/packages/fabro-api-client/src/models/node-state.ts b/lib/packages/fabro-api-client/src/models/node-state.ts index 9f7f823f2..31f37ca91 100644 --- a/lib/packages/fabro-api-client/src/models/node-state.ts +++ b/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; } -