fix(workdir): preserve mutation and command safety through diagnostics

This commit is contained in:
2026-10-10 00:00:43 +09:00
parent 6ac87c885e
commit fd759545c1
9 changed files with 497 additions and 12 deletions
+16 -3
View File
@@ -716,10 +716,23 @@ mod client {
"mismatched response after checkout mutation".into(),
)),
Ok(_) => Err(Self::mismatch("checkout_execute")),
Err(WorkdirError::Unavailable(_) | WorkdirError::Transport(_)) if mutation => {
Err(WorkdirError::OutcomeUnknown(
Err(error)
if mutation
&& matches!(
error.classification_source(),
WorkdirError::Unavailable(_) | WorkdirError::Transport(_)
) =>
{
let outcome = WorkdirError::OutcomeUnknown(
"checkout transport failed; effects may have occurred".into(),
))
);
Err(match error.denial_reason() {
Some(reason) => WorkdirError::DenialContext {
reason,
source: Box::new(outcome),
},
None => outcome,
})
}
Err(error) => Err(error),
}
+6 -1
View File
@@ -166,7 +166,12 @@ impl RoutedSessionState {
for handle in handles {
match self.session.command_status(handle.clone()).await {
Ok(status) => self.mark_terminal(&handle, status),
Err(WorkdirError::UnknownCommand(_)) => {
Err(error)
if matches!(
error.classification_source(),
WorkdirError::UnknownCommand(_)
) =>
{
self.active_commands
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
+12 -2
View File
@@ -240,8 +240,18 @@ impl WorkdirScopeLease {
.expect("scoped command set mutex poisoned")
.remove(&command_id);
}
(Ok(()), Err(WorkdirError::UnknownCommand(_)))
| (Err(WorkdirError::UnknownCommand(_)), Err(WorkdirError::UnknownCommand(_))) => {
(cancel, Err(error))
if matches!(
error.classification_source(),
WorkdirError::UnknownCommand(_)
) && match &cancel {
Ok(()) => true,
Err(error) => matches!(
error.classification_source(),
WorkdirError::UnknownCommand(_)
),
} =>
{
self.broker.authority.publish_terminal_if_missing(
&command_id,
CommandStatus::Cancelled,
@@ -0,0 +1,185 @@
//! Wire diagnostics must not change command disappearance or lease release.
//! Fake command handles only; no OS process is started, polled, or cancelled.
use async_trait::async_trait;
use std::{path::Path, sync::Arc};
use tempfile::TempDir;
use workdir::*;
#[derive(Debug)]
struct CommandFailureSession {
local: LocalWorkdirSession,
code: &'static str,
depth: u8,
cancel_ok: bool,
}
impl CommandFailureSession {
fn new(root: &Path, code: &'static str, depth: u8, cancel_ok: bool) -> Arc<Self> {
Arc::new(Self {
local: LocalWorkdirSession::new(
manifest::Scope::writable(root).unwrap(),
root.to_path_buf(),
),
code,
depth,
cancel_ok,
})
}
fn failure(&self) -> WorkdirError {
let mut error = serde_json::from_value::<http::WorkdirTransportError>(serde_json::json!({
"code": self.code, "message": "safe fixture error",
"denial_reason": (self.depth > 0).then_some(WorkdirDenialReason::ReadOnlySession),
}))
.unwrap()
.into_workdir_error();
if self.depth == 2 {
error = WorkdirError::DenialContext {
reason: WorkdirDenialReason::OsPermissionDenied,
source: Box::new(error),
};
}
error
}
}
#[async_trait]
impl WorkdirSession for CommandFailureSession {
fn workdir(&self) -> &Workdir {
self.local.workdir()
}
fn capabilities(&self) -> WorkdirSessionCapabilities {
self.local.capabilities()
}
async fn authorize_scope_path(
&self,
r: WorkdirScopeAuthorizationRequest,
) -> Result<(), WorkdirError> {
self.local.authorize_scope_path(r).await
}
async fn scope_rules_overlap(
&self,
r: WorkdirScopeOverlapRequest,
) -> Result<bool, WorkdirError> {
self.local.scope_rules_overlap(r).await
}
async fn stat(&self, r: StatRequest) -> Result<StatResult, WorkdirError> {
self.local.stat(r).await
}
async fn read(&self, r: ReadRequest) -> Result<ReadResult, WorkdirError> {
self.local.read(r).await
}
async fn write(&self, r: WriteRequest) -> Result<WriteResult, WorkdirError> {
self.local.write(r).await
}
async fn edit(&self, r: EditRequest) -> Result<EditResult, WorkdirError> {
self.local.edit(r).await
}
async fn list(&self, r: ListRequest) -> Result<ListResult, WorkdirError> {
self.local.list(r).await
}
async fn glob(&self, r: GlobRequest) -> Result<GlobResult, WorkdirError> {
self.local.glob(r).await
}
async fn grep(&self, r: GrepRequest) -> Result<GrepResult, WorkdirError> {
self.local.grep(r).await
}
async fn start_command(&self, _: CommandRequest) -> Result<CommandHandle, WorkdirError> {
Ok(CommandHandle("fake-command".into()))
}
async fn command_status(&self, _: CommandHandle) -> Result<CommandStatus, WorkdirError> {
Err(self.failure())
}
async fn command_output(&self, _: CommandOutputRequest) -> Result<CommandOutput, WorkdirError> {
Err(self.failure())
}
async fn cancel_command(&self, _: CommandHandle) -> Result<(), WorkdirError> {
if self.cancel_ok {
Ok(())
} else {
Err(self.failure())
}
}
async fn close(&self) -> Result<(), WorkdirError> {
self.local.close().await
}
}
fn command() -> CommandRequest {
CommandRequest {
command: "fake command, never executed".into(),
timeout_secs: 1,
output_limit: 100,
cwd: WorkdirPath::root(),
spill_dir: None,
tool_call_id: None,
}
}
fn scope() -> WorkdirToolScope {
WorkdirToolScope {
rules: vec![WorkdirToolScopeRule {
target: WorkdirPath::root(),
permission: WorkdirToolScopePermission::Write,
recursive: true,
symlink_policy: Default::default(),
}],
command: true,
cwd: WorkdirPath::root(),
}
}
#[tokio::test]
async fn command_diagnostics_preserve_detach_disappearance_and_busy_guard() {
for code in ["unknown_command", "unavailable"] {
for depth in [0, 1, 2] {
let dir = TempDir::new().unwrap();
let router = WorkdirSessionRouter::new();
let alias = WorkdirAttachmentAlias::new("fixture").unwrap();
router
.attach(
alias.clone(),
CommandFailureSession::new(dir.path(), code, depth, false),
)
.unwrap();
let session = router.resolve(Some("fixture")).unwrap().session;
session.start_command(command()).await.unwrap();
let result = router.detach(&alias).await;
if code == "unknown_command" {
result.unwrap_or_else(|error| panic!("depth {depth}: {error:?}"));
assert!(router.resolve(Some("fixture")).is_err());
} else {
assert!(matches!(result, Err(WorkdirError::Conflict(_))));
assert!(router.resolve(Some("fixture")).is_ok());
}
}
}
}
#[tokio::test]
async fn command_diagnostics_preserve_scope_close_and_write_lease_guard() {
for code in ["unknown_command", "unavailable"] {
for depth in [0, 1, 2] {
for cancel_ok in [false, true] {
let dir = TempDir::new().unwrap();
let broker = WorkdirToolBroker::new(CommandFailureSession::new(
dir.path(),
code,
depth,
cancel_ok,
));
let child = broker.scope(scope()).await.unwrap();
child.start_command(command()).await.unwrap();
let result = child.close().await;
if code == "unknown_command" {
result.unwrap_or_else(|error| {
panic!("depth {depth}, cancel_ok={cancel_ok}: {error:?}")
});
assert!(!child.is_active());
broker.scope(scope()).await.unwrap().close().await.unwrap();
} else {
assert!(result.is_err());
// Closing revokes new work immediately, but failed cleanup
// must retain the write lease until ownership is resolved.
assert!(!child.is_active());
assert!(broker.scope(scope()).await.is_err());
}
}
}
}
}
+110 -2
View File
@@ -4,7 +4,7 @@ use std::{
io::{Read, Write},
net::{TcpListener, TcpStream},
thread,
time::Duration,
time::{Duration, Instant},
};
use workdir::{
StatRequest, Workdir, WorkdirDenialReason as Reason, WorkdirPath, WorkdirSession,
@@ -15,7 +15,18 @@ use workdir::{
};
fn reply(listener: &TcpListener, status: &str, body: &str) {
let (mut stream, _) = listener.accept().unwrap();
listener.set_nonblocking(true).unwrap();
let deadline = Instant::now() + Duration::from_secs(5);
let (mut stream, _) = loop {
match listener.accept() {
Ok(connection) => break connection,
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
assert!(Instant::now() < deadline, "fake provider accept timed out");
thread::yield_now();
}
Err(error) => panic!("fake provider accept failed: {error}"),
}
};
stream
.set_read_timeout(Some(Duration::from_secs(5)))
.unwrap();
@@ -108,3 +119,100 @@ async fn remote_http_denials_forward_only_typed_reasons_and_old_missing_reason()
server.join().unwrap();
}
}
#[tokio::test]
async fn remote_checkout_diagnostics_preserve_mutation_outcome_and_read_failures() {
use workdir::{CheckoutOperation as Op, CheckoutRequest, WorkdirSessionCapability};
for code in [
Code::Unavailable,
Code::Transport,
Code::Denied,
Code::Conflict,
Code::OutcomeUnknown,
] {
for reason in [None, Some(Reason::ReadOnlySession)] {
for operation in [
Op::Read {
offset: 0,
limit: 1,
max_bytes: 10,
},
Op::Create {
path: WorkdirPath::new("new.txt").unwrap(),
content: b"fixture".to_vec(),
},
Op::Write {
content: b"fixture".to_vec(),
expected_hash: [0; 32],
},
Op::Edit {
old_string: "old".into(),
new_string: "new".into(),
replace_all: false,
expected_hash: [0; 32],
},
] {
let mutation = operation.capability() != WorkdirSessionCapability::Read;
let expected_code =
if mutation && matches!(code, Code::Unavailable | Code::Transport) {
Code::OutcomeUnknown
} else {
code
};
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let error = serde_json::json!({"code":code, "message":"/secret/host provider-secret", "denial_reason":reason});
let server = thread::spawn(move || {
reply(
&listener,
"200 OK",
r#"{"session_id":"s","workdir_id":"w","capabilities":{"bits":63}}"#,
);
reply(
&listener,
&format!("{} Fixture failure", code.http_status()),
&error.to_string(),
);
});
let remote = RemoteWorkdirSession::open(
reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()
.unwrap(),
format!("http://{address}").parse().unwrap(),
"request-secret",
Workdir::new("w").id().clone(),
OpenWorkdirSessionRequest::default(),
)
.await
.unwrap();
let error = remote
.checkout_execute(CheckoutRequest {
target: WorkdirPath::root(),
validator: vec![1],
operation,
})
.await
.unwrap_err();
server.join().unwrap();
let forwarded = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(
forwarded.code, expected_code,
"{code:?}, {reason:?}, mutation={mutation}: {error:?}"
);
assert_eq!(error.denial_reason(), reason);
assert_eq!(forwarded.denial_reason, reason);
assert_eq!(forwarded.code.http_status(), expected_code.http_status());
if expected_code == Code::OutcomeUnknown {
assert!(forwarded.message.contains("inspect before retrying"));
}
assert!(
!serde_json::to_string(&forwarded)
.unwrap()
.contains("secret")
);
}
}
}
}
+16 -4
View File
@@ -1225,12 +1225,23 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession {
"mismatched checkout mutation response".into(),
)),
Ok(_) => Err(Self::mismatch("checkout_execute")),
Err(workdir::WorkdirError::Unavailable(_) | workdir::WorkdirError::Transport(_))
if mutation =>
Err(error)
if mutation
&& matches!(
error.classification_source(),
workdir::WorkdirError::Unavailable(_) | workdir::WorkdirError::Transport(_)
) =>
{
Err(workdir::WorkdirError::OutcomeUnknown(
let outcome = workdir::WorkdirError::OutcomeUnknown(
"checkout provider response lost; inspect effects before retry".into(),
))
);
Err(match error.denial_reason() {
Some(reason) => workdir::WorkdirError::DenialContext {
reason,
source: Box::new(outcome),
},
None => outcome,
})
}
Err(error) => Err(error),
}
@@ -33033,6 +33044,7 @@ mod tests {
mod drive_api_tests;
mod drive_tests;
mod drive_web_roundtrip_tests;
mod external_checkout_tests;
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
mod ticket_evidence_tests;
@@ -0,0 +1,129 @@
//! External codec -> actual session dispatch -> mutation safety -> HTTP error.
//! Channel fixture only; no product process or real mutation.
use super::*;
use workdir::{
CheckoutOperation as Op, CheckoutRequest, WorkdirDenialReason as Reason, WorkdirPath,
WorkdirSession, WorkdirSessionCapability, external::ExternalWorkdirOperationOutcome,
http::WorkdirTransportErrorCode as Code,
};
#[tokio::test]
async fn external_checkout_diagnostics_preserve_mutation_outcome_and_read_failures() {
for code in [
Code::Unavailable,
Code::Transport,
Code::Denied,
Code::Conflict,
Code::OutcomeUnknown,
] {
for reason in [None, Some(Reason::ReadOnlySession)] {
for operation in [
Op::Read {
offset: 0,
limit: 1,
max_bytes: 10,
},
Op::Create {
path: WorkdirPath::new("new.txt").unwrap(),
content: b"fixture".to_vec(),
},
Op::Write {
content: b"fixture".to_vec(),
expected_hash: [0; 32],
},
Op::Edit {
old_string: "old".into(),
new_string: "new".into(),
replace_all: false,
expected_hash: [0; 32],
},
] {
let mutation = operation.capability() != WorkdirSessionCapability::Read;
let expected_code =
if mutation && matches!(code, Code::Unavailable | Code::Transport) {
Code::OutcomeUnknown
} else {
code
};
let (sender, mut commands) = tokio::sync::mpsc::channel(1);
let connection = Arc::new(ExternalProviderConnection {
grant_id: "fixture-grant".into(),
workdir_id: "fixture-workdir".into(),
provider_instance_id: "fixture-provider".into(),
generation: 1,
expires_at: None,
capabilities: workdir::WorkdirSessionCapabilities::ALL,
admission: Arc::new(tokio::sync::Semaphore::new(16)),
shutdown_confirmed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
sender,
});
let session = Arc::new(ExternalProviderWorkdirSession::new(connection, None));
let caller = tokio::spawn(async move {
session
.checkout_execute(CheckoutRequest {
target: WorkdirPath::root(),
validator: vec![1],
operation,
})
.await
});
let command =
tokio::time::timeout(std::time::Duration::from_secs(5), commands.recv())
.await
.unwrap()
.unwrap();
let ExternalProviderCommand::Operation {
operation,
response,
..
} = command
else {
panic!("expected dispatched operation");
};
assert!(matches!(
operation,
WorkdirSessionOperation::CheckoutExecute(_)
));
let failed = serde_json::from_value::<ExternalWorkdirOperationOutcome>(json!({
"outcome":"failed", "error":{"code":code, "denial_reason":reason},
}))
.unwrap();
let ExternalWorkdirOperationOutcome::Failed { error } = failed else {
panic!("expected fixture failure");
};
response.send(Err(error.into_transport_error())).unwrap();
let error = tokio::time::timeout(std::time::Duration::from_secs(5), caller)
.await
.unwrap()
.unwrap()
.unwrap_err();
assert!(
matches!(
commands.try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Empty
| tokio::sync::mpsc::error::TryRecvError::Disconnected)
),
"must not retry"
);
let forwarded = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(
forwarded.code, expected_code,
"{code:?}, {reason:?}, mutation={mutation}: {error:?}"
);
assert_eq!(error.denial_reason(), reason);
assert_eq!(forwarded.denial_reason, reason);
if expected_code == Code::OutcomeUnknown {
assert!(forwarded.message.contains("inspect before retrying"));
}
let response = InternalWorkdirOperationError::Provider(forwarded).into_response();
assert_eq!(response.status().as_u16(), expected_code.http_status());
let body = axum::body::to_bytes(response.into_body(), 2048)
.await
.unwrap();
let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(body["code"], expected_code.as_str());
assert!(body.get("denial_reason").is_none());
}
}
}
}