Merge remote-tracking branch 'origin/work/T-773-workdir-audit' into work/T-774-workdir-failure-diagnostics

# Conflicts:
#	crates/workdir/src/http.rs
This commit is contained in:
2026-10-21 06:13:35 +09:00
9 changed files with 907 additions and 63 deletions
+7 -1
View File
@@ -125,6 +125,8 @@ pub struct WorkdirOperationApiError {
pub message: String, pub message: String,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub diagnostics: Option<Vec<Diagnostic>>, 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")] #[serde(skip, default = "default_repository_error_status")]
status: u16, status: u16,
/// Server-local metadata retained by generated error response extensions, never the wire body. /// Server-local metadata retained by generated error response extensions, never the wire body.
@@ -156,6 +158,7 @@ impl WorkdirOperationApiError {
code: None, code: None,
message: error.message, message: error.message,
diagnostics: Some(error.diagnostics), diagnostics: Some(error.diagnostics),
audit_failures: Vec::new(),
status: error.status, status: error.status,
workdir_log: None, workdir_log: None,
denial_reason: None, denial_reason: None,
@@ -170,6 +173,7 @@ impl WorkdirOperationApiError {
code: Some(error.code), code: Some(error.code),
message: error.message, message: error.message,
diagnostics: None, diagnostics: None,
audit_failures: Vec::new(),
status, status,
workdir_log: None, workdir_log: None,
denial_reason: error.denial_reason, denial_reason: error.denial_reason,
@@ -7522,7 +7526,9 @@ pub struct CurrentWorkerWorkdirOperationRequest {
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[serde(transparent)] #[serde(transparent)]
pub struct CurrentWorkerWorkdirOperationResponse(pub workdir::http::WorkdirSessionOperationResult); pub struct CurrentWorkerWorkdirOperationResponse(
pub workdir::http::WorkdirSessionOperationResponse,
);
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)]
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] #[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::IsDirectory { .. }
| WorkdirError::InvalidGlob { .. } | WorkdirError::InvalidGlob { .. }
| WorkdirError::InvalidRegex { .. } => "checkout_invalid", | WorkdirError::InvalidRegex { .. } => "checkout_invalid",
WorkdirError::NotDispatched => "checkout_not_dispatched",
WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable", WorkdirError::Unavailable { .. } | WorkdirError::SessionClosed => "checkout_unavailable",
// IO/transport failures may arrive after effects have begun. Do not infer // IO/transport failures may arrive after effects have begun. Do not infer
// no-effects from their text or strip explicit OutcomeUnknown evidence. // no-effects from their text or strip explicit OutcomeUnknown evidence.
+88 -1
View File
@@ -119,6 +119,51 @@ pub enum WorkdirSessionOperationResult {
CommandCancel, 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. /// Dispatch one typed operation against any provider-neutral Workdir session.
/// ///
/// Runtime HTTP handlers, outbound External providers, and Backend brokers can /// Runtime HTTP handlers, outbound External providers, and Backend brokers can
@@ -208,6 +253,7 @@ pub enum WorkdirTransportErrorCode {
NotFound, NotFound,
Conflict, Conflict,
OutcomeUnknown, OutcomeUnknown,
NotDispatched,
Unsupported, Unsupported,
InvalidRequest, InvalidRequest,
Denied, Denied,
@@ -231,6 +277,7 @@ impl WorkdirTransportErrorCode {
Self::NotFound => "not_found", Self::NotFound => "not_found",
Self::Conflict => "conflict", Self::Conflict => "conflict",
Self::OutcomeUnknown => "outcome_unknown", Self::OutcomeUnknown => "outcome_unknown",
Self::NotDispatched => "not_dispatched",
Self::Unsupported => "unsupported", Self::Unsupported => "unsupported",
Self::InvalidRequest => "invalid_request", Self::InvalidRequest => "invalid_request",
Self::Denied => "denied", Self::Denied => "denied",
@@ -261,7 +308,7 @@ impl WorkdirTransportErrorCode {
| Self::SymlinkTargetIsDirectory | Self::SymlinkTargetIsDirectory
| Self::IsDirectory | Self::IsDirectory
| Self::SymlinkDirectoryNotTraversed => 400, | Self::SymlinkDirectoryNotTraversed => 400,
Self::Unavailable => 503, Self::Unavailable | Self::NotDispatched => 503,
Self::Io | Self::Internal | Self::OutcomeUnknown => 500, Self::Io | Self::Internal | Self::OutcomeUnknown => 500,
Self::Transport => 502, Self::Transport => 502,
} }
@@ -298,6 +345,9 @@ impl WorkdirTransportError {
Code::OutcomeUnknown, Code::OutcomeUnknown,
"Workdir operation may have effects; inspect before retrying", "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::NotFound(_) => (Code::NotFound, "Workdir path was not found"),
WorkdirError::Conflict(_) => ( WorkdirError::Conflict(_) => (
Code::Conflict, Code::Conflict,
@@ -412,6 +462,7 @@ impl WorkdirTransportError {
}, },
Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()), Code::UnknownCommand => WorkdirError::UnknownCommand("<remote>".to_string()),
Code::Unavailable => WorkdirError::Unavailable("Workdir session is unavailable".into()), Code::Unavailable => WorkdirError::Unavailable("Workdir session is unavailable".into()),
Code::NotDispatched => WorkdirError::NotDispatched,
Code::Io => WorkdirError::Io { Code::Io => WorkdirError::Io {
path: "<remote>".into(), path: "<remote>".into(),
source: std::io::Error::other("Workdir I/O operation failed"), source: std::io::Error::other("Workdir I/O operation failed"),
@@ -1215,6 +1266,37 @@ pub use client::{
mod tests { mod tests {
use super::*; 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] #[test]
fn transport_error_round_trip_keeps_public_classification() { fn transport_error_round_trip_keeps_public_classification() {
for (code, expected_status, expected_error) in [ for (code, expected_status, expected_error) in [
@@ -1234,6 +1316,11 @@ mod tests {
409, 409,
"The target file's content or existence changed since it was last observed", "The target file's content or existence changed since it was last observed",
), ),
(
WorkdirTransportErrorCode::NotDispatched,
503,
"not dispatched",
),
(WorkdirTransportErrorCode::Unsupported, 400, "unsupported"), (WorkdirTransportErrorCode::Unsupported, 400, "unsupported"),
(WorkdirTransportErrorCode::Denied, 403, "denied"), (WorkdirTransportErrorCode::Denied, 403, "denied"),
( (
+6
View File
@@ -409,6 +409,7 @@ pub enum WorkdirFailureReason {
SessionClosed, SessionClosed,
ProviderUnavailable, ProviderUnavailable,
CommandTerminalMissing, CommandTerminalMissing,
AuditStoreUnavailable,
HttpTimeout, HttpTimeout,
HttpConnect, HttpConnect,
HttpTransfer, HttpTransfer,
@@ -440,6 +441,7 @@ impl WorkdirFailureReason {
Self::SessionClosed => "session_closed", Self::SessionClosed => "session_closed",
Self::ProviderUnavailable => "provider_unavailable", Self::ProviderUnavailable => "provider_unavailable",
Self::CommandTerminalMissing => "command_terminal_missing", Self::CommandTerminalMissing => "command_terminal_missing",
Self::AuditStoreUnavailable => "audit_store_unavailable",
Self::HttpTimeout => "http_timeout", Self::HttpTimeout => "http_timeout",
Self::HttpConnect => "http_connect", Self::HttpConnect => "http_connect",
Self::HttpTransfer => "http_transfer", Self::HttpTransfer => "http_transfer",
@@ -561,6 +563,10 @@ pub enum WorkdirError {
#[error("Workdir session is unavailable: {0}")] #[error("Workdir session is unavailable: {0}")]
Unavailable(String), Unavailable(String),
/// The broker has proof that this operation never reached the provider.
#[error("Workdir operation was not dispatched")]
NotDispatched,
#[error("Workdir operation failed")] #[error("Workdir operation failed")]
OperationFailed, OperationFailed,
@@ -291,6 +291,11 @@ impl WorkspaceAttachedWorkdirSession {
.execute(request) .execute(request)
.map_err(workspace_workdir_error)?; .map_err(workspace_workdir_error)?;
if !response.is_success() { 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( return Err(
serde_json::from_str::<workdir::http::WorkdirTransportError>(&response.body) serde_json::from_str::<workdir::http::WorkdirTransportError>(&response.body)
.map(workdir::http::WorkdirTransportError::into_workdir_error) .map(workdir::http::WorkdirTransportError::into_workdir_error)
@@ -302,11 +307,14 @@ impl WorkspaceAttachedWorkdirSession {
}), }),
); );
} }
serde_json::from_str(&response.body).map_err(|error| { let response: workdir::http::WorkdirSessionOperationResponse =
WorkdirError::Transport(format!( serde_json::from_str(&response.body).map_err(|_| {
"failed to decode Workspace Workdir operation result: {error}" 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 { 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 { fn workspace_workdir_error(error: WorkspaceClientError) -> WorkdirError {
match error { match error {
WorkspaceClientError::Unavailable(message) => WorkdirError::Unavailable(message), 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] #[tokio::test]
async fn attached_session_does_not_send_worker_local_bash_spill_path() { async fn attached_session_does_not_send_worker_local_bash_spill_path() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({ let client = Arc::new(RecordingWorkspaceClient::new(vec![response(json!({
+123 -53
View File
@@ -26,6 +26,7 @@
use subjektiv::api::*; use subjektiv::api::*;
mod backend_job_profile; mod backend_job_profile;
mod external_operation_audit;
mod subject_authorization; mod subject_authorization;
mod subjektiv_job; mod subjektiv_job;
mod subjektiv_journal; mod subjektiv_journal;
@@ -933,6 +934,10 @@ struct ExternalProviderAuditContext {
/// output confirmation from the old connection; unconfirmed cleanup is not reported as success. /// output confirmation from the old connection; unconfirmed cleanup is not reported as success.
/// Only confirmed provider shutdown can replace per-command confirmation. Reconnection does /// Only confirmed provider shutdown can replace per-command confirmation. Reconnection does
/// not prove that missing commands have stopped. /// not prove that missing commands have stopped.
///
/// Required request auditing fails closed before dispatch. Completion audit failure cannot
/// undo a provider result: it is emitted independently and retained in the current API
/// response, including successful command handles. Audit loss never triggers a resend.
struct ExternalProviderWorkdirSession { struct ExternalProviderWorkdirSession {
workdir: workdir::Workdir, workdir: workdir::Workdir,
connection: Arc<ExternalProviderConnection>, connection: Arc<ExternalProviderConnection>,
@@ -991,18 +996,22 @@ impl ExternalProviderWorkdirSession {
async fn audit_operation( async fn audit_operation(
&self, &self,
operation_id: &ExternalWorkdirOperationId, operation_id: &ExternalWorkdirOperationId,
operation_kind: &str, operation_kind: &'static str,
outcome: &str, outcome: &str,
) -> std::result::Result<(), workdir::WorkdirError> { stage: workdir::http::WorkdirAuditStage,
operation_outcome: workdir::http::WorkdirAuditOperationOutcome,
) -> std::result::Result<(), workdir::http::WorkdirAuditFailure> {
let Some(audit) = &self.audit else { let Some(audit) = &self.audit else {
return Ok(()); return Ok(());
}; };
let audit = audit.clone(); let audit = audit.clone();
let operation_id_for_notification = operation_id.as_str().to_owned();
let operation_kind_for_notification = operation_kind;
let operation_id = operation_id.clone(); let operation_id = operation_id.clone();
let operation_kind = operation_kind.to_owned(); let operation_kind = operation_kind.to_owned();
let outcome = outcome.to_owned(); let outcome = outcome.to_owned();
let connection = self.connection.clone(); let connection = self.connection.clone();
audit let result = audit
.store .store
.db_executor() .db_executor()
.run(move || { .run(move || {
@@ -1024,12 +1033,17 @@ impl ExternalProviderWorkdirSession {
&Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), &Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
) )
}) })
.await .await;
.map_err(|error| { result.map_err(|_| {
workdir::WorkdirError::Unavailable(format!( let failure = workdir::http::WorkdirAuditFailure {
"External Workdir operation audit failed: {error}" operation_id: operation_id_for_notification,
)) stage,
}) operation_outcome,
reason: workdir::http::WorkdirAuditFailureReason::AuditStoreUnavailable,
};
external_operation_audit::notify(failure.clone(), operation_kind_for_notification);
failure
})
} }
async fn operate( async fn operate(
@@ -1079,8 +1093,19 @@ impl ExternalProviderWorkdirSession {
} }
let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7())) let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7()))
.map_err(workdir::WorkdirError::InvalidArgument)?; .map_err(workdir::WorkdirError::InvalidArgument)?;
self.audit_operation(&operation_id, operation_kind, "requested") use workdir::http::{WorkdirAuditOperationOutcome as Outcome, WorkdirAuditStage as Stage};
.await?; self.audit_operation(
&operation_id,
operation_kind,
"requested",
Stage::BeforeDispatch,
Outcome::NotDispatched,
)
.await
.map_err(|_| {
workdir::WorkdirError::NotDispatched
.with_failure_reason(workdir::WorkdirFailureReason::AuditStoreUnavailable)
})?;
let (response, receive) = tokio::sync::oneshot::channel(); let (response, receive) = tokio::sync::oneshot::channel();
let mut cancel_guard = ExternalOperationCancelGuard { let mut cancel_guard = ExternalOperationCancelGuard {
sender: self.connection.sender.clone(), sender: self.connection.sender.clone(),
@@ -1097,39 +1122,72 @@ impl ExternalProviderWorkdirSession {
.await .await
.is_err() .is_err()
{ {
self.audit_operation(&operation_id, operation_kind, "provider_unavailable") cancel_guard.disarm();
.await?; // A rejected channel send proves there was no provider dispatch.
return Err(workdir::WorkdirError::Unavailable( let _audit_result = self
"External Workdir provider disconnected".to_string(), .audit_operation(
)); &operation_id,
operation_kind,
"not_dispatched",
Stage::DispatchRejected,
Outcome::NotDispatched,
)
.await;
return Err(workdir::WorkdirError::NotDispatched
.with_failure_reason(workdir::WorkdirFailureReason::ProviderUnavailable));
} }
let result = match tokio::time::timeout(operation_timeout, receive).await { let (result, stage) = match tokio::time::timeout(operation_timeout, receive).await {
Ok(result) => { Ok(result) => {
cancel_guard.disarm(); cancel_guard.disarm();
match result { match result {
Ok(result) => result.map_err(WorkdirTransportError::into_workdir_error), Ok(result) => (
Err(_) => Err(workdir::WorkdirError::Unavailable( result.map_err(WorkdirTransportError::into_workdir_error),
"External Workdir provider disconnected".to_string(), Stage::ResponseReceived,
)), ),
Err(_) => (
Err(workdir::WorkdirError::OutcomeUnknown(
"External Workdir provider response lost; inspect before retrying"
.to_string(),
)),
Stage::ResponseUnknown,
),
} }
} }
Err(_) => { Err(_) => {
cancel_guard.cancel(); cancel_guard.cancel();
Err(workdir::WorkdirError::Unavailable( (
"External Workdir provider operation timed out".to_string(), Err(workdir::WorkdirError::OutcomeUnknown(
)) "External Workdir provider response timed out; inspect before retrying"
.to_string(),
)),
Stage::ResponseUnknown,
)
} }
}; };
self.audit_operation( let outcome = match &result {
&operation_id, Ok(_) => Outcome::Succeeded,
operation_kind, Err(error)
if result.is_ok() { if matches!(
"completed" error.classification_source(),
} else { workdir::WorkdirError::OutcomeUnknown(_)
"failed" ) =>
{
Outcome::Unknown
}
Err(error) => Outcome::Failed {
code: WorkdirTransportError::from_workdir_error(error).code,
}, },
) };
.await?; let audit_outcome = match outcome {
Outcome::Succeeded => "completed",
Outcome::Failed { .. } => "failed",
Outcome::Unknown => "outcome_unknown",
Outcome::NotDispatched => unreachable!("provider result follows dispatch"),
};
// Failure is notified independently, never substituted for the provider result.
let _audit_result = self
.audit_operation(&operation_id, operation_kind, audit_outcome, stage, outcome)
.await;
result result
} }
@@ -13169,26 +13227,38 @@ impl server_api::ServerApi for ServerApiContractService {
), ),
..Default::default() ..Default::default()
}; };
let Json(response) = scoped_execute_current_worker_workdir_operation( let (result, audit_failures) =
State( external_operation_audit::capture(scoped_execute_current_worker_workdir_operation(
self.workspace_api() State(
.map_err(server_api::WorkdirOperationApiError::api)? self.workspace_api()
.clone(), .map_err(server_api::WorkdirOperationApiError::api)?
), .clone(),
AxumPath(ScopedWorkspacePath { workspace_id }), ),
current_worker_contract_headers(&context).map_err(|error| { AxumPath(ScopedWorkspacePath { workspace_id }),
let mut error = server_api::WorkdirOperationApiError::api(error); current_worker_contract_headers(&context).map_err(|error| {
error.workdir_log = Some(fallback_log); let mut error = server_api::WorkdirOperationApiError::api(error);
error error.workdir_log = Some(fallback_log);
})?, error
Json(WorkspaceWorkdirSessionOperationRequest { })?,
target_workdir: request.target_workdir, Json(WorkspaceWorkdirSessionOperationRequest {
operation: request.operation, target_workdir: request.target_workdir,
}), operation: request.operation,
) }),
.await ))
.map_err(InternalWorkdirOperationError::into_contract_error)?; .await;
Ok(server_api::CurrentWorkerWorkdirOperationResponse(response)) match result {
Ok(Json(result)) => Ok(server_api::CurrentWorkerWorkdirOperationResponse(
workdir::http::WorkdirSessionOperationResponse {
result,
audit_failures,
},
)),
Err(error) => {
let mut error = error.into_contract_error();
error.audit_failures = audit_failures;
Err(error)
}
}
} }
async fn external_workdir_grant_create( async fn external_workdir_grant_create(
@@ -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));
}
@@ -1,3 +1,5 @@
mod audit_tests;
use super::*; use super::*;
use sha2::Digest; use sha2::Digest;
use std::sync::atomic::AtomicI64; use std::sync::atomic::AtomicI64;
@@ -459,7 +461,7 @@ async fn provider_error_and_disconnect_release_dispatch_admission() {
assert_json( assert_json(
result, result,
if disconnect { if disconnect {
StatusCode::SERVICE_UNAVAILABLE StatusCode::INTERNAL_SERVER_ERROR
} else { } else {
StatusCode::FORBIDDEN 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; let (long, _reply) = f.start_read("1", &mut a).await;
tokio::time::pause(); tokio::time::pause();
tokio::time::advance(std::time::Duration::from_secs(31)).await; tokio::time::advance(std::time::Duration::from_secs(31)).await;
let body = assert_json(long.await.unwrap(), StatusCode::SERVICE_UNAVAILABLE).await; let body = assert_json(long.await.unwrap(), StatusCode::INTERNAL_SERVER_ERROR).await;
assert_eq!(body["code"], "unavailable"); assert_eq!(body["code"], "outcome_unknown");
tokio::time::resume(); tokio::time::resume();
let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await; let gate = f.server.admission_for_workspace(TEST_WORKSPACE_ID).await;
drop( drop(
@@ -0,0 +1,546 @@
//! 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_write(&mut self, pending: &mut PendingHttpWrite) -> (String, 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;
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.as_str().to_owned(), 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 {
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: WorkdirSessionOperation::Write(workdir::WriteRequest {
path: workdir::WorkdirPath::new(PATH_SECRET).unwrap(),
content: WRITE_SECRET.to_vec(),
expected_hash: None,
}),
})
.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,
pending.proof.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)]) {
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("operation_kind=write"), "{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) {
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:?}");
}
#[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_eq!(body["code"], "outcome_unknown");
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");
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_failure_reason(&events, "provider_unavailable");
assert!(!h.effect_path().exists());
provider.assert_dispatches(0);
h.assert_audits(&[(&receipt_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")]);
}