Publish release-5edbadb22804 filtered source

Private source: 5edbadb2280419cf2cb42b5237835304ad7040e0

Public tree identity: sha256:e0d4d5e1c8693959e69448684a71582edfc13e9b18d7f1e733034fea5ce62cda

crates/clusterflux-client/src/lib.rs

@@ -109,6 +109,21 @@ impl ClusterfluxClient {
         }
     }
 
+    pub async fn cancel_browser_login(
+        &self,
+        transaction_id: impl Into<String>,
+    ) -> Result<(), ClientError> {
+        match self
+            .send_login(LoginRequest::CancelWebBrowserLogin {
+                transaction_id: transaction_id.into(),
+            })
+            .await?
+        {
+            WireResponse::WebBrowserLoginCancelled {} => Ok(()),
+            _ => Err(unexpected_response("web_browser_login_cancelled")),
+        }
+    }
+
     pub async fn account_status(&self) -> Result<AccountStatus, ClientError> {
         match self
             .send_authenticated(AuthenticatedRequest::AuthStatus)

crates/clusterflux-client/src/protocol.rs

@@ -122,6 +122,9 @@ pub(crate) enum AuthenticatedRequest {
 #[serde(tag = "type", rename_all = "snake_case")]
 pub(crate) enum LoginRequest {
     BeginWebBrowserLogin {},
+    CancelWebBrowserLogin {
+        transaction_id: String,
+    },
     ExchangeWebLoginHandoff {
         transaction_id: String,
         handoff_code: String,
@@ -297,6 +300,7 @@ pub(crate) enum WireResponse {
         authorization_url: String,
         expires_at_epoch_seconds: u64,
     },
+    WebBrowserLoginCancelled {},
     WebBrowserSession {
         session: WireBrowserSession,
     },
@@ -334,6 +338,7 @@ mod tests {
             "abort_process",
             "auth_status",
             "begin_web_browser_login",
+            "cancel_web_browser_login",
             "cancel_process",
             "create_artifact_download_link",
             "create_debug_epoch",

crates/clusterflux-client/tests/client_contract.rs

@@ -6,7 +6,7 @@ use clusterflux_client::{
     ClusterfluxClient, ControlTransport, MockTransport, ProjectId, SessionCredential, TenantId,
     UserId, CLIENT_API_VERSION,
 };
-use clusterflux_control::CONTROL_API_PATH;
+use clusterflux_control::{CONTROL_API_PATH, LOGIN_API_PATH};
 use clusterflux_coordinator::service::CoordinatorService;
 use serde_json::{json, Value};
 
@@ -64,6 +64,32 @@ async fn structured_errors_retain_machine_fields_and_originating_request_id() {
     assert!(!error.retryable);
 }
 
+#[tokio::test]
+async fn browser_login_cancellation_uses_the_login_boundary_and_exact_transaction() {
+    let transport = MockTransport::from_json_responses([
+        json!({ "type": "web_browser_login_cancelled" }).to_string(),
+    ]);
+    let client = ClusterfluxClient::with_transport(transport.clone());
+
+    client
+        .cancel_browser_login("login-transaction")
+        .await
+        .unwrap();
+
+    let requests = transport.requests();
+    assert_eq!(requests.len(), 1);
+    assert_eq!(requests[0].api_path, LOGIN_API_PATH);
+    let envelope: Value = serde_json::from_slice(&requests[0].body).unwrap();
+    assert_eq!(envelope["operation"], "cancel_web_browser_login");
+    assert_eq!(
+        envelope["payload"],
+        json!({
+            "type": "cancel_web_browser_login",
+            "transaction_id": "login-transaction"
+        })
+    );
+}
+
 #[tokio::test]
 async fn a_mismatched_error_request_id_is_rejected_as_a_protocol_error() {
     let transport = MockTransport::from_json_responses([json!({

crates/clusterflux-client/tests/fixtures/web_operations.json

@@ -5,6 +5,12 @@
     "request": { "type": "begin_web_browser_login" },
     "response": { "type": "web_browser_login_started", "transaction_id": "login-transaction", "authorization_url": "https://auth.clusterflux.lesstuff.com/authorize?state=opaque", "expires_at_epoch_seconds": 1000 }
   },
+  {
+    "operation": "cancel_web_browser_login",
+    "boundary": "login",
+    "request": { "type": "cancel_web_browser_login", "transaction_id": "login-transaction" },
+    "response": { "type": "web_browser_login_cancelled" }
+  },
   {
     "operation": "exchange_web_login_handoff",
     "boundary": "login",

crates/clusterflux-coordinator/src/service.rs

@@ -304,6 +304,13 @@ pub struct CoordinatorService {
         clusterflux_core::TaskInstanceId,
         String,
     )>,
+    recent_log_quota_truncated_streams: BTreeSet<(
+        TenantId,
+        ProjectId,
+        ProcessId,
+        clusterflux_core::TaskInstanceId,
+        String,
+    )>,
     next_recent_log_sequence: u64,
     debug_audit_events: VecDeque<DebugAuditEvent>,
     debug_epochs: BTreeMap<ProcessControlKey, u64>,
@@ -579,6 +586,7 @@ impl CoordinatorService {
             recent_log_dropped_through: BTreeMap::new(),
             recent_log_accounted_bytes: BTreeMap::new(),
             recent_log_truncated_streams: BTreeSet::new(),
+            recent_log_quota_truncated_streams: BTreeSet::new(),
             next_recent_log_sequence: 1,
             debug_audit_events: VecDeque::new(),
             debug_epochs: BTreeMap::new(),

crates/clusterflux-coordinator/src/service/logs.rs

@@ -40,19 +40,7 @@ impl CoordinatorService {
         validate_task_log_tail("stdout_tail", &stdout_tail)?;
         validate_task_log_tail("stderr_tail", &stderr_tail)?;
         let now_epoch_seconds = self.current_epoch_seconds()?;
-        let unaccounted_bytes = self.unaccounted_final_log_bytes(
-            &tenant,
-            &project,
-            &process,
-            &task,
-            stdout_bytes,
-            stderr_bytes,
-        )?;
-        self.quota
-            .can_charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?;
-        self.quota
-            .charge_log_bytes(&tenant, &project, unaccounted_bytes, now_epoch_seconds)?;
-        self.reconcile_final_log_stream(
+        let stdout_retained = self.accept_final_log_stream(
             &tenant,
             &project,
             &process,
@@ -62,8 +50,8 @@ impl CoordinatorService {
             &stdout_tail,
             stdout_truncated,
             now_epoch_seconds,
-        );
-        self.reconcile_final_log_stream(
+        )?;
+        let stderr_retained = self.accept_final_log_stream(
             &tenant,
             &project,
             &process,
@@ -73,18 +61,22 @@ impl CoordinatorService {
             &stderr_tail,
             stderr_truncated,
             now_epoch_seconds,
-        );
+        )?;
         Ok(CoordinatorResponse::TaskLogRecorded {
             process,
             task,
             stdout_bytes,
             stderr_bytes,
-            stdout_tail: if stdout_truncated {
+            stdout_tail: if !stdout_retained {
+                "[log output truncated at project log quota]".to_owned()
+            } else if stdout_truncated {
                 format!("{stdout_tail}\n... truncated")
             } else {
                 stdout_tail
             },
-            stderr_tail: if stderr_truncated {
+            stderr_tail: if !stderr_retained {
+                "[log output truncated at project log quota]".to_owned()
+            } else if stderr_truncated {
                 format!("{stderr_tail}\n... truncated")
             } else {
                 stderr_tail
@@ -150,11 +142,41 @@ impl CoordinatorService {
             });
         }
         let now_epoch_seconds = self.current_epoch_seconds()?;
+        if self.recent_log_quota_truncated_streams.contains(&key) {
+            if end > expected {
+                self.recent_log_accounted_bytes.insert(key, end);
+            }
+            return Ok(CoordinatorResponse::TaskLogChunkRecorded {
+                process,
+                task,
+                sequence: None,
+                next_offset: end.max(expected),
+            });
+        }
         let newly_accounted = end.saturating_sub(expected);
-        self.quota
-            .can_charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
-        self.quota
-            .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
+        if self
+            .quota
+            .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)
+            .is_err()
+        {
+            if end > expected {
+                self.recent_log_accounted_bytes.insert(key.clone(), end);
+            }
+            let sequence = self.mark_log_quota_truncated(
+                &tenant,
+                &project,
+                &process,
+                &task,
+                &stream,
+                now_epoch_seconds,
+            );
+            return Ok(CoordinatorResponse::TaskLogChunkRecorded {
+                process,
+                task,
+                sequence,
+                next_offset: end.max(expected),
+            });
+        }
         if offset > expected {
             self.record_recent_log(
                 tenant.clone(),
@@ -334,27 +356,7 @@ impl CoordinatorService {
             result,
         };
         let now_epoch_seconds = self.current_epoch_seconds()?;
-        let unaccounted_bytes = self.unaccounted_final_log_bytes(
-            &event.tenant,
-            &event.project,
-            &event.process,
-            &event.task,
-            event.stdout_bytes,
-            event.stderr_bytes,
-        )?;
-        self.quota.can_charge_log_bytes(
-            &event.tenant,
-            &event.project,
-            unaccounted_bytes,
-            now_epoch_seconds,
-        )?;
-        self.quota.charge_log_bytes(
-            &event.tenant,
-            &event.project,
-            unaccounted_bytes,
-            now_epoch_seconds,
-        )?;
-        self.reconcile_final_log_stream(
+        let stdout_retained = self.accept_final_log_stream(
             &event.tenant,
             &event.project,
             &event.process,
@@ -364,8 +366,8 @@ impl CoordinatorService {
             &event.stdout_tail,
             event.stdout_truncated,
             now_epoch_seconds,
-        );
-        self.reconcile_final_log_stream(
+        )?;
+        let stderr_retained = self.accept_final_log_stream(
             &event.tenant,
             &event.project,
             &event.process,
@@ -375,7 +377,15 @@ impl CoordinatorService {
             &event.stderr_tail,
             event.stderr_truncated,
             now_epoch_seconds,
-        );
+        )?;
+        if !stdout_retained {
+            event.stdout_tail = "[log output truncated at project log quota]".to_owned();
+            event.stdout_truncated = true;
+        }
+        if !stderr_retained {
+            event.stderr_tail = "[log output truncated at project log quota]".to_owned();
+            event.stderr_truncated = true;
+        }
         let task_key = task_control_key(
             &event.tenant,
             &event.project,
@@ -833,48 +843,94 @@ impl CoordinatorService {
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn unaccounted_final_log_bytes(
-        &self,
+    fn accept_final_log_stream(
+        &mut self,
         tenant: &TenantId,
         project: &ProjectId,
         process: &ProcessId,
         task: &TaskInstanceId,
-        stdout_bytes: u64,
-        stderr_bytes: u64,
-    ) -> Result<u64, CoordinatorServiceError> {
-        let stdout_accounted = self
+        stream: TaskLogStream,
+        total_source_bytes: u64,
+        final_tail: &str,
+        source_truncated: bool,
+        now_epoch_seconds: u64,
+    ) -> Result<bool, CoordinatorServiceError> {
+        let key = recent_log_offset_key(tenant, project, process, task, &stream);
+        let accounted = self
             .recent_log_accounted_bytes
-            .get(&recent_log_offset_key(
+            .get(&key)
+            .copied()
+            .unwrap_or(0);
+        let remaining = total_source_bytes.checked_sub(accounted).ok_or_else(|| {
+            let stream_name = match stream {
+                TaskLogStream::Stdout => "stdout",
+                TaskLogStream::Stderr => "stderr",
+            };
+            CoordinatorServiceError::Protocol(format!(
+                "final {stream_name} byte count {total_source_bytes} is below the {accounted} live bytes already accounted"
+            ))
+        })?;
+        if self.recent_log_quota_truncated_streams.contains(&key) {
+            self.recent_log_accounted_bytes
+                .insert(key, total_source_bytes);
+            return Ok(false);
+        }
+        if self
+            .quota
+            .charge_log_bytes(tenant, project, remaining, now_epoch_seconds)
+            .is_err()
+        {
+            self.recent_log_accounted_bytes
+                .insert(key, total_source_bytes);
+            self.mark_log_quota_truncated(
                 tenant,
                 project,
                 process,
                 task,
-                &TaskLogStream::Stdout,
-            ))
-            .copied()
-            .unwrap_or(0);
-        let stderr_accounted = self
-            .recent_log_accounted_bytes
-            .get(&recent_log_offset_key(
-                tenant,
-                project,
-                process,
-                task,
-                &TaskLogStream::Stderr,
-            ))
-            .copied()
-            .unwrap_or(0);
-        let stdout_remaining = stdout_bytes.checked_sub(stdout_accounted).ok_or_else(|| {
-            CoordinatorServiceError::Protocol(format!(
-                "final stdout byte count {stdout_bytes} is below the {stdout_accounted} live bytes already accounted"
-            ))
-        })?;
-        let stderr_remaining = stderr_bytes.checked_sub(stderr_accounted).ok_or_else(|| {
-            CoordinatorServiceError::Protocol(format!(
-                "final stderr byte count {stderr_bytes} is below the {stderr_accounted} live bytes already accounted"
-            ))
-        })?;
-        checked_reported_log_bytes(stdout_remaining, stderr_remaining)
+                &stream,
+                now_epoch_seconds,
+            );
+            return Ok(false);
+        }
+        self.reconcile_final_log_stream(
+            tenant,
+            project,
+            process,
+            task,
+            stream,
+            total_source_bytes,
+            final_tail,
+            source_truncated,
+            now_epoch_seconds,
+        );
+        Ok(true)
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn mark_log_quota_truncated(
+        &mut self,
+        tenant: &TenantId,
+        project: &ProjectId,
+        process: &ProcessId,
+        task: &TaskInstanceId,
+        stream: &TaskLogStream,
+        now_epoch_seconds: u64,
+    ) -> Option<u64> {
+        let key = recent_log_offset_key(tenant, project, process, task, stream);
+        if !self.recent_log_quota_truncated_streams.insert(key.clone()) {
+            return None;
+        }
+        self.recent_log_truncated_streams.insert(key);
+        Some(self.record_recent_log(
+            tenant.clone(),
+            project.clone(),
+            process.clone(),
+            task.clone(),
+            stream.clone(),
+            "[log output truncated at project log quota]".to_owned(),
+            true,
+            now_epoch_seconds,
+        ))
     }
 
     #[allow(clippy::too_many_arguments)]
@@ -1020,6 +1076,14 @@ impl CoordinatorService {
                     || entry_task != task
             },
         );
+        self.recent_log_quota_truncated_streams.retain(
+            |(entry_tenant, entry_project, entry_process, entry_task, _)| {
+                entry_tenant != tenant
+                    || entry_project != project
+                    || entry_process != process
+                    || entry_task != task
+            },
+        );
     }
 
     fn finish_task_attempt(&mut self, event: &mut TaskCompletionEvent) -> bool {
@@ -1260,17 +1324,6 @@ impl CoordinatorService {
     }
 }
 
-fn checked_reported_log_bytes(
-    stdout_bytes: u64,
-    stderr_bytes: u64,
-) -> Result<u64, CoordinatorServiceError> {
-    stdout_bytes.checked_add(stderr_bytes).ok_or_else(|| {
-        CoordinatorServiceError::Protocol(
-            "reported task log byte counts exceed the supported range".to_owned(),
-        )
-    })
-}
-
 fn recent_log_offset_key(
     tenant: &TenantId,
     project: &ProjectId,
@@ -1305,11 +1358,11 @@ fn bounded_log_tail(mut value: String, truncated: &mut bool) -> String {
     if value.len() <= MAX_TASK_LOG_TAIL_BYTES {
         return value;
     }
-    let mut boundary = MAX_TASK_LOG_TAIL_BYTES;
-    while !value.is_char_boundary(boundary) {
-        boundary -= 1;
+    let mut boundary = value.len() - MAX_TASK_LOG_TAIL_BYTES;
+    while boundary < value.len() && !value.is_char_boundary(boundary) {
+        boundary += 1;
     }
-    value.truncate(boundary);
+    value.drain(..boundary);
     *truncated = true;
     value
 }

crates/clusterflux-coordinator/src/service/quota.rs

@@ -260,22 +260,6 @@ impl CoordinatorQuota {
         self.charge(tenant, project, LimitKind::ApiCall, 1, now_epoch_seconds)
     }
 
-    pub(super) fn can_charge_log_bytes(
-        &self,
-        tenant: &TenantId,
-        project: &ProjectId,
-        bytes: u64,
-        now_epoch_seconds: u64,
-    ) -> Result<(), LimitError> {
-        self.can_charge(
-            tenant,
-            project,
-            LimitKind::LogBytes,
-            bytes,
-            now_epoch_seconds,
-        )
-    }
-
     pub(super) fn charge_log_bytes(
         &mut self,
         tenant: &TenantId,

crates/clusterflux-coordinator/src/service/summaries.rs

@@ -98,6 +98,11 @@ impl CoordinatorService {
                 entry_tenant != tenant || entry_project != project || entry_process != process
             },
         );
+        self.recent_log_quota_truncated_streams.retain(
+            |(entry_tenant, entry_project, entry_process, _, _)| {
+                entry_tenant != tenant || entry_project != project || entry_process != process
+            },
+        );
         self.process_summary_order
             .retain(|retained| retained != &key);
         self.evict_process_summaries_for_project(tenant, project);
@@ -525,6 +530,10 @@ impl CoordinatorService {
             .retain(|(tenant, project, process, _, _)| {
                 tenant != &key.0 || project != &key.1 || process != &key.2
             });
+        self.recent_log_quota_truncated_streams
+            .retain(|(tenant, project, process, _, _)| {
+                tenant != &key.0 || project != &key.1 || process != &key.2
+            });
         self.process_summary_order
             .retain(|retained| retained != key);
     }

crates/clusterflux-coordinator/src/service/tests.rs

@@ -2435,7 +2435,7 @@ fn authenticated_api_calls_are_metered_per_tenant_and_project_before_dispatch()
 }
 
 #[test]
-fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
+fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports() {
     let mut limits = ResourceLimits::unlimited();
     limits.limits.insert(LimitKind::LogBytes, 4);
     let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap();
@@ -2494,7 +2494,7 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
             backpressured: false,
         })
         .unwrap();
-    let denied = service
+    let truncated = service
         .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLog {
             tenant: "tenant".to_owned(),
             project: "project".to_owned(),
@@ -2509,8 +2509,22 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
             stderr_truncated: false,
             backpressured: true,
         })
-        .unwrap_err();
-    assert!(denied.to_string().contains("LogBytes"));
+        .unwrap();
+    let CoordinatorResponse::TaskLogRecorded {
+        stdout_tail,
+        stdout_bytes: 4,
+        ..
+    } = truncated
+    else {
+        panic!("expected a successful truncated task-log report");
+    };
+    assert_eq!(stdout_tail, "[log output truncated at project log quota]");
+    assert!(service
+        .recent_logs
+        .get(&(TenantId::from("tenant"), ProjectId::from("project")))
+        .unwrap()
+        .iter()
+        .any(|entry| entry.text.contains("project log quota") && entry.truncated));
     assert_eq!(
         service
             .quota
@@ -2519,6 +2533,140 @@ fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
     );
 }
 
+#[test]
+fn log_quota_exhaustion_cannot_strand_task_completion_or_artifact_publication() {
+    let mut limits = ResourceLimits::unlimited();
+    limits.limits.insert(LimitKind::LogBytes, 4);
+    let quota = CoordinatorQuotaConfiguration::new(limits, [(LimitKind::LogBytes, 60)]).unwrap();
+    let mut service = CoordinatorService::new_with_admin_token_database_url_and_quota(
+        7,
+        "test-admin-token",
+        None,
+        quota,
+    )
+    .unwrap();
+    service.set_server_time(30);
+    service
+        .handle_request(CoordinatorRequest::AttachNode {
+            tenant: "tenant".to_owned(),
+            project: "project".to_owned(),
+            node: "node".to_owned(),
+            public_key: test_node_public_key("node"),
+        })
+        .unwrap();
+    service
+        .handle_request(CoordinatorRequest::StartProcess {
+            launch_attempt: None,
+            tenant: "tenant".to_owned(),
+            project: "project".to_owned(),
+            actor_user: None,
+            actor_agent: None,
+            agent_public_key_fingerprint: None,
+            agent_signature: None,
+            process: "process".to_owned(),
+            restart: false,
+        })
+        .unwrap();
+    service
+        .handle_signed_node_request_auto(CoordinatorRequest::ReconnectNode {
+            tenant: "tenant".to_owned(),
+            project: "project".to_owned(),
+            node: "node".to_owned(),
+            process: "process".to_owned(),
+            epoch: 7,
+        })
+        .unwrap();
+    register_test_task_assignment(
+        &mut service,
+        "tenant",
+        "project",
+        "process",
+        "node",
+        "compile",
+        "compile-one",
+        7,
+    );
+    service.record_task_completion_event(TaskCompletionEvent {
+        tenant: TenantId::from("tenant"),
+        project: ProjectId::from("project"),
+        process: ProcessId::from("process"),
+        node: NodeId::from("coordinator-main"),
+        executor: TaskExecutor::CoordinatorMain,
+        task_definition: TaskDefinitionId::from("build"),
+        task: TaskInstanceId::from("main"),
+        attempt_id: None,
+        placement: None,
+        terminal_state: TaskTerminalState::Completed,
+        status_code: Some(0),
+        stdout_bytes: 0,
+        stderr_bytes: 0,
+        stdout_tail: String::new(),
+        stderr_tail: String::new(),
+        stdout_truncated: false,
+        stderr_truncated: false,
+        artifact_path: None,
+        artifact_digest: None,
+        artifact_size_bytes: None,
+        result: None,
+    });
+
+    service
+        .handle_signed_node_request_auto(CoordinatorRequest::TaskCompleted {
+            tenant: "tenant".to_owned(),
+            project: "project".to_owned(),
+            process: "process".to_owned(),
+            node: "node".to_owned(),
+            task: "compile-one".to_owned(),
+            terminal_state: Some(TaskTerminalState::Completed),
+            status_code: Some(0),
+            stdout_bytes: 64,
+            stderr_bytes: 0,
+            stdout_tail: "the-real-final-tail".to_owned(),
+            stderr_tail: String::new(),
+            stdout_truncated: true,
+            stderr_truncated: false,
+            artifact_path: Some("/vfs/artifacts/result.bin".to_owned()),
+            artifact_digest: Some(Digest::sha256("artifact bytes")),
+            artifact_size_bytes: Some(14),
+            result: None,
+        })
+        .unwrap();
+
+    let event = service
+        .task_events
+        .iter()
+        .find(|event| event.task == TaskInstanceId::from("compile-one"))
+        .unwrap();
+    assert_eq!(
+        event.stdout_tail,
+        "[log output truncated at project log quota]"
+    );
+    assert!(event.stdout_truncated);
+    assert!(!service
+        .active_tasks
+        .iter()
+        .any(|key| key.4 == TaskInstanceId::from("compile-one")));
+    assert!(service
+        .coordinator
+        .active_process(
+            &TenantId::from("tenant"),
+            &ProjectId::from("project"),
+            &ProcessId::from("process"),
+        )
+        .is_none());
+    assert!(matches!(
+        service
+            .handle_request(CoordinatorRequest::GetArtifact {
+                tenant: "tenant".to_owned(),
+                project: "project".to_owned(),
+                actor_user: "user".to_owned(),
+                artifact: "result.bin".to_owned(),
+            })
+            .unwrap(),
+        CoordinatorResponse::Artifact { .. }
+    ));
+}
+
 #[test]
 fn service_attaches_node_starts_process_and_records_scoped_task_event() {
     let mut service = CoordinatorService::new(7);

crates/clusterflux-node/src/assignment_runner/process_runner.rs

@@ -9,6 +9,26 @@ struct LiveLogChunk {
     truncated: bool,
 }
 
+fn append_bounded_tail(tail: &mut Vec<u8>, bytes: &[u8], maximum: usize) {
+    if maximum == 0 {
+        tail.clear();
+        return;
+    }
+    if bytes.len() >= maximum {
+        tail.clear();
+        tail.extend_from_slice(&bytes[bytes.len() - maximum..]);
+        return;
+    }
+    let overflow = tail
+        .len()
+        .saturating_add(bytes.len())
+        .saturating_sub(maximum);
+    if overflow > 0 {
+        tail.drain(..overflow);
+    }
+    tail.extend_from_slice(bytes);
+}
+
 fn redact_safe_live_log_prefix(
     pending: &[u8],
     configured_secrets: &[String],
@@ -279,9 +299,9 @@ impl CoordinatorControlledProcessRunner {
             let mut captured = Vec::new();
             let mut buffer = [0_u8; 16 * 1024];
             let stream_base = source_bytes_total.load(Ordering::Relaxed);
+            let mut source_bytes_read = 0_u64;
             let mut pending_offset = stream_base;
             let mut pending = Vec::new();
-            let mut discarded = false;
             loop {
                 let count = reader
                     .read(&mut buffer)
@@ -294,27 +314,21 @@ impl CoordinatorControlledProcessRunner {
                     Ordering::Relaxed,
                     |current| Some(current.saturating_add(count as u64)),
                 );
-                let remaining = maximum.saturating_sub(captured.len());
-                let retained = count.min(remaining);
-                if retained > 0 {
-                    captured.extend_from_slice(&buffer[..retained]);
-                    pending.extend_from_slice(&buffer[..retained]);
-                    if let Some((consumed, text)) =
-                        redact_safe_live_log_prefix(&pending, &configured_secrets, false)
-                    {
-                        let _ = sender.try_send(LiveLogChunk {
-                            stream,
-                            offset: pending_offset,
-                            source_bytes: consumed as u64,
-                            bytes: text.into_bytes(),
-                            truncated: false,
-                        });
-                        pending.drain(..consumed);
-                        pending_offset = pending_offset.saturating_add(consumed as u64);
-                    }
-                }
-                if retained < count {
-                    discarded = true;
+                source_bytes_read = source_bytes_read.saturating_add(count as u64);
+                append_bounded_tail(&mut captured, &buffer[..count], maximum);
+                pending.extend_from_slice(&buffer[..count]);
+                if let Some((consumed, text)) =
+                    redact_safe_live_log_prefix(&pending, &configured_secrets, false)
+                {
+                    let _ = sender.try_send(LiveLogChunk {
+                        stream,
+                        offset: pending_offset,
+                        source_bytes: consumed as u64,
+                        bytes: text.into_bytes(),
+                        truncated: false,
+                    });
+                    pending.drain(..consumed);
+                    pending_offset = pending_offset.saturating_add(consumed as u64);
                 }
             }
             if let Some((consumed, text)) =
@@ -328,10 +342,10 @@ impl CoordinatorControlledProcessRunner {
                     truncated: false,
                 });
             }
-            if discarded {
+            if source_bytes_read > maximum as u64 {
                 let _ = sender.try_send(LiveLogChunk {
                     stream,
-                    offset: stream_base.saturating_add(maximum as u64),
+                    offset: stream_base.saturating_add(source_bytes_read),
                     source_bytes: 0,
                     bytes: b"[log output truncated at node capture limit]".to_vec(),
                     truncated: true,
@@ -616,11 +630,11 @@ mod tests {
     }
 
     #[test]
-    fn bounded_capture_retains_the_complete_source_byte_count() {
+    fn bounded_capture_retains_the_real_tail_and_complete_source_byte_count() {
         let source_bytes = Arc::new(AtomicU64::new(5));
         let (sender, receiver) = mpsc::sync_channel(8);
         let reader = CoordinatorControlledProcessRunner::drain_bounded(
-            Cursor::new(vec![b'x'; 32]),
+            Cursor::new(b"0123456789abcdefghijklmnopqrstuv".to_vec()),
             8,
             "stdout",
             sender,
@@ -630,14 +644,14 @@ mod tests {
         let captured = reader.join().unwrap().unwrap();
         let chunks = receiver.into_iter().collect::<Vec<_>>();
 
-        assert_eq!(captured, vec![b'x'; 8]);
+        assert_eq!(captured, b"opqrstuv");
         assert_eq!(source_bytes.load(Ordering::Relaxed), 37);
         assert_eq!(
             chunks.iter().map(|chunk| chunk.source_bytes).sum::<u64>(),
-            8
+            32
         );
         assert_eq!(chunks[0].offset, 5);
-        assert_eq!(chunks.last().unwrap().offset, 13);
+        assert_eq!(chunks.last().unwrap().offset, 37);
         assert!(chunks.iter().any(|chunk| chunk.truncated));
     }
 }

crates/clusterflux-node/src/command_runner.rs

@@ -1,6 +1,6 @@
 use clusterflux_core::{
-    CommandInvocation, Digest, LogBuffer, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject,
-    VfsOverlay, VfsPath,
+    CommandInvocation, Digest, NativeCommandPolicy, NodeId, TaskInstanceId, VfsObject, VfsOverlay,
+    VfsPath,
 };
 use serde::{Deserialize, Serialize};
 
@@ -113,24 +113,20 @@ impl LocalCommandExecutor {
 }
 
 pub(super) fn capture_command_logs(
-    task: &TaskInstanceId,
+    _task: &TaskInstanceId,
     stdout: &[u8],
     stderr: &[u8],
     max_log_bytes: usize,
 ) -> CapturedCommandLogs {
-    let mut logs = LogBuffer::new(max_log_bytes);
-    logs.push(task.clone(), stdout);
-    logs.push(task.clone(), stderr);
-    let records = logs.records();
-    let stdout_record = &records[0];
-    let stderr_record = &records[1];
-    debug_assert_eq!(&stdout_record.task, task);
-    debug_assert_eq!(&stderr_record.task, task);
+    let stdout_truncated = stdout.len() > max_log_bytes;
+    let stderr_truncated = stderr.len() > max_log_bytes;
+    let stdout_start = stdout.len().saturating_sub(max_log_bytes);
+    let stderr_start = stderr.len().saturating_sub(max_log_bytes);
     CapturedCommandLogs {
-        stdout: String::from_utf8_lossy(&stdout_record.bytes).into_owned(),
-        stderr: String::from_utf8_lossy(&stderr_record.bytes).into_owned(),
-        stdout_truncated: stdout_record.truncated,
-        stderr_truncated: stderr_record.truncated,
-        backpressured: logs.backpressured(),
+        stdout: String::from_utf8_lossy(&stdout[stdout_start..]).into_owned(),
+        stderr: String::from_utf8_lossy(&stderr[stderr_start..]).into_owned(),
+        stdout_truncated,
+        stderr_truncated,
+        backpressured: stdout_truncated || stderr_truncated,
     }
 }

crates/clusterflux-node/src/lib.rs

@@ -741,7 +741,7 @@ mod tests {
     }
 
     #[test]
-    fn linux_backend_caps_logs_without_truncating_staged_artifact_bytes() {
+    fn linux_backend_retains_final_log_tail_without_truncating_staged_artifact_bytes() {
         let invocation = CommandInvocation {
             program: "cargo".to_owned(),
             args: vec!["build".to_owned()],
@@ -774,7 +774,7 @@ mod tests {
             .unwrap();
 
         assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux"));
-        assert_eq!(output.stdout, "abcd");
+        assert_eq!(output.stdout, "cdef");
         assert!(output.stdout_truncated);
         assert!(output.log_backpressured);
         assert_eq!(output.staged_artifact.as_ref().unwrap().size, 6);
@@ -992,7 +992,7 @@ mod tests {
 
     #[cfg(unix)]
     #[test]
-    fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() {
+    fn local_command_executor_retains_each_stream_tail_and_reports_backpressure() {
         let executor = LocalCommandExecutor {
             node: clusterflux_core::NodeId::from("node"),
             hosted_control_plane: false,
@@ -1021,10 +1021,10 @@ mod tests {
             .unwrap();
 
         assert_eq!(output.virtual_thread, TaskInstanceId::from("compile-linux"));
-        assert_eq!(output.stdout, "abcd");
-        assert_eq!(output.stderr, "");
+        assert_eq!(output.stdout, "cdef");
+        assert_eq!(output.stderr, "err");
         assert!(output.stdout_truncated);
-        assert!(output.stderr_truncated);
+        assert!(!output.stderr_truncated);
         assert!(output.log_backpressured);
     }