From 4872cf2aac9f562dd531da08bcc2ad19daa4befc Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 05:55:48 +0900 Subject: [PATCH] fix: retain independent Bash Drop cleanup diagnostics --- crates/tools/src/bash.rs | 89 +----- crates/tools/src/bash/cleanup.rs | 121 ++++++++ crates/tools/src/bash/cleanup/tests.rs | 397 +++++++++++++++++++++++++ crates/tools/src/bash_output.rs | 75 ++++- 4 files changed, 607 insertions(+), 75 deletions(-) create mode 100644 crates/tools/src/bash/cleanup.rs create mode 100644 crates/tools/src/bash/cleanup/tests.rs diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index c3c3a6311..abb9700dc 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -19,6 +19,8 @@ use workdir::{ WorkdirSessionRouter, }; +mod cleanup; + const DEFAULT_TIMEOUT_SECS: u64 = 120; const MAX_TIMEOUT_SECS: u64 = 600; const INLINE_BYTE_BUDGET: usize = 12 * 1024; @@ -72,80 +74,19 @@ impl Drop for CommandGuard { let Some(handle) = self.handle.take() else { return; }; - let workdir = self.session.clone(); - let state = Arc::clone(&self.state); - let execution_id = self.execution_id.clone(); - let execution_nonce = self.execution_nonce; - let mut retained = self.retained.take(); - let cursor = self.cursor; - // A dropped provider future is not terminal confirmation. Keep the live - // execution registered until cleanup has both requested cancellation and - // observed terminal command output, so cancellation/session teardown - // cannot race with an apparently empty registry. - tokio::spawn(async move { - let result = tokio::time::timeout(std::time::Duration::from_secs(10), async { - let _ = workdir.cancel_command(handle.clone()).await; - let mut cursor = cursor; - loop { - let page = command_page( - &workdir, - CommandOutputRequest { - handle: handle.clone(), - cursor, - limit: workdir::MAX_COMMAND_READ_BYTES, - wait: true, - }, - ) - .await?; - if let Some(writer) = &mut retained { - let mut next = cursor; - for chunk in &page.chunks { - if chunk.sequence != next { - return Err(workdir::WorkdirError::Unavailable( - "command cleanup sequence gap".into(), - )); - } - writer - .append_chunk(chunk) - .map_err(workdir::WorkdirError::Unavailable)?; - next = next.checked_add(chunk.bytes.len() as u64).ok_or_else(|| { - workdir::WorkdirError::Unavailable( - "command cleanup cursor overflow".into(), - ) - })?; - } - if page.next_cursor != next { - return Err(workdir::WorkdirError::Unavailable( - "command cleanup cursor gap".into(), - )); - } - } - cursor = page.next_cursor; - if let Some(terminal) = page.terminal { - if let Some(writer) = &mut retained { - writer - .finish_transport(&terminal) - .map_err(workdir::WorkdirError::Unavailable)?; - } - return Ok::<(), workdir::WorkdirError>(()); - } - } - }) - .await; - // Bounded cleanup can end with an unknown provider outcome. The writer then - // terminalizes partial evidence on drop; retaining a dead collector in `active` - // would leak registrations and falsely suggest it could still collect output. - let _ = result; - let mut state = state.lock().unwrap(); - if state - .active - .get(&execution_id) - .is_some_and(|active| active.execution_nonce == execution_nonce) - { - state.active.remove(&execution_id); - state.cancellation_requested.remove(&execution_id); - } - }); + let cleanup = cleanup::Cleanup { + session: self.session.clone(), + state: Arc::clone(&self.state), + execution_id: self.execution_id.clone(), + execution_nonce: self.execution_nonce, + handle, + cursor: self.cursor, + retained: self + .retained + .take() + .expect("started command owns output writer"), + }; + tokio::spawn(cleanup.run()); } } diff --git a/crates/tools/src/bash/cleanup.rs b/crates/tools/src/bash/cleanup.rs new file mode 100644 index 000000000..ed118c898 --- /dev/null +++ b/crates/tools/src/bash/cleanup.rs @@ -0,0 +1,121 @@ +//! Finite Drop cleanup. A cancel acknowledgement is not termination evidence; only a +//! validated terminal fence confirms the process result. No provider text is logged. + +use super::*; +use crate::bash_output::{ + CleanupCancel, CleanupDiagnostics, CleanupDrain, CleanupFailure, CleanupProviderFailure, + CleanupRegistration, +}; +use std::time::Duration; + +const CLEANUP_TIMEOUT: Duration = Duration::from_secs(10); +// Reserve time for independent terminal observation even if cancel never responds. +const CANCEL_TIMEOUT: Duration = Duration::from_secs(2); + +pub(super) struct Cleanup { + pub session: WorkdirSessionHandle, + pub state: Arc>, + pub execution_id: String, + pub execution_nonce: u64, + pub handle: CommandHandle, + pub cursor: u64, + pub retained: BashOutputWriter, +} + +impl Cleanup { + pub async fn run(mut self) { + let deadline = tokio::time::Instant::now() + CLEANUP_TIMEOUT; + let cancel = match tokio::time::timeout( + CANCEL_TIMEOUT, + self.session.cancel_command(self.handle.clone()), + ) + .await + { + Ok(Ok(())) => CleanupCancel::Acknowledged, + Ok(Err(error)) => CleanupCancel::Failed { + reason: CleanupProviderFailure::from(&error), + }, + Err(_) => CleanupCancel::DeadlineExceeded, + }; + let drain = match tokio::time::timeout_at(deadline, self.drain()).await { + Ok(Ok(())) => CleanupDrain::TerminalConfirmed, + Ok(Err(reason)) => CleanupDrain::Failed { reason }, + Err(_) => CleanupDrain::DeadlineExceeded, + }; + let deadline_expired = matches!(drain, CleanupDrain::DeadlineExceeded); + // Finalize evidence before exposing an empty registry. Keep the lock through + // publication so an older cleanup can never remove a newer generation. + let mut state = self.state.lock().unwrap(); + let registration = if state + .active + .get(&self.execution_id) + .is_some_and(|active| active.execution_nonce == self.execution_nonce) + { + state.active.remove(&self.execution_id); + state.cancellation_requested.remove(&self.execution_id); + CleanupRegistration::Released + } else { + CleanupRegistration::Superseded + }; + let diagnostics = CleanupDiagnostics { + execution_nonce: self.execution_nonce, + cancel, + drain, + deadline_expired, + registration, + }; + tracing::info!( + output_id = %self.retained.id(), + cleanup = ?diagnostics, + "Bash Drop cleanup finished; cancellation and terminal observation are independent" + ); + self.retained.finish_cleanup(diagnostics); + drop(self.retained); + } + + async fn drain(&mut self) -> Result<(), CleanupFailure> { + loop { + let page = command_page( + &self.session, + CommandOutputRequest { + handle: self.handle.clone(), + cursor: self.cursor, + limit: workdir::MAX_COMMAND_READ_BYTES, + wait: true, + }, + ) + .await + .map_err(|error| CleanupFailure::Provider { + reason: CleanupProviderFailure::from(&error), + })?; + let mut next = self.cursor; + for chunk in &page.chunks { + if chunk.sequence != next { + return Err(CleanupFailure::SequenceGap); + } + next = next + .checked_add(chunk.bytes.len() as u64) + .filter(|next| *next <= crate::bash_output::COMMAND_BYTES as u64) + .ok_or(CleanupFailure::DeliveryBudgetExceeded)?; + self.retained + .append_chunk(chunk) + .map_err(|_| CleanupFailure::RetainedChunkInvalid)?; + } + if page.next_cursor != next { + return Err(CleanupFailure::CursorGap); + } + self.cursor = next; + if let Some(terminal) = page.terminal { + self.retained + .finish_transport(&terminal) + .map_err(|_| CleanupFailure::TerminalMetadataInvalid)?; + return Ok(()); + } + // An immediately-ready provider must not starve the cleanup deadline. + tokio::task::yield_now().await; + } + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/tools/src/bash/cleanup/tests.rs b/crates/tools/src/bash/cleanup/tests.rs new file mode 100644 index 000000000..9b04ec0c0 --- /dev/null +++ b/crates/tools/src/bash/cleanup/tests.rs @@ -0,0 +1,397 @@ +//! Component tests enter through the production CommandGuard Drop, cleanup, page +//! validation and retained store. Only the Workdir provider is replaced; paused time +//! and a cancel barrier make faults/deadlines deterministic without OS timing. + +use super::*; +use std::sync::atomic::{AtomicBool, Ordering}; +use workdir::*; + +const PRIVATE: &str = "SECRET command /private/host/path output password=poison"; + +#[derive(Clone, Copy, Debug)] +enum Fault { + Cancel, + CancelHang, + Drain, + TerminalTimeout, + Sequence, + Cursor, + RetainedChunk, + TerminalMetadata, + CollectionIncomplete, + Complete, +} + +#[derive(Debug)] +struct Provider { + local: LocalWorkdirSession, + fault: Fault, + cancelling: tokio::sync::Notify, + allow_drain: AtomicBool, +} + +#[async_trait] +impl WorkdirSession for Provider { + fn workdir(&self) -> &Workdir { + self.local.workdir() + } + fn capabilities(&self) -> WorkdirSessionCapabilities { + WorkdirSessionCapabilities::ALL + } + async fn stat(&self, _: StatRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn read(&self, _: ReadRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn write(&self, _: WriteRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn edit(&self, _: EditRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn list(&self, _: ListRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn glob(&self, _: GlobRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn grep(&self, _: GrepRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn start_command(&self, _: CommandRequest) -> Result { + Err(WorkdirError::OperationFailed) + } + async fn command_status(&self, _: CommandHandle) -> Result { + Err(WorkdirError::OperationFailed) + } + + async fn command_output( + &self, + request: CommandOutputRequest, + ) -> Result { + assert_eq!(request.cursor, 3); + while !self.allow_drain.load(Ordering::Acquire) { + tokio::task::yield_now().await; + } + match self.fault { + Fault::Drain => { + return Err(WorkdirError::Transport(PRIVATE.into()) + .with_failure_reason(WorkdirFailureReason::HttpTransfer)); + } + Fault::TerminalTimeout => return std::future::pending().await, + _ => {} + } + let mut chunk = CommandByteChunk { + sequence: 3, + offset: 3, + stream: CommandStream::Stdout, + bytes: b"tail".to_vec(), + }; + if matches!(self.fault, Fault::Sequence) { + chunk.sequence = 4; + } + if matches!(self.fault, Fault::RetainedChunk) { + chunk.offset = 42; + } + let mut terminal = terminal(); + if matches!(self.fault, Fault::TerminalMetadata) { + terminal.stdout_delivered_bytes = 6; + } + if matches!(self.fault, Fault::CollectionIncomplete) { + terminal.output_collection = CommandOutputCollection::Incomplete { + failure: CommandFailure { + stage: CommandFailureStage::StdoutDrain, + reason: CommandIoErrorKind::Other, + os_code: None, + }, + }; + } + Ok(CommandOutput { + chunks: vec![chunk], + next_cursor: if matches!(self.fault, Fault::Cursor) { + 8 + } else { + 7 + }, + terminal: Some(terminal), + }) + } + + async fn cancel_command(&self, _: CommandHandle) -> Result<(), WorkdirError> { + self.cancelling.notify_one(); + match self.fault { + Fault::Cancel => Err(WorkdirError::Unavailable(PRIVATE.into())), + Fault::CancelHang => std::future::pending().await, + _ => Ok(()), + } + } + async fn close(&self) -> Result<(), WorkdirError> { + Ok(()) + } +} + +fn terminal() -> CommandTerminal { + CommandTerminal { + status: CommandStatus::Completed, + exit_code: Some(0), + process_failure: None, + output_collection: CommandOutputCollection::Complete, + stdout_bytes: 7, + stderr_bytes: 0, + stdout_delivered_bytes: 7, + stderr_delivered_bytes: 0, + truncation_reason: None, + } +} + +struct Fixture { + _root: tempfile::TempDir, + provider: Arc, + state: Arc>, + view: BashOutputView, + id: String, +} + +impl Fixture { + fn start(fault: Fault) -> (Self, CommandGuard) { + let base = + std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target/cleanup-fixtures"); + std::fs::create_dir_all(&base).unwrap(); + let root = tempfile::tempdir_in(base).unwrap(); + let provider = Arc::new(Provider { + local: LocalWorkdirSession::new( + manifest::Scope::writable(root.path()).unwrap(), + root.path().into(), + ), + fault, + cancelling: Default::default(), + allow_drain: AtomicBool::new(false), + }); + let view = BashOutputView::new(); + let mut retained = view.begin().unwrap(); + retained.append(CommandStream::Stdout, b"pre"); + let id = retained.id(); + let state = Arc::new(Mutex::new(BashExecutionState::default())); + let handle = CommandHandle(PRIVATE.into()); + state.lock().unwrap().active.insert( + PRIVATE.into(), + ActiveCommand { + call_id: PRIVATE.into(), + execution_nonce: 1, + session: provider.clone(), + handle: handle.clone(), + }, + ); + state + .lock() + .unwrap() + .cancellation_requested + .insert(PRIVATE.into()); + let guard = CommandGuard { + session: provider.clone(), + state: state.clone(), + execution_id: PRIVATE.into(), + execution_nonce: 1, + handle: Some(handle), + cursor: 3, + retained: Some(retained), + }; + ( + Self { + _root: root, + provider, + state, + view, + id, + }, + guard, + ) + } + + async fn finish(&self) -> BashOutputMetadata { + self.provider.allow_drain.store(true, Ordering::Release); + // The provider's read retry backoff and cleanup deadlines are virtual time. + for _ in 0..4 { + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(3)).await; + } + for _ in 0..1000 { + if let Some(metadata) = self.view.metadata(&self.id) { + return metadata; + } + tokio::task::yield_now().await; + } + panic!("finite cleanup did not publish partial evidence"); + } + + fn assert_evidence(&self, m: &BashOutputMetadata, terminal: bool, bytes: &[u8]) { + assert_eq!( + self.view + .read(&self.id, CommandStream::Stdout, 0, 64) + .unwrap(), + bytes + ); + assert_eq!(m.stdout_saved, bytes.len()); + assert_eq!(m.stdout_delivered, bytes.len() as u64); + assert_eq!(m.stdout_seen, bytes.len() as u64); + assert_eq!( + m.terminal.status, + terminal.then_some(CommandStatus::Completed) + ); + assert_eq!(m.terminal.exit_code, terminal.then_some(0)); + if !terminal { + assert!(m.terminal.output_collection.is_none()); + assert!(m.terminal.process_failure.is_none()); + assert!(!m.terminal.transport_complete); + } + let diagnostics = m.cleanup.as_ref().unwrap(); + assert_eq!(diagnostics.execution_nonce, 1); + let json = serde_json::to_string(diagnostics).unwrap(); + let debug = format!("{diagnostics:?}"); + for forbidden in [ + PRIVATE, "SECRET", "/private", "password", "poison", "tail", "pre", + ] { + assert!(!json.contains(forbidden), "diagnostics leaked data: {json}"); + assert!( + !debug.contains(forbidden), + "tracing payload leaked data: {debug}" + ); + } + } +} + +#[tokio::test(start_paused = true)] +async fn cancel_failure_does_not_preclude_confirmed_terminal() { + let (f, guard) = Fixture::start(Fault::Cancel); + drop(guard); + f.provider.cancelling.notified().await; + assert!(f.state.lock().unwrap().active.contains_key(PRIVATE)); + let m = f.finish().await; + f.assert_evidence(&m, true, b"pretail"); + assert!(m.terminal.transport_complete); + assert!(matches!( + m.cleanup.as_ref().unwrap().cancel, + CleanupCancel::Failed { .. } + )); + assert!(matches!( + m.cleanup.as_ref().unwrap().drain, + CleanupDrain::TerminalConfirmed + )); + assert!(!m.cleanup.as_ref().unwrap().deadline_expired); + assert!(f.state.lock().unwrap().active.is_empty()); + assert!(f.state.lock().unwrap().cancellation_requested.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn acknowledged_cancel_drain_faults_keep_distinct_reasons_and_partial_bytes() { + for (fault, reason, bytes) in [ + (Fault::Drain, "provider", b"pre".as_slice()), + (Fault::Sequence, "sequence_gap", b"pre"), + (Fault::Cursor, "cursor_gap", b"pretail"), + (Fault::RetainedChunk, "retained_chunk_invalid", b"pre"), + ( + Fault::TerminalMetadata, + "terminal_metadata_invalid", + b"pretail", + ), + ] { + let (f, guard) = Fixture::start(fault); + drop(guard); + f.provider.cancelling.notified().await; + let m = f.finish().await; + f.assert_evidence(&m, false, bytes); + let diagnostic = serde_json::to_value(m.cleanup.as_ref().unwrap()).unwrap(); + assert_eq!(diagnostic["cancel"]["outcome"], "acknowledged"); + assert_eq!(diagnostic["drain"]["outcome"], "failed"); + assert_eq!(diagnostic["drain"]["reason"]["kind"], reason); + assert_eq!(diagnostic["registration"], "released"); + assert_eq!(diagnostic["deadline_expired"], false); + assert!(f.state.lock().unwrap().active.is_empty()); + } +} + +#[tokio::test(start_paused = true)] +async fn terminal_deadline_never_claims_stop_or_leaves_active_collector() { + let (f, guard) = Fixture::start(Fault::TerminalTimeout); + drop(guard); + f.provider.cancelling.notified().await; + let m = f.finish().await; + f.assert_evidence(&m, false, b"pre"); + assert!(matches!( + m.cleanup.as_ref().unwrap().cancel, + CleanupCancel::Acknowledged + )); + assert!(matches!( + m.cleanup.as_ref().unwrap().drain, + CleanupDrain::DeadlineExceeded + )); + assert!(m.cleanup.as_ref().unwrap().deadline_expired); + assert!(f.state.lock().unwrap().active.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn cancel_deadline_still_attempts_independent_terminal_observation() { + let (f, guard) = Fixture::start(Fault::CancelHang); + drop(guard); + f.provider.cancelling.notified().await; + let m = f.finish().await; + f.assert_evidence(&m, true, b"pretail"); + assert!(matches!( + m.cleanup.as_ref().unwrap().cancel, + CleanupCancel::DeadlineExceeded + )); + assert!(matches!( + m.cleanup.as_ref().unwrap().drain, + CleanupDrain::TerminalConfirmed + )); + assert!(!m.cleanup.as_ref().unwrap().deadline_expired); + assert!(f.state.lock().unwrap().active.is_empty()); +} + +#[tokio::test(start_paused = true)] +async fn old_cleanup_preserves_new_execution_and_cancellation_registration() { + let (f, guard) = Fixture::start(Fault::Complete); + drop(guard); + f.provider.cancelling.notified().await; + f.state + .lock() + .unwrap() + .active + .get_mut(PRIVATE) + .unwrap() + .execution_nonce = 2; + let m = f.finish().await; + f.assert_evidence(&m, true, b"pretail"); + assert!(matches!( + m.cleanup.as_ref().unwrap().registration, + CleanupRegistration::Superseded + )); + assert_eq!(f.state.lock().unwrap().active[PRIVATE].execution_nonce, 2); + assert!( + f.state + .lock() + .unwrap() + .cancellation_requested + .contains(PRIVATE) + ); +} + +#[tokio::test(start_paused = true)] +async fn terminal_confirmation_is_independent_of_output_collection_completeness() { + let (f, guard) = Fixture::start(Fault::CollectionIncomplete); + drop(guard); + f.provider.cancelling.notified().await; + let m = f.finish().await; + f.assert_evidence(&m, true, b"pretail"); + assert!(!m.terminal.transport_complete); + assert!(matches!( + m.terminal.output_collection, + Some(CommandOutputCollection::Incomplete { .. }) + )); + assert!(matches!( + m.cleanup.as_ref().unwrap().drain, + CleanupDrain::TerminalConfirmed + )); + assert!(f.state.lock().unwrap().active.is_empty()); +} diff --git a/crates/tools/src/bash_output.rs b/crates/tools/src/bash_output.rs index fad3b02e9..080a87fec 100644 --- a/crates/tools/src/bash_output.rs +++ b/crates/tools/src/bash_output.rs @@ -19,7 +19,7 @@ use std::time::{Duration, Instant}; use serde::Serialize; use workdir::{CommandStatus, CommandStream}; -const COMMAND_BYTES: usize = 4 * 1024 * 1024; +pub(crate) const COMMAND_BYTES: usize = 4 * 1024 * 1024; const STORE_BYTES: usize = 32 * 1024 * 1024; const STORE_COUNT: usize = 64; const ORDER_COUNT: usize = 4096; @@ -55,6 +55,7 @@ struct Record { transport_counts: Option<(u64, u64)>, transport_truncation: Option, terminal: Option, + cleanup: Option, } #[derive(Clone, Debug, Serialize)] @@ -95,10 +96,72 @@ pub struct BashOutputMetadata { pub transport_truncation: Option, pub storage_truncated: bool, pub terminal: OutputTerminal, + pub cleanup: Option, pub text_encoding: &'static str, pub ordering: &'static str, } +/// Drop cleanup observations, not assertions that cancellation stopped a process. +/// Correlate with the enclosing output UUID and this host-assigned execution generation; +/// caller identifiers, provider handles, commands and error text are deliberately excluded. +#[derive(Clone, Debug, Serialize)] +pub struct CleanupDiagnostics { + pub execution_nonce: u64, + pub cancel: CleanupCancel, + pub drain: CleanupDrain, + pub deadline_expired: bool, + pub registration: CleanupRegistration, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "outcome", rename_all = "snake_case")] +pub enum CleanupCancel { + Acknowledged, + Failed { reason: CleanupProviderFailure }, + DeadlineExceeded, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "outcome", rename_all = "snake_case")] +pub enum CleanupDrain { + TerminalConfirmed, + Failed { reason: CleanupFailure }, + DeadlineExceeded, +} + +#[derive(Clone, Debug, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum CleanupFailure { + Provider { reason: CleanupProviderFailure }, + SequenceGap, + CursorGap, + DeliveryBudgetExceeded, + RetainedChunkInvalid, + TerminalMetadataInvalid, +} + +#[derive(Clone, Debug, Serialize)] +pub struct CleanupProviderFailure { + pub failure_reason: Option, + pub denial_reason: Option, +} + +impl From<&workdir::WorkdirError> for CleanupProviderFailure { + fn from(error: &workdir::WorkdirError) -> Self { + Self { + failure_reason: error.failure_reason(), + denial_reason: error.denial_reason(), + } + } +} + +#[derive(Clone, Copy, Debug, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum CleanupRegistration { + Released, + Superseded, +} + /// A bounded reservation acquired before command start. Drop terminalizes partial evidence. pub struct BashOutputWriter { record: Arc>, @@ -214,6 +277,7 @@ impl BashOutputView { transport_counts: None, transport_truncation: None, terminal: None, + cleanup: None, })); store.records.push_back(record.clone()); store.reserved += COMMAND_BYTES; @@ -313,6 +377,7 @@ impl Record { transport_truncation: self.transport_truncation.clone(), storage_truncated: self.truncated || self.transport_truncation.is_some(), terminal: self.terminal.clone().expect("finalized record"), + cleanup: self.cleanup.clone(), text_encoding: "utf8_lossy", ordering: "provider_read_observation", } @@ -342,6 +407,14 @@ async fn reap(weak: Weak>, mut shutdown: tokio::sync::watch::Receiv } impl BashOutputWriter { + pub(crate) fn id(&self) -> String { + self.record.lock().unwrap().id.clone() + } + + pub(crate) fn finish_cleanup(&mut self, diagnostics: CleanupDiagnostics) { + self.record.lock().unwrap().cleanup = Some(diagnostics); + } + pub fn append_chunk(&mut self, chunk: &workdir::CommandByteChunk) -> Result<(), String> { let expected = { let r = self.record.lock().unwrap();