scriptoriumadmin/fabro
1
0
Fork 0
mirror of https://github.com/fabro-sh/fabro.git synced 2026-09-11 22:53:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat(runs): merge command output streams

Browse source 
Route command stderr into stdout at execution time and expose a single output log across events, projections, API clients, and the web UI. Keep replay compatibility for older command.completed events that still contain split stdout/stderr fields.
This commit is contained in:
Bryan Helmkamp 2026-05-07 22:07:13 -07:00
parent 23cb211cce
commit befb2e00ec
No known key found for this signature in database
42 changed files with 421 additions and 701 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
apps/fabro-web/app/lib/queries.ts
View file

@ -7,7 +7,6 @@ import type {
PaginatedRunList,
PaginatedRunStageList,
CommandLogResponse,
CommandOutputStream,
RunBilling,
RunProjection,
ServerSettings,
@ -153,23 +152,21 @@ export function useRunStageEvents(id: string | undefined, stageId: string | unde
export function fetchRunCommandLog(
id: string,
stageId: string,
stream: CommandOutputStream,
offset: number,
limit?: number,
) {
return apiFetcher<CommandLogResponse>(
queryKeys.runs.stageLog(id, stageId, stream, offset, limit),
queryKeys.runs.stageLog(id, stageId, offset, limit),
);
}
export function useRunStageLog(
id: string | undefined,
stageId: string | undefined,
stream: CommandOutputStream,
enabled: boolean,
) {
return useSWR<CommandLogResponse>(
enabled && id && stageId ? queryKeys.runs.stageLog(id, stageId, stream) : null,
enabled && id && stageId ? queryKeys.runs.stageLog(id, stageId) : null,
apiFetcher,
);
}
@ -209,4 +206,4 @@ export function useServerSettings() {
return useSWR<ServerSettings>(queryKeys.settings.server(), apiFetcher, immutableOptions);
}
export { apiTextFetcher };
export { apiTextFetcher };

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

@ -8,8 +8,8 @@ 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",
expect(queryKeys.runs.stageLog("run 1", "build step@2", 12, 34)).toBe(
"/api/v1/runs/run%201/stages/build%20step%402/logs/output?offset=12&limit=34",
);
expect(queryKeys.runs.stageEvents("run 1", "build step", 7, 25)).toBe(
"/api/v1/runs/run%201/stages/build%20step/events?since_seq=7&limit=25",

12
apps/fabro-web/app/lib/query-keys.ts
View file

@ -52,15 +52,9 @@ export const queryKeys = {
since_seq: sinceSeq,
limit,
}),
stageLog: (
id: string,
stageId: string,
stream: "stdout" | "stderr",
offset = 0,
limit = 65_536,
) =>
stageLog: (id: string, stageId: string, offset = 0, limit = 65_536) =>
withQuery(
`/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/logs/${stream}`,
`/api/v1/runs/${pathSegment(id)}/stages/${pathSegment(stageId)}/logs/output`,
{ offset, limit },
),
preview: (id: string) => `/api/v1/runs/${pathSegment(id)}/preview`,
@ -81,4 +75,4 @@ export const queryKeys = {
settings: {
server: () => "/api/v1/settings",
},
};
};

15
apps/fabro-web/app/routes/run-stages.test.ts
View file

@ -84,10 +84,8 @@ describe("eventsToActivity", () => {
event: "command.completed",
node_id: "fmt",
properties: {
stdout: "blob://sha256/abc",
stderr: "blob://sha256/def",
stdout_bytes: 42,
stderr_bytes: 0,
output: "blob://sha256/abc",
output_bytes: 42,
exit_code: 0,
duration_ms: 12,
termination: "exited",
@ -101,8 +99,7 @@ describe("eventsToActivity", () => {
kind: "command",
script: "cargo fmt",
running: false,
stdoutBytes: 42,
stderrBytes: 0,
outputBytes: 42,
});
});
@ -119,8 +116,7 @@ describe("eventsToActivity", () => {
stage_id: "verify@2",
node_id: "verify",
properties: {
stdout: "hi",
stderr: "",
output: "hi",
exit_code: 0,
duration_ms: 5,
termination: "exited",
@ -425,8 +421,7 @@ describe("turnsToStageKind", () => {
event: "command.completed",
node_id: "fmt",
properties: {
stdout: "blob://sha256/abc",
stderr: "blob://sha256/def",
output: "blob://sha256/abc",
exit_code: 0,
duration_ms: 5,
termination: "exited",

37
apps/fabro-web/app/routes/run-stages.tsx
View file

@ -33,7 +33,7 @@ import {
import { STAGE_ACTIVITY_EVENT_TYPES, type StageActivityEventType } from "../lib/run-events";
import { mapRunStagesToSidebarStages } from "../lib/stage-sidebar";
import { getNumber, getString, type UnknownRecord } from "../lib/unknown";
import type { CommandOutputStream, EventEnvelope } from "@qltysh/fabro-api-client";
import type { EventEnvelope } from "@qltysh/fabro-api-client";
export const handle = { wide: true, fullHeight: true };
@ -48,8 +48,7 @@ type TurnType =
running: boolean;
exitCode: number | null;
durationMs: number;
stdoutBytes: number;
stderrBytes: number;
outputBytes: number;
};
type CommandTurn = Extract<TurnType, { kind: "command" }>;
@ -199,8 +198,7 @@ export function eventsToActivity(events: EventEnvelope[], stageId: string): Turn
running: false,
exitCode: getNumber(props, "exit_code") ?? null,
durationMs: getNumber(props, "duration_ms") ?? 0,
stdoutBytes: getNumber(props, "stdout_bytes") ?? 0,
stderrBytes: getNumber(props, "stderr_bytes") ?? 0,
outputBytes: getNumber(props, "output_bytes") ?? 0,
});
pendingCommand = undefined;
break;
@ -218,8 +216,7 @@ export function eventsToActivity(events: EventEnvelope[], stageId: string): Turn
running: true,
exitCode: null,
durationMs: 0,
stdoutBytes: 0,
stderrBytes: 0,
outputBytes: 0,
});
}
@ -763,21 +760,17 @@ function decodeBase64Utf8(b64: string): string {
function LogStream({
runId,
stageId,
stream,
label,
byteCount,
enabled,
tone,
}: {
runId: string;
stageId: string;
stream: CommandOutputStream;
label: string;
byteCount: number;
enabled: boolean;
tone?: "stderr";
}) {
const { data, error, isLoading } = useRunStageLog(runId, stageId, stream, enabled && byteCount > 0);
const { data, error, isLoading } = useRunStageLog(runId, stageId, enabled && byteCount > 0);
const text = useMemo(() => {
if (!data?.bytes_base64) return "";
try {
@ -802,16 +795,14 @@ function LogStream({
)}
</header>
<pre
className={`overflow-x-auto whitespace-pre-wrap rounded-md bg-overlay-strong p-3 font-mono text-xs leading-relaxed ${
tone === "stderr" ? "text-coral" : "text-fg-3"
}`}
className="overflow-x-auto whitespace-pre-wrap rounded-md bg-overlay-strong p-3 font-mono text-xs leading-relaxed text-fg-3"
>
{byteCount === 0 ? (
<span className="text-fg-muted">empty</span>
) : isLoading && !data ? (
<span className="text-fg-muted">loading…</span>
) : error ? (
<span className="text-coral">Failed to load {stream}.</span>
<span className="text-coral">Failed to load output.</span>
) : (
text || <span className="text-fg-muted">empty</span>
)}
@ -886,20 +877,10 @@ function CommandLogs({
<LogStream
runId={runId}
stageId={stageId}
stream="stdout"
label="Stdout"
byteCount={turn.stdoutBytes}
label="Output"
byteCount={turn.outputBytes}
enabled={!turn.running}
/>
<LogStream
runId={runId}
stageId={stageId}
stream="stderr"
label="Stderr"
byteCount={turn.stderrBytes}
enabled={!turn.running}
tone="stderr"
/>
</div>
);
}

41
docs/public/api-reference/fabro-api.yaml
View file

@ -1877,16 +1877,15 @@ paths:
schema:
$ref: "#/components/schemas/ErrorResponse"
/api/v1/runs/{id}/stages/{stageId}/logs/{stream}:
/api/v1/runs/{id}/stages/{stageId}/logs/output:
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.
description: Returns a byte-offset slice of a command stage output 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:
@ -1897,7 +1896,7 @@ paths:
schema:
$ref: "#/components/schemas/CommandLogResponse"
"400":
description: Invalid stage, stream, offset, or limit.
description: Invalid stage, offset, or limit.
headers:
x-request-id:
$ref: "#/components/headers/XRequestId"
@ -3054,15 +3053,6 @@ 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
@ -5183,13 +5173,6 @@ components:
description: Blob identifier.
example: 550e8400-e29b-41d4-a716-446655440000
CommandOutputStream:
description: Command output stream name.
type: string
enum:
- stdout
- stderr
CommandTermination:
description: Terminal state for a command execution.
type: string
@ -5202,7 +5185,6 @@ components:
description: Byte-offset command log slice.
type: object
required:
- stream
- offset
- next_offset
- total_bytes
@ -5211,8 +5193,6 @@ components:
- cas_ref
- live_streaming
properties:
stream:
$ref: "#/components/schemas/CommandOutputStream"
offset:
type: integer
minimum: 0
@ -5226,7 +5206,7 @@ components:
total_bytes:
type: integer
minimum: 0
description: Total bytes currently available for the stream.
description: Total bytes currently available for the output log.
example: 8192
bytes_base64:
type: string
@ -5234,7 +5214,7 @@ components:
example: aGVsbG8K
eof:
type: boolean
description: Whether the stream is finalized.
description: Whether the output log is finalized.
example: false
cas_ref:
oneOf:
@ -5440,18 +5420,11 @@ components:
items:
type: object
description: Per-branch result objects produced by a parallel stage.
stdout:
output:
type: ["string", "null"]
stderr:
type: ["string", "null"]
stdout_bytes:
output_bytes:
type: ["integer", "null"]
minimum: 0
stderr_bytes:
type: ["integer", "null"]
minimum: 0
streams_separated:
type: ["boolean", "null"]
live_streaming:
type: ["boolean", "null"]
termination:

12
lib/crates/fabro-agent/src/tools.rs
View file

@ -235,6 +235,7 @@ pub fn make_shell_tool_with_config(config: &SessionOptions) -> RegisteredTool {
executor: Arc::new(move |args, ctx| {
Box::pin(async move {
let command = required_str(&args, "command")?;
let command = format!("exec 2>&1\n{command}");
let timeout_ms = args
.get("timeout_ms")
.and_then(serde_json::Value::as_u64)
@ -249,7 +250,7 @@ pub fn make_shell_tool_with_config(config: &SessionOptions) -> RegisteredTool {
let result = ctx
.env
.exec_command(
command,
&command,
timeout_ms,
None,
tool_env.as_ref(),
@ -266,12 +267,11 @@ pub fn make_shell_tool_with_config(config: &SessionOptions) -> RegisteredTool {
}
let _ = write!(
output,
"Exit code: {}\nstdout:\n{}\nstderr:\n{}",
"Exit code: {}\noutput:\n{}",
result
.exit_code
.map_or_else(|| "none".to_string(), |code| code.to_string()),
result.stdout,
result.stderr
result.stdout
);
Ok(output)
})
@ -916,8 +916,8 @@ mod tests {
let tool = make_shell_tool();
let env: Arc<dyn Sandbox> = Arc::new(MockSandbox {
exec_result: ExecResult {
stdout: String::new(),
stderr: "error".into(),
stdout: "error".into(),
stderr: String::new(),
exit_code: Some(1),
termination: CommandTermination::Exited,
duration_ms: 10,

5
lib/crates/fabro-api/build.rs
View file

@ -337,11 +337,6 @@ fn main() {
("StageCompletion", "fabro_types::StageCompletion", &[]),
("StageOutcome", "fabro_types::StageOutcome", &[]),
("StageState", "fabro_types::StageState", &[]),
(
"CommandOutputStream",
"fabro_types::CommandOutputStream",
&[],
),
("CommandTermination", "fabro_types::CommandTermination", &[]),
("StageProjection", "fabro_types::StageProjection", &[]),
("SecretMetadata", "fabro_types::SecretMetadata", &[]),

13
lib/crates/fabro-api/src/lib.rs
View file

@ -29,13 +29,12 @@ pub mod types {
BlockedReason, FailureReason, RunControlAction, RunStatus, SuccessReason, TerminalStatus,
};
pub use fabro_types::{
AuthMethod, BilledTokenCounts, CommandOutputStream, CommandTermination, DiffStats,
DiffSummary, DirtyStatus, EventEnvelope, GitContext, IdpIdentity, InterviewOption,
InterviewQuestionRecord, PendingInterviewRecord, PreRunPushOutcome, Principal,
QuestionType, RepositoryReference, RunClientProvenance, RunEvent, RunProjection,
RunProvenance, RunServerProvenance, RunSummary, SecretMetadata, SecretType, ServerSettings,
StageCompletion, StageOutcome, StageProjection, StageState, SystemActorKind, UserPrincipal,
WorkflowSettings,
AuthMethod, BilledTokenCounts, CommandTermination, DiffStats, DiffSummary, DirtyStatus,
EventEnvelope, GitContext, IdpIdentity, InterviewOption, InterviewQuestionRecord,
PendingInterviewRecord, PreRunPushOutcome, Principal, QuestionType, RepositoryReference,
RunClientProvenance, RunEvent, RunProjection, RunProvenance, RunServerProvenance,
RunSummary, SecretMetadata, SecretType, ServerSettings, StageCompletion, StageOutcome,
StageProjection, StageState, SystemActorKind, UserPrincipal, WorkflowSettings,
};
pub use crate::generated::types::*;

44
lib/crates/fabro-api/tests/command_output_stream_round_trip.rs
View file

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

3
lib/crates/fabro-api/tests/run_projection_round_trip.rs
View file

@ -68,8 +68,7 @@ fn run_projection_round_trips_populated_projection() {
"script_invocation": null,
"script_timing": null,
"parallel_results": null,
"stdout": "done",
"stderr": null
"output": "done"
}
}
});

3
lib/crates/fabro-api/tests/stage_projection_round_trip.rs
View file

@ -26,8 +26,7 @@ fn stage_projection_round_trips_representative_json() {
"script_invocation": { "command": "cargo test" },
"script_timing": { "duration_ms": 42 },
"parallel_results": [{ "branch": 0, "status": "succeeded" }],
"stdout": "ok",
"stderr": "",
"output": "ok",
"termination": "exited",
"started_at": "2026-04-29T12:34:00Z",
"duration_ms": 56000,

4
lib/crates/fabro-cli/tests/it/cmd/runner.rs
View file

@ -16,7 +16,7 @@ use fabro_store::EventEnvelope;
use fabro_test::{
assert_reqwest_status, expect_reqwest_json, fabro_json_snapshot, fabro_snapshot, test_context,
};
use fabro_types::{CommandOutputStream, EventBody, FailureReason, RunEvent, StageId};
use fabro_types::{EventBody, FailureReason, RunEvent, StageId};
use httpmock::MockServer;
use super::support::{
@ -505,7 +505,7 @@ methods = ["dev-token"]
let _probe = state
.stage(&probe_stage_id)
.expect("probe node state should exist");
let stdout = command_log_text(&run_dir, &probe_stage_id, CommandOutputStream::Stdout);
let stdout = command_log_text(&run_dir, &probe_stage_id);
assert!(
stdout.contains("probe-ran"),
"probe stage should have executed, got stdout:\n{stdout}"

10
lib/crates/fabro-cli/tests/it/cmd/support.rs
View file

@ -21,7 +21,7 @@ use fabro_config::daemon::ServerDaemon;
use fabro_config::{Storage, envfile};
use fabro_store::EventEnvelope;
use fabro_test::{TestContext, expect_reqwest_status};
use fabro_types::{CommandOutputStream, RunId, StageId};
use fabro_types::{RunId, StageId};
use httpmock::{Mock, MockServer};
use serde_json::Value;
use shlex::try_quote;
@ -704,15 +704,11 @@ pub(crate) fn run_events(run_dir: &Path) -> Vec<EventEnvelope> {
crate::support::parse_event_envelopes(&response)
}
pub(crate) fn command_log_text(
run_dir: &Path,
stage_id: &StageId,
stream: CommandOutputStream,
) -> String {
pub(crate) fn command_log_text(run_dir: &Path, stage_id: &StageId) -> String {
let run_id = infer_run_id(run_dir);
let response: CommandLogResponseRecord = block_on(get_server_json(
run_dir,
&format!("/api/v1/runs/{run_id}/stages/{stage_id}/logs/{stream}?offset=0&limit=1048576"),
&format!("/api/v1/runs/{run_id}/stages/{stage_id}/logs/output?offset=0&limit=1048576"),
));
let bytes = BASE64_STANDARD
.decode(&response.bytes_base64)

21
lib/crates/fabro-dump/src/lib.rs
View file

@ -129,16 +129,10 @@ impl RunDump {
parallel_results.clone(),
));
}
if let Some(stdout) = stage.stdout.as_ref() {
if let Some(output) = stage.output.as_ref() {
entries.push(RunDumpEntry::text_path(
&base.join("stdout.log"),
stdout.clone(),
));
}
if let Some(stderr) = stage.stderr.as_ref() {
entries.push(RunDumpEntry::text_path(
&base.join("stderr.log"),
stderr.clone(),
&base.join("output.log"),
output.clone(),
));
}
}
@ -597,8 +591,7 @@ mod tests {
stage.script_invocation = Some(serde_json::json!({ "command": "cargo test" }));
stage.script_timing = Some(serde_json::json!({ "duration_ms": 10 }));
stage.parallel_results = Some(serde_json::json!([{ "stage": "fanout@1" }]));
stage.stdout = Some("stdout".to_string());
stage.stderr = Some("stderr".to_string());
stage.output = Some("output".to_string());
let dump = RunDump::from_projection(&projection).unwrap();
let paths: Vec<&str> = dump
@ -619,8 +612,7 @@ mod tests {
assert!(paths.contains(&"stages/001-build@2/script_invocation.json"));
assert!(paths.contains(&"stages/001-build@2/script_timing.json"));
assert!(paths.contains(&"stages/001-build@2/parallel_results.json"));
assert!(paths.contains(&"stages/001-build@2/stdout.log"));
assert!(paths.contains(&"stages/001-build@2/stderr.log"));
assert!(paths.contains(&"stages/001-build@2/output.log"));
assert!(!paths.contains(&"start.json"));
assert!(!paths.contains(&"status.json"));
assert!(!paths.contains(&"checkpoint.json"));
@ -648,8 +640,7 @@ mod tests {
assert_eq!(node.prompt, None);
assert_eq!(node.response, None);
assert_eq!(node.diff, None);
assert_eq!(node.stdout, None);
assert_eq!(node.stderr, None);
assert_eq!(node.output, None);
assert_eq!(
node.provider_used,
Some(serde_json::json!({ "provider": "openai" }))

73
lib/crates/fabro-retro/src/retro_agent.rs
View file

@ -24,7 +24,7 @@ You have access to the run's data files:
- `graph.fabro` — the workflow source for the run
- `checkpoints/{seq:04}.json` — zero-padded checkpoint snapshots captured during the run
- `run.log` — server/worker log output for the run when available
- `stages/{rank:03}-{node_id}@{visit}/...` — execution-order-prefixed per-stage prompt, response, status, diff, stdout/stderr, and tool metadata files
- `stages/{rank:03}-{node_id}@{visit}/...` — execution-order-prefixed per-stage prompt, response, status, diff, output, and tool metadata files
## Your task
@ -430,8 +430,7 @@ mod tests {
stage.script_invocation = Some(serde_json::json!({ "command": "cargo test" }));
stage.script_timing = Some(serde_json::json!({ "duration_ms": 10 }));
stage.parallel_results = Some(serde_json::json!([{ "stage": "fanout@1" }]));
stage.stdout = Some("stdout".to_string());
stage.stderr = Some("stderr".to_string());
stage.output = Some("output".to_string());
upload_data_files(
&sandbox,
@ -473,10 +472,10 @@ mod tests {
"done"
);
assert_eq!(
fs::read_to_string(target_dir.join("stages/001-build@2/stdout.log"))
fs::read_to_string(target_dir.join("stages/001-build@2/output.log"))
.await
.expect("stdout file should exist"),
"stdout"
.expect("output file should exist"),
"output"
);
assert_eq!(
fs::read_to_string(target_dir.join("events.jsonl"))
@ -501,7 +500,7 @@ mod tests {
}
#[tokio::test]
async fn upload_data_files_resolves_command_stdout_stderr_blob_refs() {
async fn upload_data_files_resolves_command_output_blob_refs() {
let sandbox_root = tempfile::tempdir().expect("sandbox tempdir should exist");
let sandbox: Arc<dyn Sandbox> =
Arc::new(LocalSandbox::new(sandbox_root.path().to_path_buf()));
@ -509,37 +508,28 @@ mod tests {
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 output_blob = serde_json::to_vec("resolved output").unwrap();
let output_id = fabro_types::RunBlobId::new(&output_blob);
let stage_id = StageId::new("build", 1);
let mut state = RunProjection::default();
let stdout_ref = fabro_types::format_blob_ref(&stdout_id);
let stderr_ref = fabro_types::format_blob_ref(&stderr_id);
let output_ref = fabro_types::format_blob_ref(&output_id);
let stage = state.stage_entry(stage_id.node_id(), stage_id.visit(), first_event_seq(1));
stage.script_invocation = Some(serde_json::json!({
"command": "cargo test",
"stdout": stdout_ref,
"stderr": stderr_ref,
"output": output_ref,
}));
stage.script_timing = Some(serde_json::json!({
"exit_code": 0,
"stdout": stdout_ref,
"stderr": stderr_ref,
"output": output_ref,
}));
stage.stdout = Some(stdout_ref);
stage.stderr = Some(stderr_ref);
stage.output = Some(output_ref);
let reader: BlobReader = Box::new(move |blob_id| {
let stdout_blob = stdout_blob.clone();
let stderr_blob = stderr_blob.clone();
let output_blob = output_blob.clone();
Box::pin(async move {
if blob_id == stdout_id {
Ok(Some(stdout_blob.into()))
} else if blob_id == stderr_id {
Ok(Some(stderr_blob.into()))
if blob_id == output_id {
Ok(Some(output_blob.into()))
} else {
Ok(None)
}
@ -551,16 +541,10 @@ mod tests {
.expect("retro files should upload");
assert_eq!(
fs::read_to_string(target_dir.join("stages/001-build@1/stdout.log"))
fs::read_to_string(target_dir.join("stages/001-build@1/output.log"))
.await
.expect("stdout file should exist"),
"resolved stdout"
);
assert_eq!(
fs::read_to_string(target_dir.join("stages/001-build@1/stderr.log"))
.await
.expect("stderr file should exist"),
"resolved stderr"
.expect("output file should exist"),
"resolved output"
);
let script_timing: serde_json::Value = serde_json::from_str(
@ -569,8 +553,7 @@ mod tests {
.expect("script timing should exist"),
)
.expect("script timing should parse");
assert_eq!(script_timing["stdout"], "resolved stdout");
assert_eq!(script_timing["stderr"], "resolved stderr");
assert_eq!(script_timing["output"], "resolved output");
let script_invocation: serde_json::Value = serde_json::from_str(
&fs::read_to_string(target_dir.join("stages/001-build@1/script_invocation.json"))
@ -578,8 +561,7 @@ mod tests {
.expect("script invocation should exist"),
)
.expect("script invocation should parse");
assert_eq!(script_invocation["stdout"], "resolved stdout");
assert_eq!(script_invocation["stderr"], "resolved stderr");
assert_eq!(script_invocation["output"], "resolved output");
let run_json: serde_json::Value = serde_json::from_str(
&fs::read_to_string(target_dir.join("run.json"))
@ -588,18 +570,13 @@ mod tests {
)
.expect("run.json should parse");
assert_eq!(
run_json["stages"]["build@1"]["script_timing"]["stdout"],
"resolved stdout"
run_json["stages"]["build@1"]["script_timing"]["output"],
"resolved output"
);
assert_eq!(
run_json["stages"]["build@1"]["script_timing"]["stderr"],
"resolved stderr"
run_json["stages"]["build@1"]["script_invocation"]["output"],
"resolved output"
);
assert_eq!(
run_json["stages"]["build@1"]["script_invocation"]["stdout"],
"resolved stdout"
);
assert!(run_json["stages"]["build@1"]["stdout"].is_null());
assert!(run_json["stages"]["build@1"]["stderr"].is_null());
assert!(run_json["stages"]["build@1"]["output"].is_null());
}
}

20
lib/crates/fabro-server/src/principal_middleware.rs
View file

@ -6,7 +6,7 @@ use axum::http::StatusCode;
use axum::http::request::Parts;
use axum::middleware::Next;
use axum::response::{IntoResponse, Response};
use fabro_types::{CommandOutputStream, Principal, RunBlobId, RunId, StageId, UserPrincipal};
use fabro_types::{Principal, RunBlobId, RunId, StageId, UserPrincipal};
use jsonwebtoken::decode_header;
use strum::IntoStaticStr;
@ -54,11 +54,7 @@ pub(crate) struct RequireRunScoped(pub(crate) RunId);
pub(crate) struct RequireRunBlob(pub(crate) RunId, pub(crate) RunBlobId);
pub(crate) struct RequireRunStageScoped(pub(crate) RunId, pub(crate) String);
pub(crate) struct RequireStageArtifact(pub(crate) RunId, pub(crate) StageId);
pub(crate) struct RequireCommandLog(
pub(crate) RunId,
pub(crate) StageId,
pub(crate) CommandOutputStream,
);
pub(crate) struct RequireCommandLog(pub(crate) RunId, pub(crate) StageId);
#[derive(Clone, Debug)]
pub(crate) struct AuthenticatedUser {
@ -246,18 +242,14 @@ impl FromRequestParts<Arc<AppState>> for RequireCommandLog {
parts: &mut Parts,
state: &Arc<AppState>,
) -> Result<Self, Self::Rejection> {
let Path((id, stage_id, stream)): Path<(String, String, String)> =
Path::from_request_parts(parts, state)
.await
.map_err(IntoResponse::into_response)?;
let Path((id, stage_id)): Path<(String, String)> = Path::from_request_parts(parts, state)
.await
.map_err(IntoResponse::into_response)?;
let run_id = parse_run_id_path(&id)?;
let stage_id = parse_stage_id_path(&stage_id)?;
let stream = stream
.parse::<CommandOutputStream>()
.map_err(|_| ApiError::bad_request("Invalid command log stream.").into_response())?;
require_worker_or_user_for_run(&auth_slot_from_parts(parts), &run_id)
.map_err(IntoResponse::into_response)?;
Ok(Self(run_id, stage_id, stream))
Ok(Self(run_id, stage_id))
}
}

2
lib/crates/fabro-server/src/server.rs
View file

@ -71,8 +71,6 @@ use fabro_store::{
};
#[cfg(test)]
use fabro_types::BlockedReason;
#[cfg(test)]
use fabro_types::CommandOutputStream;
use fabro_types::settings::run::RunMode;
use fabro_types::settings::server::{
GithubIntegrationSettings, GithubIntegrationStrategy, LogDestination,

2
lib/crates/fabro-server/src/server/handler/mod.rs
View file

@ -44,7 +44,7 @@ pub(super) fn demo_routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/blobs", post(not_implemented))
.route("/runs/{id}/blobs/{blobId}", get(not_implemented))
.route(
"/runs/{id}/stages/{stageId}/logs/{stream}",
"/runs/{id}/stages/{stageId}/logs/output",
get(not_implemented),
)
.route("/runs/{id}/checkpoint", get(demo::checkpoint_stub))

22
lib/crates/fabro-server/src/server/handler/runs.rs
View file

@ -16,8 +16,8 @@ use fabro_api::types::{
use fabro_config::Storage;
use fabro_interview::AnswerSubmission;
use fabro_types::{
CommandOutputStream, Principal, RunClientProvenance, RunId, RunProvenance, RunServerProvenance,
UserPrincipal, parse_blob_ref,
Principal, RunClientProvenance, RunId, RunProvenance, RunServerProvenance, UserPrincipal,
parse_blob_ref,
};
use fabro_util::version::FABRO_VERSION;
use fabro_workflow::command_log::{command_log_path, read_json_string_blob, read_log_slice};
@ -57,7 +57,7 @@ pub(super) fn routes() -> Router<Arc<AppState>> {
.route("/runs/{id}/state", get(get_run_state))
.route("/runs/{id}/logs", get(get_run_logs))
.route(
"/runs/{id}/stages/{stageId}/logs/{stream}",
"/runs/{id}/stages/{stageId}/logs/output",
get(get_run_stage_command_log),
)
.route("/runs/{id}/settings", get(get_run_settings))
@ -302,7 +302,6 @@ struct CommandLogQuery {
#[derive(Debug, serde::Serialize)]
struct CommandLogResponseBody {
stream: CommandOutputStream,
offset: u64,
next_offset: u64,
total_bytes: u64,
@ -677,7 +676,7 @@ async fn get_run_logs(
}
async fn get_run_stage_command_log(
RequireCommandLog(id, stage_id, stream): RequireCommandLog,
RequireCommandLog(id, stage_id): RequireCommandLog,
State(state): State<Arc<AppState>>,
Query(query): Query<CommandLogQuery>,
) -> Response {
@ -701,10 +700,7 @@ async fn get_run_stage_command_log(
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 stream_value = node.output.as_deref();
let cas_ref = stream_value
.filter(|value| parse_blob_ref(value).is_some())
.map(str::to_string);
@ -715,12 +711,11 @@ async fn get_run_stage_command_log(
.run_scratch(&id)
.root()
.to_path_buf();
let scratch_path = command_log_path(&run_dir, &stage_id, stream);
let scratch_path = command_log_path(&run_dir, &stage_id);
match read_log_slice(&scratch_path, query.offset, limit).await {
Ok((bytes, total_bytes)) => {
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Sliced { bytes, total_bytes },
@ -746,7 +741,6 @@ async fn get_run_stage_command_log(
}
};
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(text.as_bytes()),
@ -758,7 +752,6 @@ async fn get_run_stage_command_log(
if let Some(inline_text) = stream_value {
return build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(inline_text.as_bytes()),
@ -769,7 +762,6 @@ async fn get_run_stage_command_log(
}
build_command_log_response(
stream,
query.offset,
limit,
LogSource::Full(&[]),
@ -788,7 +780,6 @@ enum LogSource<'a> {
}
fn build_command_log_response(
stream: CommandOutputStream,
requested_offset: u64,
limit: u64,
source: LogSource<'_>,
@ -812,7 +803,6 @@ fn build_command_log_response(
}
};
Json(CommandLogResponseBody {
stream,
offset,
next_offset: offset + u64::try_from(body_bytes.len()).unwrap_or(u64::MAX),
total_bytes,

82
lib/crates/fabro-server/src/server/tests.rs
View file

@ -4173,7 +4173,7 @@ async fn get_run_stage_command_log_returns_scratch_slice() {
.run_scratch(&run_id)
.root()
.to_path_buf();
let log_path = command_log_path(&run_dir, &stage_id, CommandOutputStream::Stdout);
let log_path = command_log_path(&run_dir, &stage_id);
tokio::fs::create_dir_all(log_path.parent().unwrap())
.await
.unwrap();
@ -4182,7 +4182,7 @@ async fn get_run_stage_command_log_returns_scratch_slice() {
let req = Request::builder()
.method("GET")
.uri(api(&format!(
"/runs/{run_id}/stages/{stage_id}/logs/stdout?offset=6&limit=5"
"/runs/{run_id}/stages/{stage_id}/logs/output?offset=6&limit=5"
)))
.body(Body::empty())
.unwrap();
@ -4193,7 +4193,7 @@ async fn get_run_stage_command_log_returns_scratch_slice() {
.decode(body["bytes_base64"].as_str().unwrap())
.unwrap();
assert_eq!(body["stream"], "stdout");
assert!(body.get("stream").is_none());
assert_eq!(body["offset"], 6);
assert_eq!(body["next_offset"], 11);
assert_eq!(body["total_bytes"], 11);
@ -4209,16 +4209,11 @@ async fn get_run_stage_command_log_returns_cas_slice() {
let app = crate::test_support::build_test_router(Arc::clone(&state));
let run_id = RunId::new();
let run_store = state.store.create_run(&run_id).await.unwrap();
let stdout_blob = run_store
let output_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}");
let output_ref = format!("blob://sha256/{output_blob}");
for event in [
workflow_event::Event::RunSubmitted {
definition_blob: None,
@ -4232,16 +4227,13 @@ async fn get_run_stage_command_log_returns_cas_slice() {
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,
termination: CommandTermination::Exited,
stdout_bytes: 11,
stderr_bytes: 0,
streams_separated: true,
live_streaming: false,
node_id: "script_node".to_string(),
output: output_ref.clone(),
exit_code: Some(0),
duration_ms: 5,
termination: CommandTermination::Exited,
output_bytes: 11,
live_streaming: false,
},
] {
workflow_event::append_event(&run_store, &run_id, &event)
@ -4252,7 +4244,7 @@ async fn get_run_stage_command_log_returns_cas_slice() {
let req = Request::builder()
.method("GET")
.uri(api(&format!(
"/runs/{run_id}/stages/script_node@1/logs/stdout?offset=6&limit=5"
"/runs/{run_id}/stages/script_node@1/logs/output?offset=6&limit=5"
)))
.body(Body::empty())
.unwrap();
@ -4263,13 +4255,13 @@ async fn get_run_stage_command_log_returns_cas_slice() {
.decode(body["bytes_base64"].as_str().unwrap())
.unwrap();
assert_eq!(body["stream"], "stdout");
assert!(body.get("stream").is_none());
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["cas_ref"], output_ref);
assert_eq!(body["live_streaming"], false);
}
@ -4280,16 +4272,11 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
let run_id = RunId::new();
let stage_id = StageId::new("script_node", 1);
let run_store = state.store.create_run(&run_id).await.unwrap();
let stdout_blob = run_store
let output_blob = run_store
.write_blob(&serde_json::to_vec("cas log").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}");
let output_ref = format!("blob://sha256/{output_blob}");
for event in [
workflow_event::Event::RunSubmitted {
definition_blob: None,
@ -4303,16 +4290,13 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
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,
termination: CommandTermination::Exited,
stdout_bytes: 7,
stderr_bytes: 0,
streams_separated: true,
live_streaming: false,
node_id: "script_node".to_string(),
output: output_ref.clone(),
exit_code: Some(0),
duration_ms: 5,
termination: CommandTermination::Exited,
output_bytes: 7,
live_streaming: false,
},
] {
workflow_event::append_event(&run_store, &run_id, &event)
@ -4324,7 +4308,7 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
.run_scratch(&run_id)
.root()
.to_path_buf();
let log_path = command_log_path(&run_dir, &stage_id, CommandOutputStream::Stdout);
let log_path = command_log_path(&run_dir, &stage_id);
tokio::fs::create_dir_all(log_path.parent().unwrap())
.await
.unwrap();
@ -4333,7 +4317,7 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
let req = Request::builder()
.method("GET")
.uri(api(&format!(
"/runs/{run_id}/stages/{stage_id}/logs/stdout?offset=0&limit=64"
"/runs/{run_id}/stages/{stage_id}/logs/output?offset=0&limit=64"
)))
.body(Body::empty())
.unwrap();
@ -4344,13 +4328,13 @@ async fn get_run_stage_command_log_prefers_scratch_when_cas_ref_exists() {
.decode(body["bytes_base64"].as_str().unwrap())
.unwrap();
assert_eq!(body["stream"], "stdout");
assert!(body.get("stream").is_none());
assert_eq!(body["offset"], 0);
assert_eq!(body["next_offset"], 11);
assert_eq!(body["total_bytes"], 11);
assert_eq!(bytes, b"scratch log");
assert_eq!(body["eof"], true);
assert_eq!(body["cas_ref"], stdout_ref);
assert_eq!(body["cas_ref"], output_ref);
assert_eq!(body["live_streaming"], false);
}
@ -4366,7 +4350,7 @@ async fn get_run_stage_command_log_returns_not_found_for_missing_stage() {
let req = Request::builder()
.method("GET")
.uri(api(&format!("/runs/{run_id}/stages/missing@1/logs/stdout")))
.uri(api(&format!("/runs/{run_id}/stages/missing@1/logs/output")))
.body(Body::empty())
.unwrap();
@ -6004,7 +5988,7 @@ async fn worker_token_controls_command_log_route() {
.clone()
.oneshot(bearer_request(
Method::GET,
&format!("/runs/{run_id}/stages/code@1/logs/stdout"),
&format!("/runs/{run_id}/stages/code@1/logs/output"),
&worker_token,
Body::empty(),
))
@ -6016,7 +6000,7 @@ async fn worker_token_controls_command_log_route() {
.clone()
.oneshot(bearer_request(
Method::GET,
&format!("/runs/{run_id}/stages/code@1/logs/stdout"),
&format!("/runs/{run_id}/stages/code@1/logs/output"),
&user_jwt,
Body::empty(),
))
@ -6028,7 +6012,7 @@ async fn worker_token_controls_command_log_route() {
.clone()
.oneshot(bearer_request(
Method::GET,
&format!("/runs/{run_id}/stages/code@1/logs/stdout"),
&format!("/runs/{run_id}/stages/code@1/logs/output"),
&mismatched_worker_token,
Body::empty(),
))
@ -6040,7 +6024,7 @@ async fn worker_token_controls_command_log_route() {
.oneshot(
Request::builder()
.method(Method::GET)
.uri(api(&format!("/runs/{run_id}/stages/code@1/logs/stdout")))
.uri(api(&format!("/runs/{run_id}/stages/code@1/logs/output")))
.body(Body::empty())
.unwrap(),
)

16
lib/crates/fabro-server/tests/it/scenario/usage.rs
View file

@ -17,6 +17,14 @@ const COMMAND_DOT: &str = r#"digraph Test {
start -> echo_task -> exit
}"#;
const WAIT_DOT: &str = r#"digraph Test {
graph [goal="Test"]
start [shape=Mdiamond]
wait_task [shape=insulator, duration="1ms"]
exit [shape=Msquare]
start -> wait_task -> exit
}"#;
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn aggregate_billing_increments_after_run_completes() {
let state = test_app_state_with_options(test_settings(), 5);
@ -59,15 +67,13 @@ async fn run_billing_includes_completed_non_llm_stages() {
let state = test_app_state_with_options(test_settings(), 5);
let app = test_app_with_scheduler(state);
let run_id =
create_and_start_run_from_manifest(&app, minimal_manifest_json_with_dry_run(MINIMAL_DOT))
.await;
let run_id = create_and_start_run_from_manifest(&app, minimal_manifest_json(WAIT_DOT)).await;
let status = wait_for_run_status(&app, &run_id, &["succeeded", "failed"]).await;
assert_eq!(status, "succeeded");
let billing = run_billing(&app, &run_id).await;
assert_non_llm_billing(&billing, &["start"]);
assert_non_llm_billing(&billing, &["wait_task"]);
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@ -81,7 +87,7 @@ async fn run_billing_includes_completed_command_stages() {
assert_eq!(status, "succeeded");
let billing = run_billing(&app, &run_id).await;
assert_non_llm_billing(&billing, &["echo_task", "start"]);
assert_non_llm_billing(&billing, &["echo_task"]);
}
async fn run_billing(app: &axum::Router, run_id: &str) -> serde_json::Value {

48
lib/crates/fabro-store/src/run_state.rs
View file

@ -386,11 +386,8 @@ impl RunProjectionReducer for RunProjection {
let Some(stage) = stage_at_stored_or_current_visit(self, stored, event.seq) else {
return Ok(());
};
stage.stdout = Some(props.stdout.clone());
stage.stderr = Some(props.stderr.clone());
stage.stdout_bytes = Some(props.stdout_bytes);
stage.stderr_bytes = Some(props.stderr_bytes);
stage.streams_separated = Some(props.streams_separated);
stage.output = Some(props.output.clone());
stage.output_bytes = Some(props.output_bytes);
stage.live_streaming = Some(props.live_streaming);
stage.termination = Some(props.termination);
stage.script_timing = Some(script_timing);
@ -402,8 +399,7 @@ impl RunProjectionReducer for RunProjection {
apply_agent_cli_terminal(
stage,
props,
&props.stdout,
&props.stderr,
merge_agent_cli_output(&props.stdout, &props.stderr),
CommandTermination::Exited,
)?;
}
@ -414,8 +410,7 @@ impl RunProjectionReducer for RunProjection {
apply_agent_cli_terminal(
stage,
props,
&props.stdout,
&props.stderr,
merge_agent_cli_output(&props.stdout, &props.stderr),
CommandTermination::Cancelled,
)?;
}
@ -426,8 +421,7 @@ impl RunProjectionReducer for RunProjection {
apply_agent_cli_terminal(
stage,
props,
&props.stdout,
&props.stderr,
merge_agent_cli_output(&props.stdout, &props.stderr),
CommandTermination::TimedOut,
)?;
}
@ -718,19 +712,26 @@ fn provider_used_from_agent_cli_started(props: &AgentCliStartedProps) -> Value {
fn apply_agent_cli_terminal(
stage: &mut StageProjection,
props: &impl serde::Serialize,
stdout: &str,
stderr: &str,
output: String,
termination: CommandTermination,
) -> Result<()> {
let script_timing = serde_json::to_value(props)
.map_err(|err| Error::InvalidEvent(format!("invalid agent.cli terminal payload: {err}")))?;
stage.stdout = Some(stdout.to_string());
stage.stderr = Some(stderr.to_string());
stage.output = Some(output);
stage.termination = Some(termination);
stage.script_timing = Some(script_timing);
Ok(())
}
fn merge_agent_cli_output(stdout: &str, stderr: &str) -> String {
match (stdout.is_empty(), stderr.is_empty()) {
(true, true) => String::new(),
(false, true) => stdout.to_string(),
(true, false) => stderr.to_string(),
(false, false) => format!("{stdout}\n{stderr}"),
}
}
#[cfg(test)]
mod tests {
use std::collections::{BTreeMap, HashMap};
@ -911,7 +912,7 @@ mod tests {
"build@2": {
"first_event_seq": 1,
"diff": "diff --git a/file b/file",
"stdout": "done"
"output": "done"
}
}
}))
@ -928,7 +929,7 @@ mod tests {
serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap();
let serialized = serde_json::to_value(&state).unwrap();
let round_tripped_node = round_tripped.stage(&stage_id).unwrap();
assert_eq!(round_tripped_node.stdout.as_deref(), Some("done"));
assert_eq!(round_tripped_node.output.as_deref(), Some("done"));
assert_eq!(round_tripped.list_node_visits("build"), vec![2]);
assert_eq!(
round_tripped.pending_control,
@ -955,7 +956,7 @@ mod tests {
restart_failure_signatures: HashMap::new(),
node_visits: HashMap::from([("build".to_string(), 2usize)]),
})];
state.stage_entry("build", 2, first_event_seq(7)).stdout = Some("done".to_string());
state.stage_entry("build", 2, first_event_seq(7)).output = Some("done".to_string());
let round_tripped: RunProjection =
serde_json::from_value(serde_json::to_value(&state).unwrap()).unwrap();
@ -964,7 +965,7 @@ mod tests {
round_tripped
.stage(&StageId::new("build", 2))
.unwrap()
.stdout
.output
.as_deref(),
Some("done")
);
@ -1127,8 +1128,7 @@ mod tests {
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.stdout.as_deref(), Some("done"));
assert_eq!(stage.stderr.as_deref(), Some("warn"));
assert_eq!(stage.output.as_deref(), Some("done\nwarn"));
assert_eq!(stage.termination, Some(CommandTermination::Exited));
assert_eq!(
stage.script_timing.as_ref().unwrap()["duration_ms"],
@ -1155,8 +1155,7 @@ mod tests {
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.stdout.as_deref(), Some("partial"));
assert_eq!(stage.stderr.as_deref(), Some("cancelled"));
assert_eq!(stage.output.as_deref(), Some("partial\ncancelled"));
assert_eq!(stage.termination, Some(CommandTermination::Cancelled));
assert_eq!(
stage.script_timing.as_ref().unwrap()["duration_ms"],
@ -1183,8 +1182,7 @@ mod tests {
.unwrap();
let stage = state.stage(&stage_id).unwrap();
assert_eq!(stage.stdout.as_deref(), Some("partial"));
assert_eq!(stage.stderr.as_deref(), Some("timeout"));
assert_eq!(stage.output.as_deref(), Some("partial\ntimeout"));
assert_eq!(stage.termination, Some(CommandTermination::TimedOut));
assert_eq!(
stage.script_timing.as_ref().unwrap()["duration_ms"],

3
lib/crates/fabro-store/src/serializable_projection.rs
View file

@ -14,8 +14,7 @@ impl Serialize for SerializableProjection<'_> {
stage.prompt = None;
stage.response = None;
stage.diff = None;
stage.stdout = None;
stage.stderr = None;
stage.output = None;
}
projection.serialize(serializer)

6
lib/crates/fabro-store/tests/serializable_projection.rs
View file

@ -118,8 +118,7 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() {
stage.parallel_results = Some(json!([{ "stage": "fanout@1" }]));
stage.duration_ms = Some(1234);
stage.usage = Some(sample_usage());
stage.stdout = Some("stdout".to_string());
stage.stderr = Some("stderr".to_string());
stage.output = Some("output".to_string());
let serialized = serde_json::to_value(SerializableProjection(&projection))
.expect("projection should serialize");
@ -144,8 +143,7 @@ fn serializable_projection_round_trips_and_trims_bulky_node_fields() {
assert_eq!(node.prompt, None);
assert_eq!(node.response, None);
assert_eq!(node.diff, None);
assert_eq!(node.stdout, None);
assert_eq!(node.stderr, None);
assert_eq!(node.output, None);
assert_eq!(node.first_event_seq, first_event_seq(2));
assert_eq!(
node.completion

108
lib/crates/fabro-types/src/run_event/misc.rs
View file

@ -1,4 +1,4 @@
use serde::{Deserialize, Serialize};
use serde::{Deserialize, Serialize, de};
use serde_json::Value;
use super::ExecOutputTail;
@ -204,22 +204,104 @@ pub struct CommandStartedProps {
pub timeout_ms: Option<u64>,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[derive(Debug, Clone, PartialEq, Serialize)]
pub struct CommandCompletedProps {
pub stdout: String,
pub stderr: String,
pub output: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub exit_code: Option<i32>,
pub duration_ms: u64,
pub termination: CommandTermination,
pub exit_code: Option<i32>,
pub duration_ms: u64,
pub termination: CommandTermination,
#[serde(default)]
pub stdout_bytes: u64,
pub output_bytes: u64,
#[serde(default)]
pub stderr_bytes: u64,
#[serde(default)]
pub streams_separated: bool,
#[serde(default)]
pub live_streaming: bool,
pub live_streaming: bool,
}
impl<'de> Deserialize<'de> for CommandCompletedProps {
fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
where
D: serde::Deserializer<'de>,
{
#[derive(Deserialize)]
struct Wire {
#[serde(default)]
output: Option<String>,
#[serde(default)]
stdout: Option<String>,
#[serde(default)]
stderr: Option<String>,
#[serde(default)]
exit_code: Option<i32>,
duration_ms: u64,
termination: CommandTermination,
#[serde(default)]
output_bytes: Option<u64>,
#[serde(default)]
stdout_bytes: Option<u64>,
#[serde(default)]
stderr_bytes: Option<u64>,
#[serde(default)]
live_streaming: bool,
}
let wire = Wire::deserialize(deserializer)?;
let (output, output_bytes) = if let Some(output) = wire.output {
(output, wire.output_bytes.unwrap_or(0))
} else {
let stdout_bytes = wire.stdout_bytes.unwrap_or(0);
let stderr_bytes = wire.stderr_bytes.unwrap_or(0);
let legacy_output = if stdout_bytes == 0 && stderr_bytes > 0 && wire.stderr.is_some() {
wire.stderr
} else {
wire.stdout.or(wire.stderr)
}
.ok_or_else(|| de::Error::missing_field("output"))?;
let legacy_bytes = if stdout_bytes == 0 && stderr_bytes > 0 {
stderr_bytes
} else {
stdout_bytes
};
(legacy_output, legacy_bytes)
};
Ok(Self {
output,
exit_code: wire.exit_code,
duration_ms: wire.duration_ms,
termination: wire.termination,
output_bytes,
live_streaming: wire.live_streaming,
})
}
}
#[cfg(test)]
mod tests {
use serde_json::json;
use super::*;
#[test]
fn command_completed_deserializes_legacy_stdout_stderr_shape() {
let props: CommandCompletedProps = serde_json::from_value(json!({
"stdout": "blob://sha256/stdout",
"stderr": "blob://sha256/stderr",
"exit_code": 1,
"duration_ms": 42,
"termination": "exited",
"stdout_bytes": 0,
"stderr_bytes": 12,
"streams_separated": true,
"live_streaming": true
}))
.unwrap();
assert_eq!(props.output, "blob://sha256/stderr");
assert_eq!(props.output_bytes, 12);
assert_eq!(props.exit_code, Some(1));
assert_eq!(props.termination, CommandTermination::Exited);
assert!(props.live_streaming);
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]

16
lib/crates/fabro-types/src/run_projection.rs
View file

@ -52,14 +52,9 @@ pub struct StageProjection {
pub script_invocation: Option<serde_json::Value>,
pub script_timing: Option<serde_json::Value>,
pub parallel_results: Option<serde_json::Value>,
pub stdout: Option<String>,
pub stderr: Option<String>,
pub output: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stdout_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub stderr_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub streams_separated: Option<bool>,
pub output_bytes: Option<u64>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub live_streaming: Option<bool>,
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -99,11 +94,8 @@ impl StageProjection {
script_invocation: None,
script_timing: None,
parallel_results: None,
stdout: None,
stderr: None,
stdout_bytes: None,
stderr_bytes: None,
streams_separated: None,
output: None,
output_bytes: None,
live_streaming: None,
termination: None,
started_at: None,

7
lib/crates/fabro-workflow/src/artifact.rs
View file

@ -122,7 +122,7 @@ pub async fn resolve_context_for_edge_selection(
run_store: &RunStoreHandle,
) -> Result<Context> {
let mut values = context.snapshot();
for key in [context::keys::COMMAND_OUTPUT, context::keys::COMMAND_STDERR] {
for key in [context::keys::COMMAND_OUTPUT] {
if let Some(Value::String(current)) = values.get_mut(key) {
*current = resolve_text_or_blob_ref_str(current, run_store).await?;
}
@ -274,10 +274,7 @@ fn resolve_execution_value<'a>(
Box::pin(async move {
match value {
Value::String(current) => {
if matches!(
key,
Some(context::keys::COMMAND_OUTPUT | context::keys::COMMAND_STDERR)
) {
if matches!(key, Some(context::keys::COMMAND_OUTPUT)) {
*current = resolve_text_or_blob_ref_str(current, run_store).await?;
} else if let Some(blob_id) = parse_blob_ref(current) {
*current = materialize_blob_ref(&blob_id, run_store, env, run_dir).await?;

79
lib/crates/fabro-workflow/src/command_log.rs
View file

@ -3,7 +3,7 @@ use std::sync::Arc;
use fabro_config::RunScratch;
use fabro_store::stage_storage_segment;
use fabro_types::{CommandOutputStream, StageId, format_blob_ref};
use fabro_types::{StageId, format_blob_ref};
use serde_json::Value;
use tokio::fs::{self, File, OpenOptions};
use tokio::io::{AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
@ -14,26 +14,20 @@ 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 output_ref: String,
pub output_bytes: u64,
pub output_text: String,
}
pub struct CommandLogRecorder {
stdout: Mutex<File>,
stderr: Mutex<File>,
stdout_path: PathBuf,
stderr_path: PathBuf,
output: Mutex<File>,
output_path: PathBuf,
}
impl CommandLogRecorder {
pub async fn create(run_dir: &Path, stage_id: &StageId) -> Result<Arc<Self>> {
let stdout_path = command_log_path(run_dir, stage_id, CommandOutputStream::Stdout);
let stderr_path = command_log_path(run_dir, stage_id, CommandOutputStream::Stderr);
if let Some(parent) = stdout_path.parent() {
let output_path = command_log_path(run_dir, stage_id);
if let Some(parent) = output_path.parent() {
fs::create_dir_all(parent).await.map_err(|err| {
Error::Io(format!(
"creating command log directory {}: {err}",
@ -41,82 +35,59 @@ impl CommandLogRecorder {
))
})?;
}
let stdout = open_truncated(&stdout_path).await?;
let stderr = open_truncated(&stderr_path).await?;
let output = open_truncated(&output_path).await?;
Ok(Arc::new(Self {
stdout: Mutex::new(stdout),
stderr: Mutex::new(stderr),
stdout_path,
stderr_path,
output: Mutex::new(output),
output_path,
}))
}
pub async fn append(&self, stream: CommandOutputStream, bytes: &[u8]) -> Result<()> {
pub async fn append(&self, 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,
};
let mut file = self.output.lock().await;
file.write_all(bytes)
.await
.map_err(|err| Error::Io(format!("writing command {stream} log failed: {err}")))?;
.map_err(|err| Error::Io(format!("writing command output log failed: {err}")))?;
Ok(())
}
pub async fn finalize(&self, run_store: &RunStoreHandle) -> Result<FinalizedCommandLogs> {
self.flush_all().await?;
let (stdout_text, stdout_bytes) = read_lossy_text(&self.stdout_path).await?;
let (stderr_text, stderr_bytes) = read_lossy_text(&self.stderr_path).await?;
let stdout_ref = write_json_string_blob(run_store, &stdout_text).await?;
let stderr_ref = write_json_string_blob(run_store, &stderr_text).await?;
let (output_text, output_bytes) = read_lossy_text(&self.output_path).await?;
let output_ref = write_json_string_blob(run_store, &output_text).await?;
Ok(FinalizedCommandLogs {
stdout_ref,
stderr_ref,
stdout_bytes,
stderr_bytes,
stdout_text,
stderr_text,
output_ref,
output_bytes,
output_text,
})
}
pub async fn discard(self: Arc<Self>) -> Result<()> {
self.flush_all().await?;
let stdout_path = self.stdout_path.clone();
let stderr_path = self.stderr_path.clone();
let output_path = self.output_path.clone();
drop(self);
remove_if_exists(&stdout_path).await?;
remove_if_exists(&stderr_path).await
remove_if_exists(&output_path).await
}
async fn flush_all(&self) -> Result<()> {
self.stdout
self.output
.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}")))?;
.map_err(|err| Error::Io(format!("flushing command output log failed: {err}")))?;
Ok(())
}
}
pub fn command_log_path(
run_dir: &Path,
stage_id: &StageId,
stream: CommandOutputStream,
) -> PathBuf {
pub fn command_log_path(run_dir: &Path, stage_id: &StageId) -> PathBuf {
RunScratch::new(run_dir)
.runtime_dir()
.join("stages")
.join(stage_storage_segment(stage_id))
.join(stream.command_log_relative_path())
.join("output.log")
}
pub async fn read_log_slice(

1
lib/crates/fabro-workflow/src/context.rs
View file

@ -31,7 +31,6 @@ pub mod keys {
// --- command.* keys ---
pub const COMMAND_OUTPUT: &str = "command.output";
pub const COMMAND_STDERR: &str = "command.stderr";
// --- human.gate.* keys ---
pub const HUMAN_GATE_SELECTED: &str = "human.gate.selected";

22
lib/crates/fabro-workflow/src/event/convert.rs
View file

@ -965,26 +965,20 @@ fn event_body_from_event(event: &Event) -> EventBody {
timeout_ms: *timeout_ms,
}),
Event::CommandCompleted {
stdout,
stderr,
output,
exit_code,
duration_ms,
termination,
stdout_bytes,
stderr_bytes,
streams_separated,
output_bytes,
live_streaming,
..
} => EventBody::CommandCompleted(fabro_types::CommandCompletedProps {
stdout: stdout.clone(),
stderr: stderr.clone(),
exit_code: *exit_code,
duration_ms: *duration_ms,
termination: *termination,
stdout_bytes: *stdout_bytes,
stderr_bytes: *stderr_bytes,
streams_separated: *streams_separated,
live_streaming: *live_streaming,
output: output.clone(),
exit_code: *exit_code,
duration_ms: *duration_ms,
termination: *termination,
output_bytes: *output_bytes,
live_streaming: *live_streaming,
}),
Event::AgentCliStarted {
visit,

23
lib/crates/fabro-workflow/src/event/events.rs
View file

@ -519,17 +519,14 @@ pub enum Event {
timeout_ms: Option<u64>,
},
CommandCompleted {
node_id: String,
stdout: String,
stderr: String,
node_id: String,
output: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
exit_code: Option<i32>,
duration_ms: u64,
termination: CommandTermination,
stdout_bytes: u64,
stderr_bytes: u64,
streams_separated: bool,
live_streaming: bool,
exit_code: Option<i32>,
duration_ms: u64,
termination: CommandTermination,
output_bytes: u64,
live_streaming: bool,
},
AgentCliStarted {
node_id: String,
@ -1304,8 +1301,7 @@ impl Event {
exit_code,
duration_ms,
termination,
stdout_bytes,
stderr_bytes,
output_bytes,
..
} => {
debug!(
@ -1313,8 +1309,7 @@ impl Event {
exit_code,
duration_ms,
termination = %termination,
stdout_bytes,
stderr_bytes,
output_bytes,
"Command completed"
);
}

17
lib/crates/fabro-workflow/src/git.rs
View file

@ -508,16 +508,13 @@ 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,
termination: CommandTermination::Exited,
stdout_bytes: 3,
stderr_bytes: 0,
streams_separated: true,
live_streaming: true,
node_id: "work".into(),
output: "hi\n".into(),
exit_code: Some(0),
duration_ms: 10,
termination: CommandTermination::Exited,
output_bytes: 3,
live_streaming: true,
})
.await
.unwrap();

126
lib/crates/fabro-workflow/src/handler/command.rs
View file

@ -49,9 +49,6 @@ impl Handler for CommandHandler {
outcome
.context_updates
.insert(keys::COMMAND_OUTPUT.to_string(), serde_json::json!(""));
outcome
.context_updates
.insert(keys::COMMAND_STDERR.to_string(), serde_json::json!(""));
Ok(outcome)
}
@ -91,6 +88,7 @@ impl Handler for CommandHandler {
} else {
script.to_string()
};
let command = format!("exec 2>&1\n{command}");
let stage_scope = StageScope::for_handler(context, &node.id);
services.run.emitter.emit_scoped(
&Event::CommandStarted {
@ -114,11 +112,11 @@ impl Handler for CommandHandler {
let recorder = CommandLogRecorder::create(run_dir, &stage_id).await?;
let output_callback: CommandOutputCallback = {
let recorder = recorder.clone();
std::sync::Arc::new(move |stream, bytes| {
std::sync::Arc::new(move |_stream, bytes| {
let recorder = recorder.clone();
Box::pin(async move {
recorder
.append(stream, &bytes)
.append(&bytes)
.await
.map_err(|err| fabro_sandbox::Error::message(err.to_string()))
})
@ -150,29 +148,26 @@ impl Handler for CommandHandler {
services.run.emitter.emit_scoped(
&Event::CommandCompleted {
node_id: node.id.clone(),
stdout: finalized.stdout_ref.clone(),
stderr: finalized.stderr_ref.clone(),
exit_code: result.exit_code,
duration_ms: result.duration_ms,
termination: result.termination,
stdout_bytes: finalized.stdout_bytes,
stderr_bytes: finalized.stderr_bytes,
streams_separated: streaming.streams_separated,
live_streaming: streaming.live_streaming,
node_id: node.id.clone(),
output: finalized.output_ref.clone(),
exit_code: result.exit_code,
duration_ms: result.duration_ms,
termination: result.termination,
output_bytes: finalized.output_bytes,
live_streaming: streaming.live_streaming,
},
&stage_scope,
);
if result.termination == CommandTermination::TimedOut {
let mut reason = format!("Script timed out after {timeout_ms}ms: {script}");
append_output_tails(&mut reason, &finalized.stdout_text, &finalized.stderr_text);
append_output_tail(&mut reason, &finalized.output_text);
return Err(Error::handler(reason));
}
if result.termination == CommandTermination::Cancelled {
let mut reason = format!("Script cancelled: {script}");
append_output_tails(&mut reason, &finalized.stdout_text, &finalized.stderr_text);
append_output_tail(&mut reason, &finalized.output_text);
return Err(Error::handler(reason));
}
@ -180,11 +175,7 @@ impl Handler for CommandHandler {
let mut outcome = Outcome::success();
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!(finalized.stdout_ref),
);
outcome.context_updates.insert(
keys::COMMAND_STDERR.to_string(),
serde_json::json!(finalized.stderr_ref),
serde_json::json!(finalized.output_ref),
);
outcome.notes = Some(format!("Script completed: {script}"));
Ok(outcome)
@ -193,31 +184,22 @@ impl Handler for CommandHandler {
"Script failed with exit code: {}",
result.exit_code.unwrap_or(-1)
);
append_output_tails(&mut reason, &finalized.stdout_text, &finalized.stderr_text);
append_output_tail(&mut reason, &finalized.output_text);
let mut outcome = Outcome::fail_classify(reason);
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!(finalized.stdout_ref),
);
outcome.context_updates.insert(
keys::COMMAND_STDERR.to_string(),
serde_json::json!(finalized.stderr_ref),
serde_json::json!(finalized.output_ref),
);
Ok(outcome)
}
}
}
fn append_output_tails(reason: &mut String, stdout: &str, stderr: &str) {
let stdout_tail = tail_bytes(stdout, 4096);
let stderr_tail = tail_bytes(stderr, 4096);
if !stdout_tail.trim().is_empty() {
reason.push_str("\n\n## stdout\n");
reason.push_str(&stdout_tail);
}
if !stderr_tail.trim().is_empty() {
reason.push_str("\n\n## stderr\n");
reason.push_str(&stderr_tail);
fn append_output_tail(reason: &mut String, output: &str) {
let output_tail = tail_bytes(output, 4096);
if !output_tail.trim().is_empty() {
reason.push_str("\n\n## output\n");
reason.push_str(&output_tail);
}
}
@ -240,7 +222,7 @@ mod tests {
use bytes::Bytes;
use fabro_graphviz::graph::AttrValue;
use fabro_store::{Database, RunDatabase, StageId};
use fabro_types::{CommandOutputStream, fixtures};
use fabro_types::fixtures;
use object_store::memory::InMemory;
use tokio::sync::Mutex;
@ -375,10 +357,7 @@ mod tests {
outcome.context_updates.get(keys::COMMAND_OUTPUT),
Some(&serde_json::json!(""))
);
assert_eq!(
outcome.context_updates.get(keys::COMMAND_STDERR),
Some(&serde_json::json!(""))
);
assert!(!outcome.context_updates.contains_key("command.stderr"));
}
#[tokio::test]
@ -435,8 +414,7 @@ mod tests {
.await
.contains("hello")
);
let command_stderr = outcome.context_updates.get(keys::COMMAND_STDERR).unwrap();
assert_eq!(command_text(&services, command_stderr).await, "");
assert!(!outcome.context_updates.contains_key("command.stderr"));
}
#[tokio::test]
@ -508,7 +486,7 @@ mod tests {
let snapshot = run_store.state().await.unwrap();
let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap();
let json = node_state.script_invocation.as_ref().unwrap();
assert_eq!(json["command"], "echo hello");
assert_eq!(json["command"], "exec 2>&1\necho hello");
assert_eq!(json["language"], "shell");
assert_eq!(json["timeout_ms"], serde_json::Value::Null);
}
@ -539,13 +517,13 @@ mod tests {
let snapshot = run_store.state().await.unwrap();
let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap();
let json = node_state.script_invocation.as_ref().unwrap();
assert_eq!(json["command"], "echo hello");
assert_eq!(json["command"], "exec 2>&1\necho hello");
assert_eq!(json["language"], "shell");
assert_eq!(json["timeout_ms"], 5000);
}
#[tokio::test]
async fn writes_stdout_and_stderr_logs() {
async fn writes_output_log() {
let handler = CommandHandler;
let mut node = Node::new("script_node");
node.attrs.insert(
@ -565,18 +543,14 @@ mod tests {
let snapshot = run_store.state().await.unwrap();
let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap();
let stdout = node_state.stdout.as_deref().unwrap();
assert_eq!(command_log_text(&services, stdout).await.trim(), "hello");
let stderr = node_state.stderr.as_deref().unwrap();
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));
let output = node_state.output.as_deref().unwrap();
assert_eq!(command_log_text(&services, output).await.trim(), "hello");
assert_eq!(node_state.output_bytes, Some(6));
assert_eq!(node_state.live_streaming, Some(true));
}
#[tokio::test]
async fn writes_stderr_log_on_failure() {
async fn writes_stderr_to_output_log_on_failure() {
let handler = CommandHandler;
let mut node = Node::new("script_node");
node.attrs.insert(
@ -596,8 +570,8 @@ mod tests {
let snapshot = run_store.state().await.unwrap();
let node_state = snapshot.stage(&StageId::new("script_node", 1)).unwrap();
let stderr = node_state.stderr.as_deref().unwrap();
assert_eq!(command_log_text(&services, stderr).await.trim(), "oops");
let output = node_state.output.as_deref().unwrap();
assert_eq!(command_log_text(&services, output).await.trim(), "oops");
}
#[tokio::test]
@ -824,7 +798,7 @@ mod tests {
}
#[tokio::test]
async fn script_handler_captures_stderr() {
async fn script_handler_merges_stderr_into_output() {
let handler = CommandHandler;
let mut node = Node::new("script_node");
node.attrs.insert(
@ -841,13 +815,13 @@ mod tests {
.await
.unwrap();
assert_eq!(outcome.status, StageOutcome::Succeeded);
let command_stderr = outcome.context_updates.get(keys::COMMAND_STDERR).unwrap();
let command_output = outcome.context_updates.get(keys::COMMAND_OUTPUT).unwrap();
assert!(
command_text(&services, command_stderr)
command_text(&services, command_output)
.await
.contains("err"),
"command.stderr should contain 'err', got: {:?}",
command_stderr
"command.output should contain 'err', got: {:?}",
command_output
);
}
@ -1034,8 +1008,8 @@ mod tests {
);
assert_eq!(
spy.captured_command().as_deref(),
Some("echo hello"),
"sandbox should receive the script as the command"
Some("exec 2>&1\necho hello"),
"sandbox should receive the wrapped script as the command"
);
}
@ -1077,7 +1051,7 @@ mod tests {
assert_eq!(outcome.status, StageOutcome::Succeeded);
let captured = spy.captured_command().unwrap();
assert!(
captured.starts_with("python3 -c ") && captured.contains("print"),
captured.starts_with("exec 2>&1\npython3 -c ") && captured.contains("print"),
"sandbox command should invoke python3 with the script, got: {captured}"
);
}
@ -1238,11 +1212,11 @@ mod tests {
assert!(message.contains("timed out"), "got: {message}");
assert!(
message.contains("partial stdout"),
"timeout error should include stdout tail, got: {message}"
"timeout error should include output tail, got: {message}"
);
assert!(
message.contains("partial stderr"),
"timeout error should include stderr tail, got: {message}"
"timeout error should include merged output tail, got: {message}"
);
}
@ -1271,7 +1245,7 @@ mod tests {
}
#[tokio::test]
async fn script_handler_failure_includes_stdout() {
async fn script_handler_failure_includes_output() {
let handler = CommandHandler;
let mut node = Node::new("script_node");
node.attrs.insert(
@ -1293,11 +1267,11 @@ mod tests {
let reason = outcome.failure_reason().unwrap();
assert!(
reason.contains("build output"),
"failure_reason should contain stdout, got: {reason}"
"failure_reason should contain output, got: {reason}"
);
assert!(
reason.contains("oops"),
"failure_reason should contain stderr, got: {reason}"
"failure_reason should contain merged stderr, got: {reason}"
);
assert!(
reason.contains("exit code: 1"),
@ -1326,12 +1300,8 @@ mod tests {
assert!(err.to_string().contains("Failed to spawn script"));
let stage_id = StageId::new("script_node", 1);
assert!(
!command_log_path(run_dir.path(), &stage_id, CommandOutputStream::Stdout).exists(),
"spawn failure should remove pre-created stdout scratch log"
);
assert!(
!command_log_path(run_dir.path(), &stage_id, CommandOutputStream::Stderr).exists(),
"spawn failure should remove pre-created stderr scratch log"
!command_log_path(run_dir.path(), &stage_id).exists(),
"spawn failure should remove pre-created output scratch log"
);
}
@ -1363,7 +1333,7 @@ mod tests {
command_text(&services, command_output)
.await
.contains("build output"),
"command.output should contain stdout, got: {command_output:?}"
"command.output should contain output, got: {command_output:?}"
);
}
}

125
lib/crates/fabro-workflow/src/handler/llm/preamble.rs
View file

@ -151,7 +151,6 @@ fn tail_lines(text: &str, max_lines: usize, indent: &str) -> String {
fn stage_rendered_keys(node_id: &str, outcome: &Outcome) -> HashSet<String> {
let candidates = [
keys::COMMAND_OUTPUT.to_string(),
keys::COMMAND_STDERR.to_string(),
keys::LAST_STAGE.to_string(),
keys::LAST_RESPONSE.to_string(),
keys::response_key(node_id),
@ -182,25 +181,14 @@ fn render_compact_stage_details(
lines.push(format!(" - Script: `{cmd}`"));
}
}
if let Some(stdout_val) = outcome.context_updates.get(keys::COMMAND_OUTPUT) {
let stdout = format_value(stdout_val);
if stdout.trim().is_empty() {
lines.push(" - Stdout: (empty)".to_string());
if let Some(output_val) = outcome.context_updates.get(keys::COMMAND_OUTPUT) {
let output = format_value(output_val);
if output.trim().is_empty() {
lines.push(" - Output: (empty)".to_string());
} else {
lines.push(" - Stdout:".to_string());
lines.push(" - Output:".to_string());
lines.push(" ```".to_string());
lines.push(tail_lines(stdout.trim(), COMPACT_OUTPUT_MAX_LINES, " "));
lines.push(" ```".to_string());
}
}
if let Some(stderr_val) = outcome.context_updates.get(keys::COMMAND_STDERR) {
let stderr = format_value(stderr_val);
if stderr.trim().is_empty() {
lines.push(" - Stderr: (empty)".to_string());
} else {
lines.push(" - Stderr:".to_string());
lines.push(" ```".to_string());
lines.push(tail_lines(stderr.trim(), COMPACT_OUTPUT_MAX_LINES, " "));
lines.push(tail_lines(output.trim(), COMPACT_OUTPUT_MAX_LINES, " "));
lines.push(" ```".to_string());
}
}
@ -254,37 +242,18 @@ fn render_summary_high_stage_section(
lines.push(format!("- Script: `{cmd}`"));
}
}
if let Some(stdout_val) = outcome.context_updates.get(keys::COMMAND_OUTPUT) {
if let Some(path) = artifact_path(stdout_val) {
lines.push(format!("- Stdout: {}", format_artifact_reference(path)));
if let Some(output_val) = outcome.context_updates.get(keys::COMMAND_OUTPUT) {
if let Some(path) = artifact_path(output_val) {
lines.push(format!("- Output: {}", format_artifact_reference(path)));
} else {
let stdout = format_value(stdout_val);
if stdout.trim().is_empty() {
lines.push("- Stdout: (empty)".to_string());
let output = format_value(output_val);
if output.trim().is_empty() {
lines.push("- Output: (empty)".to_string());
} else {
lines.push("- Stdout:".to_string());
lines.push("- Output:".to_string());
lines.push(" ```".to_string());
lines.push(tail_lines(
stdout.trim(),
SUMMARY_HIGH_OUTPUT_MAX_LINES,
" ",
));
lines.push(" ```".to_string());
}
}
}
if let Some(stderr_val) = outcome.context_updates.get(keys::COMMAND_STDERR) {
if let Some(path) = artifact_path(stderr_val) {
lines.push(format!("- Stderr: {}", format_artifact_reference(path)));
} else {
let stderr = format_value(stderr_val);
if stderr.trim().is_empty() {
lines.push("- Stderr: (empty)".to_string());
} else {
lines.push("- Stderr:".to_string());
lines.push(" ```".to_string());
lines.push(tail_lines(
stderr.trim(),
output.trim(),
SUMMARY_HIGH_OUTPUT_MAX_LINES,
" ",
));
@ -877,7 +846,7 @@ mod tests {
// --- compact handler-specific details ---
#[test]
fn compact_command_stage_shows_command_stdout_stderr() {
fn compact_command_stage_shows_command_output() {
let mut graph = Graph::new("test");
let mut run_tests = Node::new("run_tests");
run_tests.attrs.insert(
@ -898,9 +867,6 @@ mod tests {
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!("10 passed\n"),
);
outcome
.context_updates
.insert(keys::COMMAND_STDERR.to_string(), serde_json::json!(""));
node_outcomes.insert("run_tests".to_string(), outcome);
let preamble = build_preamble(
@ -915,11 +881,11 @@ mod tests {
preamble.contains("Script: `echo '10 passed'`"),
"should show script command"
);
assert!(preamble.contains("Stdout:"), "should show stdout label");
assert!(preamble.contains("10 passed"), "should show stdout content");
assert!(preamble.contains("Output:"), "should show output label");
assert!(preamble.contains("10 passed"), "should show output content");
assert!(
preamble.contains("Stderr: (empty)"),
"should show empty stderr"
!preamble.contains("Stderr:"),
"should not show stderr label"
);
}
@ -1028,16 +994,12 @@ mod tests {
// command.output is set in context (the engine copies context_updates to
// context)
context.set(keys::COMMAND_OUTPUT, serde_json::json!("hi\n"));
context.set(keys::COMMAND_STDERR, serde_json::json!(""));
let completed_nodes = vec!["step".to_string()];
let mut node_outcomes: HashMap<String, Outcome> = HashMap::new();
let mut outcome = Outcome::success();
outcome
.context_updates
.insert(keys::COMMAND_OUTPUT.to_string(), serde_json::json!("hi\n"));
outcome
.context_updates
.insert(keys::COMMAND_STDERR.to_string(), serde_json::json!(""));
node_outcomes.insert("step".to_string(), outcome);
let preamble = build_preamble(
@ -1175,10 +1137,10 @@ mod tests {
preamble.contains("Script: `cargo test`"),
"should show script command"
);
// Low mode should NOT include stdout/stderr
// Low mode should NOT include output
assert!(
!preamble.contains("Stdout:"),
"should not show stdout in low mode"
!preamble.contains("Output:"),
"should not show output in low mode"
);
}
@ -1326,9 +1288,6 @@ mod tests {
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!("All tests passed\n"),
);
outcome
.context_updates
.insert(keys::COMMAND_STDERR.to_string(), serde_json::json!(""));
node_outcomes.insert("run_tests".to_string(), outcome);
let preamble = build_preamble(
@ -1345,7 +1304,7 @@ mod tests {
);
assert!(
preamble.contains("All tests passed"),
"should show stdout via compact renderer"
"should show output via compact renderer"
);
assert!(
!preamble.contains("set command.output"),
@ -1529,11 +1488,7 @@ mod tests {
let mut outcome = Outcome::success();
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!("All tests passed\n"),
);
outcome.context_updates.insert(
keys::COMMAND_STDERR.to_string(),
serde_json::json!("warning: unused var\n"),
serde_json::json!("All tests passed\nwarning: unused var\n"),
);
node_outcomes.insert("run_tests".to_string(), outcome);
@ -1556,11 +1511,11 @@ mod tests {
);
assert!(
preamble.contains("All tests passed"),
"should include stdout"
"should include output"
);
assert!(
preamble.contains("warning: unused var"),
"should include stderr"
"should include merged stderr"
);
}
@ -2136,7 +2091,7 @@ mod tests {
}
#[test]
fn compact_command_stage_truncates_long_stdout() {
fn compact_command_stage_truncates_long_output() {
let mut graph = Graph::new("test");
let mut build = Node::new("build");
build.attrs.insert(
@ -2153,14 +2108,14 @@ mod tests {
let completed_nodes = vec!["build".to_string()];
let mut node_outcomes: HashMap<String, Outcome> = HashMap::new();
let mut outcome = Outcome::success();
// Generate >25 lines of stdout
let long_stdout: String = (1..=30)
// Generate >25 lines of output
let long_output: String = (1..=30)
.map(|i| format!("output line {i}"))
.collect::<Vec<_>>()
.join("\n");
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!(long_stdout),
serde_json::json!(long_output),
);
node_outcomes.insert("build".to_string(), outcome);
@ -2174,7 +2129,7 @@ mod tests {
assert!(
preamble.contains("(5 lines omitted)"),
"should show omission indicator for long stdout, got:\n{preamble}"
"should show omission indicator for long output, got:\n{preamble}"
);
assert!(
preamble.contains("output line 30"),
@ -2187,7 +2142,7 @@ mod tests {
}
#[test]
fn summary_high_command_stage_truncates_long_stdout() {
fn summary_high_command_stage_truncates_long_output() {
let mut graph = Graph::new("test");
let mut build = Node::new("build");
build.attrs.insert(
@ -2204,14 +2159,14 @@ mod tests {
let completed_nodes = vec!["build".to_string()];
let mut node_outcomes: HashMap<String, Outcome> = HashMap::new();
let mut outcome = Outcome::success();
// Generate >50 lines of stdout
let long_stdout: String = (1..=60)
// Generate >50 lines of output
let long_output: String = (1..=60)
.map(|i| format!("output line {i}"))
.collect::<Vec<_>>()
.join("\n");
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!(long_stdout),
serde_json::json!(long_output),
);
node_outcomes.insert("build".to_string(), outcome);
@ -2225,7 +2180,7 @@ mod tests {
assert!(
preamble.contains("(10 lines omitted)"),
"should show omission indicator for long stdout, got:\n{preamble}"
"should show omission indicator for long output, got:\n{preamble}"
);
assert!(
preamble.contains("output line 60"),
@ -2238,7 +2193,7 @@ mod tests {
}
#[test]
fn summary_high_artifact_stdout_not_truncated() {
fn summary_high_artifact_output_not_truncated() {
let mut graph = Graph::new("test");
let mut build = Node::new("build");
build.attrs.insert(
@ -2251,10 +2206,10 @@ mod tests {
let completed_nodes = vec!["build".to_string()];
let mut node_outcomes: HashMap<String, Outcome> = HashMap::new();
let mut outcome = Outcome::success();
// Artifact pointer — should NOT be truncated
// Artifact pointer should not be truncated.
outcome.context_updates.insert(
keys::COMMAND_OUTPUT.to_string(),
serde_json::json!("file:///tmp/artifacts/stdout.txt"),
serde_json::json!("file:///tmp/artifacts/output.txt"),
);
node_outcomes.insert("build".to_string(), outcome);
@ -2271,7 +2226,7 @@ mod tests {
"artifact pointers should not be truncated, got:\n{preamble}"
);
assert!(
preamble.contains("/tmp/artifacts/stdout.txt"),
preamble.contains("/tmp/artifacts/output.txt"),
"should show artifact path"
);
}

8
lib/crates/fabro-workflow/tests/it/integration.rs
View file

@ -8762,8 +8762,8 @@ async fn fidelity_prompt_compact() {
"compact: should show script sub-item for run_tests"
);
assert!(
prompt.contains("Stdout:"),
"compact: should show stdout sub-item for run_tests"
prompt.contains("Output:"),
"compact: should show output sub-item for run_tests"
);
// Original prompt at the end
@ -8833,8 +8833,8 @@ async fn fidelity_prompt_summary_medium() {
"summary:medium: should show script sub-item for run_tests"
);
assert!(
prompt.contains("Stdout:"),
"summary:medium: should show stdout sub-item for run_tests"
prompt.contains("Output:"),
"summary:medium: should show output sub-item for run_tests"
);
// Original prompt at the end

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

@ -42,7 +42,6 @@ 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/command-termination.ts
models/completion-content-part.ts
models/completion-message.ts

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

@ -28,8 +28,6 @@ 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';
@ -185,27 +183,23 @@ export const RunInternalsApiAxiosParamCreator = function (configuration?: Config
};
},
/**
* 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.
* Returns a byte-offset slice of a command stage output log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
* @summary Tail Command Log
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {CommandOutputStream} stream Command output stream to read.
* @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
* @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getRunStageCommandLog: async (id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
getRunStageCommandLog: async (id: string, stageId: string, offset?: number, limit?: number, options: RawAxiosRequestConfig = {}): Promise<RequestArgs> => {
// verify required parameter 'id' is not null or undefined
assertParamExists('getRunStageCommandLog', 'id', id)
// verify required parameter 'stageId' is not null or undefined
assertParamExists('getRunStageCommandLog', 'stageId', stageId)
// verify required parameter 'stream' is not null or undefined
assertParamExists('getRunStageCommandLog', 'stream', stream)
const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/logs/{stream}`
const localVarPath = `/api/v1/runs/{id}/stages/{stageId}/logs/output`
.replace(`{${"id"}}`, encodeURIComponent(String(id)))
.replace(`{${"stageId"}}`, encodeURIComponent(String(stageId)))
.replace(`{${"stream"}}`, encodeURIComponent(String(stream)));
.replace(`{${"stageId"}}`, encodeURIComponent(String(stageId)));
// use dummy base URL string because the URL constructor only accepts absolute URLs.
const localVarUrlObj = new URL(localVarPath, DUMMY_BASE_URL);
let baseOptions;
@ -859,18 +853,17 @@ export const RunInternalsApiFp = function(configuration?: Configuration) {
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.
* Returns a byte-offset slice of a command stage output log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
* @summary Tail Command Log
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {CommandOutputStream} stream Command output stream to read.
* @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
* @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
async getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<CommandLogResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.getRunStageCommandLog(id, stageId, stream, offset, limit, options);
async getRunStageCommandLog(id: string, stageId: string, offset?: number, limit?: number, options?: RawAxiosRequestConfig): Promise<(axios?: AxiosInstance, basePath?: string) => AxiosPromise<CommandLogResponse>> {
const localVarAxiosArgs = await localVarAxiosParamCreator.getRunStageCommandLog(id, stageId, 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);
@ -1090,18 +1083,17 @@ export const RunInternalsApiFactory = function (configuration?: Configuration, b
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.
* Returns a byte-offset slice of a command stage output log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
* @summary Tail Command Log
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {CommandOutputStream} stream Command output stream to read.
* @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
* @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise<CommandLogResponse> {
return localVarFp.getRunStageCommandLog(id, stageId, stream, offset, limit, options).then((request) => request(axios, basePath));
getRunStageCommandLog(id: string, stageId: string, offset?: number, limit?: number, options?: RawAxiosRequestConfig): AxiosPromise<CommandLogResponse> {
return localVarFp.getRunStageCommandLog(id, stageId, offset, limit, options).then((request) => request(axios, basePath));
},
/**
* Returns the internal event-sourced run projection. This is not a stable public contract.
@ -1283,18 +1275,17 @@ export class RunInternalsApi extends BaseAPI {
}
/**
* 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.
* Returns a byte-offset slice of a command stage output log. Bytes are base64-encoded and are not snapped to UTF-8 boundaries.
* @summary Tail Command Log
* @param {string} id Unique run identifier (ULID).
* @param {string} stageId Identifier of a stage within a run\&#39;s workflow graph, serialized as &#x60;node_id@visit&#x60;.
* @param {CommandOutputStream} stream Command output stream to read.
* @param {number} [offset] Byte offset to start reading from. Defaults to &#x60;0&#x60;.
* @param {number} [limit] Maximum bytes to return. Defaults to 65536 and is capped at 1048576.
* @param {*} [options] Override http request option.
* @throws {RequiredError}
*/
public getRunStageCommandLog(id: string, stageId: string, stream: CommandOutputStream, offset?: number, limit?: number, options?: RawAxiosRequestConfig) {
return RunInternalsApiFp(this.configuration).getRunStageCommandLog(id, stageId, stream, offset, limit, options).then((request) => request(this.axios, this.basePath));
public getRunStageCommandLog(id: string, stageId: string, offset?: number, limit?: number, options?: RawAxiosRequestConfig) {
return RunInternalsApiFp(this.configuration).getRunStageCommandLog(id, stageId, offset, limit, options).then((request) => request(this.axios, this.basePath));
}
/**

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

@ -13,15 +13,11 @@
*/
// 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.
*/
@ -31,7 +27,7 @@ export interface CommandLogResponse {
*/
'next_offset': number;
/**
* Total bytes currently available for the stream.
* Total bytes currently available for the output log.
*/
'total_bytes': number;
/**
@ -39,7 +35,7 @@ export interface CommandLogResponse {
*/
'bytes_base64': string;
/**
* Whether the stream is finalized.
* Whether the output log is finalized.
*/
'eof': boolean;
'cas_ref': string | null;
@ -49,5 +45,3 @@ export interface CommandLogResponse {
'live_streaming': boolean;
}

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

@ -1,29 +0,0 @@
/* 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];

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

@ -22,7 +22,6 @@ 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 './command-termination';
export * from './completion-content-part';
export * from './completion-message';

7
lib/packages/fabro-api-client/src/models/stage-projection.ts
View file

@ -48,11 +48,8 @@ export interface StageProjection {
* Per-branch result objects produced by a parallel stage.
*/
'parallel_results'?: Array<object> | null;
'stdout'?: string | null;
'stderr'?: string | null;
'stdout_bytes'?: number | null;
'stderr_bytes'?: number | null;
'streams_separated'?: boolean | null;
'output'?: string | null;
'output_bytes'?: number | null;
'live_streaming'?: boolean | null;
'termination'?: CommandTermination | null;
/**