Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

fix: retain command failure stages and output collection state

Browse Source
This commit is contained in:
Hare
2026-10-21 05:40:39 +09:00
parent aa0fb2f787
commit 62b2d27b4a
14 changed files with 1329 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workdir/Cargo.toml
+1
View File
@@ -23,5 +23,6 @@ thiserror.workspace = true
tokio = { workspace = true, features = ["io-util", "process", "rt", "sync", "time"] }
[dev-dependencies]
axum.workspace = true
serde_json.workspace = true
tempfile.workspace = true
crates/workdir/src/external.rs
+4
View File
@@ -924,6 +924,8 @@ mod tests {
chunks: vec![],
next_cursor: 0,
terminal: Some(crate::CommandTerminal {
process_failure: None,
output_collection: crate::CommandOutputCollection::Complete,
status: crate::CommandStatus::Completed,
exit_code: Some(0),
stdout_bytes: 0,
@@ -1082,6 +1084,8 @@ mod tests {
}],
next_cursor: 3,
terminal: Some(crate::CommandTerminal {
process_failure: None,
output_collection: crate::CommandOutputCollection::Complete,
status: crate::CommandStatus::Completed,
exit_code: Some(0),
stdout_bytes: 100 * 1024 * 1024,
crates/workdir/src/local/command_transport.rs
+172 -48
View File
@@ -3,6 +3,7 @@
//! A replay page survives a lost HTTP response until its next cursor is sent.
use super::*;
use crate::{CommandFailure, CommandFailureStage, CommandIoErrorKind, CommandOutputCollection};
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::sync::mpsc;
@@ -146,6 +147,33 @@ pub(super) fn start(
telemetry: CommandTelemetry,
environment: BTreeMap<String, String>,
resources: Vec<Arc<dyn WorkdirSessionResource>>,
) -> LocalCommand {
let execution_id = command_id.clone();
let execution_telemetry = telemetry.clone();
start_with(
command_id,
telemetry,
resources,
move |cancel, sender| async move {
execute(
cwd,
request,
&execution_id,
&execution_telemetry,
environment,
cancel,
&sender,
)
.await
},
)
}
fn start_with<F: std::future::Future<Output = crate::CommandTerminal> + Send + 'static>(
command_id: String,
telemetry: CommandTelemetry,
resources: Vec<Arc<dyn WorkdirSessionResource>>,
execution: impl FnOnce(watch::Receiver<bool>, mpsc::Sender<Message>) -> F + Send + 'static,
) -> LocalCommand {
let (sender, receiver) = mpsc::channel(QUEUE_CHUNKS);
let (cancel, cancel_rx) = watch::channel(false);
@@ -160,16 +188,7 @@ pub(super) fn start(
}));
let task = tokio::spawn(async move {
let _resources = resources;
let terminal = execute(
cwd,
request,
&command_id,
&telemetry,
environment,
cancel_rx,
&sender,
)
.await;
let terminal = execution(cancel_rx, sender.clone()).await;
telemetry.terminal(&command_id, terminal.status, terminal.exit_code);
// Closing the byte queue fences this already-published terminal, without
// requiring spare queue capacity or a receiver to release task resources.
@@ -192,10 +211,18 @@ struct ByteAccounting {
}
impl ByteAccounting {
fn terminal(&self, status: CommandStatus, exit_code: Option<i32>) -> crate::CommandTerminal {
fn terminal(
&self,
status: CommandStatus,
exit_code: Option<i32>,
process_failure: Option<CommandFailure>,
output_collection: CommandOutputCollection,
) -> crate::CommandTerminal {
crate::CommandTerminal {
status,
exit_code,
process_failure,
output_collection,
stdout_bytes: self.stdout,
stderr_bytes: self.stderr,
stdout_delivered_bytes: self.stdout_delivered,
@@ -215,7 +242,7 @@ async fn pump<R: AsyncRead + Unpin>(
command_id: &str,
mut remaining: u64,
accounting: &mut ByteAccounting,
) -> std::io::Result<()> {
) -> Result<(), CommandFailure> {
let mut stdout_done = false;
let mut stderr_done = false;
let mut stdout_decoder = CommandOutputDecoder::default();
@@ -228,7 +255,11 @@ async fn pump<R: AsyncRead + Unpin>(
// Before quota, reserve first to retain bounded pipe backpressure.
let permit = if remaining > 0 {
let Ok(permit) = sender.reserve().await else {
return Ok(());
return Err(CommandFailure {
stage: CommandFailureStage::OutputDelivery,
reason: CommandIoErrorKind::Unknown,
os_code: None,
});
};
Some(permit)
} else {
@@ -242,7 +273,15 @@ async fn pump<R: AsyncRead + Unpin>(
CommandStream::Stdout => &stdout_buffer,
CommandStream::Stderr => &stderr_buffer,
};
let count = result?;
let count = result.map_err(|error| {
CommandFailure::io(
match stream {
CommandStream::Stdout => CommandFailureStage::StdoutRead,
CommandStream::Stderr => CommandFailureStage::StderrRead,
},
&error,
)
})?;
let decoder = match stream {
CommandStream::Stdout => &mut stdout_decoder,
CommandStream::Stderr => &mut stderr_decoder,
@@ -288,7 +327,7 @@ async fn execute(
command_id: &str,
telemetry: &CommandTelemetry,
environment: BTreeMap<String, String>,
mut cancel: watch::Receiver<bool>,
cancel: watch::Receiver<bool>,
sender: &mpsc::Sender<Message>,
) -> crate::CommandTerminal {
telemetry.started(command_id, request.tool_call_id);
@@ -304,74 +343,159 @@ async fn execute(
.kill_on_drop(true);
#[cfg(unix)]
command.as_std_mut().process_group(0);
let mut accounting = ByteAccounting::default();
let Ok(mut child) = command.spawn() else {
return accounting.terminal(CommandStatus::Failed, None);
let mut child = match command.spawn() {
Ok(child) => child,
Err(error) => {
return ByteAccounting::default().terminal(
CommandStatus::Failed,
None,
Some(CommandFailure::io(CommandFailureStage::Spawn, &error)),
CommandOutputCollection::NotStarted,
);
}
};
let mut process_group = CommandProcessGroupGuard::new(child.id());
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
let (status, exit_code) = {
supervise(
child,
stdout,
stderr,
request.timeout_secs,
request.transport_byte_budget,
command_id,
telemetry,
cancel,
sender,
)
.await
}
// Substitute only syscalls and readers: injected faults run the production supervisor.
trait CommandProcess: Send {
fn id(&self) -> Option<u32>;
fn start_kill(&mut self);
fn wait(
&mut self,
) -> impl std::future::Future<Output = std::io::Result<std::process::ExitStatus>> + Send;
}
impl CommandProcess for tokio::process::Child {
fn id(&self) -> Option<u32> {
self.id()
}
fn start_kill(&mut self) {
let _ = self.start_kill();
}
async fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
self.wait().await
}
}
fn process_result(
result: std::io::Result<std::process::ExitStatus>,
) -> (CommandStatus, Option<i32>, Option<CommandFailure>) {
match result {
Ok(exit) => (
if exit.success() {
CommandStatus::Completed
} else {
CommandStatus::Failed
},
exit.code(),
None,
),
Err(error) => (
CommandStatus::Failed,
None,
Some(CommandFailure::io(CommandFailureStage::Wait, &error)),
),
}
}
#[allow(clippy::too_many_arguments)]
async fn supervise(
mut child: impl CommandProcess,
stdout: impl AsyncRead + Unpin,
stderr: impl AsyncRead + Unpin,
timeout_secs: u64,
byte_budget: u64,
command_id: &str,
telemetry: &CommandTelemetry,
mut cancel: watch::Receiver<bool>,
sender: &mpsc::Sender<Message>,
) -> crate::CommandTerminal {
let mut process_group = CommandProcessGroupGuard::new(child.id());
let mut accounting = ByteAccounting::default();
let (status, exit_code, process_failure, output_collection) = {
let output = pump(
stdout,
stderr,
sender,
telemetry,
command_id,
request.transport_byte_budget,
byte_budget,
&mut accounting,
);
tokio::pin!(output);
let mut drained = false;
let timeout = tokio::time::sleep(Duration::from_secs(request.timeout_secs.max(1)));
let mut collected = None;
let timeout = tokio::time::sleep(Duration::from_secs(timeout_secs.max(1)));
tokio::pin!(timeout);
let terminal = loop {
tokio::select! {
result = &mut output, if !drained => {
drained = true;
if result.is_err() || sender.is_closed() {
result = &mut output, if collected.is_none() => {
let failed = result.is_err();
collected = Some(result);
if failed {
process_group.terminate();
let _ = child.start_kill();
let _ = child.wait().await;
break (CommandStatus::Failed, None);
child.start_kill();
break process_result(child.wait().await);
}
}
exit = child.wait() => {
process_group.terminate();
break match exit {
Ok(exit) => (if exit.success() { CommandStatus::Completed } else { CommandStatus::Failed }, exit.code()),
Err(_) => (CommandStatus::Failed, None),
};
break process_result(exit);
}
_ = &mut timeout => {
process_group.terminate();
let _ = child.start_kill();
let code = child.wait().await.ok().and_then(|exit| exit.code());
break (CommandStatus::TimedOut, code);
child.start_kill();
let (_, code, failure) = process_result(child.wait().await);
break (CommandStatus::TimedOut, code, failure);
}
_ = sender.closed() => {
process_group.terminate();
let _ = child.start_kill();
let _ = child.wait().await;
break (CommandStatus::Cancelled, None);
child.start_kill();
let (_, code, failure) = process_result(child.wait().await);
break (CommandStatus::Cancelled, code, failure);
}
changed = cancel.changed() => {
if changed.is_err() || *cancel.borrow() {
process_group.terminate();
let _ = child.start_kill();
let code = child.wait().await.ok().and_then(|exit| exit.code());
break (CommandStatus::Cancelled, code);
child.start_kill();
let (_, code, failure) = process_result(child.wait().await);
break (CommandStatus::Cancelled, code, failure);
}
}
}
};
// Even after kill, preserve the bounded pipe tail before publishing terminal.
if !drained {
let _ = output.await;
}
terminal
// Exit is not pipe EOF. Retain read failures first observed during drain.
let result = match collected {
Some(result) => result,
None => output.await.map_err(|mut failure| {
failure.stage = match failure.stage {
CommandFailureStage::StdoutRead => CommandFailureStage::StdoutDrain,
CommandFailureStage::StderrRead => CommandFailureStage::StderrDrain,
stage => stage,
};
failure
}),
};
let collection = match result {
Ok(()) => CommandOutputCollection::Complete,
Err(failure) => CommandOutputCollection::Incomplete { failure },
};
(terminal.0, terminal.1, terminal.2, collection)
};
accounting.terminal(status, exit_code)
accounting.terminal(status, exit_code, process_failure, output_collection)
}
#[cfg(test)]
crates/workdir/src/local/command_transport/tests.rs
+263
View File
@@ -520,3 +520,266 @@ async fn slow_receiver_backpressures_before_quota_but_full_queue_does_not_block_
assert_eq!(queued_bytes, QUEUE_CHUNKS * COMMAND_EVENT_CHUNK_BYTES);
}
}
#[derive(Clone, Default)]
struct Gate(Arc<std::sync::Mutex<(bool, Option<std::task::Waker>)>>);
impl Gate {
fn open(&self) {
let mut state = self.0.lock().unwrap();
state.0 = true;
if let Some(waker) = state.1.take() {
waker.wake();
}
}
fn poll(&self, cx: &mut std::task::Context<'_>) -> std::task::Poll<()> {
let mut state = self.0.lock().unwrap();
if state.0 {
std::task::Poll::Ready(())
} else {
state.1 = Some(cx.waker().clone());
std::task::Poll::Pending
}
}
}
struct FaultProcess {
ready: Gate,
exited: Gate,
wait_error: Option<std::io::ErrorKind>,
}
impl CommandProcess for FaultProcess {
fn id(&self) -> Option<u32> {
None
}
fn start_kill(&mut self) {
self.ready.open();
}
async fn wait(&mut self) -> std::io::Result<std::process::ExitStatus> {
std::future::poll_fn(|cx| self.ready.poll(cx)).await;
self.exited.open();
match self.wait_error {
Some(kind) => Err(std::io::Error::new(
kind,
"/private/path TOKEN=secret command-body",
)),
None => Ok(std::process::ExitStatus::default()),
}
}
}
struct FaultReader {
prefix: &'static [u8],
prefix_read: Gate,
fail_ready: Gate,
}
impl AsyncRead for FaultReader {
fn poll_read(
mut self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buffer: &mut tokio::io::ReadBuf<'_>,
) -> std::task::Poll<std::io::Result<()>> {
if !self.prefix.is_empty() {
let count = self.prefix.len().min(buffer.remaining());
buffer.put_slice(&self.prefix[..count]);
self.prefix = &self.prefix[count..];
self.prefix_read.open();
return std::task::Poll::Ready(Ok(()));
}
std::task::ready!(self.fail_ready.poll(cx));
std::task::Poll::Ready(Err(std::io::Error::new(
std::io::ErrorKind::PermissionDenied,
"/private/path TOKEN=secret command-body",
)))
}
}
// Faults replace only OS wait/read boundaries, not collection or finalization.
#[tokio::test]
async fn read_and_post_exit_drain_faults_keep_prefix_exit_and_safe_original_reason() {
for stream in [CommandStream::Stdout, CommandStream::Stderr] {
for after_exit in [false, true] {
let ready = Gate::default();
let exited = Gate::default();
let fail_ready = if after_exit {
exited.clone()
} else {
let gate = Gate::default();
gate.open();
gate
};
let reader = FaultReader {
prefix: b"partial",
prefix_read: if after_exit {
ready.clone()
} else {
Gate::default()
},
fail_ready,
};
let process = FaultProcess {
ready,
exited,
wait_error: None,
};
let (sender, mut receiver) = mpsc::channel(QUEUE_CHUNKS);
let (_cancel, cancel_rx) = watch::channel(false);
let telemetry = CommandTelemetry::new();
let terminal = tokio::time::timeout(Duration::from_secs(5), async {
match stream {
CommandStream::Stdout => {
supervise(
process,
reader,
&b""[..],
10,
100,
"fault",
&telemetry,
cancel_rx,
&sender,
)
.await
}
CommandStream::Stderr => {
supervise(
process,
&b""[..],
reader,
10,
100,
"fault",
&telemetry,
cancel_rx,
&sender,
)
.await
}
}
})
.await
.unwrap();
let Message::Bytes(chunk) = receiver.recv().await.unwrap();
assert_eq!(chunk.bytes, b"partial");
assert_eq!(chunk.stream, stream);
assert_eq!(chunk.sequence, 0);
assert_eq!(chunk.offset, 0);
assert!(receiver.try_recv().is_err());
assert_eq!(terminal.status, CommandStatus::Completed);
assert_eq!(terminal.exit_code, Some(0));
assert_eq!(terminal.process_failure, None);
assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 7);
assert_eq!(
terminal.stdout_delivered_bytes + terminal.stderr_delivered_bytes,
7
);
assert_eq!(terminal.truncation_reason, None);
assert_eq!(
terminal.output_collection,
CommandOutputCollection::Incomplete {
failure: CommandFailure {
stage: match (stream, after_exit) {
(CommandStream::Stdout, false) => CommandFailureStage::StdoutRead,
(CommandStream::Stdout, true) => CommandFailureStage::StdoutDrain,
(CommandStream::Stderr, false) => CommandFailureStage::StderrRead,
(CommandStream::Stderr, true) => CommandFailureStage::StderrDrain,
},
reason: CommandIoErrorKind::PermissionDenied,
os_code: None
},
}
);
let wire = serde_json::to_string(&terminal).unwrap();
for secret in ["/private", "TOKEN", "secret", "command-body"] {
assert!(!wire.contains(secret));
}
}
}
}
#[tokio::test]
async fn wait_fault_keeps_collected_bytes_and_does_not_invent_exit_information() {
let ready = Gate::default();
ready.open();
let process = FaultProcess {
ready,
exited: Gate::default(),
wait_error: Some(std::io::ErrorKind::InvalidData),
};
let (sender, mut receiver) = mpsc::channel(QUEUE_CHUNKS);
let (_cancel, cancel_rx) = watch::channel(false);
let terminal = supervise(
process,
&b"out"[..],
&b"err"[..],
10,
100,
"wait-fault",
&CommandTelemetry::new(),
cancel_rx,
&sender,
)
.await;
assert_eq!(terminal.status, CommandStatus::Failed);
assert_eq!(terminal.exit_code, None);
assert_eq!(
terminal.process_failure,
Some(CommandFailure {
stage: CommandFailureStage::Wait,
reason: CommandIoErrorKind::InvalidData,
os_code: None
})
);
assert_eq!(
terminal.output_collection,
CommandOutputCollection::Complete
);
assert_eq!(terminal.stdout_bytes, 3);
assert_eq!(terminal.stderr_bytes, 3);
let mut chunks = Vec::new();
while let Ok(Message::Bytes(chunk)) = receiver.try_recv() {
chunks.push(chunk);
}
assert_eq!(stream_bytes(&chunks, CommandStream::Stdout), b"out");
assert_eq!(stream_bytes(&chunks, CommandStream::Stderr), b"err");
assert!(!serde_json::to_string(&terminal).unwrap().contains("secret"));
}
#[tokio::test]
async fn spawn_fault_retains_safe_reason_without_claiming_output_collection() {
let root = TempDir::new().unwrap();
let (sender, _receiver) = mpsc::channel(QUEUE_CHUNKS);
let (_cancel, cancel_rx) = watch::channel(false);
let terminal = execute(
root.path().join("missing-secret-path"),
Harness::request("secret-command", 10, 100),
"spawn-fault",
&CommandTelemetry::new(),
BTreeMap::new(),
cancel_rx,
&sender,
)
.await;
assert_eq!(
terminal.process_failure,
Some(CommandFailure {
stage: CommandFailureStage::Spawn,
reason: CommandIoErrorKind::NotFound,
os_code: Some(libc::ENOENT)
})
);
assert_eq!(
terminal.output_collection,
CommandOutputCollection::NotStarted
);
assert_eq!(terminal.exit_code, None);
assert_eq!(terminal.status, CommandStatus::Failed);
assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 0);
assert!(!serde_json::to_string(&terminal).unwrap().contains("secret"));
}
#[cfg(feature = "http-client")]
mod http;
crates/workdir/src/local/command_transport/tests/http.rs
+238
View File
@@ -0,0 +1,238 @@
//! Loopback integration of the real supervisor, byte queue/replay, local session,
//! HTTP dispatcher/client. OS wait/read are
//! deterministic faults; session open and start bind that injected process fixture.
//! Authorization is not asserted here (Runtime's signed HTTP suite owns it).
use super::*;
use crate::http::*;
use axum::{Json, Router, extract::State, routing::post};
struct Server(tokio::task::JoinHandle<()>);
impl Drop for Server {
fn drop(&mut self) {
self.0.abort();
}
}
async fn operation(
State(session): State<Arc<LocalWorkdirSession>>,
Json(request): Json<WorkdirSessionOperationRequest>,
) -> Json<WorkdirSessionOperationResult> {
if matches!(request.operation, WorkdirSessionOperation::CommandStart(_)) {
return Json(WorkdirSessionOperationResult::CommandStart(CommandHandle(
"injected".into(),
)));
}
Json(
dispatch_workdir_session_operation(session.as_ref(), request.operation)
.await
.unwrap(),
)
}
#[tokio::test]
async fn pipe_and_wait_faults_cross_real_http_with_prefix_and_terminal_replay() {
tokio::time::timeout(Duration::from_secs(20), async {
for (stream, after_exit, wait_fault) in [
(CommandStream::Stdout, false, false),
(CommandStream::Stderr, false, false),
(CommandStream::Stdout, true, false),
(CommandStream::Stderr, true, false),
(CommandStream::Stdout, false, true),
] {
let root = TempDir::new().unwrap();
let session = Arc::new(LocalWorkdirSession::new(
Scope::writable(root.path()).unwrap(),
root.path().into(),
));
let ready = Gate::default();
let exited = Gate::default();
if wait_fault {
ready.open();
}
let process = FaultProcess {
ready: ready.clone(),
exited: exited.clone(),
wait_error: wait_fault.then_some(std::io::ErrorKind::InvalidData),
};
let fail_ready = if after_exit {
exited
} else {
let gate = Gate::default();
gate.open();
gate
};
let reader = FaultReader {
prefix: b"partial",
prefix_read: if after_exit { ready } else { Gate::default() },
fail_ready,
};
let command = start_with(
"injected".into(),
CommandTelemetry::new(),
vec![],
move |cancel, sender| async move {
let telemetry = CommandTelemetry::new();
if wait_fault {
supervise(
process,
&b"partial"[..],
&b""[..],
10,
100,
"injected",
&telemetry,
cancel,
&sender,
)
.await
} else {
match stream {
CommandStream::Stdout => {
supervise(
process,
reader,
&b""[..],
10,
100,
"injected",
&telemetry,
cancel,
&sender,
)
.await
}
CommandStream::Stderr => {
supervise(
process,
&b""[..],
reader,
10,
100,
"injected",
&telemetry,
cancel,
&sender,
)
.await
}
}
}
},
);
session
.inner
.commands
.lock()
.await
.insert("injected".into(), command);
let app = Router::new()
.route(
"/v1/working-directories/{id}/sessions",
post(|| async {
Json(OpenWorkdirSessionResponse {
session_id: WorkdirSessionId::new("fault-session").unwrap(),
workdir_id: crate::Workdir::new("fault-workdir").id().clone(),
capabilities: crate::WorkdirSessionCapabilities::ALL,
})
}),
)
.route("/v1/workdir-sessions/{id}/operations", post(operation))
.with_state(session.clone());
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let _server = Server(tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
}));
let remote: crate::WorkdirSessionHandle = Arc::new(
RemoteWorkdirSession::open(
reqwest::Client::builder().no_proxy().build().unwrap(),
base.parse().unwrap(),
"fixture-token",
crate::Workdir::new("fault-workdir").id().clone(),
OpenWorkdirSessionRequest::default(),
)
.await
.unwrap(),
);
let handle = CommandHandle("injected".into());
let mut request = CommandOutputRequest {
handle,
cursor: 0,
limit: 1,
wait: true,
};
let mut bytes = Vec::new();
let terminal = loop {
let page = remote.command_output(request.clone()).await.unwrap();
assert_eq!(remote.command_output(request.clone()).await.unwrap(), page);
for chunk in &page.chunks {
assert_eq!(chunk.sequence, request.cursor);
assert_eq!(chunk.offset, request.cursor);
assert_eq!(chunk.stream, stream);
bytes.extend_from_slice(&chunk.bytes);
request.cursor += chunk.bytes.len() as u64;
}
assert_eq!(page.next_cursor, request.cursor);
if let Some(terminal) = page.terminal {
break terminal;
}
};
assert_eq!(bytes, b"partial");
assert_eq!(terminal.truncation_reason, None);
assert_eq!(terminal.stdout_bytes + terminal.stderr_bytes, 7);
assert_eq!(
terminal.stdout_delivered_bytes + terminal.stderr_delivered_bytes,
7
);
if wait_fault {
assert_eq!(terminal.status, CommandStatus::Failed);
assert_eq!(terminal.exit_code, None);
assert_eq!(
terminal.process_failure,
Some(CommandFailure {
stage: CommandFailureStage::Wait,
reason: CommandIoErrorKind::InvalidData,
os_code: None
})
);
assert_eq!(
terminal.output_collection,
CommandOutputCollection::Complete
);
} else {
assert_eq!(terminal.status, CommandStatus::Completed);
assert_eq!(terminal.exit_code, Some(0));
let stage = match (stream, after_exit) {
(CommandStream::Stdout, false) => CommandFailureStage::StdoutRead,
(CommandStream::Stderr, false) => CommandFailureStage::StderrRead,
(CommandStream::Stdout, true) => CommandFailureStage::StdoutDrain,
(CommandStream::Stderr, true) => CommandFailureStage::StderrDrain,
};
assert_eq!(
terminal.output_collection,
CommandOutputCollection::Incomplete {
failure: CommandFailure {
stage,
reason: CommandIoErrorKind::PermissionDenied,
os_code: None
}
}
);
}
let wire = serde_json::to_string(&terminal).unwrap();
for secret in [
"/private",
"TOKEN",
"secret",
"command-body",
"fixture-token",
] {
assert!(!wire.contains(secret));
}
session.close().await.unwrap();
}
})
.await
.expect("bounded loopback fault projection");
}
crates/workdir/src/operation.rs
+257 -10
View File
@@ -132,12 +132,151 @@ pub enum CommandOutputTruncation {
ByteBudget,
}
/// Raw totals include discarded pipe bytes. Delivered totals count only prefixes
/// admitted to the transport queue, not page reads or retries by a consumer.
/// Closed diagnostic vocabulary: never serialize OS messages, paths, or errno text.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum CommandIoErrorKind {
NotFound,
PermissionDenied,
ConnectionRefused,
ConnectionReset,
ConnectionAborted,
NotConnected,
BrokenPipe,
AlreadyExists,
WouldBlock,
InvalidInput,
InvalidData,
TimedOut,
WriteZero,
Interrupted,
UnexpectedEof,
OutOfMemory,
HostUnreachable,
NetworkUnreachable,
AddrInUse,
AddrNotAvailable,
NetworkDown,
NotADirectory,
IsADirectory,
DirectoryNotEmpty,
ReadOnlyFilesystem,
StaleNetworkFileHandle,
StorageFull,
NotSeekable,
QuotaExceeded,
FileTooLarge,
ResourceBusy,
ExecutableFileBusy,
Deadlock,
CrossesDevices,
TooManyLinks,
InvalidFilename,
ArgumentListTooLong,
Unsupported,
Other,
Unknown,
}
impl From<std::io::ErrorKind> for CommandIoErrorKind {
fn from(kind: std::io::ErrorKind) -> Self {
use std::io::ErrorKind as Io;
match kind {
Io::NotFound => Self::NotFound,
Io::PermissionDenied => Self::PermissionDenied,
Io::ConnectionRefused => Self::ConnectionRefused,
Io::ConnectionReset => Self::ConnectionReset,
Io::ConnectionAborted => Self::ConnectionAborted,
Io::NotConnected => Self::NotConnected,
Io::BrokenPipe => Self::BrokenPipe,
Io::AlreadyExists => Self::AlreadyExists,
Io::WouldBlock => Self::WouldBlock,
Io::InvalidInput => Self::InvalidInput,
Io::InvalidData => Self::InvalidData,
Io::TimedOut => Self::TimedOut,
Io::WriteZero => Self::WriteZero,
Io::Interrupted => Self::Interrupted,
Io::UnexpectedEof => Self::UnexpectedEof,
Io::OutOfMemory => Self::OutOfMemory,
Io::HostUnreachable => Self::HostUnreachable,
Io::NetworkUnreachable => Self::NetworkUnreachable,
Io::AddrInUse => Self::AddrInUse,
Io::AddrNotAvailable => Self::AddrNotAvailable,
Io::NetworkDown => Self::NetworkDown,
Io::NotADirectory => Self::NotADirectory,
Io::IsADirectory => Self::IsADirectory,
Io::DirectoryNotEmpty => Self::DirectoryNotEmpty,
Io::ReadOnlyFilesystem => Self::ReadOnlyFilesystem,
Io::StaleNetworkFileHandle => Self::StaleNetworkFileHandle,
Io::StorageFull => Self::StorageFull,
Io::NotSeekable => Self::NotSeekable,
Io::QuotaExceeded => Self::QuotaExceeded,
Io::FileTooLarge => Self::FileTooLarge,
Io::ResourceBusy => Self::ResourceBusy,
Io::ExecutableFileBusy => Self::ExecutableFileBusy,
Io::Deadlock => Self::Deadlock,
Io::CrossesDevices => Self::CrossesDevices,
Io::TooManyLinks => Self::TooManyLinks,
Io::InvalidFilename => Self::InvalidFilename,
Io::ArgumentListTooLong => Self::ArgumentListTooLong,
Io::Unsupported => Self::Unsupported,
Io::Other => Self::Other,
_ => Self::Unknown,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum CommandFailureStage {
Spawn,
Wait,
StdoutRead,
StderrRead,
StdoutDrain,
StderrDrain,
OutputDelivery,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CommandFailure {
pub stage: CommandFailureStage,
pub reason: CommandIoErrorKind,
/// Observed numeric OS code, if supplied by the source. Never inferred from text.
pub os_code: Option<i32>,
}
impl CommandFailure {
pub(crate) fn io(stage: CommandFailureStage, error: &std::io::Error) -> Self {
Self {
stage,
reason: error.kind().into(),
os_code: error.raw_os_error(),
}
}
}
/// EOF collection is independent of exit status and the explicit byte budget.
/// Incomplete counts are lower bounds on bytes actually produced by the process.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)]
pub enum CommandOutputCollection {
NotStarted,
Complete,
Incomplete { failure: CommandFailure },
}
/// Raw totals include discarded *observed* pipe bytes, not an unknown failed tail.
/// Delivered totals count only prefixes admitted to the transport queue, not page
/// reads or retries. Neither terminal arrival nor exit zero implies EOF collection.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct CommandTerminal {
pub status: CommandStatus,
pub exit_code: Option<i32>,
pub process_failure: Option<CommandFailure>,
pub output_collection: CommandOutputCollection,
pub stdout_bytes: u64,
pub stderr_bytes: u64,
pub stdout_delivered_bytes: u64,
@@ -145,6 +284,54 @@ pub struct CommandTerminal {
pub truncation_reason: Option<CommandOutputTruncation>,
}
impl CommandTerminal {
/// Validate provider-authored metadata before publishing it as collected evidence.
pub fn metadata_valid(&self) -> bool {
let truncated = self.stdout_bytes > self.stdout_delivered_bytes
|| self.stderr_bytes > self.stderr_delivered_bytes;
if self.status == CommandStatus::Running
|| self.stdout_delivered_bytes > self.stdout_bytes
|| self.stderr_delivered_bytes > self.stderr_bytes
|| self.truncation_reason != truncated.then_some(CommandOutputTruncation::ByteBudget)
{
return false;
}
match self.process_failure {
Some(CommandFailure {
stage: CommandFailureStage::Spawn,
..
}) => {
self.status == CommandStatus::Failed
&& self.exit_code.is_none()
&& self.output_collection == CommandOutputCollection::NotStarted
&& self.stdout_bytes == 0
&& self.stderr_bytes == 0
}
Some(CommandFailure {
stage: CommandFailureStage::Wait,
..
}) => {
self.status != CommandStatus::Completed
&& self.exit_code.is_none()
&& self.collection_started()
}
Some(_) => false,
None => self.collection_started(),
}
}
fn collection_started(&self) -> bool {
match self.output_collection {
CommandOutputCollection::NotStarted => false,
CommandOutputCollection::Complete => true,
CommandOutputCollection::Incomplete { failure } => !matches!(
failure.stage,
CommandFailureStage::Spawn | CommandFailureStage::Wait
),
}
}
}
/// Sending next_cursor acknowledges this page. Only the current page may be
/// retried; acknowledged bytes expire. Terminal fences all captured bytes,
/// including on timeout/cancel. Consumers must pull while the process runs.
@@ -186,19 +373,13 @@ impl CommandOutput {
self.next_cursor <= MAX_COMMAND_TRANSPORT_BYTES
&& end.is_none_or(|end| end == self.next_cursor)
&& self.terminal.as_ref().is_none_or(|terminal| {
let truncated = terminal.stdout_bytes > terminal.stdout_delivered_bytes
|| terminal.stderr_bytes > terminal.stderr_delivered_bytes;
stream_ends[0].is_none_or(|end| end == terminal.stdout_delivered_bytes)
terminal.metadata_valid()
&& stream_ends[0].is_none_or(|end| end == terminal.stdout_delivered_bytes)
&& stream_ends[1].is_none_or(|end| end == terminal.stderr_delivered_bytes)
&& terminal.status != CommandStatus::Running
&& terminal.stdout_delivered_bytes <= terminal.stdout_bytes
&& terminal.stderr_delivered_bytes <= terminal.stderr_bytes
&& terminal
.stdout_delivered_bytes
.checked_add(terminal.stderr_delivered_bytes)
== Some(self.next_cursor)
&& terminal.truncation_reason
== truncated.then_some(CommandOutputTruncation::ByteBudget)
})
}
}
@@ -262,4 +443,70 @@ mod tests {
.contains(&serde_json::json!("transport_byte_budget"))
);
}
#[test]
fn failure_wire_is_closed_and_does_not_infer_os_codes_from_custom_messages() {
let error = std::io::Error::new(
std::io::ErrorKind::Other,
"/host/private TOKEN=secret EACCES command",
);
let failure = CommandFailure::io(CommandFailureStage::Wait, &error);
assert_eq!(
serde_json::to_value(failure).unwrap(),
serde_json::json!({"stage":"wait","reason":"other","os_code":null})
);
let os_error = std::io::Error::from_raw_os_error(libc::EIO);
let actual = CommandFailure::io(CommandFailureStage::StdoutDrain, &os_error);
assert_eq!(actual.os_code, Some(libc::EIO));
for invalid in [
serde_json::json!({"stage":"wait","reason":"/host/private TOKEN=secret"}),
serde_json::json!({"stage":"/host/private","reason":"other"}),
serde_json::json!({"stage":"wait","reason":"other","message":"secret"}),
] {
assert!(serde_json::from_value::<CommandFailure>(invalid).is_err());
}
}
#[test]
fn provider_terminal_rejects_inconsistent_failure_domains_and_unstarted_byte_claims() {
let wire = serde_json::json!({"status":"completed","exit_code":0,"process_failure":null,"output_collection":{"state":"incomplete","failure":{"stage":"stdout_drain","reason":"other","os_code":null}},"stdout_bytes":3,"stderr_bytes":0,"stdout_delivered_bytes":3,"stderr_delivered_bytes":0,"truncation_reason":null});
let terminal: CommandTerminal = serde_json::from_value(wire.clone()).unwrap();
assert!(terminal.metadata_valid());
for (field, value) in [
(
"output_collection",
serde_json::json!({"state":"not_started"}),
),
(
"output_collection",
serde_json::json!({"state":"incomplete","failure":{"stage":"wait","reason":"other"}}),
),
(
"process_failure",
serde_json::json!({"stage":"stdout_read","reason":"other"}),
),
(
"process_failure",
serde_json::json!({"stage":"spawn","reason":"not_found"}),
),
("truncation_reason", serde_json::json!("byte_budget")),
] {
let mut invalid = wire.clone();
invalid[field] = value;
assert!(
!serde_json::from_value::<CommandTerminal>(invalid)
.unwrap()
.metadata_valid()
);
}
let mut unclassified = wire.clone();
unclassified
.as_object_mut()
.unwrap()
.remove("output_collection");
assert!(serde_json::from_value::<CommandTerminal>(unclassified).is_err());
let mut raw = wire;
raw["message"] = serde_json::json!("/private TOKEN=secret");
assert!(serde_json::from_value::<CommandTerminal>(raw).is_err());
}
}