Spool native cargo build and test streams to persistent world logs

cli/src/cargo.rs

@@ -9,7 +9,7 @@ use std::{
     path::{Path, PathBuf},
     sync::Arc,
 };
-use tokio::{io::AsyncReadExt, process::Command, time::Instant};
+use tokio::{process::Command, time::Instant};
 
 /// Default wall-clock budget after acquiring the build mutex (not queue time).
 pub const DEFAULT_TIMEOUT_SECS: u64 = 900;
@@ -58,9 +58,78 @@ pub struct TestResults {
     pub failed: u64,
     pub output: String,
 }
+/// Raw, unfiltered output shared by the test and release phases of one build.
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct BuildLogs {
+    pub stdout: StreamLog,
+    pub stderr: StreamLog,
+    pub total_bytes: u64,
+}
+#[derive(Clone, Debug, serde::Serialize)]
+pub struct StreamLog {
+    pub path: String,
+    pub bytes: u64,
+}
+/// Persistent descriptors are inherited by cargo, so cancellation cannot lose
+/// bytes already written by cargo or its descendants. No buffered writer is used.
+struct LogFiles {
+    stdout: std::fs::File,
+    stderr: std::fs::File,
+    stdout_path: PathBuf,
+    stderr_path: PathBuf,
+}
+impl LogFiles {
+    fn create(world: &Path) -> Result<Self> {
+        let directory = world
+            .join("logs/cargo")
+            .join(chrono::Utc::now().format("%Y-%m-%d").to_string());
+        std::fs::create_dir_all(&directory).context("create cargo log directory")?;
+        let id = uuid::Uuid::now_v7();
+        let stdout_path = directory.join(format!("{id}.stdout.log"));
+        let stderr_path = directory.join(format!("{id}.stderr.log"));
+        let create = |path: &Path| {
+            std::fs::OpenOptions::new()
+                .create_new(true)
+                .append(true)
+                .open(path)
+                .with_context(|| format!("create {}", path.display()))
+        };
+        Ok(Self {
+            stdout: create(&stdout_path)?,
+            stderr: create(&stderr_path)?,
+            stdout_path,
+            stderr_path,
+        })
+    }
+    fn info(&self, world: &Path) -> Result<BuildLogs> {
+        let stream = |file: &std::fs::File, path: &Path| -> Result<StreamLog> {
+            Ok(StreamLog {
+                path: format!("/{}", path.strip_prefix(world)?.display()),
+                bytes: file.metadata()?.len(),
+            })
+        };
+        let stdout = stream(&self.stdout, &self.stdout_path)?;
+        let stderr = stream(&self.stderr, &self.stderr_path)?;
+        Ok(BuildLogs {
+            total_bytes: stdout.bytes + stderr.bytes,
+            stdout,
+            stderr,
+        })
+    }
+    fn read(&self, path: &Path, offset: u64) -> Result<Vec<u8>> {
+        use std::io::{Read, Seek, SeekFrom};
+        let mut file =
+            std::fs::File::open(path).with_context(|| format!("read {}", path.display()))?;
+        file.seek(SeekFrom::Start(offset))?;
+        let mut bytes = Vec::new();
+        file.read_to_end(&mut bytes)?;
+        Ok(bytes)
+    }
+}
 /// A successful build, usable without a running host.
 #[derive(Debug)]
 pub struct BuildOutput {
+    pub logs: Option<BuildLogs>,
     pub wasm: PathBuf,
     pub warnings: String,
     pub tests: TestResults,
@@ -68,6 +137,7 @@ pub struct BuildOutput {
 /// A failed compiler or test phase, with its complete diagnostics.
 #[derive(Debug)]
 pub struct BuildFailure {
+    pub logs: Option<BuildLogs>,
     pub reason: String,
     pub tests: TestResults,
     pub output: String,
@@ -83,6 +153,7 @@ impl std::error::Error for BuildFailure {}
 impl From<anyhow::Error> for BuildFailure {
     fn from(error: anyhow::Error) -> Self {
         Self {
+            logs: None,
             reason: "build preparation failed".into(),
             output: format!("{error:#}"),
             tests: TestResults::default(),
@@ -179,7 +250,7 @@ impl Cargo {
         host.register_native(ID,Arc::new(move |_context,method,params| {
             let builder=builder.clone();
             Box::pin(async move {
-                if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings"],"properties":{"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
+                if method=="rpc.discover" { return Ok(json!({"openrpc":"1.2.6","info":{"title":"cargo","version":"native"},"methods":[{"name":"build","paramStructure":"by-name","description":"Copies source, runs bounded WASI tests, then builds a release component. Deadlines kill cargo and its descendants; cancellation releases the build mutex.","params":[{"name":"plugin","required":true,"schema":{"type":"string"}},{"name":"timeout_secs","required":false,"description":"Total wall-clock limit in seconds for tests and release build after acquiring the build mutex (queue time excluded). Default 900; minimum 1. Cannot be disabled.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TIMEOUT_SECS}},{"name":"test_timeout_secs","required":false,"description":"Wall-clock limit in seconds for each WASI test instance. Default 120; minimum 1. A timed-out test is named and fails; remaining tests continue.","schema":{"type":"integer","minimum":1,"default":DEFAULT_TEST_TIMEOUT_SECS}}],"result":{"name":"build","schema":{"type":"object","required":["wasm","tests","warnings","logs"],"properties":{"logs":{"type":"object","required":["stdout","stderr","total_bytes"],"properties":{"stdout":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"stderr":{"type":"object","required":["path","bytes"],"properties":{"path":{"type":"string"},"bytes":{"type":"integer","minimum":0}}},"total_bytes":{"type":"integer","minimum":0}}},"wasm":{"type":"string"},"warnings":{"type":"string"},"tests":{"type":"object","required":["passed","failed","output"],"properties":{"passed":{"type":"integer","minimum":0},"failed":{"type":"integer","minimum":0},"output":{"type":"string"}}}}}},"errors":[{"code":-32004,"message":"Build failed"}]}]})); }
                 if method!="build" { return Err(crate::error(-32601,"cargo method not found")); }
                 let plugin=params["plugin"].as_str().ok_or_else(||crate::error(-32602,"plugin must be a string"))?;
                 let options = BuildOptions {
@@ -187,8 +258,8 @@ impl Cargo {
                     test_timeout_secs: seconds(&params, "test_timeout_secs", DEFAULT_TEST_TIMEOUT_SECS)?,
                 };
                 match builder.build_with_options(&builder.world.plugin_dir(plugin).join("source"), options).await {
-                    Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings})),
-                    Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step}})),
+                    Ok(output)=>Ok(json!({"wasm":format!("/{}",output.wasm.strip_prefix(&builder.world.root).unwrap().display()),"tests":output.tests,"warnings":output.warnings,"logs":output.logs})),
+                    Err(failure)=>Err(json!({"code":-32004,"message":failure.reason,"data":{"reason":failure.reason,"tests":failure.tests,"output":failure.output,"elapsed_secs":failure.elapsed_secs,"step":failure.step,"logs":failure.logs}})),
                 }
             })
         }));
@@ -203,6 +274,26 @@ impl Cargo {
         &self,
         source: &Path,
         options: BuildOptions,
+    ) -> std::result::Result<BuildOutput, BuildFailure> {
+        let logs = LogFiles::create(&self.world.root)?;
+        let result = self.build_logged(source, options, &logs).await;
+        let info = logs.info(&self.world.root)?;
+        match result {
+            Ok(mut output) => {
+                output.logs = Some(info);
+                Ok(output)
+            }
+            Err(mut failure) => {
+                failure.logs = Some(info);
+                Err(failure)
+            }
+        }
+    }
+    async fn build_logged(
+        &self,
+        source: &Path,
+        options: BuildOptions,
+        logs: &LogFiles,
     ) -> std::result::Result<BuildOutput, BuildFailure> {
         if options.timeout_secs == 0 || options.test_timeout_secs == 0 {
             return Err(anyhow::anyhow!("timeouts must be at least 1 second").into());
@@ -285,7 +376,15 @@ impl Cargo {
             "target.wasm32-wasip2.runner={}",
             json!([self.runner, "test-runner"])
         ));
-        let output = run_command(test, started, deadline, "tests", TestResults::default()).await?;
+        let output = run_command(
+            test,
+            started,
+            deadline,
+            "tests",
+            TestResults::default(),
+            logs,
+        )
+        .await?;
         let stdout = String::from_utf8_lossy(&output.stdout);
         let stderr = String::from_utf8_lossy(&output.stderr);
         let tests = TestResults {
@@ -299,6 +398,7 @@ impl Cargo {
         };
         if !output.status.success() {
             return Err(BuildFailure {
+                logs: None,
                 reason: "WASI tests failed".into(),
                 output: tests.output.clone(),
                 tests,
@@ -318,7 +418,15 @@ impl Cargo {
             ])
             .arg(&target)
             .current_dir(&work);
-        let output = run_command(build, started, deadline, "release build", tests.clone()).await?;
+        let output = run_command(
+            build,
+            started,
+            deadline,
+            "release build",
+            tests.clone(),
+            logs,
+        )
+        .await?;
         let result = async {
             let stdout = String::from_utf8_lossy(&output.stdout);
             let stderr = String::from_utf8_lossy(&output.stderr);
@@ -352,6 +460,7 @@ impl Cargo {
             })
             .await??;
             Ok(BuildOutput {
+                logs: None,
                 wasm: destination,
                 warnings: strip_progress(&diagnostics),
                 tests: tests.clone(),
@@ -359,6 +468,7 @@ impl Cargo {
         }
         .await;
         result.map_err(|error: anyhow::Error| BuildFailure {
+            logs: None,
             reason: "release build failed".into(),
             output: format!("{error:#}"),
             tests,
@@ -384,28 +494,23 @@ async fn run_command(
     deadline: Instant,
     step: &'static str,
     mut tests: TestResults,
+    logs: &LogFiles,
 ) -> std::result::Result<std::process::Output, BuildFailure> {
     command
         .process_group(0)
         .kill_on_drop(true)
         .stdin(std::process::Stdio::null())
-        .stdout(std::process::Stdio::piped())
-        .stderr(std::process::Stdio::piped());
+        // Child writes go straight to persistent files: no pipe buffers or reader
+        // tasks can be dropped on cancellation. Parsing happens only afterwards.
+        .stdout(logs.stdout.try_clone().context("clone stdout log")?)
+        .stderr(logs.stderr.try_clone().context("clone stderr log")?);
+    let out_start = logs.stdout.metadata().context("stat stdout log")?.len();
+    let err_start = logs.stderr.metadata().context("stat stderr log")?.len();
     let mut child = command.spawn().context("start cargo")?;
     let group = ProcessGroup(child.id().unwrap());
-    let mut stdout = child.stdout.take().unwrap();
-    let mut stderr = child.stderr.take().unwrap();
-    let mut out = Vec::new();
-    let mut err = Vec::new();
+
     let (status, timed_out) = {
-        let collect = async {
-            let (status, _, _) = tokio::try_join!(
-                child.wait(),
-                stdout.read_to_end(&mut out),
-                stderr.read_to_end(&mut err)
-            )?;
-            Ok::<_, std::io::Error>(status)
-        };
+        let collect = child.wait();
         tokio::pin!(collect);
         tokio::select! {
             biased;
@@ -419,6 +524,10 @@ async fn run_command(
         }
     };
     let status = status.map_err(anyhow::Error::from)?;
+    // Also stop any descendants that kept the streams open after cargo exited.
+    drop(group);
+    let out = logs.read(&logs.stdout_path, out_start)?;
+    let err = logs.read(&logs.stderr_path, err_start)?;
     if timed_out {
         let output = format!(
             "{}\n{}",
@@ -441,6 +550,7 @@ async fn run_command(
             tests.output = output.clone();
         }
         return Err(BuildFailure {
+            logs: None,
             reason: "timeout".into(),
             tests,
             output,
@@ -599,6 +709,103 @@ fn strip_progress(output: &str) -> String {
 mod timeout_tests {
     use super::*;
 
+    /// Streams retain exact bytes, including non-UTF8, progress and JSON; phases append.
+    #[tokio::test]
+    async fn raw_streams_append_before_parsing() {
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
+        for _ in 0..2 {
+            let mut command = Command::new("sh");
+            command.args(["-c", "printf '\\377raw\\r'; printf '   Compiling progress\\n{\"reason\":\"compiler-artifact\"}\\n' >&2; exit 3"]);
+            let now = Instant::now();
+            let output = tokio::time::timeout(
+                Duration::from_secs(5),
+                run_command(
+                    command,
+                    now,
+                    now + Duration::from_secs(3),
+                    "tests",
+                    TestResults::default(),
+                    &logs,
+                ),
+            )
+            .await
+            .unwrap()
+            .unwrap();
+            assert!(!output.status.success());
+            assert_eq!(output.stdout, b"\xffraw\r");
+            assert!(String::from_utf8_lossy(&output.stderr).contains("Compiling progress"));
+        }
+        assert_eq!(
+            std::fs::read(&logs.stdout_path).unwrap(),
+            b"\xffraw\r\xffraw\r"
+        );
+        let info = logs.info(directory.path()).unwrap();
+        assert_eq!(info.stdout.bytes, 10);
+        assert_eq!(info.total_bytes, info.stdout.bytes + info.stderr.bytes);
+        assert!(info.stdout.path.starts_with("/logs/cargo/"));
+        assert!(!directory.path().join("state/cargo").exists());
+    }
+
+    /// Cancellation leaves kernel-written data intact, even without a newline.
+    #[tokio::test]
+    async fn cancelled_command_keeps_logs() {
+        use notify::Watcher;
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let mut watcher =
+            notify::recommended_watcher(move |event: notify::Result<notify::Event>| {
+                if let Ok(event) = event {
+                    if event
+                        .paths
+                        .iter()
+                        .any(|path| path.file_name().is_some_and(|name| name == "ready"))
+                    {
+                        let _ = tx.send(());
+                    }
+                }
+            })
+            .unwrap();
+        watcher
+            .watch(directory.path(), notify::RecursiveMode::NonRecursive)
+            .unwrap();
+        let mut command = Command::new("sh");
+        command.args(["-c", "printf 'partial\\377'; printf 'stderr\\r' >&2; touch \"$1/ready\"; exec sleep 1000", "sh"]).arg(directory.path());
+        let now = Instant::now();
+        let task = tokio::spawn(async move {
+            run_command(
+                command,
+                now,
+                now + Duration::from_secs(60),
+                "tests",
+                TestResults::default(),
+                &logs,
+            )
+            .await
+        });
+        tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        task.abort();
+        assert!(task.await.unwrap_err().is_cancelled());
+        let paths: Vec<_> = std::fs::read_dir(directory.path().join("logs/cargo"))
+            .unwrap()
+            .flat_map(|day| std::fs::read_dir(day.unwrap().path()).unwrap())
+            .map(|file| file.unwrap().path())
+            .collect();
+        assert_eq!(paths.len(), 2);
+        for path in paths {
+            let expected: &[u8] = if path.to_string_lossy().ends_with(".stdout.log") {
+                b"partial\xff"
+            } else {
+                b"stderr\r"
+            };
+            assert_eq!(std::fs::read(path).unwrap(), expected);
+        }
+    }
+
     /// Deadline errors retain the shared start time, phase, test totals, and output tail.
     #[tokio::test]
     async fn timeout_diagnostics() {
@@ -607,6 +814,8 @@ mod timeout_tests {
             "-c",
             "i=0; while [ $i -lt 100 ]; do echo diagnostic-$i; i=$((i+1)); done; exec sleep 1000",
         ]);
+        let directory = tempfile::tempdir().unwrap();
+        let logs = LogFiles::create(directory.path()).unwrap();
         let now = Instant::now();
         let failure = run_command(
             command,
@@ -618,6 +827,7 @@ mod timeout_tests {
                 failed: 0,
                 output: "successful tests".into(),
             },
+            &logs,
         )
         .await
         .unwrap_err();
@@ -629,5 +839,8 @@ mod timeout_tests {
         assert_eq!(failure.output.lines().count(), 50);
         assert!(failure.output.contains("diagnostic-99"));
         assert!(!failure.output.contains("diagnostic-0\n"));
+        let raw = std::fs::read_to_string(&logs.stdout_path).unwrap();
+        assert_eq!(raw.lines().count(), 100);
+        assert!(raw.starts_with("diagnostic-0\n"));
     }
 }