crates/clusterflux-client/src/lib.rs

@@ -109,21 +109,6 @@ 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,9 +122,6 @@ 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,
@@ -300,7 +297,6 @@ pub(crate) enum WireResponse {
         authorization_url: String,
         expires_at_epoch_seconds: u64,
     },
-    WebBrowserLoginCancelled {},
     WebBrowserSession {
         session: WireBrowserSession,
     },
@@ -338,7 +334,6 @@ 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, LOGIN_API_PATH};
+use clusterflux_control::CONTROL_API_PATH;
 use clusterflux_coordinator::service::CoordinatorService;
 use serde_json::{json, Value};
 
@@ -64,32 +64,6 @@ 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,12 +5,6 @@
     "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,13 +304,6 @@ 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>,
@@ -586,7 +579,6 @@ 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,7 +40,19 @@ 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 stdout_retained = self.accept_final_log_stream(
+        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(
             &tenant,
             &project,
             &process,
@@ -50,8 +62,8 @@ impl CoordinatorService {
             &stdout_tail,
             stdout_truncated,
             now_epoch_seconds,
-        )?;
+        );
-        let stderr_retained = self.accept_final_log_stream(
+        self.reconcile_final_log_stream(
             &tenant,
             &project,
             &process,
@@ -61,22 +73,18 @@ impl CoordinatorService {
             &stderr_tail,
             stderr_truncated,
             now_epoch_seconds,
-        )?;
+        );
         Ok(CoordinatorResponse::TaskLogRecorded {
             process,
             task,
             stdout_bytes,
             stderr_bytes,
-            stdout_tail: if !stdout_retained {
+            stdout_tail: if stdout_truncated {
-                "[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_retained {
+            stderr_tail: if stderr_truncated {
-                "[log output truncated at project log quota]".to_owned()
-            } else if stderr_truncated {
                 format!("{stderr_tail}\n... truncated")
             } else {
                 stderr_tail
@@ -142,41 +150,11 @@ 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);
-        if self
+        self.quota
-            .quota
+            .can_charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
-            .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)
+        self.quota
-            .is_err()
+            .charge_log_bytes(&tenant, &project, newly_accounted, now_epoch_seconds)?;
-        {
-            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(),
@@ -356,7 +334,27 @@ impl CoordinatorService {
             result,
         };
         let now_epoch_seconds = self.current_epoch_seconds()?;
-        let stdout_retained = self.accept_final_log_stream(
+        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(
             &event.tenant,
             &event.project,
             &event.process,
@@ -366,8 +364,8 @@ impl CoordinatorService {
             &event.stdout_tail,
             event.stdout_truncated,
             now_epoch_seconds,
-        )?;
+        );
-        let stderr_retained = self.accept_final_log_stream(
+        self.reconcile_final_log_stream(
             &event.tenant,
             &event.project,
             &event.process,
@@ -377,15 +375,7 @@ 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,
@@ -843,94 +833,48 @@ impl CoordinatorService {
     }
 
     #[allow(clippy::too_many_arguments)]
-    fn accept_final_log_stream(
+    fn unaccounted_final_log_bytes(
-        &mut self,
+        &self,
         tenant: &TenantId,
         project: &ProjectId,
         process: &ProcessId,
         task: &TaskInstanceId,
-        stream: TaskLogStream,
+        stdout_bytes: u64,
-        total_source_bytes: u64,
+        stderr_bytes: u64,
-        final_tail: &str,
+    ) -> Result<u64, CoordinatorServiceError> {
-        source_truncated: bool,
+        let stdout_accounted = self
-        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(&key)
+            .get(&recent_log_offset_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,
-                &stream,
+                &TaskLogStream::Stdout,
-                now_epoch_seconds,
+            ))
-            );
+            .copied()
-            return Ok(false);
+            .unwrap_or(0);
-        }
+        let stderr_accounted = self
-        self.reconcile_final_log_stream(
+            .recent_log_accounted_bytes
-            tenant,
+            .get(&recent_log_offset_key(
-            project,
+                tenant,
-            process,
+                project,
-            task,
+                process,
-            stream,
+                task,
-            total_source_bytes,
+                &TaskLogStream::Stderr,
-            final_tail,
+            ))
-            source_truncated,
+            .copied()
-            now_epoch_seconds,
+            .unwrap_or(0);
-        );
+        let stdout_remaining = stdout_bytes.checked_sub(stdout_accounted).ok_or_else(|| {
-        Ok(true)
+            CoordinatorServiceError::Protocol(format!(
-    }
+                "final stdout byte count {stdout_bytes} is below the {stdout_accounted} live bytes already accounted"
-
+            ))
-    #[allow(clippy::too_many_arguments)]
+        })?;
-    fn mark_log_quota_truncated(
+        let stderr_remaining = stderr_bytes.checked_sub(stderr_accounted).ok_or_else(|| {
-        &mut self,
+            CoordinatorServiceError::Protocol(format!(
-        tenant: &TenantId,
+                "final stderr byte count {stderr_bytes} is below the {stderr_accounted} live bytes already accounted"
-        project: &ProjectId,
+            ))
-        process: &ProcessId,
+        })?;
-        task: &TaskInstanceId,
+        checked_reported_log_bytes(stdout_remaining, stderr_remaining)
-        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)]
@@ -1076,14 +1020,6 @@ 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 {
@@ -1324,6 +1260,17 @@ 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,
@@ -1358,11 +1305,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 = value.len() - MAX_TASK_LOG_TAIL_BYTES;
+    let mut boundary = MAX_TASK_LOG_TAIL_BYTES;
-    while boundary < value.len() && !value.is_char_boundary(boundary) {
+    while !value.is_char_boundary(boundary) {
-        boundary += 1;
+        boundary -= 1;
     }
-    value.drain(..boundary);
+    value.truncate(boundary);
     *truncated = true;
     value
 }

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

@@ -260,6 +260,22 @@ 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,11 +98,6 @@ 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);
@@ -530,10 +525,6 @@ 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_truncates_at_scoped_quota_without_failing_reports() {
+fn signed_node_log_ingestion_checks_scoped_quota_before_accepting_bytes() {
     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_truncates_at_scoped_quota_without_failing_reports()
             backpressured: false,
         })
         .unwrap();
-    let truncated = service
+    let denied = service
         .handle_signed_node_request_auto(CoordinatorRequest::ReportTaskLog {
             tenant: "tenant".to_owned(),
             project: "project".to_owned(),
@@ -2509,22 +2509,8 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports()
             stderr_truncated: false,
             backpressured: true,
         })
-        .unwrap();
+        .unwrap_err();
-    let CoordinatorResponse::TaskLogRecorded {
+    assert!(denied.to_string().contains("LogBytes"));
-        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
@@ -2533,140 +2519,6 @@ fn signed_node_log_ingestion_truncates_at_scoped_quota_without_failing_reports()
     );
 }
 
-#[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,26 +9,6 @@ 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],
@@ -299,9 +279,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)
@@ -314,21 +294,27 @@ impl CoordinatorControlledProcessRunner {
                     Ordering::Relaxed,
                     |current| Some(current.saturating_add(count as u64)),
                 );
-                source_bytes_read = source_bytes_read.saturating_add(count as u64);
+                let remaining = maximum.saturating_sub(captured.len());
-                append_bounded_tail(&mut captured, &buffer[..count], maximum);
+                let retained = count.min(remaining);
-                pending.extend_from_slice(&buffer[..count]);
+                if retained > 0 {
-                if let Some((consumed, text)) =
+                    captured.extend_from_slice(&buffer[..retained]);
-                    redact_safe_live_log_prefix(&pending, &configured_secrets, false)
+                    pending.extend_from_slice(&buffer[..retained]);
-                {
+                    if let Some((consumed, text)) =
-                    let _ = sender.try_send(LiveLogChunk {
+                        redact_safe_live_log_prefix(&pending, &configured_secrets, false)
-                        stream,
+                    {
-                        offset: pending_offset,
+                        let _ = sender.try_send(LiveLogChunk {
-                        source_bytes: consumed as u64,
+                            stream,
-                        bytes: text.into_bytes(),
+                            offset: pending_offset,
-                        truncated: false,
+                            source_bytes: consumed as u64,
-                    });
+                            bytes: text.into_bytes(),
-                    pending.drain(..consumed);
+                            truncated: false,
-                    pending_offset = pending_offset.saturating_add(consumed as u64);
+                        });
+                        pending.drain(..consumed);
+                        pending_offset = pending_offset.saturating_add(consumed as u64);
+                    }
+                }
+                if retained < count {
+                    discarded = true;
                 }
             }
             if let Some((consumed, text)) =
@@ -342,10 +328,10 @@ impl CoordinatorControlledProcessRunner {
                     truncated: false,
                 });
             }
-            if source_bytes_read > maximum as u64 {
+            if discarded {
                 let _ = sender.try_send(LiveLogChunk {
                     stream,
-                    offset: stream_base.saturating_add(source_bytes_read),
+                    offset: stream_base.saturating_add(maximum as u64),
                     source_bytes: 0,
                     bytes: b"[log output truncated at node capture limit]".to_vec(),
                     truncated: true,
@@ -630,11 +616,11 @@ mod tests {
     }
 
     #[test]
-    fn bounded_capture_retains_the_real_tail_and_complete_source_byte_count() {
+    fn bounded_capture_retains_the_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(b"0123456789abcdefghijklmnopqrstuv".to_vec()),
+            Cursor::new(vec![b'x'; 32]),
             8,
             "stdout",
             sender,
@@ -644,14 +630,14 @@ mod tests {
         let captured = reader.join().unwrap().unwrap();
         let chunks = receiver.into_iter().collect::<Vec<_>>();
 
-        assert_eq!(captured, b"opqrstuv");
+        assert_eq!(captured, vec![b'x'; 8]);
         assert_eq!(source_bytes.load(Ordering::Relaxed), 37);
         assert_eq!(
             chunks.iter().map(|chunk| chunk.source_bytes).sum::<u64>(),
-            32
+            8
         );
         assert_eq!(chunks[0].offset, 5);
-        assert_eq!(chunks.last().unwrap().offset, 37);
+        assert_eq!(chunks.last().unwrap().offset, 13);
         assert!(chunks.iter().any(|chunk| chunk.truncated));
     }
 }

crates/clusterflux-node/src/command_runner.rs

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

crates/clusterflux-node/src/lib.rs

@@ -741,7 +741,7 @@ mod tests {
     }
 
     #[test]
-    fn linux_backend_retains_final_log_tail_without_truncating_staged_artifact_bytes() {
+    fn linux_backend_caps_logs_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, "cdef");
+        assert_eq!(output.stdout, "abcd");
         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_retains_each_stream_tail_and_reports_backpressure() {
+    fn local_command_executor_caps_logs_and_reports_backpressure_by_virtual_thread() {
         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, "cdef");
+        assert_eq!(output.stdout, "abcd");
-        assert_eq!(output.stderr, "err");
+        assert_eq!(output.stderr, "");
         assert!(output.stdout_truncated);
-        assert!(!output.stderr_truncated);
+        assert!(output.stderr_truncated);
         assert!(output.log_backpressured);
     }