From b1361d86e9f8f6ce733f22fc4bba8c341a5a36a9 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 21 Oct 2026 06:39:54 +0900 Subject: [PATCH] fix: distinguish external Workdir admission dispatch and response failures Integrate the published T773 audit contract without overwriting operation outcomes. Preserve bounded admission, send refusal, response loss/timeout and invalid result reasons independently of audit failures and request stages. Signed HTTP/SQLite/log fault tests pass (10), API logging 14 and relevant focused assertions pass. Root check, docs, generated artifacts, fmt/diff and state-contract naming checks pass. Full six-package normal suite ran with no-fail-fast but is not green: long Unix socket fixture paths and existing shared close assertion plus environment-isolation failures are recorded in target/t774 logs. Isolated focused checks resolved the latter; socket validation requires a granted shorter test root. This is a partial checkpoint, not a reviewed completion. --- crates/workdir/src/lib.rs | 22 +++ crates/workspace-server/src/hosts.rs | 11 +- crates/workspace-server/src/server.rs | 32 +++- .../workspace_admission_tests/audit_tests.rs | 161 +++++++++++++++++- 4 files changed, 212 insertions(+), 14 deletions(-) diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 7aecce776..c3c3d97e8 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -408,6 +408,11 @@ impl From<&str> for WorkdirDenial { pub enum WorkdirFailureReason { SessionClosed, ProviderUnavailable, + ProviderGrantExpired, + ProviderAdmissionClosed, + ProviderDispatchRejected, + ProviderResponseLost, + ProviderOperationTimeout, CommandTerminalMissing, AuditStoreUnavailable, HttpTimeout, @@ -440,6 +445,11 @@ 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", @@ -474,6 +484,10 @@ impl WorkdirFailureReason { 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 @@ -505,6 +519,10 @@ impl WorkdirFailureReason { pub enum WorkdirFailureStage { Session, Provider, + ProviderAdmission, + ProviderDispatch, + ProviderResponse, + Audit, Command, Http, ResponseValidation, @@ -518,6 +536,10 @@ impl WorkdirFailureStage { 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", diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 7fbb8bf25..da0b02671 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6854,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(); } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 753526c71..11a33bbc6 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1062,7 +1062,8 @@ impl ExternalProviderWorkdirSession { { return Err(workdir::WorkdirError::Unavailable( "External Workdir grant has expired".to_string(), - )); + ) + .with_failure_reason(workdir::WorkdirFailureReason::ProviderGrantExpired)); } // Cancellation must reach the provider even when waiting output readers // occupy every data-operation slot. The socket separately bounds control requests. @@ -1079,6 +1080,7 @@ impl ExternalProviderWorkdirSession { workdir::WorkdirError::Unavailable( "External Workdir provider admission is closed".to_string(), ) + .with_failure_reason(workdir::WorkdirFailureReason::ProviderAdmissionClosed) })?, ) }; @@ -1089,7 +1091,8 @@ impl ExternalProviderWorkdirSession { { return Err(workdir::WorkdirError::Unavailable( "External Workdir grant has expired".to_string(), - )); + ) + .with_failure_reason(workdir::WorkdirFailureReason::ProviderGrantExpired)); } let operation_id = ExternalWorkdirOperationId::new(format!("op-{}", Uuid::now_v7())) .map_err(workdir::WorkdirError::InvalidArgument)?; @@ -1134,7 +1137,7 @@ impl ExternalProviderWorkdirSession { ) .await; return Err(workdir::WorkdirError::NotDispatched - .with_failure_reason(workdir::WorkdirFailureReason::ProviderUnavailable)); + .with_failure_reason(workdir::WorkdirFailureReason::ProviderDispatchRejected)); } let (result, stage) = match tokio::time::timeout(operation_timeout, receive).await { Ok(result) => { @@ -1148,7 +1151,8 @@ impl ExternalProviderWorkdirSession { Err(workdir::WorkdirError::OutcomeUnknown( "External Workdir provider response lost; inspect before retrying" .to_string(), - )), + ) + .with_failure_reason(workdir::WorkdirFailureReason::ProviderResponseLost)), Stage::ResponseUnknown, ), } @@ -1159,7 +1163,8 @@ impl ExternalProviderWorkdirSession { Err(workdir::WorkdirError::OutcomeUnknown( "External Workdir provider response timed out; inspect before retrying" .to_string(), - )), + ) + .with_failure_reason(workdir::WorkdirFailureReason::ProviderOperationTimeout)), Stage::ResponseUnknown, ) } @@ -1328,6 +1333,7 @@ impl ExternalProviderWorkdirSession { workdir::WorkdirError::Unavailable(format!( "External Workdir provider returned a mismatched result; expected {expected}" )) + .with_failure_reason(workdir::WorkdirFailureReason::OperationResultMismatch) } } @@ -1457,7 +1463,8 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession { } Ok(_) if mutation => Err(workdir::WorkdirError::OutcomeUnknown( "mismatched checkout mutation response".into(), - )), + ) + .with_failure_reason(workdir::WorkdirFailureReason::OperationResultMismatch)), Ok(_) => Err(Self::mismatch("checkout_execute")), Err(error) if mutation @@ -37863,10 +37870,19 @@ mod tests { ))) .unwrap(); } + let error = write.await.unwrap().unwrap_err(); assert!(matches!( - write.await.unwrap(), - Err(workdir::WorkdirError::OutcomeUnknown(_)) + error.classification_source(), + workdir::WorkdirError::OutcomeUnknown(_) )); + assert_eq!( + error.failure_reason(), + Some(if lose_response { + workdir::WorkdirFailureReason::ProviderResponseLost + } else { + workdir::WorkdirFailureReason::OperationResultMismatch + }) + ); } } diff --git a/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs index 212198b6f..a6f76f548 100644 --- a/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs +++ b/crates/workspace-server/src/server/tests/workspace_admission_tests/audit_tests.rs @@ -260,6 +260,9 @@ impl AuditHttpFixture { PATH_SECRET, PROVIDER_SECRET, 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}"); @@ -363,13 +366,33 @@ fn assert_audit_failure( operation_id.to_owned() } -fn assert_failure_reason(events: &[Value], reason: &str) { +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] @@ -436,7 +459,14 @@ async fn lost_write_response_after_audit_failure_stays_unknown_without_repeating // 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"); + 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, @@ -467,7 +497,7 @@ async fn before_write_audit_failure_is_not_dispatched_and_has_no_provider_effect json!({"kind":"not_dispatched"}), "NotDispatched", ); - assert_failure_reason(&events, "audit_store_unavailable"); + assert_failure_reason(&events, "audit_store_unavailable", "audit"); assert!( body.get("failure_reason").is_none(), "internal reason leaked: {body}" @@ -494,12 +524,135 @@ async fn rejected_write_send_with_audit_failure_is_not_dispatched_not_unknown() json!({"kind":"not_dispatched"}), "NotDispatched", ); - assert_failure_reason(&events, "provider_unavailable"); + 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;