test(cli): read command logs via endpoint

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

@@ -17,15 +17,15 @@ use fabro_store::EventEnvelope;
 use fabro_test::{
     assert_reqwest_status, expect_reqwest_json, fabro_json_snapshot, fabro_snapshot, test_context,
 };
-use fabro_types::{EventBody, FailureReason, RunEvent, StageId};
+use fabro_types::{CommandOutputStream, EventBody, FailureReason, RunEvent, StageId};
 use hkdf::Hkdf;
 use httpmock::MockServer;
 use jsonwebtoken::{Algorithm, EncodingKey, Header};
 use sha2::Sha256;
 
 use super::support::{
-    find_run_dir, local_dev_token, output_stderr, run_events, run_state, server_endpoint,
-    server_target, wait_for_event_names, wait_for_status, write_gated_workflow,
+    command_log_text, find_run_dir, local_dev_token, output_stderr, run_events, run_state,
+    server_endpoint, server_target, wait_for_event_names, wait_for_status, write_gated_workflow,
 };
 use crate::support::unique_run_id;
 
@@ -549,20 +549,16 @@ methods = ["dev-token"]
     wait_for_status(&run_dir, &["succeeded"]);
 
     let state = run_state(&run_dir);
+    let probe_stage_id = StageId::new("probe", 1);
     let _probe = state
-        .node(&StageId::new("probe", 1))
+        .node(&probe_stage_id)
         .expect("probe node state should exist");
-    let stdout = state
-        .checkpoint
-        .as_ref()
-        .and_then(|checkpoint| checkpoint.context_values.get("command.output"))
-        .and_then(serde_json::Value::as_str)
-        .expect("probe command output should exist");
+    let stdout = command_log_text(&run_dir, &probe_stage_id, CommandOutputStream::Stdout);
     assert!(
         stdout.contains("probe-ran"),
         "probe stage should have executed, got stdout:\n{stdout}"
     );
-    assert_no_worker_env_leak("probe stdout", stdout);
+    assert_no_worker_env_leak("probe stdout", &stdout);
     assert_no_worker_env_leak(
         "run state",
         &serde_json::to_string(&state).expect("run state should serialize"),

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

@@ -14,12 +14,14 @@ use std::process::Output;
 use std::sync::atomic::{AtomicU64, Ordering};
 use std::time::{Duration, Instant};
 
+use base64::Engine as _;
+use base64::engine::general_purpose::STANDARD as BASE64_STANDARD;
 use fabro_config::bind::Bind;
 use fabro_config::daemon::ServerDaemon;
 use fabro_config::{Storage, envfile};
 use fabro_store::EventEnvelope;
 use fabro_test::{TestContext, expect_reqwest_status};
-use fabro_types::RunId;
+use fabro_types::{CommandOutputStream, RunId, StageId};
 use httpmock::{Mock, MockServer};
 use serde_json::Value;
 use shlex::try_quote;
@@ -39,6 +41,11 @@ struct RunSummaryRecord {
     labels: std::collections::HashMap<String, String>,
 }
 
+#[derive(Debug, serde::Deserialize)]
+struct CommandLogResponseRecord {
+    bytes_base64: String,
+}
+
 pub(crate) struct RunSetup {
     pub(crate) run_id:  String,
     pub(crate) run_dir: PathBuf,
@@ -697,6 +704,22 @@ 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 {
+    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"),
+    ));
+    let bytes = BASE64_STANDARD
+        .decode(&response.bytes_base64)
+        .expect("command log bytes should decode");
+    String::from_utf8(bytes).expect("command log should be UTF-8")
+}
+
 #[expect(
     clippy::disallowed_methods,
     reason = "This sync integration helper polls stored events without requiring a Tokio runtime."