feat(sandbox): write agent output to the container log (#4005)

Since #2726 the canonical main process's stdout and stderr are captured
in pipes that feed only the in-memory replay buffer used by sandbox
connect. Agent output therefore never reaches the container's own stdout
and stderr, so it is missing from kubectl logs, docker logs, and podman
logs and from anything that collects container logs. Before #2726 the
entrypoint inherited the container's descriptors and its output appeared
there.

Copy the main process's output to the launcher's stdout and stderr in
addition to the replay buffer, restoring the earlier behavior:

- Output is copied byte for byte to the matching stream from a
  forwarder thread per stream, after it is published to the replay
  buffer. When the container runtime falls behind on a stream, that
  stream's reader waits instead of dropping output, so backpressure
  reaches the agent as it did with inherited descriptors, while the
  other stream and attachments keep receiving output.
- Before the main process's exit is published, the output readers
  finish and queued output is drained to the container log, so an
  agent's final lines are not lost at shutdown. A 30 second deadline
  covers both; when it expires, readers waiting on the container log
  are released and drain the pipes into the replay buffer only, so a
  stalled container log cannot block exit reporting.
- PTY-mode processes are not copied. The terminal stream carries escape
  sequences and echoed input, and terminal commands never reached the
  container log before #2726.
- Exec, SSH, and SFTP sessions are not copied.

Launcher log lines keep their existing format and remain in the
container's stderr. They are written as whole lines, and a newline is
inserted first when the agent left stderr mid-line, so launcher and
agent lines do not merge.

The Docker and VM drivers appended the tail of the workload's output to
failure messages: Docker the workload container's log, and the VM driver
the guest console, which carries the launcher's stdout and stderr. Those
messages land in the sandbox's Ready condition and in platform events
that the gateway republishes to the sandbox event stream. With agent
output in that log, those messages would carry arbitrary agent output,
including anything sensitive the agent prints, into gateway status and
events. The supervisor starts its health endpoint only after the agent
starts, so every Docker failure path could include agent output, and the
VM driver reports one whenever the VM or host supervisor exits. Forward
only the supervisor's log tail, matching the Podman driver, which reads
the workload log solely to match fixed launcher markers and never
forwards raw workload output. The workload's output remains available
through docker logs and the VM's rootfs-console.log.

Document where main process output appears in the logging docs and the
cluster debugging skill.

Closes #3928

Signed-off-by: Kris Hicks <khicks@nvidia.com>

Author: krishicks

crates/openshell-driver-docker/src/lib.rs

@@ -5288,12 +5288,6 @@ async fn spawn_docker_control_process(
                 if !log_tail.is_empty() {
                     write!(message, "; log tail: {log_tail}").ok();
                 }
-                let sandbox_log_tail =
-                    docker_container_log_tail(&monitored_docker, &failure_context.container_id)
-                        .await;
-                if !sandbox_log_tail.is_empty() {
-                    write!(message, "; sandbox log tail: {sandbox_log_tail}").ok();
-                }
                 let _ = monitored_docker.remove_container(
                     &monitored_supervisor_id,
                     Some(RemoveContainerOptionsBuilder::default().force(true).build()),
@@ -5344,11 +5338,9 @@ async fn wait_for_docker_supervisor_ready(
                 Status::internal(format!("inspect Docker sandbox container: {error}"))
             })?;
         if sandbox.state.unwrap_or_default().running == Some(false) {
-            let sandbox_log_tail = docker_container_log_tail(docker, sandbox_id).await;
-            return Err(Status::unavailable(format!(
-                "Docker sandbox exited before supervisor became ready{}",
-                format_named_log_tail("sandbox log tail", &sandbox_log_tail)
-            )));
+            return Err(Status::unavailable(
+                "Docker sandbox exited before supervisor became ready",
+            ));
         }
         let inspected = docker
             .inspect_container(supervisor_id, None)
@@ -5361,12 +5353,10 @@ async fn wait_for_docker_supervisor_ready(
             Some(HealthStatusEnum::HEALTHY) => return Ok(()),
             _ if state.running == Some(false) => {
                 let log_tail = docker_container_log_tail(docker, supervisor_id).await;
-                let sandbox_log_tail = docker_container_log_tail(docker, sandbox_id).await;
-                warn!(sandbox_id, supervisor_id, supervisor_logs = %log_tail, sandbox_logs = %sandbox_log_tail, "Docker supervisor exited before becoming ready");
+                warn!(sandbox_id, supervisor_id, supervisor_logs = %log_tail, "Docker supervisor exited before becoming ready");
                 return Err(Status::unavailable(format!(
-                    "Docker supervisor exited before becoming ready{}{}",
-                    format_log_tail(&log_tail),
-                    format_named_log_tail("sandbox log tail", &sandbox_log_tail)
+                    "Docker supervisor exited before becoming ready{}",
+                    format_log_tail(&log_tail)
                 )));
             }
             _ => tokio::time::sleep(Duration::from_millis(100)).await,
@@ -5375,13 +5365,9 @@ async fn wait_for_docker_supervisor_ready(
 }
 
 fn format_log_tail(log_tail: &str) -> String {
-    format_named_log_tail("log tail", log_tail)
-}
-
-fn format_named_log_tail(label: &str, log_tail: &str) -> String {
-    // gRPC status messages travel in HTTP/2 headers. Two 16 KiB container
-    // tails exceed the client's 16 KiB header budget and hide the real error
-    // behind PROTOCOL_ERROR. Allow for up to 3x percent-encoding expansion.
+    // gRPC status messages travel in HTTP/2 headers. A full 16 KiB container
+    // tail can exceed the client's 16 KiB header budget and hide the real
+    // error behind PROTOCOL_ERROR. Allow for up to 3x percent-encoding expansion.
     const MAX_STATUS_LOG_TAIL_BYTES: usize = 1024;
     if log_tail.is_empty() {
         String::new()
@@ -5390,9 +5376,9 @@ fn format_named_log_tail(label: &str, log_tail: &str) -> String {
         while !log_tail.is_char_boundary(start) {
             start += 1;
         }
-        format!("; {label}: [truncated] {}", &log_tail[start..])
+        format!("; log tail: [truncated] {}", &log_tail[start..])
     } else {
-        format!("; {label}: {log_tail}")
+        format!("; log tail: {log_tail}")
     }
 }
 

crates/openshell-driver-docker/src/tests.rs

@@ -27,15 +27,14 @@ use tempfile::TempDir;
 #[test]
 fn startup_error_log_tails_fit_grpc_header_budget() {
     // Multibyte text exercises both the UTF-8 cut and worst-case gRPC message
-    // percent encoding. Preserve the final diagnostic from each container.
+    // percent encoding. Preserve the supervisor's final diagnostic.
     let logs = format!("{}\nstartup timed out", "🦀".repeat(8192));
     let message = format!(
-        "Docker supervisor exited before becoming ready{}{}",
+        "Docker supervisor exited before becoming ready{}",
         format_log_tail(&logs),
-        format_named_log_tail("sandbox log tail", &logs),
     );
-    assert_eq!(message.matches("[truncated]").count(), 2);
-    assert_eq!(message.matches("startup timed out").count(), 2);
+    assert_eq!(message.matches("[truncated]").count(), 1);
+    assert_eq!(message.matches("startup timed out").count(), 1);
     let response = Status::unavailable(message).into_http::<()>();
     let header_bytes: usize = response
         .headers()

crates/openshell-driver-vm/src/driver.rs

@@ -4171,16 +4171,6 @@ impl VmDriver {
                     || format!("{component} process exited"),
                     |code| format!("{component} process exited with status {code}"),
                 );
-                if component == "VM"
-                    && let Some(state_dir) = state_dir.as_deref()
-                    && let Some(console) = read_vm_console_tail(
-                        &state_dir.join("rootfs-console.log"),
-                        VM_CONSOLE_DIAGNOSTIC_BYTES,
-                    )
-                {
-                    write!(message, "; guest console tail:\n{console}")
-                        .expect("writing to String cannot fail");
-                }
                 if component == "host supervisor"
                     && let Some(state_dir) = state_dir.as_deref()
                     && let Some(stderr) = read_vm_console_tail(
@@ -4191,16 +4181,6 @@ impl VmDriver {
                     write!(message, "; supervisor stderr tail:\n{stderr}")
                         .expect("writing to String cannot fail");
                 }
-                if component == "host supervisor"
-                    && let Some(state_dir) = state_dir.as_deref()
-                    && let Some(console) = read_vm_console_tail(
-                        &state_dir.join("rootfs-console.log"),
-                        VM_CONSOLE_DIAGNOSTIC_BYTES,
-                    )
-                {
-                    write!(message, "; guest console tail:\n{console}")
-                        .expect("writing to String cannot fail");
-                }
                 if let Some(snapshot) = self
                     .set_snapshot_condition(
                         &sandbox_id,
@@ -7517,6 +7497,63 @@ mod tests {
         }));
     }
 
+    #[tokio::test]
+    async fn process_exit_status_omits_guest_console_output() {
+        let running_child = || {
+            Command::new("sleep")
+                .arg("30")
+                .stdin(Stdio::null())
+                .stdout(Stdio::null())
+                .stderr(Stdio::null())
+                .kill_on_drop(true)
+                .spawn()
+                .unwrap()
+        };
+        for (component, child) in [
+            ("VM", spawn_exited_child()),
+            ("host supervisor", running_child()),
+        ] {
+            let temp = tempfile::tempdir().unwrap();
+            std::fs::write(temp.path().join("rootfs-console.log"), "agent output\n").unwrap();
+            std::fs::write(temp.path().join("supervisor.err.log"), "supervisor error\n").unwrap();
+            let driver = test_driver_with_extensions(LifecycleExtensionRegistry::new());
+            let mut events = driver.events.subscribe();
+            insert_test_record(&driver, "sb-exit", temp.path().to_path_buf(), child).await;
+
+            driver.monitor_sandbox("sb-exit".to_string()).await;
+
+            let condition = driver.registry.lock().await["sb-exit"]
+                .snapshot
+                .status
+                .as_ref()
+                .and_then(|status| {
+                    status
+                        .conditions
+                        .iter()
+                        .find(|condition| condition.reason == "ProcessExited")
+                        .cloned()
+                })
+                .expect("ProcessExited condition");
+            let mut event_message = None;
+            while let Ok(event) = events.try_recv() {
+                if let Some(watch_sandboxes_event::Payload::PlatformEvent(platform)) = event.payload
+                    && let Some(event) = platform.event
+                    && event.reason == "ProcessExited"
+                {
+                    event_message = Some(event.message);
+                }
+            }
+            let event_message = event_message.expect("ProcessExited platform event");
+            for message in [&condition.message, &event_message] {
+                assert!(message.starts_with(&format!("{component} process exited")));
+                assert!(!message.contains("agent output"), "{component}: {message}");
+            }
+            if component == "host supervisor" {
+                assert!(condition.message.contains("supervisor error"));
+            }
+        }
+    }
+
     #[tokio::test]
     async fn background_provisioning_does_not_extend_the_rpc_span_lifetime() {
         let traced = TestTracing::new();