fix: preserve Workdir failure diagnostics and audit outcomes

This commit is contained in:
2026-10-22 00:16:11 +09:00
18 changed files with 2727 additions and 341 deletions
+7 -1
View File
@@ -125,6 +125,8 @@ pub struct WorkdirOperationApiError {
pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub diagnostics: Option<Vec<Diagnostic>>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub audit_failures: Vec<workdir::http::WorkdirAuditFailure>,
#[serde(skip, default = "default_repository_error_status")]
status: u16,
/// Server-local metadata retained by generated error response extensions, never the wire body.
@@ -156,6 +158,7 @@ impl WorkdirOperationApiError {
code: None,
message: error.message,
diagnostics: Some(error.diagnostics),
audit_failures: Vec::new(),
status: error.status,
workdir_log: None,
denial_reason: None,
@@ -170,6 +173,7 @@ impl WorkdirOperationApiError {
code: Some(error.code),
message: error.message,
diagnostics: None,
audit_failures: Vec::new(),
status,
workdir_log: None,
denial_reason: error.denial_reason,
@@ -7522,7 +7526,9 @@ pub struct CurrentWorkerWorkdirOperationRequest {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(transparent)]
pub struct CurrentWorkerWorkdirOperationResponse(pub workdir::http::WorkdirSessionOperationResult);
pub struct CurrentWorkerWorkdirOperationResponse(
pub workdir::http::WorkdirSessionOperationResponse,
);
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
+1
View File
@@ -167,6 +167,7 @@ pub(crate) fn checked_error(error: workdir::WorkdirError) -> ToolError {
| WorkdirError::IsDirectory { .. }
| WorkdirError::InvalidGlob { .. }
| WorkdirError::InvalidRegex { .. } => "checkout_invalid",
WorkdirError::NotDispatched => "checkout_not_dispatched",
WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable",
// IO/transport failures may arrive after effects have begun. Do not infer
// no-effects from their text or strip explicit OutcomeUnknown evidence.
+1 -3
View File
@@ -556,15 +556,13 @@ mod tests {
));
}
#[tokio::test]
async fn checkout_hardlink_save_isolates_alias_and_rejects_symlinks_special_files() {
async fn checkout_hardlink_save_isolates_alias_and_rejects_symlinks() {
let root = tempfile::tempdir().unwrap();
std::fs::write(root.path().join("a"), "original").unwrap();
std::fs::hard_link(root.path().join("a"), root.path().join("alias")).unwrap();
std::os::unix::fs::symlink("a", root.path().join("link")).unwrap();
let _socket = std::os::unix::net::UnixListener::bind(root.path().join("socket")).unwrap();
let session = session(root.path());
assert!(session.checkout_observe(path("link")).await.is_err());
assert!(session.checkout_observe(path("socket")).await.is_err());
let read = checked_read(&session, session.checkout_observe(path("a")).await.unwrap()).await;
session
.checkout_execute(CheckoutRequest {
+101 -9
View File
@@ -119,6 +119,51 @@ pub enum WorkdirSessionOperationResult {
CommandCancel,
}
/// Audit persistence is independent of the provider result and never authorizes a resend.
/// Only server-generated correlation and closed classifications cross this boundary.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(deny_unknown_fields)]
pub struct WorkdirAuditFailure {
pub operation_id: String,
pub stage: WorkdirAuditStage,
pub operation_outcome: WorkdirAuditOperationOutcome,
pub reason: WorkdirAuditFailureReason,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum WorkdirAuditStage {
BeforeDispatch,
DispatchRejected,
ResponseReceived,
ResponseUnknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(tag = "kind", rename_all = "snake_case", deny_unknown_fields)]
pub enum WorkdirAuditOperationOutcome {
NotDispatched,
Succeeded,
Failed { code: WorkdirTransportErrorCode },
Unknown,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum WorkdirAuditFailureReason {
AuditStoreUnavailable,
}
/// A successful operation can still have failed audit persistence. Consumers must
/// surface that degradation separately, without discarding the result or resending.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct WorkdirSessionOperationResponse {
#[serde(flatten)]
pub result: WorkdirSessionOperationResult,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub audit_failures: Vec<WorkdirAuditFailure>,
}
/// Dispatch one typed operation against any provider-neutral Workdir session.
///
/// Runtime HTTP handlers, outbound External providers, and Backend brokers can
@@ -208,6 +253,7 @@ pub enum WorkdirTransportErrorCode {
NotFound,
Conflict,
OutcomeUnknown,
NotDispatched,
Unsupported,
InvalidRequest,
Denied,
@@ -231,6 +277,7 @@ impl WorkdirTransportErrorCode {
Self::NotFound => "not_found",
Self::Conflict => "conflict",
Self::OutcomeUnknown => "outcome_unknown",
Self::NotDispatched => "not_dispatched",
Self::Unsupported => "unsupported",
Self::InvalidRequest => "invalid_request",
Self::Denied => "denied",
@@ -261,7 +308,7 @@ impl WorkdirTransportErrorCode {
| Self::SymlinkTargetIsDirectory
| Self::IsDirectory
| Self::SymlinkDirectoryNotTraversed => 400,
Self::Unavailable => 503,
Self::Unavailable | Self::NotDispatched => 503,
Self::Io | Self::Internal | Self::OutcomeUnknown => 500,
Self::Transport => 502,
}
@@ -298,6 +345,9 @@ impl WorkdirTransportError {
Code::OutcomeUnknown,
"Workdir operation may have effects; inspect before retrying",
),
WorkdirError::NotDispatched => {
(Code::NotDispatched, "Workdir operation was not dispatched")
}
WorkdirError::NotFound(_) => (Code::NotFound, "Workdir path was not found"),
WorkdirError::Conflict(_) => (
Code::Conflict,
@@ -375,12 +425,14 @@ impl WorkdirTransportError {
use WorkdirTransportErrorCode as Code;
let error = match self.code {
Code::NotFound => WorkdirError::NotFound("<remote>".into()),
Code::Conflict => WorkdirError::Conflict(self.message),
Code::Conflict => WorkdirError::Conflict(
"The target file's content or existence changed since it was last observed; read the file again before retrying".into(),
),
Code::OutcomeUnknown => WorkdirError::OutcomeUnknown(
"Workdir operation may have effects; inspect before retrying".into(),
),
Code::Unsupported => WorkdirError::UnsupportedOperation(self.message),
Code::InvalidRequest => WorkdirError::InvalidArgument(self.message),
Code::Unsupported => WorkdirError::UnsupportedOperation("Workdir operation is not supported".into()),
Code::InvalidRequest => WorkdirError::InvalidArgument("Workdir operation request is invalid".into()),
Code::Denied => WorkdirError::Denied(crate::WorkdirDenial {
reason: self.denial_reason,
// Provider-authored text is not trusted diagnostic data.
@@ -409,12 +461,13 @@ impl WorkdirTransportError {
target: "<remote-target>".into(),
},
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
Code::Unavailable => WorkdirError::Unavailable(self.message),
Code::Unavailable => WorkdirError::Unavailable("Workdir session is unavailable".into()),
Code::NotDispatched => WorkdirError::NotDispatched,
Code::Io => WorkdirError::Io {
path: "<remote>".into(),
source: std::io::Error::other(self.message),
source: std::io::Error::other("Workdir I/O operation failed"),
},
Code::Transport => WorkdirError::Transport(self.message),
Code::Transport => WorkdirError::Transport("Workdir transport failed".into()),
Code::Internal => WorkdirError::OperationFailed,
};
let error = if let Some(reason) = self.failure_reason {
@@ -549,7 +602,8 @@ mod client {
if opened.workdir_id.as_str() != workdir_id.as_str() {
return Err(WorkdirError::Unavailable(
"Runtime opened a session for a different Workdir".to_string(),
));
)
.with_failure_reason(crate::WorkdirFailureReason::SessionIdentityMismatch));
}
Ok(Self {
client,
@@ -608,6 +662,7 @@ mod client {
WorkdirError::Unavailable(format!(
"Runtime returned a mismatched Workdir operation result; expected {expected}"
))
.with_failure_reason(crate::WorkdirFailureReason::OperationResultMismatch)
}
}
@@ -770,7 +825,8 @@ mod client {
}
Ok(_) if mutation => Err(WorkdirError::OutcomeUnknown(
"mismatched response after checkout mutation".into(),
)),
)
.with_failure_reason(crate::WorkdirFailureReason::OperationResultMismatch)),
Ok(_) => Err(Self::mismatch("checkout_execute")),
Err(error)
if mutation
@@ -1210,6 +1266,37 @@ pub use client::{
mod tests {
use super::*;
#[test]
fn operation_response_keeps_success_and_independent_audit_failure() {
let original =
WorkdirSessionOperationResult::CommandStart(crate::CommandHandle("command-a".into()));
let response = WorkdirSessionOperationResponse {
result: original.clone(),
audit_failures: vec![WorkdirAuditFailure {
operation_id: "op-00000000-0000-7000-8000-000000000001".into(),
stage: WorkdirAuditStage::ResponseReceived,
operation_outcome: WorkdirAuditOperationOutcome::Succeeded,
reason: WorkdirAuditFailureReason::AuditStoreUnavailable,
}],
};
let json = serde_json::to_value(&response).unwrap();
assert_eq!(json["operation"], "command_start");
assert_eq!(json["result"], "command-a");
assert_eq!(
json["audit_failures"][0]["operation_outcome"]["kind"],
"succeeded"
);
assert_eq!(
serde_json::from_value::<WorkdirSessionOperationResponse>(json).unwrap(),
response
);
let normal = serde_json::to_value(&original).unwrap();
let decoded: WorkdirSessionOperationResponse =
serde_json::from_value(normal.clone()).unwrap();
assert!(decoded.audit_failures.is_empty());
assert_eq!(serde_json::to_value(decoded).unwrap(), normal);
}
#[test]
fn transport_error_round_trip_keeps_public_classification() {
for (code, expected_status, expected_error) in [
@@ -1229,6 +1316,11 @@ mod tests {
409,
"The target file's content or existence changed since it was last observed",
),
(
WorkdirTransportErrorCode::NotDispatched,
503,
"not dispatched",
),
(WorkdirTransportErrorCode::Unsupported, 400, "unsupported"),
(WorkdirTransportErrorCode::Denied, 403, "denied"),
(
+123
View File
@@ -408,12 +408,36 @@ impl From<&str> for WorkdirDenial {
pub enum WorkdirFailureReason {
SessionClosed,
ProviderUnavailable,
ProviderGrantExpired,
ProviderAdmissionClosed,
ProviderDispatchRejected,
ProviderResponseLost,
ProviderOperationTimeout,
CommandTerminalMissing,
AuditStoreUnavailable,
HttpTimeout,
HttpConnect,
HttpTransfer,
HttpResponseInvalid,
HttpResponseTooLarge,
OperationResultMismatch,
SessionIdentityMismatch,
SessionOpenTaskFailed,
SessionOpenFailed,
AuthorizationMissing,
AuthorizationStale,
AuthorizationUnavailable,
SigningIdentityUnavailable,
AuthorizationSignFailed,
RuntimeStopped,
RuntimeStateUnavailable,
RuntimeBackendUnavailable,
RuntimeOwnerMismatch,
RuntimeInvalidRequest,
WorkingDirectoryNotFound,
WorkingDirectoryNotActive,
WorkingDirectoryRecordInvalid,
WorkingDirectoryAccessUnavailable,
}
impl WorkdirFailureReason {
@@ -421,12 +445,107 @@ impl WorkdirFailureReason {
match self {
Self::SessionClosed => "session_closed",
Self::ProviderUnavailable => "provider_unavailable",
Self::ProviderGrantExpired => "provider_grant_expired",
Self::ProviderAdmissionClosed => "provider_admission_closed",
Self::ProviderDispatchRejected => "provider_dispatch_rejected",
Self::ProviderResponseLost => "provider_response_lost",
Self::ProviderOperationTimeout => "provider_operation_timeout",
Self::CommandTerminalMissing => "command_terminal_missing",
Self::AuditStoreUnavailable => "audit_store_unavailable",
Self::HttpTimeout => "http_timeout",
Self::HttpConnect => "http_connect",
Self::HttpTransfer => "http_transfer",
Self::HttpResponseInvalid => "http_response_invalid",
Self::HttpResponseTooLarge => "http_response_too_large",
Self::OperationResultMismatch => "operation_result_mismatch",
Self::SessionIdentityMismatch => "session_identity_mismatch",
Self::SessionOpenTaskFailed => "session_open_task_failed",
Self::SessionOpenFailed => "session_open_failed",
Self::AuthorizationMissing => "authorization_missing",
Self::AuthorizationStale => "authorization_stale",
Self::AuthorizationUnavailable => "authorization_unavailable",
Self::SigningIdentityUnavailable => "signing_identity_unavailable",
Self::AuthorizationSignFailed => "authorization_sign_failed",
Self::RuntimeStopped => "runtime_stopped",
Self::RuntimeStateUnavailable => "runtime_state_unavailable",
Self::RuntimeBackendUnavailable => "runtime_backend_unavailable",
Self::RuntimeOwnerMismatch => "runtime_owner_mismatch",
Self::RuntimeInvalidRequest => "runtime_invalid_request",
Self::WorkingDirectoryNotFound => "working_directory_not_found",
Self::WorkingDirectoryNotActive => "working_directory_not_active",
Self::WorkingDirectoryRecordInvalid => "working_directory_record_invalid",
Self::WorkingDirectoryAccessUnavailable => "working_directory_access_unavailable",
}
}
/// The observed failing boundary, not a dispatch or effects assertion.
pub const fn stage(self) -> WorkdirFailureStage {
use WorkdirFailureStage as Stage;
match self {
Self::SessionClosed => Stage::Session,
Self::ProviderUnavailable => Stage::Provider,
Self::ProviderGrantExpired | Self::ProviderAdmissionClosed => Stage::ProviderAdmission,
Self::ProviderDispatchRejected => Stage::ProviderDispatch,
Self::ProviderResponseLost | Self::ProviderOperationTimeout => Stage::ProviderResponse,
Self::AuditStoreUnavailable => Stage::Audit,
Self::CommandTerminalMissing => Stage::Command,
Self::HttpTimeout | Self::HttpConnect | Self::HttpTransfer => Stage::Http,
Self::HttpResponseInvalid
| Self::HttpResponseTooLarge
| Self::OperationResultMismatch => Stage::ResponseValidation,
Self::SessionIdentityMismatch
| Self::SessionOpenTaskFailed
| Self::SessionOpenFailed
| Self::WorkingDirectoryNotFound
| Self::WorkingDirectoryNotActive
| Self::WorkingDirectoryRecordInvalid
| Self::WorkingDirectoryAccessUnavailable => Stage::SessionOpen,
Self::AuthorizationMissing
| Self::AuthorizationStale
| Self::AuthorizationUnavailable
| Self::SigningIdentityUnavailable
| Self::AuthorizationSignFailed => Stage::AuthorizationPreparation,
Self::RuntimeStopped
| Self::RuntimeStateUnavailable
| Self::RuntimeBackendUnavailable
| Self::RuntimeOwnerMismatch
| Self::RuntimeInvalidRequest => Stage::Runtime,
}
}
}
/// Internal operational boundary derived only from a bounded reason.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum WorkdirFailureStage {
Session,
Provider,
ProviderAdmission,
ProviderDispatch,
ProviderResponse,
Audit,
Command,
Http,
ResponseValidation,
SessionOpen,
AuthorizationPreparation,
Runtime,
}
impl WorkdirFailureStage {
pub const fn as_str(self) -> &'static str {
match self {
Self::Session => "session",
Self::Provider => "provider",
Self::ProviderAdmission => "provider_admission",
Self::ProviderDispatch => "provider_dispatch",
Self::ProviderResponse => "provider_response",
Self::Audit => "audit",
Self::Command => "command",
Self::Http => "http",
Self::ResponseValidation => "response_validation",
Self::SessionOpen => "session_open",
Self::AuthorizationPreparation => "authorization_preparation",
Self::Runtime => "runtime",
}
}
}
@@ -466,6 +585,10 @@ pub enum WorkdirError {
#[error("Workdir session is unavailable: {0}")]
Unavailable(String),
/// The broker has proof that this operation never reached the provider.
#[error("Workdir operation was not dispatched")]
NotDispatched,
#[error("Workdir operation failed")]
OperationFailed,
@@ -83,3 +83,34 @@ fn provider_cannot_inject_free_text_failure_reason() {
.is_err()
);
}
#[test]
fn provider_authored_operational_messages_never_enter_error_display_or_debug() {
for code in [
Code::Conflict,
Code::OutcomeUnknown,
Code::Unsupported,
Code::InvalidRequest,
Code::Unavailable,
Code::Io,
Code::Transport,
Code::Internal,
] {
let wire = serde_json::json!({
"code": code, "message":"/host/private command-secret body-secret proof-secret key-secret Authorization: Bearer bearer-secret",
"failure_reason": "session_open_task_failed"
});
let decoded: WorkdirTransportError = serde_json::from_value(wire).unwrap();
let error = decoded.into_workdir_error();
for text in [
format!("{error}"),
format!("{error:?}"),
serde_json::to_string(&WorkdirTransportError::from_workdir_error(&error)).unwrap(),
] {
assert!(!text.contains("secret"), "{text}");
assert!(!text.contains("/host/private"), "{text}");
}
assert_eq!(error.failure_reason(), Some(Reason::SessionOpenTaskFailed));
assert_eq!(WorkdirTransportError::from_workdir_error(&error).code, code);
}
}
@@ -216,3 +216,106 @@ async fn remote_checkout_diagnostics_preserve_mutation_outcome_and_read_failures
}
}
}
#[tokio::test]
async fn mismatched_http_results_keep_cause_and_mutation_uncertainty() {
use workdir::{
CheckoutOperation, CheckoutRequest, CommandHandle, CommandOutputRequest,
WorkdirFailureReason as Failure,
};
for case in ["variant", "bytes", "mutation", "identity"] {
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
let address = listener.local_addr().unwrap();
let server = thread::spawn(move || {
reply(
&listener,
"200 OK",
if case == "identity" {
r#"{"session_id":"s","workdir_id":"wrong-secret","capabilities":{"bits":63}}"#
} else {
r#"{"session_id":"s","workdir_id":"w","capabilities":{"bits":63}}"#
},
);
if case != "identity" {
let result = if case == "bytes" {
serde_json::json!({"operation":"command_output","result":{
"chunks": [], "next_cursor": 123, "terminal": null
}})
} else {
serde_json::to_value(
workdir::http::WorkdirSessionOperationResult::CommandCancel,
)
.unwrap()
};
reply(&listener, "200 OK", &result.to_string());
}
});
let opened = 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;
let error = if case == "identity" {
opened.unwrap_err()
} else {
let remote = opened.unwrap();
match case {
"variant" => remote
.stat(StatRequest {
path: WorkdirPath::new("file-secret").unwrap(),
})
.await
.unwrap_err(),
"bytes" => remote
.command_output(CommandOutputRequest {
handle: CommandHandle("command-secret".into()),
cursor: 0,
limit: 16,
wait: false,
})
.await
.unwrap_err(),
"mutation" => remote
.checkout_execute(CheckoutRequest {
target: WorkdirPath::new("file-secret").unwrap(),
validator: vec![1],
operation: CheckoutOperation::Write {
content: b"body-secret".to_vec(),
expected_hash: [0; 32],
},
})
.await
.unwrap_err(),
_ => unreachable!(),
}
};
server.join().unwrap();
let expected = if case == "identity" {
Failure::SessionIdentityMismatch
} else {
Failure::OperationResultMismatch
};
assert_eq!(error.failure_reason(), Some(expected), "{case}: {error:?}");
let transport = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(
transport.code,
if case == "mutation" {
Code::OutcomeUnknown
} else {
Code::Unavailable
}
);
let wire = serde_json::to_string(&transport).unwrap();
assert!(!wire.contains("secret"), "{wire}");
assert!(!format!("{error:?}").contains("secret"));
let forwarded = transport.into_workdir_error();
assert_eq!(forwarded.failure_reason(), Some(expected));
}
}
+24 -2
View File
@@ -2187,7 +2187,25 @@ impl RuntimeHttpWorkdirError {
}
fn runtime(error: RuntimeError) -> Self {
match error {
use workdir::WorkdirFailureReason as Reason;
let reason = match &error {
RuntimeError::WorkingDirectory(diagnostic) => match diagnostic.code.as_str() {
"working_directory_not_found" => Reason::WorkingDirectoryNotFound,
"working_directory_not_active" => Reason::WorkingDirectoryNotActive,
"working_directory_record_invalid" => Reason::WorkingDirectoryRecordInvalid,
"working_directory_repository_access_unavailable" => {
Reason::WorkingDirectoryAccessUnavailable
}
_ => Reason::SessionOpenFailed,
},
RuntimeError::RuntimeStopped => Reason::RuntimeStopped,
RuntimeError::StatePoisoned => Reason::RuntimeStateUnavailable,
RuntimeError::ExecutionBackendUnavailable { .. } => Reason::RuntimeBackendUnavailable,
RuntimeError::WorkspaceOwnerMismatch { .. } => Reason::RuntimeOwnerMismatch,
RuntimeError::InvalidRequest(_) => Reason::RuntimeInvalidRequest,
_ => Reason::SessionOpenFailed,
};
let mut response = match error {
RuntimeError::WorkingDirectory(diagnostic)
if diagnostic.code == "working_directory_not_found" =>
{
@@ -2200,7 +2218,9 @@ impl RuntimeHttpWorkdirError {
WorkdirTransportErrorCode::Unavailable,
"Workdir session is unavailable",
),
}
};
response.payload.failure_reason = Some(reason);
response
}
}
@@ -2222,6 +2242,7 @@ impl IntoResponse for RuntimeHttpWorkdirError {
status = self.status.as_u16(),
code = self.payload.code.as_str(),
failure_reason = reason.as_str(),
failure_stage = reason.stage().as_str(),
"Workdir provider operation failed"
);
}
@@ -2434,6 +2455,7 @@ pub enum RuntimeHttpServerError {
#[cfg(test)]
mod tests {
mod command_transport_tests;
mod runtime_workdir_failure_tests;
#[cfg(feature = "fs-store")]
mod session_material_tests;
@@ -0,0 +1,528 @@
//! Session-open integration through the production signed router, Runtime owner/binding
//! checks, HTTP DTO and tracing boundary. Only the execution backend is replaced
//! for deterministic lookup/open faults and inert owner-binding Worker fixtures;
//! no handler, middleware or grant is replaced. Runs in the normal http-server lib
//! suite with isolated loopback sockets.
use super::*;
use crate::working_directory::WorkingDirectoryDiagnostic;
use std::io::{self, Write};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
const OPEN_PATH: &str = "/v1/working-directories/private-workdir-marker/sessions";
const PRIVATE_DIAGNOSTIC: &str = concat!(
"raw-runtime-error-marker /private/host-path-marker ",
"command=private-command-marker contents=private-file-marker ",
"proof=private-proof-marker Authorization: Bearer private-header-marker ",
"working_directory_not_found authorization_stale runtime_stopped",
);
const PRIVATE_MARKERS: &[&str] = &[
"raw-runtime-error-marker",
"/private/host-path-marker",
"private-command-marker",
"private-file-marker",
"private-proof-marker",
"private-header-marker",
"private-workdir-marker",
"private-owner-marker",
];
#[derive(Clone, Default)]
struct LogCapture(Arc<Mutex<Vec<u8>>>);
impl Write for LogCapture {
fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> io::Result<()> {
Ok(())
}
}
impl LogCapture {
fn subscriber(&self) -> impl tracing::Subscriber + Send + Sync + 'static {
let writer = self.clone();
tracing_subscriber::fmt()
.with_ansi(false)
.without_time()
.with_max_level(tracing::Level::WARN)
.with_writer(move || writer.clone())
.finish()
}
fn take(&self) -> String {
String::from_utf8(std::mem::take(&mut *self.0.lock().unwrap())).unwrap()
}
}
fn assert_private_absent(text: &str) {
for marker in PRIVATE_MARKERS {
assert!(
!text.contains(marker),
"private input escaped error boundary"
);
}
}
fn assert_failure_log(log: &str, status: StatusCode, code: &str, reason: &str, stage: &str) {
let events: Vec<_> = log
.lines()
.filter(|line| line.contains("workdir_provider_failure"))
.collect();
assert_eq!(events.len(), 1, "one diagnostic per failed request: {log}");
let event = events[0];
for field in [
"WARN".to_string(),
"worker_runtime::workdir".to_string(),
format!("status={}", status.as_u16()),
format!("code=\"{code}\""),
format!("failure_reason=\"{reason}\""),
format!("failure_stage=\"{stage}\""),
] {
assert!(event.contains(&field), "missing {field}: {event}");
}
assert_private_absent(log);
}
fn assert_failure_body(body: &str, code: &str, message: &str, reason: &str) {
assert_private_absent(body);
// Exact safe transport vocabulary; raw diagnostics and stage are not wire fields.
assert_eq!(
serde_json::from_str::<serde_json::Value>(body).unwrap(),
serde_json::json!({"code": code, "message": message, "failure_reason": reason}),
);
}
struct Harness {
server: tokio::task::JoinHandle<()>,
client: reqwest::Client,
base: String,
identity: RuntimeIdentityMaterial,
trust: WorkspaceIssuerTrustRecord,
}
impl Drop for Harness {
fn drop(&mut self) {
self.server.abort();
}
}
impl Harness {
async fn new(runtime: Runtime) -> Self {
let identity = RuntimeIdentityMaterial::generate("runtime-failure-test-key").unwrap();
let trust = workspace_trust(&identity);
let app = runtime_http_router_with_workspace_auth(
runtime,
None,
workspace_test_auth(vec![trust.clone()]),
);
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let base = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
Self {
server,
client: reqwest::Client::builder()
.no_proxy()
.timeout(Duration::from_secs(5))
.build()
.unwrap(),
base,
identity,
trust,
}
}
async fn send(
&self,
claims: &WorkspaceCapabilityClaims,
body: Vec<u8>,
workspace: &str,
) -> (StatusCode, String, String) {
let token = issue_workspace_capability_token(&self.identity.signing_key().unwrap(), claims)
.unwrap();
let response = self
.client
.post(format!("{}{OPEN_PATH}", self.base))
.bearer_auth(&token)
.header(header::CONTENT_TYPE, "application/json")
.header(RUNTIME_WORKSPACE_SCOPE_HEADER, workspace)
.header("x-private-test-header", "private-header-marker")
.body(body)
.send()
.await
.unwrap();
let status = StatusCode::from_u16(response.status().as_u16()).unwrap();
let headers = format!("{:?}", response.headers());
let body = response.text().await.unwrap();
assert_private_absent(&headers);
for output in [&body, &headers] {
assert!(!output.contains(&token), "signed bearer token leaked");
assert!(
!output.contains(&self.identity.private_key),
"signing key leaked"
);
}
(status, body, token)
}
async fn open(&self) -> (StatusCode, String, String) {
let body = serde_json::to_vec(&OpenWorkdirSessionRequest::default()).unwrap();
let claims = signed_claims(&self.trust, "POST", OPEN_PATH, &body);
self.send(&claims, body, "workspace-a").await
}
}
#[derive(Clone, Copy)]
enum FailureAt {
Lookup,
Open,
}
struct FailingBackend {
at: FailureAt,
code: &'static str,
lookups: AtomicUsize,
opens: AtomicUsize,
}
impl FailingBackend {
fn runtime(at: FailureAt, code: &'static str) -> (Runtime, Arc<Self>) {
let backend = Arc::new(Self {
at,
code,
lookups: AtomicUsize::new(0),
opens: AtomicUsize::new(0),
});
let runtime =
Runtime::with_execution_backend(RuntimeOptions::default(), backend.clone()).unwrap();
(runtime, backend)
}
fn failure(&self) -> WorkingDirectoryDiagnostic {
WorkingDirectoryDiagnostic::rejected(self.code, PRIVATE_DIAGNOSTIC)
}
}
impl WorkerExecutionBackend for FailingBackend {
fn backend_id(&self) -> &str {
"runtime-failure-test"
}
fn spawn_worker(&self, request: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
AcceptingBackend.spawn_worker(request)
}
fn dispatch_input(&self, worker: &WorkerRef, input: WorkerInput) -> WorkerExecutionResult {
AcceptingBackend.dispatch_input(worker, input)
}
fn stop_worker(&self, worker: &WorkerRef) -> WorkerExecutionResult {
AcceptingBackend.stop_worker(worker)
}
fn working_directory(
&self,
id: &str,
) -> Result<WorkingDirectoryStatus, WorkingDirectoryDiagnostic> {
self.lookups.fetch_add(1, Ordering::SeqCst);
match self.at {
FailureAt::Lookup => Err(self.failure()),
FailureAt::Open => Ok(WorkingDirectoryStatus {
summary: crate::catalog::WorkingDirectorySummary {
working_directory_id: id.into(),
display_name: None,
repository_id: "repository-test".into(),
creation_selector: None,
creation_ref: None,
creation_tree: None,
current_selector: None,
current_ref: None,
current_tree: None,
observed_at_epoch_seconds: None,
materializer_kind: crate::catalog::MaterializerKind::RuntimeGitClone,
cleanup_target: None,
status: crate::catalog::WorkingDirectoryStatusKind::Active,
cleanliness: None,
occupied_by: None,
},
}),
}
}
fn open_workdir_session(
&self,
_: &str,
) -> Result<WorkdirSessionHandle, WorkingDirectoryDiagnostic> {
self.opens.fetch_add(1, Ordering::SeqCst);
Err(self.failure())
}
}
// Current-thread execution keeps the server tasks inside the thread-local tracing
// capture across awaits, without replacing the process-global subscriber.
#[tokio::test(flavor = "current_thread")]
async fn signed_http_session_open_preserves_runtime_availability_and_owner_failures() {
let capture = LogCapture::default();
let _guard = tracing::subscriber::set_default(capture.subscriber());
tracing::callsite::rebuild_interest_cache();
tokio::time::timeout(Duration::from_secs(15), async {
let stopped = Runtime::new_memory();
stopped.stop_runtime().unwrap();
let (owned, backend) = FailingBackend::runtime(FailureAt::Lookup, "must_not_reach_backend");
let mut request = task_request("bind owner");
request.config_bundle = None;
request.memory_settings.as_mut().unwrap().workspace_id = "workspace-a".into();
owned
.create_worker_scoped(
&RuntimeWorkspaceScope::new("workspace-a", "private-owner-marker"),
request,
)
.unwrap();
for (runtime, status, code, message, reason) in [
(
stopped,
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
"runtime_stopped",
),
(
Runtime::new_memory(),
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
"runtime_backend_unavailable",
),
(
owned,
StatusCode::NOT_FOUND,
"not_found",
"Workdir session was not found",
"runtime_owner_mismatch",
),
] {
let harness = Harness::new(runtime).await;
capture.take();
let (actual, body, token) = harness.open().await;
assert_eq!(actual, status);
assert_failure_body(&body, code, message, reason);
let log = capture.take();
assert_failure_log(&log, status, code, reason, "runtime");
assert!(!log.contains(&token));
assert!(!log.contains(&harness.identity.private_key));
}
assert_eq!(backend.lookups.load(Ordering::SeqCst), 0);
assert_eq!(backend.opens.load(Ordering::SeqCst), 0);
})
.await
.expect("bounded Runtime failure HTTP requests");
}
#[tokio::test(flavor = "current_thread")]
async fn signed_http_session_open_allowlists_diagnostic_codes_without_guessing_unknown_causes() {
let capture = LogCapture::default();
let _guard = tracing::subscriber::set_default(capture.subscriber());
tracing::callsite::rebuild_interest_cache();
tokio::time::timeout(Duration::from_secs(20), async {
for (diagnostic, reason) in [
("working_directory_not_found", "working_directory_not_found"),
(
"working_directory_not_active",
"working_directory_not_active",
),
(
"working_directory_record_invalid",
"working_directory_record_invalid",
),
(
"working_directory_repository_access_unavailable",
"working_directory_access_unavailable",
),
// Plausible provider text/code is not evidence of a typed cause.
("authorization_stale", "session_open_failed"),
(
"working_directory_not_found_private-proof-marker",
"session_open_failed",
),
] {
for at in [FailureAt::Lookup, FailureAt::Open] {
let (runtime, backend) = FailingBackend::runtime(at, diagnostic);
let harness = Harness::new(runtime).await;
capture.take();
let (status, body, token) = harness.open().await;
let (expected, code, message) = if diagnostic == "working_directory_not_found" {
(
StatusCode::NOT_FOUND,
"not_found",
"Workdir session was not found",
)
} else {
(
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
)
};
assert_eq!(status, expected, "{diagnostic}");
assert_failure_body(&body, code, message, reason);
let log = capture.take();
assert_failure_log(&log, status, code, reason, "session_open");
assert!(!log.contains(&token));
assert!(!log.contains(&harness.identity.private_key));
assert_eq!(backend.lookups.load(Ordering::SeqCst), 1, "no hidden retry");
assert_eq!(
backend.opens.load(Ordering::SeqCst),
usize::from(matches!(at, FailureAt::Open)),
);
}
}
})
.await
.expect("bounded diagnostic fault matrix");
}
#[tokio::test(flavor = "current_thread")]
async fn signed_http_session_open_rejects_bad_capability_scope_and_owner_binding_before_backend() {
let capture = LogCapture::default();
let _guard = tracing::subscriber::set_default(capture.subscriber());
tracing::callsite::rebuild_interest_cache();
tokio::time::timeout(Duration::from_secs(15), async {
let (runtime, backend) =
FailingBackend::runtime(FailureAt::Open, "working_directory_record_invalid");
let mut request = task_request("foreign workspace binding");
request.config_bundle = None;
request.memory_settings.as_mut().unwrap().workspace_id = "other-workspace".into();
let foreign_worker = runtime
.create_worker_scoped(
&RuntimeWorkspaceScope::new("other-workspace", "https://backend.test"),
request,
)
.unwrap()
.worker_ref;
let harness = Harness::new(runtime).await;
for case in ["permission", "body", "workspace", "owner_binding"] {
let body = if case == "owner_binding" {
serde_json::to_vec(&OpenWorkdirSessionRequest {
owner_worker_id: Some(foreign_worker.worker_id.to_string()),
})
.unwrap()
} else {
b"{}".to_vec()
};
let mut claims = signed_claims(&harness.trust, "POST", OPEN_PATH, &body);
if case == "permission" {
claims.operation = "workers:read".into();
}
if case == "workspace" {
claims.issuer_workspace_id = "other-workspace".into();
}
let sent = if case == "body" {
b"{ }".to_vec()
} else {
body
};
capture.take();
let (status, body, token) = harness.send(&claims, sent, "workspace-a").await;
assert_eq!(
status,
if case == "owner_binding" {
StatusCode::NOT_FOUND
} else {
StatusCode::UNAUTHORIZED
},
"{case}",
);
assert_private_absent(&body);
let log = capture.take();
assert_private_absent(&log);
assert!(!log.contains(&token));
assert!(!log.contains(&harness.identity.private_key));
if case == "owner_binding" {
assert_failure_body(
&body,
"not_found",
"Workdir session was not found",
"working_directory_not_found",
);
assert_failure_log(
&log,
status,
"not_found",
"working_directory_not_found",
"session_open",
);
} else {
let wire: serde_json::Value = serde_json::from_str(&body).unwrap();
assert_eq!(wire["error"]["code"], "unauthorized");
assert!(
!log.contains("workdir_provider_failure"),
"auth rejection is not a provider fault",
);
}
}
assert_eq!(backend.lookups.load(Ordering::SeqCst), 0);
assert_eq!(backend.opens.load(Ordering::SeqCst), 0);
})
.await
.expect("bounded authorization rejection requests");
}
#[tokio::test(flavor = "current_thread")]
async fn runtime_error_response_keeps_raw_details_private_and_unclassified_causes_generic() {
let capture = LogCapture::default();
let _guard = tracing::subscriber::set_default(capture.subscriber());
tracing::callsite::rebuild_interest_cache();
for (error, status, code, message, reason, stage) in [
(
RuntimeError::ExecutionBackendUnavailable {
message: PRIVATE_DIAGNOSTIC.into(),
},
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
"runtime_backend_unavailable",
"runtime",
),
(
RuntimeError::StatePoisoned,
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
"runtime_state_unavailable",
"runtime",
),
(
RuntimeError::InvalidRequest(PRIVATE_DIAGNOSTIC.into()),
StatusCode::BAD_REQUEST,
"invalid_request",
"Workdir operation request is invalid",
"runtime_invalid_request",
"runtime",
),
(
RuntimeError::StoreCorrupt {
operation: "session_open",
path: PathBuf::from("/private/host-path-marker"),
message: PRIVATE_DIAGNOSTIC.into(),
},
StatusCode::SERVICE_UNAVAILABLE,
"unavailable",
"Workdir session is unavailable",
"session_open_failed",
"session_open",
),
] {
capture.take();
let response = RuntimeHttpWorkdirError::runtime(error).into_response();
assert_eq!(response.status(), status);
let bytes = to_bytes(response.into_body(), 16384).await.unwrap();
let body = std::str::from_utf8(&bytes).unwrap();
assert_failure_body(body, code, message, reason);
assert_failure_log(&capture.take(), status, code, reason, stage);
}
}
@@ -66,13 +66,32 @@ impl DeferredWorkdirSession {
let open = self.open.clone();
let session = tokio::task::spawn_blocking(move || open())
.await
.map_err(|_| WorkdirError::Unavailable("Workdir session opener failed".into()))?
.map_err(WorkdirError::Unavailable)?;
.map_err(|_| {
WorkdirError::Unavailable("Workdir session opener failed".into())
.with_failure_reason(WorkdirFailureReason::SessionOpenTaskFailed)
})?
.map_err(|_| {
// The opener's string contract cannot prove a more specific cause.
WorkdirError::Unavailable("Workdir session could not be opened".into())
.with_failure_reason(WorkdirFailureReason::SessionOpenFailed)
})?;
if session.workdir().id() != self.workdir.id() {
session.close().await?;
return Err(WorkdirError::Unavailable(
"Workdir session identity mismatch".into(),
));
if let Err(error) = session.close().await {
// Retain a failed-close owner for explicit close; never use it.
state.closed = true;
state.session = Some(session);
tracing::warn!(
target: "worker_runtime::workdir",
event = "workdir_session_identity_cleanup_failed",
failure_reason = error.failure_reason().map(WorkdirFailureReason::as_str),
denial_reason = error.denial_reason().map(WorkdirDenialReason::as_str),
"Mismatched Workdir session cleanup failed"
);
}
return Err(
WorkdirError::Unavailable("Workdir session identity mismatch".into())
.with_failure_reason(WorkdirFailureReason::SessionIdentityMismatch),
);
}
if let Some(mut source) = session.subscribe_command_events() {
let events = self.events.clone();
@@ -224,10 +243,16 @@ mod tests {
},
);
assert_eq!(attempts.load(Ordering::SeqCst), 0);
let failed = session.read(read_request()).await.unwrap_err();
assert!(matches!(
session.read(read_request()).await,
Err(WorkdirError::Unavailable(_))
failed.classification_source(),
WorkdirError::Unavailable(_)
));
assert_eq!(
failed.failure_reason(),
Some(WorkdirFailureReason::SessionOpenFailed)
);
assert_eq!(attempts.load(Ordering::SeqCst), 1);
assert_eq!(
session.read(read_request()).await.unwrap().bytes,
b"restored"
@@ -253,4 +278,126 @@ mod tests {
);
session.close().await.unwrap();
}
#[tokio::test]
async fn opener_task_and_open_errors_keep_distinct_safe_causes() {
for panic in [false, true] {
let attempts = Arc::new(AtomicUsize::new(0));
let observed = attempts.clone();
let session = DeferredWorkdirSession::handle(
"saved",
WorkdirSessionCapabilities::READ_ONLY,
move || {
observed.fetch_add(1, Ordering::SeqCst);
if panic {
panic!(
"/host/private bearer-secret command-secret proof-secret key-secret"
);
}
Err("/host/private bearer-secret command-secret proof-secret key-secret".into())
},
);
let error = session.read(read_request()).await.unwrap_err();
assert_eq!(attempts.load(Ordering::SeqCst), 1);
let reason = if panic {
WorkdirFailureReason::SessionOpenTaskFailed
} else {
WorkdirFailureReason::SessionOpenFailed
};
assert_eq!(error.failure_reason(), Some(reason));
assert_eq!(reason.stage().as_str(), "session_open");
let transport = workdir::http::WorkdirTransportError::from_workdir_error(&error);
assert_eq!(transport.code.http_status(), 503);
assert!(
!serde_json::to_string(&transport)
.unwrap()
.contains("secret")
);
assert!(!format!("{error:?}").contains("secret"));
session.close().await.unwrap();
assert_eq!(attempts.load(Ordering::SeqCst), 1);
}
}
#[derive(Debug)]
struct MismatchedSession {
workdir: Workdir,
closes: Arc<AtomicUsize>,
fail_close: bool,
}
macro_rules! mismatched_session {
($( $method:ident($request:ident: $request_type:ty) -> $result:ty; )*) => {
#[async_trait]
impl WorkdirSession for MismatchedSession {
fn workdir(&self) -> &Workdir { &self.workdir }
fn capabilities(&self) -> WorkdirSessionCapabilities { WorkdirSessionCapabilities::READ_ONLY }
$(async fn $method(&self, _: $request_type) -> Result<$result, WorkdirError> {
panic!("mismatched session must never be used");
})*
async fn close(&self) -> Result<(), WorkdirError> {
let attempt = self.closes.fetch_add(1, Ordering::SeqCst);
if self.fail_close && attempt == 0 {
Err(WorkdirError::Transport("/host/private key-secret".into())
.with_failure_reason(WorkdirFailureReason::HttpTransfer))
} else { Ok(()) }
}
}
};
}
mismatched_session! {
stat(request: StatRequest) -> StatResult;
read(request: ReadRequest) -> ReadResult;
write(request: WriteRequest) -> WriteResult;
edit(request: EditRequest) -> EditResult;
list(request: ListRequest) -> ListResult;
glob(request: GlobRequest) -> GlobResult;
grep(request: GrepRequest) -> GrepResult;
start_command(request: CommandRequest) -> CommandHandle;
command_status(handle: CommandHandle) -> CommandStatus;
command_output(request: CommandOutputRequest) -> CommandOutput;
cancel_command(handle: CommandHandle) -> ();
}
#[tokio::test]
async fn identity_mismatch_preserves_primary_cause_and_failed_close_owner() {
for fail_close in [false, true] {
let closes = Arc::new(AtomicUsize::new(0));
let raw: WorkdirSessionHandle = Arc::new(MismatchedSession {
workdir: Workdir::new("wrong-secret"),
closes: closes.clone(),
fail_close,
});
let attempts = Arc::new(AtomicUsize::new(0));
let observed = attempts.clone();
let session = DeferredWorkdirSession::handle(
"saved",
WorkdirSessionCapabilities::READ_ONLY,
move || {
observed.fetch_add(1, Ordering::SeqCst);
Ok(raw.clone())
},
);
let error = session.read(read_request()).await.unwrap_err();
assert_eq!(
error.failure_reason(),
Some(WorkdirFailureReason::SessionIdentityMismatch)
);
assert!(!format!("{error:?}").contains("secret"));
assert_eq!(closes.load(Ordering::SeqCst), 1);
if fail_close {
assert!(matches!(
session.read(read_request()).await,
Err(WorkdirError::SessionClosed)
));
assert_eq!(attempts.load(Ordering::SeqCst), 1);
}
session.close().await.unwrap();
assert_eq!(
closes.load(Ordering::SeqCst),
if fail_close { 2 } else { 1 }
);
assert_eq!(attempts.load(Ordering::SeqCst), 1);
}
}
}
@@ -291,6 +291,11 @@ impl WorkspaceAttachedWorkdirSession {
.execute(request)
.map_err(workspace_workdir_error)?;
if !response.is_success() {
if let Ok(error) =
serde_json::from_str::<server_api::WorkdirOperationApiError>(&response.body)
{
notify_workdir_audit_failures(&error.audit_failures);
}
return Err(
serde_json::from_str::<workdir::http::WorkdirTransportError>(&response.body)
.map(workdir::http::WorkdirTransportError::into_workdir_error)
@@ -302,11 +307,14 @@ impl WorkspaceAttachedWorkdirSession {
}),
);
}
serde_json::from_str(&response.body).map_err(|error| {
WorkdirError::Transport(format!(
"failed to decode Workspace Workdir operation result: {error}"
))
})
let response: workdir::http::WorkdirSessionOperationResponse =
serde_json::from_str(&response.body).map_err(|_| {
WorkdirError::Transport(
"failed to decode Workspace Workdir operation result".into(),
)
})?;
notify_workdir_audit_failures(&response.audit_failures);
Ok(response.result)
}
fn mismatch(expected: &str) -> WorkdirError {
@@ -316,6 +324,27 @@ impl WorkspaceAttachedWorkdirSession {
}
}
// Notification is separate from the operation return: callers retain successful command
// handles, checkout observations and original failures, and must not resend for audit loss.
fn notify_workdir_audit_failures(failures: &[workdir::http::WorkdirAuditFailure]) {
use sha2::Digest;
for failure in failures {
// Response strings are untrusted. Never copy them into the receiving process log.
let operation_id_hash = sha2::Sha256::digest(failure.operation_id.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>();
tracing::error!(
event = "external_workdir_audit_failed",
operation_id_hash,
audit_stage = ?failure.stage,
audit_reason = ?failure.reason,
operation_outcome = ?failure.operation_outcome,
"External Workdir audit persistence failed; operation result is independent; do not resend"
);
}
}
fn workspace_workdir_error(error: WorkspaceClientError) -> WorkdirError {
match error {
WorkspaceClientError::Unavailable(message) => WorkdirError::Unavailable(message),
@@ -1653,6 +1682,67 @@ mod tests {
}
}
#[tokio::test]
async fn attached_session_keeps_operation_results_when_audit_persistence_failed() {
let failure = json!({
"operation_id":"op-safe-id", "stage":"response_received",
"reason":"audit_store_unavailable", "operation_outcome":{"kind":"succeeded"}
});
let mut responses = vec![response(json!({
"operation":"command_start", "result":"retained-command",
"audit_failures":[failure]
}))];
for (code, status, outcome) in [
("denied", 403, json!({"kind":"failed","code":"denied"})),
("outcome_unknown", 500, json!({"kind":"unknown"})),
("not_dispatched", 503, json!({"kind":"not_dispatched"})),
] {
let mut failure = failure.clone();
failure["operation_outcome"] = outcome;
responses.push(WorkspaceResponse { status, body: json!({
"code":code, "message":"safe provider classification", "audit_failures":[failure]
}).to_string() });
}
let client = Arc::new(RecordingWorkspaceClient::new(responses));
let session = WorkspaceAttachedWorkdirSession::handle(client.clone(), "checkout");
let handle = session
.start_command(CommandRequest {
command: "true".into(),
timeout_secs: 1,
transport_byte_budget: workdir::MAX_COMMAND_TRANSPORT_BYTES,
cwd: workdir::WorkdirPath::root(),
tool_call_id: None,
})
.await
.unwrap();
assert_eq!(handle.0, "retained-command");
let request = workdir::CheckoutRequest {
target: workdir::WorkdirPath::new("file.txt").unwrap(),
validator: vec![1],
operation: workdir::CheckoutOperation::Write {
content: b"new".to_vec(),
expected_hash: [0; 32],
},
};
assert!(matches!(
session.checkout_execute(request.clone()).await,
Err(WorkdirError::Denied(_))
));
assert!(matches!(
session.checkout_execute(request.clone()).await,
Err(WorkdirError::OutcomeUnknown(_))
));
assert!(matches!(
session.checkout_execute(request).await,
Err(WorkdirError::NotDispatched)
));
assert_eq!(
client.requests().len(),
4,
"audit loss must not trigger a resend"
);
}
#[tokio::test]
async fn attached_session_does_not_send_worker_local_bash_spill_path() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
+25 -3
View File
@@ -3995,12 +3995,27 @@ impl WorkdirHttpAuthorization for RemoteWorkdirAuthorization {
return authorizer
.issue_async(method, path_and_query, operation, None, body)
.await
.map_err(|error| WorkdirError::Unavailable(error.message));
.map_err(|error| {
use workdir::WorkdirFailureReason as Reason;
let reason = match error.code.as_str() {
"workspace_runtime_authorization_stale" => Reason::AuthorizationStale,
"workspace_signing_identity_unavailable" => {
Reason::SigningIdentityUnavailable
}
"workspace_runtime_authorization_sign_failed" => {
Reason::AuthorizationSignFailed
}
_ => Reason::AuthorizationUnavailable,
};
WorkdirError::Unavailable("Workdir authorization is unavailable".into())
.with_failure_reason(reason)
});
}
authorization.fallback_bearer_token.ok_or_else(|| {
WorkdirError::Unavailable(
"remote Runtime does not have bearer authorization configured".to_string(),
)
.with_failure_reason(workdir::WorkdirFailureReason::AuthorizationMissing)
})
})
}
@@ -6839,9 +6854,16 @@ mod tests {
.await
.unwrap_err();
assert!(matches!(
error,
WorkdirError::Unavailable(message) if message == Error::DbCapacityExceeded.to_string()
error.classification_source(),
WorkdirError::Unavailable(_)
));
assert_eq!(
error.failure_reason(),
Some(workdir::WorkdirFailureReason::AuthorizationUnavailable)
);
let transport = workdir::http::WorkdirTransportError::from_workdir_error(&error);
assert_eq!(transport.message, "Workdir session is unavailable");
assert!(!format!("{error:?}").contains(&Error::DbCapacityExceeded.to_string()));
drop(release);
executor.drain().await.unwrap();
}
@@ -315,3 +315,71 @@ async fn capability_claims_must_match_current_workspace_identity_before_signing(
.unwrap();
verify(&fixture.verifier(), &token).unwrap();
}
#[tokio::test]
async fn remote_workdir_authorization_failures_keep_safe_reason_without_dispatch_or_fallback() {
use workdir::WorkdirFailureReason as Reason;
use workdir::http::{OpenWorkdirSessionRequest, RemoteWorkdirSession, WorkdirTransportError};
for case in ["stale", "missing_material", "missing_authorization"] {
let fixture = Fixture::new().await;
if case == "stale" {
fixture
.store
.revoke_workspace_runtime_binding(WORKSPACE, RUNTIME, "revoked")
.unwrap();
} else if case == "missing_material" {
fixture
.materials
.delete(&fixture.identity.private_material_ref)
.unwrap();
}
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
listener.set_nonblocking(true).unwrap();
let authorization = RemoteWorkdirAuthorization {
workspace_authorization: (case != "missing_authorization")
.then(|| fixture.authorizer.clone()),
fallback_bearer_token: (case != "missing_authorization")
.then(|| "bearer-secret".into()),
};
let error = RemoteWorkdirSession::open_with_authorization(
reqwest::Client::builder().no_proxy().build().unwrap(),
format!("http://{}", listener.local_addr().unwrap())
.parse()
.unwrap(),
Arc::new(authorization),
workdir::Workdir::new("workdir-secret").id().clone(),
OpenWorkdirSessionRequest::default(),
)
.await
.unwrap_err();
let reason = match case {
"stale" => Reason::AuthorizationStale,
"missing_material" => Reason::AuthorizationUnavailable,
_ => Reason::AuthorizationMissing,
};
assert_eq!(error.failure_reason(), Some(reason));
assert_eq!(reason.stage().as_str(), "authorization_preparation");
let transport = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(transport.code.http_status(), 503);
assert_eq!(transport.message, "Workdir session is unavailable");
assert!(
!serde_json::to_string(&transport)
.unwrap()
.contains("secret")
);
let debug = format!("{error:?}");
for secret in [
"secret",
"workspace-signing/",
"private material",
"Authorization:",
"Bearer ",
] {
assert!(!debug.contains(secret), "{case}: {debug}");
}
assert!(
matches!(listener.accept(), Err(error) if error.kind() == std::io::ErrorKind::WouldBlock),
"authorization failure sent HTTP or fell back"
);
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,36 @@
//! Per-request audit notifications, independent of provider result classification.
//!
//! The scope follows the operation future, not a session-wide last-result slot: concurrent
//! requests cannot consume one another's diagnostics. Every failure is also emitted immediately
//! for direct session callers and requests cancelled before a response. This is notification,
//! not a replacement durable audit ledger or crash-recovery/effect-deduplication guarantee.
use std::{cell::RefCell, future::Future};
use workdir::http::WorkdirAuditFailure;
tokio::task_local! {
static FAILURES: RefCell<Vec<WorkdirAuditFailure>>;
}
pub(super) async fn capture<T>(future: impl Future<Output = T>) -> (T, Vec<WorkdirAuditFailure>) {
FAILURES
.scope(RefCell::new(Vec::new()), async {
let result = future.await;
let failures = FAILURES.with(|failures| failures.take());
(result, failures)
})
.await
}
pub(super) fn notify(failure: WorkdirAuditFailure, operation: &'static str) {
tracing::error!(
event = "external_workdir_audit_failed",
operation,
operation_id = %failure.operation_id,
audit_stage = ?failure.stage,
audit_reason = ?failure.reason,
operation_outcome = ?failure.operation_outcome,
"External Workdir audit persistence failed; operation result is independent; do not resend"
);
let _ = FAILURES.try_with(|failures| failures.borrow_mut().push(failure));
}
@@ -605,6 +605,10 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
workdir::WorkdirFailureReason::SessionClosed,
workdir::WorkdirFailureReason::CommandTerminalMissing,
workdir::WorkdirFailureReason::HttpTimeout,
workdir::WorkdirFailureReason::OperationResultMismatch,
workdir::WorkdirFailureReason::SessionOpenTaskFailed,
workdir::WorkdirFailureReason::AuthorizationStale,
workdir::WorkdirFailureReason::RuntimeStopped,
] {
let request = runtime_source_request(
&identity,
@@ -653,6 +657,7 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
);
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["failure_reason"], reason.as_str());
assert_eq!(events[0]["failure_stage"], reason.stage().as_str());
assert_eq!(events[0]["operation"], "command_output");
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
let log = serde_json::to_string(&events).unwrap();
@@ -1249,3 +1254,54 @@ async fn non_api_workspace_paths_are_not_logged_as_ticket_api_errors() {
assert_eq!(status, StatusCode::BAD_REQUEST);
assert!(events.is_empty());
}
#[tokio::test]
async fn session_open_failure_causes_survive_registry_api_conversion_without_raw_diagnostics() {
use workdir::WorkdirFailureReason as Reason;
for reason in [
Reason::SessionOpenFailed,
Reason::SessionOpenTaskFailed,
Reason::SessionIdentityMismatch,
Reason::AuthorizationStale,
Reason::AuthorizationUnavailable,
Reason::RuntimeStopped,
Reason::WorkingDirectoryRecordInvalid,
Reason::OperationResultMismatch,
] {
let mut fixture = manual_worker_assignment_fixture().await;
fixture.runtime.workdir_session_factory = Some(Arc::new(move |_| {
Err(workdir::WorkdirError::Unavailable(
"/host/private body-secret proof-secret key-secret credential-secret".into(),
)
.with_failure_reason(reason))
}));
fixture
.api
.runtime
.register_or_replace(fixture.runtime.clone());
let identity =
worker_runtime::auth::RuntimeIdentityMaterial::generate(&fixture.worker.runtime_id)
.unwrap();
configure_runtime_request_auth(&mut fixture.api, &identity, &fixture.worker.runtime_id);
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations");
let request = runtime_source_request(&identity, Some(&fixture.worker.worker_id), "POST", &path, serde_json::to_vec(&json!({
"target_workdir": "checkout",
"operation": {"operation": "read", "request": {"path":"README.md", "offset":0, "limit":10, "max_bytes":1024}}
})).unwrap());
let (status, body, events) =
request_with_logs(build_router(fixture.api.clone()), request).await;
assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}");
assert_eq!(
body["message"],
"workdir_session_open_failed: Workdir session is unavailable"
);
assert!(body.get("failure_reason").is_none(), "{body}");
assert!(!body.to_string().contains("secret"));
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["workdir_stage"], "session_open");
assert_eq!(events[0]["failure_reason"], reason.as_str());
assert_eq!(events[0]["failure_stage"], reason.stage().as_str());
assert_eq!(events[0]["message"], "Workdir operation failed");
assert_eq!(events[0]["diagnostics"], "[]");
}
}
@@ -1,3 +1,5 @@
mod audit_tests;
use super::*;
use sha2::Digest;
use std::sync::atomic::AtomicI64;
@@ -459,7 +461,7 @@ async fn provider_error_and_disconnect_release_dispatch_admission() {
assert_json(
result,
if disconnect {
StatusCode::SERVICE_UNAVAILABLE
StatusCode::INTERNAL_SERVER_ERROR
} else {
StatusCode::FORBIDDEN
},
@@ -575,8 +577,8 @@ async fn provider_timeout_releases_admission_without_wall_clock_wait() {
let (long, _reply) = f.start_read("1", &mut a).await;
tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(31)).await;
let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await;
assert_eq!(body["code"], "unavailable");
let body = assert_json(long.await.unwrap(), StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown");
tokio::time::resume();
let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await;
drop(
@@ -0,0 +1,885 @@
//! Integration coverage for signed loopback HTTP -> production admission/auth -> external
//! Workdir session -> real SQLite audit persistence. Only the external provider boundary
//! is substituted with a controlled command receiver. Temporary DBs, listeners and bounded
//! requests are isolated per case; these run in the normal workspace-server library suite.
use super::*;
use std::io::Write;
use tracing::instrument::WithSubscriber;
const AUDIT_SECRET: &str = "audit-store-password-secret";
const WRITE_SECRET: &[u8] = b"write-content-secret";
const PATH_SECRET: &str = "private-request-path-secret.txt";
const PROVIDER_SECRET: &str = "/host/private/provider-secret";
type ProviderReply =
oneshot::Sender<std::result::Result<WorkdirSessionOperationResult, WorkdirTransportError>>;
#[derive(Clone, Default)]
struct CapturedAuditLog(Arc<std::sync::Mutex<Vec<u8>>>);
impl Write for CapturedAuditLog {
fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
self.0.lock().unwrap().extend_from_slice(bytes);
Ok(bytes.len())
}
fn flush(&mut self) -> std::io::Result<()> {
Ok(())
}
}
struct PendingHttpWrite {
task: tokio::task::JoinHandle<(StatusCode, Value)>,
proof: String,
}
impl Drop for PendingHttpWrite {
fn drop(&mut self) {
self.task.abort();
}
}
struct ProviderProbe {
receiver: mpsc::Receiver<ExternalProviderCommand>,
dispatches: usize,
}
impl ProviderProbe {
async fn receive_operation(
&mut self,
pending: &mut PendingHttpWrite,
) -> (String, WorkdirSessionOperation, ProviderReply) {
let command = tokio::time::timeout(BOUND, async {
tokio::select! {
command = self.receiver.recv() => command.expect("provider channel closed"),
response = &mut pending.task => {
panic!("HTTP completed before write dispatch: {:?}", response.unwrap());
}
}
})
.await
.expect("signed HTTP write never reached provider");
let ExternalProviderCommand::Operation {
operation_id,
operation,
response,
} = command
else {
panic!("expected write dispatch, received cancellation")
};
self.dispatches += 1;
(operation_id.as_str().to_owned(), operation, response)
}
async fn receive_write(&mut self, pending: &mut PendingHttpWrite) -> (String, ProviderReply) {
let (operation_id, operation, response) = self.receive_operation(pending).await;
let WorkdirSessionOperation::Write(request) = operation else {
panic!("expected mutation, got {operation:?}")
};
assert_eq!(
request.path,
workdir::WorkdirPath::new(PATH_SECRET).unwrap()
);
assert_eq!(request.content, WRITE_SECRET);
assert_eq!(request.expected_hash, None);
(operation_id, response)
}
fn assert_dispatches(&mut self, expected: usize) {
while let Ok(command) = self.receiver.try_recv() {
if matches!(command, ExternalProviderCommand::Operation { .. }) {
self.dispatches += 1;
}
}
assert_eq!(
self.dispatches, expected,
"unexpected mutation dispatch/retry"
);
}
}
impl Fixture {
fn attach_audited_writer(&self) -> ProviderProbe {
let receiver = self.attach("1");
// Setup changes only: real signed requests still traverse production authorization.
// All three capability sources must agree before the session is first resolved.
let connection = rusqlite::Connection::open(&self.api.config.database_path).unwrap();
assert_eq!(
connection
.execute(
"UPDATE external_workdir_grants SET permissions='read_write' \
WHERE workspace_id=?1 AND grant_id='external-1'",
[TEST_WORKSPACE_ID],
)
.unwrap(),
1
);
assert_eq!(
connection
.execute(
"UPDATE worker_workdir_links SET capabilities='read_write' \
WHERE workspace_id=?1 AND runtime_id='workdirless-runtime' \
AND worker_id='1' AND alias='attachment' AND unlinked_at IS NULL",
[TEST_WORKSPACE_ID],
)
.unwrap(),
1
);
let mut registry = self.api.external_workdir_providers.lock().unwrap();
let provider = registry.connections.get_mut("external-1").unwrap();
Arc::get_mut(provider).unwrap().capabilities =
workdir::WorkdirSessionCapabilities::READ_WRITE;
ProviderProbe {
receiver,
dispatches: 0,
}
}
}
struct AuditHttpFixture {
fixture: Fixture,
logs: CapturedAuditLog,
client: reqwest::Client,
base_url: String,
server: tokio::task::JoinHandle<()>,
}
impl Drop for AuditHttpFixture {
fn drop(&mut self) {
self.server.abort();
}
}
impl AuditHttpFixture {
async fn new() -> (Self, ProviderProbe) {
let fixture = Fixture::new().await;
let provider = fixture.attach_audited_writer();
let logs = CapturedAuditLog::default();
let writer = logs.clone();
let subscriber = tracing_subscriber::fmt()
.without_time()
.with_ansi(false)
.json()
.flatten_event(true)
.with_writer(move || writer.clone())
.finish();
let dispatch = tracing::Dispatch::new(subscriber);
// Axum spawns connection tasks: scope the subscriber around each request rather
// than relying on the listener task's thread-local subscriber being inherited.
let app = fixture.app.clone().layer(axum::middleware::from_fn(
move |request: Request<Body>, next: axum::middleware::Next| {
let dispatch = dispatch.clone();
async move { next.run(request).with_subscriber(dispatch).await }
},
));
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let base_url = format!("http://{}", listener.local_addr().unwrap());
let server = tokio::spawn(async move {
axum::serve(listener, app).await.unwrap();
});
let client = reqwest::Client::builder()
.no_proxy()
.timeout(BOUND)
.build()
.unwrap();
(
Self {
fixture,
logs,
client,
base_url,
server,
},
provider,
)
}
fn fail_audit(&self, before_dispatch: bool) {
let filter = if before_dispatch { "=" } else { "<>" };
rusqlite::Connection::open(&self.fixture.api.config.database_path)
.unwrap()
.execute_batch(&format!(
"CREATE TRIGGER fail_operation_audit BEFORE INSERT ON audit_events \
WHEN NEW.action='external_workdir_operation' AND NEW.outcome {filter} 'requested' \
BEGIN SELECT RAISE(FAIL, '{AUDIT_SECRET}'); END;"
))
.unwrap();
}
fn start_write(&self) -> PendingHttpWrite {
self.start_operation(WorkdirSessionOperation::Write(workdir::WriteRequest {
path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(),
content: WRITE_SECRET.to_vec(),
expected_hash: None,
}))
}
fn start_operation(&self, operation: WorkdirSessionOperation) -> PendingHttpWrite {
let path = format!("/api/w/{TEST_WORKSPACE_ID}/workers/self/workdir-session/operations");
let body = serde_json::to_vec(&WorkspaceWorkdirSessionOperationRequest {
target_workdir: "attachment".into(),
operation,
})
.unwrap();
let request = self.fixture.request(
Some("1"),
&path,
body.clone(),
START,
worker_runtime::auth::WORKSPACE_REQUEST_PERMISSION,
);
let proof = request.headers()[worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER]
.to_str()
.unwrap()
.to_owned();
let url = format!("{}{path}", self.base_url);
let client = self.client.clone();
let headers = request.headers().clone();
let task = tokio::spawn(async move {
let response = client
.post(url)
.headers(headers)
.body(body)
.send()
.await
.unwrap();
let status = response.status();
assert_eq!(response.headers()[CONTENT_TYPE], "application/json");
let bytes = response.bytes().await.unwrap();
(status, serde_json::from_slice(&bytes).unwrap())
});
PendingHttpWrite { task, proof }
}
async fn finish(
&self,
mut pending: PendingHttpWrite,
expected_status: StatusCode,
) -> (Value, Vec<Value>) {
let (status, body) = tokio::time::timeout(BOUND, &mut pending.task)
.await
.expect("HTTP did not finish; possible automatic retry")
.unwrap();
assert_eq!(status, expected_status, "{body}");
let text = String::from_utf8(self.logs.0.lock().unwrap().clone()).unwrap();
let encoded_content = serde_json::to_string(WRITE_SECRET).unwrap();
for output in [body.to_string(), text.clone()] {
for secret in [
AUDIT_SECRET,
std::str::from_utf8(WRITE_SECRET).unwrap(),
encoded_content.as_str(),
PATH_SECRET,
PROVIDER_SECRET,
PROVIDER_SECRET.trim_start_matches('/'),
pending.proof.as_str(),
worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER,
self.fixture.identity.private_key.as_str(),
self.fixture.identity.public_key.as_str(),
self.fixture._root.path().to_str().unwrap(),
] {
assert!(!output.contains(secret), "sensitive data escaped: {output}");
}
}
let events = text
.lines()
.map(|line| serde_json::from_str(line).unwrap())
.collect();
(body, events)
}
fn assert_audits(&self, expected: &[(&str, &str)]) {
self.assert_operation_audits(expected, "write");
}
fn assert_operation_audits(&self, expected: &[(&str, &str)], operation: &str) {
let connection =
rusqlite::Connection::open(&self.fixture.api.config.database_path).unwrap();
let mut statement = connection
.prepare(
"SELECT request_id, outcome, summary FROM audit_events \
WHERE workspace_id=?1 AND action='external_workdir_operation' \
AND target_id='external-1' ORDER BY rowid",
)
.unwrap();
let rows: Vec<(String, String, String)> = statement
.query_map([TEST_WORKSPACE_ID], |row| {
Ok((row.get(0)?, row.get(1)?, row.get(2)?))
})
.unwrap()
.collect::<rusqlite::Result<_>>()
.unwrap();
let actual: Vec<(&str, &str)> = rows
.iter()
.map(|(operation_id, outcome, _)| (operation_id.as_str(), outcome.as_str()))
.collect();
assert_eq!(actual, expected, "persisted audits: {rows:?}");
for (_, _, summary) in rows {
assert!(
summary.contains(&format!("operation_kind={operation}")),
"{summary}"
);
for secret in [
AUDIT_SECRET,
PATH_SECRET,
PROVIDER_SECRET,
"write-content-secret",
] {
assert!(
!summary.contains(secret),
"sensitive audit summary: {summary}"
);
}
}
}
fn apply_provider_write(&self) {
std::fs::write(self.effect_path(), WRITE_SECRET).unwrap();
}
fn effect_path(&self) -> std::path::PathBuf {
self.fixture._root.path().join("provider-effect")
}
}
fn write_result() -> WorkdirSessionOperationResult {
WorkdirSessionOperationResult::Write(workdir::WriteResult {
bytes_written: WRITE_SECRET.len(),
created: true,
})
}
fn assert_audit_failure(
body: &Value,
events: &[Value],
stage: &str,
debug_stage: &str,
outcome: Value,
debug_outcome: &str,
) -> String {
let failures = body["audit_failures"]
.as_array()
.expect("independent audit receipt");
assert_eq!(failures.len(), 1, "{body}");
let operation_id = failures[0]["operation_id"].as_str().unwrap();
assert!(operation_id.starts_with("op-"), "{body}");
assert_eq!(
failures[0],
json!({
"operation_id": operation_id,
"stage": stage,
"reason": "audit_store_unavailable",
"operation_outcome": outcome,
})
);
let notifications: Vec<&Value> = events
.iter()
.filter(|event| event["event"] == "external_workdir_audit_failed")
.collect();
assert_eq!(notifications.len(), 1, "{events:?}");
let event = notifications[0];
assert_eq!(event["operation_id"], operation_id);
assert_eq!(event["operation"], "write");
assert_eq!(event["audit_stage"], debug_stage);
assert_eq!(event["audit_reason"], "AuditStoreUnavailable");
assert_eq!(event["operation_outcome"], debug_outcome);
operation_id.to_owned()
}
fn assert_failure_reason(events: &[Value], reason: &str, stage: &str) {
let errors: Vec<&Value> = events
.iter()
.filter(|event| event["event"] == "api_error")
.collect();
assert_eq!(errors.len(), 1, "{events:?}");
assert_eq!(errors[0]["failure_reason"], reason, "{events:?}");
assert_eq!(errors[0]["failure_stage"], stage, "{events:?}");
}
fn assert_operational_failure(
body: &Value,
events: &[Value],
reason: &str,
stage: &str,
code: &str,
message: &str,
) {
assert_failure_reason(events, reason, stage);
assert_eq!(body["code"], code, "{body}");
assert_eq!(body["message"], message, "{body}");
for field in ["failure_reason", "failure_stage"] {
assert!(
body.get(field).is_none(),
"internal diagnostic escaped: {body}"
);
}
}
#[tokio::test]
async fn successful_write_after_audit_failure_preserves_result_and_does_not_repeat_mutation() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
reply.send(Ok(write_result())).unwrap();
let (mut body, events) = h.finish(pending, StatusCode::OK).await;
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"succeeded"}),
"Succeeded",
);
assert_eq!(receipt_id, dispatched_id);
body.as_object_mut().unwrap().remove("audit_failures");
assert_eq!(body, serde_json::to_value(write_result()).unwrap());
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn denied_write_after_audit_failure_preserves_provider_error_and_separate_receipt() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
let mut original = WorkdirTransportError::from_workdir_error(&workdir::WorkdirError::Denied(
PROVIDER_SECRET.into(),
));
let expected_message = original.message.clone();
original.message = PROVIDER_SECRET.into();
reply.send(Err(original)).unwrap();
let (body, events) = h.finish(pending, StatusCode::FORBIDDEN).await;
assert_eq!(body["code"], "denied");
assert_eq!(body["message"], expected_message);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"failed", "code":"denied"}),
"Failed { code: Denied }",
);
assert_eq!(receipt_id, dispatched_id);
assert!(!h.effect_path().exists());
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn lost_write_response_after_audit_failure_stays_unknown_without_repeating_effect() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
// The provider applied the effect, but the server cannot infer that from response loss.
drop(reply);
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_operational_failure(
&body,
&events,
"provider_response_lost",
"provider_response",
"outcome_unknown",
"Workdir operation may have effects; inspect before retrying",
);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_unknown",
"ResponseUnknown",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn before_write_audit_failure_is_not_dispatched_and_has_no_provider_effect() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(true);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_eq!(body["code"], "not_dispatched");
assert_audit_failure(
&body,
&events,
"before_dispatch",
"BeforeDispatch",
json!({"kind":"not_dispatched"}),
"NotDispatched",
);
assert_failure_reason(&events, "audit_store_unavailable", "audit");
assert!(
body.get("failure_reason").is_none(),
"internal reason leaked: {body}"
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn rejected_write_send_with_audit_failure_is_not_dispatched_not_unknown() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
provider.receiver.close();
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_eq!(body["code"], "not_dispatched");
let receipt_id = assert_audit_failure(
&body,
&events,
"dispatch_rejected",
"DispatchRejected",
json!({"kind":"not_dispatched"}),
"NotDispatched",
);
assert_operational_failure(
&body,
&events,
"provider_dispatch_rejected",
"provider_dispatch",
"not_dispatched",
"Workdir operation was not dispatched",
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[(&receipt_id, "requested")]);
}
#[tokio::test]
async fn expired_connection_grant_is_not_dispatched_and_logs_admission_reason() {
let (h, mut provider) = AuditHttpFixture::new().await;
{
let mut registry = h.fixture.api.external_workdir_providers.lock().unwrap();
let connection = registry.connections.get_mut("external-1").unwrap();
// Keep the persisted grant online so the real handler reaches operate's
// connection-expiry fence rather than the separate session-open fence.
Arc::get_mut(connection).unwrap().expires_at =
Some(Utc::now() - chrono::Duration::seconds(1));
}
h.fail_audit(false);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_operational_failure(
&body,
&events,
"provider_grant_expired",
"provider_admission",
"unavailable",
"Workdir session is unavailable",
);
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn closed_provider_admission_is_not_dispatched_and_logs_admission_reason() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fixture
.api
.external_workdir_providers
.lock()
.unwrap()
.connection("external-1")
.unwrap()
.admission
.close();
h.fail_audit(false);
let (body, events) = h
.finish(h.start_write(), StatusCode::SERVICE_UNAVAILABLE)
.await;
assert_operational_failure(
&body,
&events,
"provider_admission_closed",
"provider_admission",
"unavailable",
"Workdir session is unavailable",
);
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[]);
}
#[tokio::test]
async fn timed_out_write_after_audit_failure_stays_unknown_and_logs_timeout_reason() {
let (mut h, mut provider) = AuditHttpFixture::new().await;
// Keep the HTTP client's independent deadline beyond the unchanged 30s
// provider timeout. Pause only after the signed HTTP request was dispatched,
// avoiding automatic virtual-time advancement during socket/SQLite setup.
h.client = reqwest::Client::builder()
.no_proxy()
.timeout(std::time::Duration::from_secs(40))
.build()
.unwrap();
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(31)).await;
tokio::time::resume();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_operational_failure(
&body,
&events,
"provider_operation_timeout",
"provider_response",
"outcome_unknown",
"Workdir operation may have effects; inspect before retrying",
);
let receipt_id = assert_audit_failure(
&body,
&events,
"response_unknown",
"ResponseUnknown",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
assert!(reply.is_closed(), "timeout retained its response receiver");
assert!(matches!(
provider.receiver.try_recv().unwrap(),
ExternalProviderCommand::Cancel { operation_id }
if operation_id.as_str() == dispatched_id.as_str()
));
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[tokio::test]
async fn successful_write_persists_requested_and_completed_with_same_operation_id() {
let (h, mut provider) = AuditHttpFixture::new().await;
let mut pending = h.start_write();
let (operation_id, reply) = provider.receive_write(&mut pending).await;
h.apply_provider_write();
reply.send(Ok(write_result())).unwrap();
let (body, events) = h.finish(pending, StatusCode::OK).await;
assert_eq!(body, serde_json::to_value(write_result()).unwrap());
assert!(
events
.iter()
.all(|event| event["event"] != "external_workdir_audit_failed")
);
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
h.assert_audits(&[(&operation_id, "requested"), (&operation_id, "completed")]);
}
#[tokio::test]
async fn provider_reported_unknown_preserves_received_response_and_unknown_effects() {
let (h, mut provider) = AuditHttpFixture::new().await;
h.fail_audit(false);
let mut pending = h.start_write();
let (dispatched_id, reply) = provider.receive_write(&mut pending).await;
reply
.send(Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::OutcomeUnknown(PROVIDER_SECRET.into()),
)))
.unwrap();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown");
let receipt_id = assert_audit_failure(
&body,
&events,
"response_received",
"ResponseReceived",
json!({"kind":"unknown"}),
"Unknown",
);
assert_eq!(receipt_id, dispatched_id);
provider.assert_dispatches(1);
h.assert_audits(&[(&dispatched_id, "requested")]);
}
#[derive(Clone, Copy, Debug)]
enum CheckoutReplyFault {
WrongOperation,
WrongPath,
EmptyValidator,
WrongOutput,
ProviderUnavailable,
ProviderTransport,
}
async fn assert_checkout_unknown_audit(fault: CheckoutReplyFault, fail_completion: bool) {
let (h, mut provider) = AuditHttpFixture::new().await;
if fail_completion {
h.fail_audit(false);
}
let target = workdir::WorkdirPath::new(PATH_SECRET).unwrap();
let mut pending = h.start_operation(WorkdirSessionOperation::CheckoutExecute(
workdir::CheckoutRequest {
target: target.clone(),
validator: vec![1],
operation: workdir::CheckoutOperation::Write {
content: WRITE_SECRET.to_vec(),
expected_hash: [0; 32],
},
},
));
let (operation_id, operation, reply) = provider.receive_operation(&mut pending).await;
let WorkdirSessionOperation::CheckoutExecute(request) = operation else {
panic!("expected checkout mutation, got {operation:?}");
};
assert_eq!(request.target, target);
assert_eq!(request.validator, vec![1]);
assert!(
matches!(request.operation, workdir::CheckoutOperation::Write { ref content, .. } if content == WRITE_SECRET)
);
h.apply_provider_write();
let result = match fault {
CheckoutReplyFault::WrongOperation => Ok(write_result()),
CheckoutReplyFault::ProviderUnavailable => Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::Unavailable(PROVIDER_SECRET.into()),
)),
CheckoutReplyFault::ProviderTransport => Err(WorkdirTransportError::from_workdir_error(
&workdir::WorkdirError::Transport(PROVIDER_SECRET.into()),
)),
_ => {
let mut result = workdir::CheckoutResult {
observation: workdir::CheckoutObservation {
path: target,
validator: vec![2],
kind: workdir::EntryKind::File,
size: WRITE_SECRET.len() as u64,
capabilities: workdir::WorkdirSessionCapabilities::READ_WRITE,
},
output: workdir::CheckoutOutput::Write(workdir::WriteResult {
bytes_written: WRITE_SECRET.len(),
created: false,
}),
};
match fault {
CheckoutReplyFault::WrongPath => {
result.observation.path =
workdir::WorkdirPath::new(PROVIDER_SECRET.trim_start_matches('/')).unwrap()
}
CheckoutReplyFault::EmptyValidator => result.observation.validator.clear(),
CheckoutReplyFault::WrongOutput => {
result.output = workdir::CheckoutOutput::Edit(workdir::EditResult {
replacements: 1,
content_hash: [0; 32],
bytes_written: 1,
})
}
_ => unreachable!(),
}
Ok(WorkdirSessionOperationResult::CheckoutExecute(result))
}
};
reply.send(result).unwrap();
let (body, events) = h.finish(pending, StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "outcome_unknown", "{fault:?}: {body}");
assert_eq!(
body["message"],
"Workdir operation may have effects; inspect before retrying"
);
if !matches!(
fault,
CheckoutReplyFault::ProviderUnavailable | CheckoutReplyFault::ProviderTransport
) {
assert_failure_reason(&events, "operation_result_mismatch", "response_validation");
}
if fail_completion {
assert_eq!(
body["audit_failures"],
json!([{
"operation_id": operation_id,
"stage": "response_received",
"reason": "audit_store_unavailable",
"operation_outcome": {"kind":"unknown"},
}]),
"{fault:?}: {body}"
);
let audit_logs: Vec<_> = events
.iter()
.filter(|e| e["event"] == "external_workdir_audit_failed")
.collect();
assert_eq!(audit_logs.len(), 1, "{events:?}");
let event = audit_logs[0];
assert_eq!(event["operation_id"], operation_id);
assert_eq!(event["operation"], "checkout_execute");
assert_eq!(event["audit_stage"], "ResponseReceived");
assert_eq!(event["operation_outcome"], "Unknown");
h.assert_operation_audits(&[(&operation_id, "requested")], "checkout_execute");
} else {
assert!(body.get("audit_failures").is_none(), "{body}");
assert!(
events
.iter()
.all(|e| e["event"] != "external_workdir_audit_failed")
);
h.assert_operation_audits(
&[
(&operation_id, "requested"),
(&operation_id, "outcome_unknown"),
],
"checkout_execute",
);
}
assert_eq!(std::fs::read(h.effect_path()).unwrap(), WRITE_SECRET);
provider.assert_dispatches(1);
}
#[tokio::test]
async fn checkout_invalid_reply_records_unknown_in_healthy_completion_audit() {
for fault in [
CheckoutReplyFault::WrongOperation,
CheckoutReplyFault::WrongPath,
CheckoutReplyFault::EmptyValidator,
CheckoutReplyFault::WrongOutput,
] {
assert_checkout_unknown_audit(fault, false).await;
}
}
#[tokio::test]
async fn checkout_invalid_reply_records_unknown_in_failed_completion_receipt_and_log() {
for fault in [
CheckoutReplyFault::WrongOperation,
CheckoutReplyFault::WrongPath,
CheckoutReplyFault::EmptyValidator,
CheckoutReplyFault::WrongOutput,
] {
assert_checkout_unknown_audit(fault, true).await;
}
}
#[tokio::test]
async fn checkout_uncertain_reply_normalization_matches_audit_outcome() {
for fault in [
CheckoutReplyFault::ProviderUnavailable,
CheckoutReplyFault::ProviderTransport,
] {
for fail_completion in [false, true] {
assert_checkout_unknown_audit(fault, fail_completion).await;
}
}
}