diff --git a/fleet/arcbox-fleet-agent/src/attach.rs b/fleet/arcbox-fleet-agent/src/attach.rs
index 0bf65f217..07b64e325 100644
--- a/fleet/arcbox-fleet-agent/src/attach.rs
+++ b/fleet/arcbox-fleet-agent/src/attach.rs
@@ -39,6 +39,11 @@ const VERDICT_RESEND_INTERVAL: Duration = Duration::from_secs(10);
 const INITIAL_BACKOFF: Duration = Duration::from_secs(1);
 const MAX_BACKOFF: Duration = Duration::from_secs(60);
 const OUTBOUND_CAPACITY: usize = 64;
+/// Depth of the runner-output queue. Deep enough to ride out a brief stall in
+/// the gRPC stream, shallow enough that a job outrunning the network is shed
+/// promptly instead of accumulating in agent memory. Separate from
+/// `OUTBOUND_CAPACITY` so output can never occupy a slot a heartbeat needs.
+const LOG_CAPACITY: usize = 256;
 const MACHINE_TOKEN_HEADER: &str = "x-arcbox-machine-token";
 /// How long to wait for runners to be torn down and reaped on shutdown before
 /// giving up. Killing a process group or container is near-instant, so this is a
@@ -124,7 +129,22 @@ fn telemetry_to_control(t: &HostTelemetry) -> control_proto::HostTelemetry {
     }
 }
 
-/// Build the [`RunnerSupervisor`] and its egress queue. The verdict-resend
+/// The [`RunnerSupervisor`] and the two outbound queues it feeds.
+///
+/// A struct rather than a tuple because the two receivers have the same type
+/// but opposite delivery contracts — swapping them would make runner output
+/// durable and verdicts lossy, and compile fine.
+pub struct SupervisorHandles {
+    pub supervisor: RunnerSupervisor,
+    /// Runner lifecycle events: delivered with an awaited send and held in the
+    /// `pending` slot across a reconnect, so a verdict is never lost.
+    pub egress_rx: mpsc::Receiver<AttachRequest>,
+    /// Runner output: delivered with `try_send` and dropped when the stream
+    /// cannot take it. See [`crate::joblog`].
+    pub log_rx: mpsc::Receiver<AttachRequest>,
+}
+
+/// Build the [`RunnerSupervisor`] and its outbound queues. The verdict-resend
 /// loop is started by [`run`] instead, so it shares the attachment's shutdown
 /// token and is reaped with it.
 ///
@@ -136,32 +156,43 @@ pub fn spawn_supervisor(
     config: &AgentConfig,
     backends: Arc<Backends>,
     state: AgentState,
-) -> (RunnerSupervisor, mpsc::Receiver<AttachRequest>) {
+) -> SupervisorHandles {
     let (egress_tx, egress_rx) = mpsc::channel::<AttachRequest>(OUTBOUND_CAPACITY);
-    let supervisor =
-        RunnerSupervisor::new(egress_tx, config.runner_script.clone(), backends, state);
-    (supervisor, egress_rx)
+    let (log_tx, log_rx) = mpsc::channel::<AttachRequest>(LOG_CAPACITY);
+    let supervisor = RunnerSupervisor::new(
+        egress_tx,
+        log_tx,
+        config.runner_script.clone(),
+        backends,
+        state,
+    );
+    SupervisorHandles {
+        supervisor,
+        egress_rx,
+        log_rx,
+    }
 }
 
 /// Connect and serve the attach stream, reconnecting on any failure until
 /// `shutdown` fires, then stop runners cleanly.
 ///
-/// `supervisor` and `egress_rx` come from [`spawn_supervisor`] and are reused
-/// across reconnects, so in-flight jobs survive a dropped connection and
-/// their verdicts reach the next live stream. On shutdown the loop exits and
-/// hands off to [`RunnerSupervisor::shutdown`], which tears down any
-/// in-flight runners.
+/// `supervisor`, `egress_rx` and `log_rx` come from [`spawn_supervisor`] and
+/// are reused across reconnects, so in-flight jobs survive a dropped
+/// connection and their verdicts reach the next live stream. On shutdown the
+/// loop exits and hands off to [`RunnerSupervisor::shutdown`], which tears
+/// down any in-flight runners.
 #[allow(
     clippy::too_many_arguments,
     reason = "the reconnect loop genuinely needs all of: endpoint config, credential, the \
-              persistent supervisor, the cross-reconnect egress queue, the backend \
-              registry, the shutdown token, and the observable state handle"
+              persistent supervisor, the cross-reconnect egress and runner-output queues, \
+              the backend registry, the shutdown token, and the observable state handle"
 )]
 pub async fn run(
     config: AgentConfig,
     credential: Credential,
     supervisor: RunnerSupervisor,
     mut egress_rx: mpsc::Receiver<AttachRequest>,
+    mut log_rx: mpsc::Receiver<AttachRequest>,
     backends: Arc<Backends>,
     shutdown: CancellationToken,
     state: AgentState,
@@ -193,6 +224,7 @@ pub async fn run(
             &credential,
             &supervisor,
             &mut egress_rx,
+            &mut log_rx,
             &mut pending,
             &mut backoff,
             &backends,
@@ -317,21 +349,22 @@ pub async fn run(
 ///
 /// Outbound traffic is multiplexed onto a fresh per-connection request channel:
 /// connection-scoped heartbeats are sent directly, while runner lifecycle
-/// events are forwarded from the shared egress queue. Inbound orders are routed
-/// to the persistent `supervisor`.
+/// events are forwarded from the shared egress queue and runner output from the
+/// shared log queue. Inbound orders are routed to the persistent `supervisor`.
 #[allow(
     clippy::too_many_arguments,
     reason = "one connection's lifecycle genuinely needs all of: endpoint config, \
-              credential, the persistent supervisor, the cross-reconnect egress queue \
-              and its pending slot, the mutable backoff, advertised capabilities, the \
-              shutdown token, the observable state handle, and the cross-reconnect \
-              update-drain flag"
+              credential, the persistent supervisor, the cross-reconnect egress and \
+              runner-output queues and the egress pending slot, the mutable backoff, \
+              advertised capabilities, the shutdown token, the observable state handle, \
+              and the cross-reconnect update-drain flag"
 )]
 async fn connect_and_serve(
     config: &AgentConfig,
     credential: &Credential,
     supervisor: &RunnerSupervisor,
     egress_rx: &mut mpsc::Receiver<AttachRequest>,
+    log_rx: &mut mpsc::Receiver<AttachRequest>,
     pending: &mut Option<AttachRequest>,
     backoff: &mut Duration,
     backends: &Backends,
@@ -496,6 +529,24 @@ async fn connect_and_serve(
                 // happen while the agent runs; treat it as a clean shutdown.
                 None => break Ok(StreamEnd::Closed),
             },
+            chunk = log_rx.recv() => match chunk {
+                // Runner output, forwarded on the opposite contract to the arm
+                // above: `try_send`, never an awaited send, and never parked in
+                // `pending`. Awaiting here would let a chatty job hold the
+                // request channel long enough to delay the heartbeat, which the
+                // gateway reads as a dead machine; `pending` is the single slot
+                // that makes a verdict survive a reconnect and must not be spent
+                // on a log chunk. A chunk that does not fit is dropped, and the
+                // consumer sees the gap in `seq`.
+                Some(msg) => {
+                    if req_tx.try_send(msg).is_err() {
+                        tracing::trace!("runner log chunk dropped; request channel full");
+                    }
+                }
+                // Same reasoning as the egress arm: the supervisor holds the
+                // only log sender.
+                None => break Ok(StreamEnd::Closed),
+            },
             message = inbound.message() => match message {
                 Ok(Some(message)) => {
                     tracing::debug!(msg = ?message.msg, "inbound attach message");
@@ -722,14 +773,14 @@ mod tests {
             machine_token: "flt_revoked".to_owned(),
         };
         let backends = Backends::fixed(Vec::new(), state.clone());
-        let (supervisor, egress_rx) =
-            spawn_supervisor(&config, Arc::clone(&backends), state.clone());
+        let handles = spawn_supervisor(&config, Arc::clone(&backends), state.clone());
         let shutdown = CancellationToken::new();
         let run = tokio::spawn(run(
             config,
             credential,
-            supervisor,
-            egress_rx,
+            handles.supervisor,
+            handles.egress_rx,
+            handles.log_rx,
             backends,
             shutdown.clone(),
             state.clone(),
@@ -851,6 +902,7 @@ mod tests {
         let backends = Backends::fixed(Vec::new(), AgentState::new(&seed()));
         let mut backends_rx = backends.subscribe();
         let (_egress_tx, mut egress_rx) = mpsc::channel::<AttachRequest>(1);
+        let (_log_tx, mut log_rx) = mpsc::channel::<AttachRequest>(1);
         let mut pending = None;
         let mut backoff = INITIAL_BACKOFF;
         let shutdown = CancellationToken::new();
@@ -872,6 +924,7 @@ mod tests {
                 &credential,
                 &supervisor,
                 &mut egress_rx,
+                &mut log_rx,
                 &mut pending,
                 &mut backoff,
                 &backends,
@@ -985,14 +1038,14 @@ mod tests {
             ..config()
         };
         let backends = Backends::new(false, None, None, None, state.clone());
-        let (supervisor, egress_rx) =
-            spawn_supervisor(&config, Arc::clone(&backends), state.clone());
+        let handles = spawn_supervisor(&config, Arc::clone(&backends), state.clone());
         let shutdown = CancellationToken::new();
         let run_task = tokio::spawn(run(
             config,
             credential(),
-            supervisor,
-            egress_rx,
+            handles.supervisor,
+            handles.egress_rx,
+            handles.log_rx,
             Arc::clone(&backends),
             shutdown.clone(),
             state,
@@ -1089,7 +1142,7 @@ mod tests {
     fn supervisor() -> RunnerSupervisor {
         let (events, _rx) = mpsc::channel(1);
         let state = AgentState::new(&seed());
-        RunnerSupervisor::new(
+        RunnerSupervisor::without_logs(
             events,
             None,
             Backends::fixed(Vec::new(), state.clone()),
@@ -1097,6 +1150,168 @@ mod tests {
         )
     }
 
+    /// A gateway that accepts every `Attach` and records every subsequent
+    /// inbound message, holding the stream open until the client leaves.
+    struct MessageRecordingGateway {
+        seen: Arc<tokio::sync::Mutex<Vec<attach_request::Msg>>>,
+    }
+
+    #[tonic::async_trait]
+    impl arcbox_fleet_proto::v1::fleet_gateway_service_server::FleetGatewayService
+        for MessageRecordingGateway
+    {
+        async fn enroll(
+            &self,
+            _: Request<arcbox_fleet_proto::v1::EnrollRequest>,
+        ) -> Result<tonic::Response<arcbox_fleet_proto::v1::EnrollResponse>, tonic::Status>
+        {
+            Err(tonic::Status::unimplemented("not used by this test"))
+        }
+
+        type AttachStream = tokio_stream::wrappers::ReceiverStream<
+            Result<arcbox_fleet_proto::v1::AttachResponse, tonic::Status>,
+        >;
+
+        async fn attach(
+            &self,
+            request: Request<tonic::Streaming<AttachRequest>>,
+        ) -> Result<tonic::Response<Self::AttachStream>, tonic::Status> {
+            let mut inbound = request.into_inner();
+            let first = inbound
+                .message()
+                .await
+                .map_err(|e| tonic::Status::internal(e.to_string()))?
+                .ok_or_else(|| tonic::Status::internal("closed before Attach"))?;
+            if !matches!(first.msg, Some(attach_request::Msg::Attach(_))) {
+                return Err(tonic::Status::internal("first message was not Attach"));
+            }
+
+            let (tx, rx) = mpsc::channel(4);
+            let _ = tx
+                .send(Ok(arcbox_fleet_proto::v1::AttachResponse {
+                    msg: Some(attach_response::Msg::AttachAccepted(
+                        arcbox_fleet_proto::v1::AttachAccepted {},
+                    )),
+                }))
+                .await;
+            let seen = Arc::clone(&self.seen);
+            tokio::spawn(async move {
+                let _hold = tx;
+                while let Ok(Some(message)) = inbound.message().await {
+                    if let Some(msg) = message.msg {
+                        seen.lock().await.push(msg);
+                    }
+                }
+            });
+            Ok(tonic::Response::new(
+                tokio_stream::wrappers::ReceiverStream::new(rx),
+            ))
+        }
+
+        async fn unenroll(
+            &self,
+            _: Request<arcbox_fleet_proto::v1::UnenrollRequest>,
+        ) -> Result<tonic::Response<arcbox_fleet_proto::v1::UnenrollResponse>, tonic::Status>
+        {
+            Err(tonic::Status::unimplemented("not used by this test"))
+        }
+    }
+
+    /// Runner output must reach the gateway on the same stream as verdicts,
+    /// while staying out of the machinery that makes verdicts durable.
+    ///
+    /// The `pending` assertion is the one that matters: it is the single slot
+    /// that carries an undelivered verdict across a reconnect, and the log arm
+    /// sits directly beside the egress arm that fills it. A log chunk parked
+    /// there would evict a verdict — the exact bug a copy-paste of the
+    /// neighbouring arm would introduce, and one nothing else would catch.
+    #[tokio::test]
+    async fn log_chunks_reach_the_stream_without_consuming_the_verdict_pending_slot() {
+        use arcbox_fleet_proto::v1::fleet_gateway_service_server::FleetGatewayServiceServer;
+
+        let seen = Arc::new(tokio::sync::Mutex::new(Vec::new()));
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let gateway = format!("http://{}", listener.local_addr().unwrap());
+        tokio::spawn(
+            tonic::transport::Server::builder()
+                .add_service(FleetGatewayServiceServer::new(MessageRecordingGateway {
+                    seen: Arc::clone(&seen),
+                }))
+                .serve_with_incoming(tokio_stream::wrappers::TcpListenerStream::new(listener)),
+        );
+
+        let state = AgentState::new(&PersistedSettings {
+            gateway: gateway.clone(),
+            ..seed()
+        });
+        let supervisor = supervisor();
+        let backends = Backends::fixed(Vec::new(), AgentState::new(&seed()));
+        let mut backends_rx = backends.subscribe();
+        let (_egress_tx, mut egress_rx) = mpsc::channel::<AttachRequest>(1);
+
+        // Queue output before the stream exists, the way a job that started on
+        // a previous connection would.
+        let (log_tx, mut log_rx) = mpsc::channel::<AttachRequest>(LOG_CAPACITY);
+        let sink = crate::joblog::JobLogSink::new("rjob_a", log_tx);
+        sink.write(b"hello from the runner\n");
+
+        let mut pending = None;
+        let mut backoff = INITIAL_BACKOFF;
+        let shutdown = CancellationToken::new();
+        let config = AgentConfig {
+            gateway,
+            ..config()
+        };
+        let credential = credential();
+        let mut drained_for_update = false;
+
+        // Leave the stream once the chunk has had time to land; the loop only
+        // ends on shutdown, so drive it with one.
+        let stopper = shutdown.clone();
+        tokio::spawn(async move {
+            tokio::time::sleep(Duration::from_millis(500)).await;
+            stopper.cancel();
+        });
+
+        tokio::time::timeout(
+            Duration::from_secs(5),
+            connect_and_serve(
+                &config,
+                &credential,
+                &supervisor,
+                &mut egress_rx,
+                &mut log_rx,
+                &mut pending,
+                &mut backoff,
+                &backends,
+                &mut backends_rx,
+                &shutdown,
+                &state,
+                &mut drained_for_update,
+            ),
+        )
+        .await
+        .expect("the serve loop must exit on shutdown, not hang")
+        .expect("clean exit");
+
+        let seen = seen.lock().await;
+        let logs: Vec<_> = seen
+            .iter()
+            .filter_map(|msg| match msg {
+                attach_request::Msg::RunnerLog(chunk) => Some(chunk),
+                _ => None,
+            })
+            .collect();
+        assert_eq!(logs.len(), 1, "the queued chunk must reach the gateway");
+        assert_eq!(logs[0].job_id, "rjob_a");
+        assert_eq!(logs[0].data, b"hello from the runner\n");
+
+        assert!(
+            pending.is_none(),
+            "a log chunk must never occupy the slot that makes a verdict survive a reconnect"
+        );
+    }
+
     /// The connect + Attach-RPC handshake has no cancellation awareness of
     /// its own (see this fn's own doc in the non-test code above) — without
     /// racing it against `shutdown`, a reconnect attempt could complete the
@@ -1110,6 +1325,7 @@ mod tests {
         let backends = Backends::fixed(Vec::new(), AgentState::new(&seed()));
         let mut backends_rx = backends.subscribe();
         let (_egress_tx, mut egress_rx) = mpsc::channel::<AttachRequest>(1);
+        let (_log_tx, mut log_rx) = mpsc::channel::<AttachRequest>(1);
         let mut pending = None;
         let mut backoff = INITIAL_BACKOFF;
         let shutdown = CancellationToken::new();
@@ -1126,6 +1342,7 @@ mod tests {
                 &credential,
                 &supervisor,
                 &mut egress_rx,
+                &mut log_rx,
                 &mut pending,
                 &mut backoff,
                 &backends,
diff --git a/fleet/arcbox-fleet-agent/src/control/mod.rs b/fleet/arcbox-fleet-agent/src/control/mod.rs
index 7c5df4ab3..92bdba8d2 100644
--- a/fleet/arcbox-fleet-agent/src/control/mod.rs
+++ b/fleet/arcbox-fleet-agent/src/control/mod.rs
@@ -320,14 +320,16 @@ impl AgentSupervisor {
         // process-lifetime state, so clear any stale draining flag rather than
         // letting a re-enroll inherit it.
         self.agent_state.set_draining(false);
-        let (supervisor, egress_rx) =
+        let handles =
             attach::spawn_supervisor(&self.config, self.backends(), self.agent_state.clone());
+        let supervisor = handles.supervisor;
         let shutdown = self.process_shutdown.child_token();
         let task = tokio::spawn(attach::run(
             self.config.clone(),
             credential.clone(),
             supervisor.clone(),
-            egress_rx,
+            handles.egress_rx,
+            handles.log_rx,
             self.backends(),
             shutdown.clone(),
             self.agent_state.clone(),
@@ -960,7 +962,7 @@ mod tests {
             Ok(())
         });
         let (events, _events_rx) = tokio::sync::mpsc::channel(1);
-        let runner = crate::runner::RunnerSupervisor::new(
+        let runner = crate::runner::RunnerSupervisor::without_logs(
             events,
             None,
             Backends::new(false, None, None, None, agent_state.clone()),
@@ -1106,7 +1108,7 @@ mod tests {
             Ok(())
         });
         let (events, _events_rx) = tokio::sync::mpsc::channel(1);
-        let runner = crate::runner::RunnerSupervisor::new(
+        let runner = crate::runner::RunnerSupervisor::without_logs(
             events,
             None,
             Backends::new(false, None, None, None, agent_state.clone()),
diff --git a/fleet/arcbox-fleet-agent/src/docker.rs b/fleet/arcbox-fleet-agent/src/docker.rs
index 86c833a06..582de9e59 100644
--- a/fleet/arcbox-fleet-agent/src/docker.rs
+++ b/fleet/arcbox-fleet-agent/src/docker.rs
@@ -8,12 +8,14 @@ use anyhow::{Context, Result};
 use bollard::Docker;
 use bollard::models::ContainerCreateBody;
 use bollard::query_parameters::{
-    CreateContainerOptions, CreateImageOptions, RemoveContainerOptions, WaitContainerOptions,
+    CreateContainerOptions, CreateImageOptions, LogsOptions, RemoveContainerOptions,
+    WaitContainerOptions,
 };
 use tokio_stream::StreamExt;
 use tracing::{debug, info, warn};
 
 use crate::host;
+use crate::joblog::JobLogSink;
 
 /// Everything the Docker runner needs to execute one job.
 pub struct RunSpec<'a> {
@@ -40,6 +42,38 @@ pub struct RunningContainer {
 }
 
 impl RunningContainer {
+    /// Stream this container's combined output into `sink` for the rest of its
+    /// life, in a detached task.
+    ///
+    /// Detaching is safe because the stream is bounded by the container: it
+    /// ends when the container exits or is removed, which every exit path from
+    /// [`wait`](Self::wait) leads to. Docker holds the output itself, so unlike
+    /// the pipe-backed backends nothing blocks if this task falls behind.
+    pub fn follow_logs(&self, sink: JobLogSink) {
+        let client = self.client.clone();
+        let id = self.id.clone();
+        tokio::spawn(async move {
+            let options = LogsOptions {
+                follow: true,
+                stdout: true,
+                stderr: true,
+                ..Default::default()
+            };
+            let mut stream = client.logs(&id, Some(options));
+            while let Some(frame) = stream.next().await {
+                match frame {
+                    Ok(output) => sink.write(&output.into_bytes()),
+                    // Teardown closes the stream mid-read; the job's own exit
+                    // path reports the outcome, so this is not an error here.
+                    Err(e) => {
+                        debug!(container = %id, error = %e, "container log stream ended");
+                        break;
+                    }
+                }
+            }
+        });
+    }
+
     /// Block until the container exits and return its exit code. No cleanup —
     /// follow with [`remove`](Self::remove). The agent reports no outcome
     /// upstream (the GitHub webhook is authoritative); the code is for logging.
diff --git a/fleet/arcbox-fleet-agent/src/interop.rs b/fleet/arcbox-fleet-agent/src/interop.rs
index faa9ec2c7..7591659cc 100644
--- a/fleet/arcbox-fleet-agent/src/interop.rs
+++ b/fleet/arcbox-fleet-agent/src/interop.rs
@@ -30,6 +30,8 @@ use anyhow::{Context, Result, bail, ensure};
 use tokio::io::{AsyncBufReadExt, BufReader};
 use tracing::{debug, warn};
 
+use crate::joblog::JobLogSink;
+
 /// Fixed Windows locations of the interop tools, translated through
 /// `wslpath` at startup so a non-default automount root still resolves.
 /// Windows PowerShell 5.1 ships with every Windows 10/11 install (unlike
@@ -162,7 +164,10 @@ impl InteropRunner {
     /// side and carries the PID that can cancel it. Any failure past the
     /// spawn reaps the relay before returning, so an error never leaks a
     /// process.
-    pub async fn spawn(&self, encoded_jit_config: &str) -> Result<InteropJob> {
+    ///
+    /// Everything the wrapper prints after the handshake is the runner's own
+    /// output and goes to `sink`.
+    pub async fn spawn(&self, encoded_jit_config: &str, sink: JobLogSink) -> Result<InteropJob> {
         validate_jit_config(&self.script, encoded_jit_config)?;
 
         let mut child = tokio::process::Command::new(&self.powershell)
@@ -170,10 +175,18 @@ impl InteropRunner {
             .arg(wrapper_command(&self.script, encoded_jit_config))
             .stdin(Stdio::null())
             .stdout(Stdio::piped())
+            .stderr(Stdio::piped())
             .kill_on_drop(true)
             .spawn()
             .context("spawning the interop wrapper")?;
 
+        // Piped, so it must be drained for the same reason stdout is. The
+        // handshake only reads stdout, so this starts immediately rather than
+        // after it.
+        if let Some(stderr) = child.stderr.take() {
+            sink.pipe(stderr);
+        }
+
         let stdout = child
             .stdout
             .take()
@@ -206,11 +219,21 @@ impl InteropRunner {
         // Keep the pipe drained for the rest of the job — the runner's
         // output flows through the wrapper's console — so a chatty runner
         // can never fill the pipe and wedge itself. Ends at EOF when the
-        // relay exits; detaching the handle is deliberate.
+        // relay exits; detaching the handle is deliberate. The reader stays
+        // line-oriented rather than switching to `sink.pipe`: the handshake
+        // already wrapped stdout in a `BufReader`, which may hold buffered
+        // bytes that a raw read of the underlying pipe would skip.
         tokio::spawn(async move {
             loop {
                 match lines.next_line().await {
-                    Ok(Some(line)) => debug!(line, "windows runner output"),
+                    // Restore the terminator `next_line` stripped, and ship the
+                    // line whole: two writes would split every line across two
+                    // chunks, doubling the message count for no benefit.
+                    Ok(Some(line)) => {
+                        let mut framed = line.into_bytes();
+                        framed.push(b'\n');
+                        sink.write(&framed);
+                    }
                     Ok(None) => break,
                     Err(e) => {
                         debug!(error = %e, "windows runner output stream failed");
@@ -435,7 +458,7 @@ mod tests {
     /// spawns long-existing Windows binaries, never freshly written ones.
     async fn spawn_retrying(runner: &InteropRunner, jit: &str) -> Result<InteropJob> {
         for _ in 0..100 {
-            match runner.spawn(jit).await {
+            match runner.spawn(jit, JobLogSink::discarding()).await {
                 Err(e) if format!("{e:#}").contains("Text file busy") => {
                     tokio::time::sleep(Duration::from_millis(20)).await;
                 }
@@ -589,7 +612,10 @@ mod tests {
         std::fs::write(&staged, "@echo off\r\nping -n 60 127.0.0.1 >NUL\r\n").unwrap();
 
         let runner = InteropRunner::new(&script).await.expect("probe");
-        let mut job = runner.spawn("dGVzdA==").await.expect("spawn");
+        let mut job = runner
+            .spawn("dGVzdA==", JobLogSink::discarding())
+            .await
+            .expect("spawn");
         assert!(job.windows_pid() > 0);
 
         let start = tokio::time::Instant::now();
diff --git a/fleet/arcbox-fleet-agent/src/joblog.rs b/fleet/arcbox-fleet-agent/src/joblog.rs
new file mode 100644
index 000000000..e866bdf3a
--- /dev/null
+++ b/fleet/arcbox-fleet-agent/src/joblog.rs
@@ -0,0 +1,297 @@
+//! Per-job capture of runner output, shipped to the gateway as
+//! [`RunnerLogChunk`]s for live viewing while the job runs.
+//!
+//! This is the one lossy path in the agent, and deliberately so. The attach
+//! stream is a control plane: its heartbeat writes to a 64-slot egress channel
+//! with a blocking send, and the gateway flips a machine `Offline` after 60s
+//! without one — which stops placement offering it work. Runner output can
+//! arrive orders of magnitude faster than that channel drains, so a sink that
+//! queued or awaited would trade a machine's liveness for its logs. Instead
+//! every write is admitted through a per-job rate budget and a non-blocking
+//! `try_send`, and anything that does not fit is dropped and counted.
+//!
+//! Backpressure therefore never propagates back to the runner: a chatty job
+//! loses log lines, not throughput, and the consumer sees the loss as a `seq`
+//! gap plus a `dropped_bytes` count on the next chunk.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Instant;
+
+use arcbox_fleet_proto::v1::{AttachRequest, RunnerLogChunk, attach_request};
+use tokio::io::{AsyncRead, AsyncReadExt};
+use tokio::sync::mpsc;
+
+/// Largest payload a single chunk carries. Sized so one chunk is a comfortable
+/// gRPC message and a burst of them still fits the egress channel.
+const MAX_CHUNK_BYTES: usize = 32 * 1024;
+
+/// Sustained per-job output rate. Well above what a normal CI job produces
+/// (a verbose build is a few KiB/s), low enough that a runaway job printing in
+/// a tight loop cannot saturate the machine's control stream.
+const BUDGET_BYTES_PER_SEC: u64 = 256 * 1024;
+
+/// Burst allowance on top of the sustained rate, so the bulk output of a
+/// single build step is not shed just for arriving all at once.
+const BUDGET_BURST_BYTES: u64 = 1024 * 1024;
+
+/// Read buffer for [`JobLogSink::pipe`]. Matches the chunk size so a full read
+/// maps to exactly one chunk.
+const PIPE_BUFFER_BYTES: usize = MAX_CHUNK_BYTES;
+
+/// Captures one job's output and ships it as [`RunnerLogChunk`]s.
+///
+/// Cheap to clone — a job's stdout and stderr readers share one sink, so their
+/// interleaving on the wire matches the order the agent observed, and `seq`
+/// stays monotonic across both.
+#[derive(Clone)]
+pub struct JobLogSink {
+    inner: Arc<Inner>,
+}
+
+struct Inner {
+    /// Prefixed runner job id (`rjob_...`) stamped on every chunk.
+    job_id: String,
+    /// Log-only egress channel, separate from the supervisor's event channel so
+    /// output can never occupy a slot a verdict or heartbeat needs.
+    tx: mpsc::Sender<AttachRequest>,
+    /// Next chunk's sequence number. Consumed even by a chunk that fails to
+    /// send, so a drop shows up downstream as a gap.
+    seq: AtomicU64,
+    /// Bytes shed since the last chunk that went out, reported on the next one.
+    dropped: AtomicU64,
+    budget: Mutex<TokenBucket>,
+}
+
+impl JobLogSink {
+    /// Open a sink for `job_id` feeding `tx`.
+    pub fn new(job_id: &str, tx: mpsc::Sender<AttachRequest>) -> Self {
+        Self {
+            inner: Arc::new(Inner {
+                job_id: job_id.to_owned(),
+                tx,
+                seq: AtomicU64::new(0),
+                dropped: AtomicU64::new(0),
+                budget: Mutex::new(TokenBucket::new()),
+            }),
+        }
+    }
+
+    /// A sink with no receiver, so every write is shed. For tests that drive a
+    /// runner path without asserting on its output.
+    #[cfg(test)]
+    pub fn discarding() -> Self {
+        let (tx, _) = mpsc::channel(1);
+        Self::new("rjob_test", tx)
+    }
+
+    /// Ship `data` as one or more chunks. Never blocks and never fails: bytes
+    /// that exceed the rate budget, or that meet a full channel, are counted
+    /// into the next chunk's `dropped_bytes` instead.
+    pub fn write(&self, data: &[u8]) {
+        for slice in data.chunks(MAX_CHUNK_BYTES) {
+            self.write_chunk(slice);
+        }
+    }
+
+    fn write_chunk(&self, slice: &[u8]) {
+        let len = slice.len() as u64;
+        if !self
+            .inner
+            .budget
+            .lock()
+            .expect("budget mutex poisoned")
+            .take(len)
+        {
+            self.inner.dropped.fetch_add(len, Ordering::Relaxed);
+            return;
+        }
+        // Read rather than swap: the count must survive a failed send, and
+        // only the chunk that actually goes out may clear it.
+        let reported = self.inner.dropped.load(Ordering::Relaxed);
+        let chunk = RunnerLogChunk {
+            job_id: self.inner.job_id.clone(),
+            seq: self.inner.seq.fetch_add(1, Ordering::Relaxed),
+            data: slice.to_vec(),
+            dropped_bytes: reported,
+        };
+        let message = AttachRequest {
+            msg: Some(attach_request::Msg::RunnerLog(chunk)),
+        };
+        match self.inner.tx.try_send(message) {
+            // Subtract rather than store 0: a concurrent writer may have added
+            // to the count between the load above and here.
+            Ok(()) => {
+                self.inner.dropped.fetch_sub(reported, Ordering::Relaxed);
+            }
+            Err(_) => {
+                self.inner.dropped.fetch_add(len, Ordering::Relaxed);
+            }
+        }
+    }
+
+    /// Drain `reader` into this sink until EOF, in a detached task.
+    ///
+    /// Detaching is deliberate and matches the lifetime of what is being read:
+    /// the reader is a pipe from the runner process, so it reaches EOF when
+    /// that process exits or is torn down, and the task ends with it. A read
+    /// error ends the task too — a broken pipe is teardown, not a fault.
+    pub fn pipe<R>(&self, mut reader: R)
+    where
+        R: AsyncRead + Unpin + Send + 'static,
+    {
+        let sink = self.clone();
+        tokio::spawn(async move {
+            let mut buffer = vec![0u8; PIPE_BUFFER_BYTES];
+            loop {
+                match reader.read(&mut buffer).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(read) => sink.write(&buffer[..read]),
+                }
+            }
+        });
+    }
+}
+
+/// Leaky-bucket rate limiter over bytes, refilled continuously from elapsed
+/// wall time. Holding the lock never spans an await — every caller is the
+/// synchronous [`JobLogSink::write_chunk`].
+struct TokenBucket {
+    tokens: u64,
+    last: Instant,
+}
+
+impl TokenBucket {
+    fn new() -> Self {
+        Self {
+            // Start full so a short job's entire output is admitted.
+            tokens: BUDGET_BURST_BYTES,
+            last: Instant::now(),
+        }
+    }
+
+    /// Admit `want` bytes, or refuse them outright. Never partially admits:
+    /// a chunk is shed whole so the reported byte count stays exact.
+    fn take(&mut self, want: u64) -> bool {
+        self.refill(Instant::now());
+        if self.tokens >= want {
+            self.tokens -= want;
+            true
+        } else {
+            false
+        }
+    }
+
+    fn refill(&mut self, now: Instant) {
+        let elapsed = now.saturating_duration_since(self.last).as_secs_f64();
+        self.last = now;
+        let refilled = (elapsed * BUDGET_BYTES_PER_SEC as f64) as u64;
+        self.tokens = self.tokens.saturating_add(refilled).min(BUDGET_BURST_BYTES);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// Pull every chunk currently queued, as `(seq, data, dropped_bytes)`.
+    fn drain(rx: &mut mpsc::Receiver<AttachRequest>) -> Vec<(u64, Vec<u8>, u64)> {
+        let mut chunks = Vec::new();
+        while let Ok(message) = rx.try_recv() {
+            match message.msg {
+                Some(attach_request::Msg::RunnerLog(chunk)) => {
+                    chunks.push((chunk.seq, chunk.data, chunk.dropped_bytes));
+                }
+                other => panic!("expected RunnerLog, got {other:?}"),
+            }
+        }
+        chunks
+    }
+
+    #[tokio::test]
+    async fn splits_oversized_writes_and_numbers_them_in_order() {
+        let (tx, mut rx) = mpsc::channel(16);
+        let sink = JobLogSink::new("rjob_a", tx);
+
+        sink.write(&vec![b'x'; MAX_CHUNK_BYTES + 1]);
+
+        let chunks = drain(&mut rx);
+        assert_eq!(chunks.len(), 2);
+        assert_eq!(chunks[0].0, 0);
+        assert_eq!(chunks[0].1.len(), MAX_CHUNK_BYTES);
+        assert_eq!(chunks[1].0, 1);
+        assert_eq!(chunks[1].1.len(), 1);
+        assert!(chunks.iter().all(|(_, _, dropped)| *dropped == 0));
+    }
+
+    /// A full channel must not block or error — the bytes are shed and
+    /// reported on the next chunk that gets through, and the burnt sequence
+    /// number leaves the gap that tells the consumer output was lost.
+    #[tokio::test]
+    async fn full_channel_sheds_and_reports_on_the_next_chunk() {
+        let (tx, mut rx) = mpsc::channel(1);
+        let sink = JobLogSink::new("rjob_a", tx);
+
+        sink.write(b"first");
+        sink.write(b"lost");
+
+        // Free the single slot, then send again: the drop is now reportable.
+        let queued = drain(&mut rx);
+        assert_eq!(queued.len(), 1);
+        assert_eq!(queued[0], (0, b"first".to_vec(), 0));
+
+        sink.write(b"third");
+        let chunks = drain(&mut rx);
+        assert_eq!(chunks.len(), 1);
+        let (seq, data, dropped) = &chunks[0];
+        assert_eq!(*seq, 2, "the shed chunk burnt seq 1, leaving a gap");
+        assert_eq!(data, b"third");
+        assert_eq!(*dropped, 4, "the shed \"lost\" bytes are reported here");
+    }
+
+    /// Sustained output past the burst allowance is shed rather than queued,
+    /// so a runaway job cannot grow the agent's memory or stall the stream.
+    #[tokio::test]
+    async fn output_beyond_the_burst_allowance_is_shed() {
+        let (tx, mut rx) = mpsc::channel(1024);
+        let sink = JobLogSink::new("rjob_a", tx);
+
+        let over_budget = (BUDGET_BURST_BYTES as usize) + (4 * MAX_CHUNK_BYTES);
+        sink.write(&vec![b'x'; over_budget]);
+
+        let chunks = drain(&mut rx);
+        let admitted: usize = chunks.iter().map(|(_, data, _)| data.len()).sum();
+        assert!(
+            admitted <= BUDGET_BURST_BYTES as usize + MAX_CHUNK_BYTES,
+            "admitted {admitted} bytes, above the burst allowance"
+        );
+        assert!(admitted > 0, "the burst allowance must admit something");
+
+        // The shed bytes are pending, reported once the budget refills.
+        assert_eq!(
+            sink.inner.dropped.load(Ordering::Relaxed) as usize,
+            over_budget - admitted
+        );
+    }
+
+    #[tokio::test]
+    async fn pipe_forwards_reader_contents_until_eof() {
+        let (tx, mut rx) = mpsc::channel(16);
+        let sink = JobLogSink::new("rjob_a", tx.clone());
+
+        sink.pipe(std::io::Cursor::new(b"runner output".to_vec()));
+        // Drop the local sender so the channel closes once the piped task
+        // finishes, making the drain below deterministic.
+        drop(tx);
+        drop(sink);
+
+        let mut seen = Vec::new();
+        while let Some(message) = rx.recv().await {
+            match message.msg {
+                Some(attach_request::Msg::RunnerLog(chunk)) => seen.extend(chunk.data),
+                other => panic!("expected RunnerLog, got {other:?}"),
+            }
+        }
+        assert_eq!(seen, b"runner output");
+    }
+}
diff --git a/fleet/arcbox-fleet-agent/src/main.rs b/fleet/arcbox-fleet-agent/src/main.rs
index a795fca0a..a06052f5d 100644
--- a/fleet/arcbox-fleet-agent/src/main.rs
+++ b/fleet/arcbox-fleet-agent/src/main.rs
@@ -36,6 +36,7 @@ mod enroll;
 mod fsutil;
 mod host;
 mod interop;
+mod joblog;
 #[cfg(test)]
 mod mock_daemon;
 mod runner;
@@ -326,13 +327,14 @@ async fn run(command: Command, config: AgentConfig) -> Result<()> {
                 shutdown.clone(),
             );
 
-            let (supervisor, egress_rx) =
+            let handles =
                 attach::spawn_supervisor(&config, Arc::clone(&backends), agent_state.clone());
             attach::run(
                 config,
                 credential,
-                supervisor,
-                egress_rx,
+                handles.supervisor,
+                handles.egress_rx,
+                handles.log_rx,
                 backends,
                 shutdown,
                 agent_state,
diff --git a/fleet/arcbox-fleet-agent/src/runner.rs b/fleet/arcbox-fleet-agent/src/runner.rs
index 95224b2d5..7857edd94 100644
--- a/fleet/arcbox-fleet-agent/src/runner.rs
+++ b/fleet/arcbox-fleet-agent/src/runner.rs
@@ -16,6 +16,7 @@
 //! so a busy host simply rejects and the platform re-offers elsewhere.
 
 use std::path::{Path, PathBuf};
+use std::process::Stdio;
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -33,6 +34,7 @@ use tracing::{info, warn};
 use crate::backends::Backends;
 use crate::docker::RunSpec;
 use crate::host;
+use crate::joblog::JobLogSink;
 use crate::state::AgentState;
 
 /// Watchdog on a VM job's runtime: GitHub concludes jobs at 6 h, so a
@@ -152,6 +154,9 @@ pub struct RunnerSupervisor {
 struct Inner {
     /// Outbound channel to the gateway (offer verdicts).
     events: mpsc::Sender<AttachRequest>,
+    /// Outbound channel for runner output. Separate from `events` so a chatty
+    /// job can never delay a verdict or a heartbeat; see [`crate::joblog`].
+    logs: mpsc::Sender<AttachRequest>,
     /// Verdicts sent but not yet acknowledged by the gateway, keyed by
     /// `offer_token`. Resent until an `OfferVerdictAck` arrives so a gateway
     /// crash-after-read cannot lose the verdict. There is no local expiry: the
@@ -209,14 +214,15 @@ impl Drop for ReleaseGuard {
 }
 
 impl RunnerSupervisor {
-    /// Create a supervisor that emits verdicts on `events`. Routing comes
-    /// live from `backends` — the same registry the advertised capability
-    /// set derives from, so routing and advertisement agree by construction.
-    /// `load_ceiling`/`mem_floor_mib` are not parameters: `admit()` reads
-    /// them live from `state`, which is the single source of truth settings
-    /// write through.
+    /// Create a supervisor that emits verdicts on `events` and runner output
+    /// on `logs`. Routing comes live from `backends` — the same registry the
+    /// advertised capability set derives from, so routing and advertisement
+    /// agree by construction. `load_ceiling`/`mem_floor_mib` are not
+    /// parameters: `admit()` reads them live from `state`, which is the single
+    /// source of truth settings write through.
     pub fn new(
         events: mpsc::Sender<AttachRequest>,
+        logs: mpsc::Sender<AttachRequest>,
         runner_script: Option<PathBuf>,
         backends: Arc<Backends>,
         state: AgentState,
@@ -224,6 +230,7 @@ impl RunnerSupervisor {
         Self {
             inner: Arc::new(Inner {
                 events,
+                logs,
                 outstanding: DashMap::new(),
                 in_flight: DashMap::new(),
                 runner_script,
@@ -509,6 +516,26 @@ impl RunnerSupervisor {
         }
     }
 
+    /// Test constructor with no runner-output consumer, so every chunk is
+    /// shed. For the many tests that exercise admission, verdicts and job
+    /// lifecycle without asserting on output.
+    #[cfg(test)]
+    pub fn without_logs(
+        events: mpsc::Sender<AttachRequest>,
+        runner_script: Option<PathBuf>,
+        backends: Arc<Backends>,
+        state: AgentState,
+    ) -> Self {
+        Self::new(events, mpsc::channel(1).0, runner_script, backends, state)
+    }
+
+    /// Open the runner-output sink for a job. Created before the runner starts
+    /// so no backend has to retrofit capture onto an already-running process;
+    /// a job that never starts simply writes nothing to it.
+    fn log_sink(&self, job_id: &str) -> JobLogSink {
+        JobLogSink::new(job_id, self.inner.logs.clone())
+    }
+
     /// Start the runner for an accepted offer, then accept once it is actually
     /// running. A failure to start rejects instead, so the platform re-offers.
     async fn run_job(
@@ -613,8 +640,9 @@ impl RunnerSupervisor {
             Canceled,
             Watchdog,
         }
+        let sink = self.log_sink(job_id);
         let exited = {
-            let wait = std::pin::pin!(running.wait());
+            let wait = std::pin::pin!(running.wait(&sink));
             tokio::select! {
                 exit = wait => VmExit::Exited(exit),
                 () = cancel.cancelled() => VmExit::Canceled,
@@ -667,7 +695,8 @@ impl RunnerSupervisor {
         // on degraded interop and must not make the agent deaf to
         // CancelRunner — mirror the VM startup path.
         let spawned = {
-            let spawn = std::pin::pin!(interop.spawn(&order.encoded_jit_config));
+            let spawn =
+                std::pin::pin!(interop.spawn(&order.encoded_jit_config, self.log_sink(job_id)));
             tokio::select! {
                 result = spawn => Some(result),
                 () = cancel.cancelled() => None,
@@ -749,6 +778,18 @@ impl RunnerSupervisor {
             }
         };
 
+        // Drain both pipes for the job's lifetime. This is not optional: with
+        // `Stdio::piped()` a runner that outruns the reader would block on a
+        // full pipe buffer, so capture and liveness are the same concern here.
+        let sink = self.log_sink(job_id);
+        let inner = child.inner();
+        if let Some(stdout) = inner.stdout.take() {
+            sink.pipe(stdout);
+        }
+        if let Some(stderr) = inner.stderr.take() {
+            sink.pipe(stderr);
+        }
+
         // group_spawn is synchronous, but cancellation is handled on another
         // runtime thread. Arbitrate through the slot before accepting.
         if !self.accept_started(job_id, token) {
@@ -833,6 +874,12 @@ impl RunnerSupervisor {
             }
         };
 
+        // Follow from here rather than before `start`: the log stream is keyed
+        // on the container, which only exists now. Docker retains output from
+        // the container's first instant, so nothing produced during startup is
+        // missed.
+        running.follow_logs(self.log_sink(job_id));
+
         // The container is running. Cancellation and acceptance arbitrate
         // through the slot before any verdict is sent.
         if !self.accept_started(job_id, token) {
@@ -993,9 +1040,20 @@ impl RunnerSupervisor {
 /// point (`run.sh`); no `.current_dir()` is set because the wrapper script
 /// locates its own sibling files via `$0`'s dirname, not the caller's
 /// working directory.
+///
+/// Both output streams are piped rather than inherited, so they can be
+/// attributed to this job and streamed to the gateway. The caller must drain
+/// them (see [`RunnerSupervisor::run_host_job`]) or a chatty runner blocks on a
+/// full pipe. This is why host-runner output no longer appears in the service
+/// manager's stdout/stderr capture, where it used to land interleaved and
+/// unattributed.
 fn runner_command(script: &Path, encoded_jit_config: &str) -> tokio::process::Command {
     let mut command = tokio::process::Command::new(script);
-    command.arg("--jitconfig").arg(encoded_jit_config);
+    command
+        .arg("--jitconfig")
+        .arg(encoded_jit_config)
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped());
     command
 }
 
@@ -1043,7 +1101,7 @@ mod tests {
     fn supervisor(capabilities: Vec<Capability>) -> RunnerSupervisor {
         let (events, _rx) = mpsc::channel(1);
         let state = AgentState::new(&seed());
-        RunnerSupervisor::new(
+        RunnerSupervisor::without_logs(
             events,
             Some(PathBuf::from("/nonexistent")),
             Backends::fixed(capabilities, state.clone()),
@@ -1091,10 +1149,88 @@ mod tests {
             interop,
             state.clone(),
         );
-        let sup = RunnerSupervisor::new(events, None, backends, state);
+        let sup = RunnerSupervisor::without_logs(events, None, backends, state);
         (sup, rx)
     }
 
+    /// Like [`windows_supervisor_with_rx`], but also hands back the
+    /// runner-output receiver so a test can assert on captured output.
+    fn windows_supervisor_with_logs(
+        interop: Option<InteropRunner>,
+    ) -> (RunnerSupervisor, mpsc::Receiver<AttachRequest>) {
+        let (events, _events_rx) = mpsc::channel(8);
+        let (logs, logs_rx) = mpsc::channel(64);
+        let state = AgentState::new(&crate::settings::PersistedSettings {
+            load_ceiling: f64::MAX,
+            mem_floor_mib: 0,
+            ..seed()
+        });
+        let backends = Backends::fixed_with_interop(
+            vec![capability("windows", "amd64", Backend::HostRunner)],
+            interop,
+            state.clone(),
+        );
+        // The events receiver is dropped, so verdicts are shed — this test
+        // is about output, and the supervisor must not depend on a live
+        // verdict consumer to capture it.
+        let sup = RunnerSupervisor::new(events, logs, None, backends, state);
+        (sup, logs_rx)
+    }
+
+    /// Output a started runner produces must reach the log queue as chunks
+    /// tagged with the job that produced them. Drives the whole path —
+    /// `handle_provision` → backend routing → interop drain → `JobLogSink` —
+    /// rather than calling the sink directly, so a backend that captures
+    /// nothing fails here.
+    #[tokio::test]
+    async fn started_runner_output_reaches_the_log_queue() {
+        let dir = tempfile::tempdir().unwrap();
+        let powershell = interop_stub(
+            dir.path(),
+            "powershell",
+            "echo 'WINPID=4242'\necho 'building the thing'\nexit 0",
+        );
+        let taskkill = interop_stub(dir.path(), "taskkill", "exit 0");
+        let interop = InteropRunner::with_paths(powershell, taskkill, r"C:\r\run.cmd");
+
+        // Same ETXTBSY warm-up as the routing test below: a concurrently
+        // forking test can hold the freshly written stub open for writing.
+        for _ in 0..100 {
+            match interop
+                .spawn("dGVzdA==", crate::joblog::JobLogSink::discarding())
+                .await
+            {
+                Ok(mut job) => {
+                    let _ = job.wait().await;
+                    break;
+                }
+                Err(_) => tokio::time::sleep(Duration::from_millis(20)).await,
+            }
+        }
+
+        let (sup, mut logs_rx) = windows_supervisor_with_logs(Some(interop));
+        sup.handle_provision(ProvisionRunner {
+            job_id: "rjob_win".to_owned(),
+            os: "windows".to_owned(),
+            arch: "amd64".to_owned(),
+            encoded_jit_config: "dGVzdA==".to_owned(),
+            offer_token: "tok1".to_owned(),
+        });
+
+        let chunk = tokio::time::timeout(Duration::from_secs(5), logs_rx.recv())
+            .await
+            .expect("runner output within the handshake budget")
+            .expect("log queue open");
+        match chunk.msg {
+            Some(attach_request::Msg::RunnerLog(chunk)) => {
+                assert_eq!(chunk.job_id, "rjob_win");
+                assert_eq!(chunk.data, b"building the thing\n");
+                assert_eq!(chunk.dropped_bytes, 0);
+            }
+            other => panic!("expected RunnerLog, got {other:?}"),
+        }
+    }
+
     /// Write an executable stub standing in for powershell/taskkill.
     fn interop_stub(dir: &std::path::Path, name: &str, body: &str) -> PathBuf {
         use std::os::unix::fs::PermissionsExt;
@@ -1119,7 +1255,7 @@ mod tests {
         // concurrently forking test can hold it open for writing until its
         // own exec) so the spawn inside handle_provision can't hit it.
         for _ in 0..100 {
-            match interop.spawn("dGVzdA==").await {
+            match interop.spawn("dGVzdA==", JobLogSink::discarding()).await {
                 Ok(mut job) => {
                     let _ = job.wait().await;
                     break;
@@ -1315,7 +1451,7 @@ mod tests {
     fn supervisor_with_rx(capacity: usize) -> (RunnerSupervisor, mpsc::Receiver<AttachRequest>) {
         let (events, rx) = mpsc::channel(capacity);
         let state = AgentState::new(&seed());
-        let sup = RunnerSupervisor::new(
+        let sup = RunnerSupervisor::without_logs(
             events,
             Some(PathBuf::from("/nonexistent")),
             Backends::fixed(
@@ -1340,7 +1476,7 @@ mod tests {
             mem_floor_mib: 0,
             ..seed()
         });
-        let sup = RunnerSupervisor::new(
+        let sup = RunnerSupervisor::without_logs(
             events,
             Some(PathBuf::from("/nonexistent")),
             Backends::fixed(
@@ -1693,7 +1829,7 @@ mod tests {
     async fn shutdown_does_not_flip_observable_draining_but_drain_does() {
         let drained = AgentState::new(&seed());
         let (events, _rx) = mpsc::channel(1);
-        RunnerSupervisor::new(
+        RunnerSupervisor::without_logs(
             events,
             None,
             Backends::fixed(Vec::new(), drained.clone()),
@@ -1707,7 +1843,7 @@ mod tests {
 
         let torn_down = AgentState::new(&seed());
         let (events, _rx) = mpsc::channel(1);
-        RunnerSupervisor::new(
+        RunnerSupervisor::without_logs(
             events,
             None,
             Backends::fixed(Vec::new(), torn_down.clone()),
diff --git a/fleet/arcbox-fleet-agent/src/vm.rs b/fleet/arcbox-fleet-agent/src/vm.rs
index 0b3c109a7..4adccdd76 100644
--- a/fleet/arcbox-fleet-agent/src/vm.rs
+++ b/fleet/arcbox-fleet-agent/src/vm.rs
@@ -37,6 +37,7 @@ use tonic::transport::Channel;
 use tracing::{debug, info, warn};
 
 use crate::host;
+use crate::joblog::JobLogSink;
 
 /// Guest login and runner location — the ArcBox macOS runner image contract
 /// (see the module doc).
@@ -380,13 +381,20 @@ impl RunningVm {
     /// follow with [`destroy`](Self::destroy). The agent reports no outcome
     /// upstream (the GitHub webhook is authoritative); the code is for
     /// logging.
-    pub async fn wait(&mut self) -> Option<u32> {
+    ///
+    /// The guest is destroyed on every exit path, so this session is the only
+    /// chance to observe the runner's output: it is forwarded to `sink` as it
+    /// arrives rather than collected, since nothing here should hold a whole
+    /// job's log in memory.
+    pub async fn wait(&mut self, sink: &JobLogSink) -> Option<u32> {
         let mut code = None;
         while let Some(msg) = self.channel.wait().await {
             match msg {
-                // Runner output stays in the disposable guest; nothing to
-                // collect here.
-                ChannelMsg::Data { .. } | ChannelMsg::ExtendedData { .. } => {}
+                // stdout and the stderr extension both carry runner output;
+                // the guest's own interleaving is the order worth preserving.
+                ChannelMsg::Data { data } | ChannelMsg::ExtendedData { data, .. } => {
+                    sink.write(&data);
+                }
                 ChannelMsg::ExitStatus { exit_status } => code = Some(exit_status),
                 _ => {}
             }
@@ -508,7 +516,7 @@ mod tests {
             )
             .await
             .expect("provision guest and exec");
-        assert_eq!(running.wait().await, Some(0));
+        assert_eq!(running.wait(&JobLogSink::discarding()).await, Some(0));
         running.destroy().await;
     }
 
diff --git a/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto b/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto
index 7a4a97e44..c654432d5 100644
--- a/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto
+++ b/fleet/arcbox-fleet-proto/proto/arcbox/fleet/v1/fleet.proto
@@ -137,6 +137,7 @@ message AttachRequest {
     RunnerAccepted runner_accepted = 3;
     RunnerRejected runner_rejected = 4;
     RunnerExited runner_exited = 5;
+    RunnerLogChunk runner_log = 6;
   }
 }
 
@@ -210,6 +211,31 @@ message RunnerExited {
   string ack_token = 2;
 }
 
+// A slice of a started runner's combined stdout/stderr, for live viewing while
+// the job runs. Deliberately the one lossy message on this stream: unlike
+// verdicts it is never tracked for resend and never acked, because the attach
+// stream is a control plane whose heartbeat must not queue behind runner
+// output. A chunk that cannot be sent immediately is dropped, and the bytes it
+// carried are reported in the next chunk's `dropped_bytes`. Nothing persists it
+// — a consumer sees only what arrives while it is subscribed, and GitHub
+// remains the authoritative record of a job's log.
+message RunnerLogChunk {
+  // Prefixed runner job id (`rjob_...`).
+  string job_id = 1;
+  // Per-job counter starting at 0, incremented for every chunk the agent
+  // emits. Consumers detect loss as a gap, never as reordering: one attach
+  // stream delivers in order.
+  uint64 seq = 2;
+  // Raw output bytes. Not line-aligned and not guaranteed UTF-8 — a chunk
+  // boundary can split a line or a multi-byte character.
+  bytes data = 3;
+  // Best-effort count of bytes shed since the previous chunk on this job —
+  // the rate-budget and queue-full drops the agent observed directly. Zero in
+  // the steady state. A `seq` gap is the authoritative loss signal: a chunk
+  // discarded further down the agent's egress path is not counted here.
+  uint64 dropped_bytes = 4;
+}
+
 message AttachResponse {
   oneof msg {
     // Handshake success — client proceeds to the heartbeat loop.