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

fix: integrate structured Workdir refusal diagnostics

Browse Source 
Merge independently approved T-731 source 8af492ff into current develop after T-732. No conflicts or source edits. Integration root locked/offline check, eight affected crates, Workdir default/http-client, Server 742+11 (one existing ignored), API schema freshness, fmt/diff passed. Historical Workdir403 remains unidentified; HTTP-only feature-gating limitation remains separate.
This commit is contained in:
Hare
2026-10-10 14:40:05 +09:00
parent ac01f64f4d 8af492ff84
commit 72ce817de4
34 changed files with 3028 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/workspace-server/src/hosts.rs
+14 -3
View File
@@ -794,6 +794,10 @@ pub struct RuntimeSessionAttachment {
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RuntimeRegistryError {
WorkdirSessionOpenFailed {
runtime_id: String,
transport: workdir::http::WorkdirTransportError,
},
InvalidIdentifier {
kind: &'static str,
value: String,
@@ -823,11 +827,19 @@ impl RuntimeRegistryError {
worker.worker_id, worker.runtime_id
),
Self::RuntimeOperationFailed { message, .. } => message.clone(),
Self::WorkdirSessionOpenFailed { transport, .. } => transport.message.clone(),
}
}
pub fn into_error(self) -> Error {
match self {
Self::WorkdirSessionOpenFailed {
runtime_id,
transport,
} => Error::WorkdirSessionOpenFailed {
runtime_id,
transport,
},
Self::InvalidIdentifier { kind, value } => Error::InvalidRuntimeIdentifier {
kind: kind.to_string(),
value,
@@ -1885,10 +1897,9 @@ impl RuntimeRegistry {
runtime
.open_workdir_session(working_directory_id, owner_worker_id)
.await
.map_err(|error| RuntimeRegistryError::RuntimeOperationFailed {
.map_err(|error| RuntimeRegistryError::WorkdirSessionOpenFailed {
runtime_id: runtime_id.to_string(),
code: "workdir_session_open_failed".to_string(),
message: error.to_string(),
transport: workdir::http::WorkdirTransportError::from_workdir_error(&error),
})
}
crates/workspace-server/src/lib.rs
+5
View File
@@ -106,6 +106,11 @@ pub enum Error {
code: String,
message: String,
},
#[error("runtime `{runtime_id}` operation failed (workdir_session_open_failed): {}", transport.message)]
WorkdirSessionOpenFailed {
runtime_id: String,
transport: workdir::http::WorkdirTransportError,
},
#[error("runtime `{runtime_id}` does not support `{capability}`")]
RuntimeCapabilityUnsupported {
runtime_id: String,
crates/workspace-server/src/server.rs
+231 -67
View File
@@ -1225,12 +1225,23 @@ impl workdir::WorkdirSession for ExternalProviderWorkdirSession {
"mismatched checkout mutation response".into(),
)),
Ok(_) => Err(Self::mismatch("checkout_execute")),
Err(workdir::WorkdirError::Unavailable(_) | workdir::WorkdirError::Transport(_))
if mutation =>
Err(error)
if mutation
&& matches!(
error.classification_source(),
workdir::WorkdirError::Unavailable(_) | workdir::WorkdirError::Transport(_)
) =>
{
Err(workdir::WorkdirError::OutcomeUnknown(
let outcome = workdir::WorkdirError::OutcomeUnknown(
"checkout provider response lost; inspect effects before retry".into(),
))
);
Err(match error.denial_reason() {
Some(reason) => workdir::WorkdirError::DenialContext {
reason,
source: Box::new(outcome),
},
None => outcome,
})
}
Err(error) => Err(error),
}
@@ -11436,6 +11447,15 @@ impl server_api::ServerApi for ServerApiContractService {
server_api::CurrentWorkerWorkdirOperationResponse,
server_api::WorkdirOperationApiError,
> {
let fallback_log = server_api::WorkdirOperationLogContext {
operation: workdir_operation_label(&request.operation),
stage: "worker_identity",
workspace_id_hash: crate::worker_source::diagnostic_id_hash(&workspace_id),
attachment_alias_hash: crate::worker_source::diagnostic_id_hash(
&request.target_workdir,
),
..Default::default()
};
let Json(response) = scoped_execute_current_worker_workdir_operation(
State(
self.workspace_api()
@@ -11443,22 +11463,18 @@ impl server_api::ServerApi for ServerApiContractService {
.clone(),
),
AxumPath(ScopedWorkspacePath { workspace_id }),
current_worker_contract_headers(&context)
.map_err(server_api::WorkdirOperationApiError::api)?,
current_worker_contract_headers(&context).map_err(|error| {
let mut error = server_api::WorkdirOperationApiError::api(error);
error.workdir_log = Some(fallback_log);
error
})?,
Json(WorkspaceWorkdirSessionOperationRequest {
target_workdir: request.target_workdir,
operation: request.operation,
}),
)
.await
.map_err(|error| match error {
InternalWorkdirOperationError::Api(error) => {
server_api::WorkdirOperationApiError::api(error.into_repository_api_error())
}
InternalWorkdirOperationError::Provider(error) => {
server_api::WorkdirOperationApiError::provider(error)
}
})?;
.map_err(InternalWorkdirOperationError::into_contract_error)?;
Ok(server_api::CurrentWorkerWorkdirOperationResponse(response))
}
@@ -11700,6 +11716,10 @@ fn build_inner_router(api: WorkspaceApi) -> Router {
async fn log_failed_api_response(request: Request, next: Next) -> Response {
let method = request.method().clone();
let uri = request.uri().clone();
let request_token_id_hash = request
.extensions()
.get::<crate::worker_source::VerifiedRuntimeRequestSource>()
.and_then(|source| source.token_id_hash.clone());
let response = next.run(request).await;
let status = response.status();
@@ -11720,11 +11740,23 @@ async fn log_failed_api_response(request: Request, next: Next) -> Response {
.get::<ApiErrorLog>()
.or(generated_error.as_ref());
let event = api_failure_log_event(&method, &uri, status, error);
let context = error.and_then(|error| error.workdir_log.as_ref());
let workdir_path = context.map(|_| bounded_auth_log_value(uri.path(), 512));
let workdir_method = context.map(|_| bounded_auth_log_value(method.as_str(), 32));
tracing::error!(
target: "yoi::api",
event = event.event,
method = event.method,
path = event.path,
method = workdir_method.as_deref().unwrap_or(event.method),
path = workdir_path.as_deref().unwrap_or(event.path),
operation = context.map(|ctx| ctx.operation),
workdir_stage = context.map(|ctx| ctx.stage),
workspace_id_hash = context.map(|ctx| ctx.workspace_id_hash.as_str()),
runtime_id_hash = context.and_then(|ctx| ctx.runtime_id_hash.as_deref()),
worker_id_hash = context.and_then(|ctx| ctx.worker_id_hash.as_deref()),
workdir_id_hash = context.and_then(|ctx| ctx.workdir_id_hash.as_deref()),
attachment_alias_hash = context.map(|ctx| ctx.attachment_alias_hash.as_str()),
request_token_id_hash = context.and(request_token_id_hash.as_deref()),
denial_reason = error.and_then(|error| error.denial_reason).map(|reason| reason.as_str()),
status = event.status,
kind = event.kind.unwrap_or("unknown"),
message = event.message.unwrap_or(""),
@@ -17202,9 +17234,10 @@ fn ticket_notification_delivery_error_category(
Err(RuntimeRegistryError::UnknownRuntime(_)) => Some("unknown_runtime"),
Err(RuntimeRegistryError::UnknownHost(_)) => Some("unknown_host"),
Err(RuntimeRegistryError::UnknownWorker { .. }) => Some("unknown_worker"),
Err(RuntimeRegistryError::RuntimeOperationFailed { .. }) => {
Some("runtime_operation_failed")
}
Err(
RuntimeRegistryError::WorkdirSessionOpenFailed { .. }
| RuntimeRegistryError::RuntimeOperationFailed { .. },
) => Some("runtime_operation_failed"),
}
}
@@ -18195,6 +18228,33 @@ fn validated_current_worker_attachment(
enum InternalWorkdirOperationError {
Api(ApiError),
Provider(WorkdirTransportError),
Contextual {
error: Box<InternalWorkdirOperationError>,
context: server_api::WorkdirOperationLogContext,
},
}
impl InternalWorkdirOperationError {
fn into_contract_error(self) -> server_api::WorkdirOperationApiError {
match self {
Self::Api(error) => {
let reason = match &error.error {
Error::WorkdirSessionOpenFailed { transport, .. } => transport.denial_reason,
_ => None,
};
let mut error =
server_api::WorkdirOperationApiError::api(error.into_repository_api_error());
error.denial_reason = reason;
error
}
Self::Provider(error) => server_api::WorkdirOperationApiError::provider(error),
Self::Contextual { error, context } => {
let mut error = error.into_contract_error();
error.workdir_log = Some(context);
error
}
}
}
}
impl From<ApiError> for InternalWorkdirOperationError {
@@ -18217,21 +18277,35 @@ impl From<WorkdirTransportError> for InternalWorkdirOperationError {
impl IntoResponse for InternalWorkdirOperationError {
fn into_response(self) -> Response {
match self {
Self::Api(error) => error.into_response(),
Self::Provider(error) => {
let status = StatusCode::from_u16(error.code.http_status())
.expect("Workdir transport error status is valid");
let log = ApiErrorLog {
kind: format!("workdir_session_operation_{}", error.code.as_str()),
message: error.message.clone(),
diagnostics: Vec::new(),
};
let mut response = (status, Json(error)).into_response();
response.extensions_mut().insert(log);
response
}
}
let error = self.into_contract_error();
let status = StatusCode::from_u16(api_macros::HttpError::status_code(&error))
.expect("Workdir error status is valid");
let log = ApiErrorLog::from_workdir_operation_error(&error, status);
let mut response = (status, Json(error)).into_response();
response.extensions_mut().insert(log);
response
}
}
fn workdir_operation_label(operation: &WorkdirSessionOperation) -> &'static str {
match operation {
WorkdirSessionOperation::AuthorizeScope(_) => "authorize_scope",
WorkdirSessionOperation::ScopeRulesOverlap(_) => "scope_rules_overlap",
WorkdirSessionOperation::CheckoutSearch(_) => "checkout_search",
WorkdirSessionOperation::CheckoutObserve(_) => "checkout_observe",
WorkdirSessionOperation::CheckoutExecute(_) => "checkout_execute",
WorkdirSessionOperation::Stat(_) => "stat",
WorkdirSessionOperation::Read(_) => "read",
WorkdirSessionOperation::ReadBytes(_) => "read_bytes",
WorkdirSessionOperation::Write(_) => "write",
WorkdirSessionOperation::Edit(_) => "edit",
WorkdirSessionOperation::List(_) => "list",
WorkdirSessionOperation::Glob(_) => "glob",
WorkdirSessionOperation::Grep(_) => "grep",
WorkdirSessionOperation::CommandStart(_) => "command_start",
WorkdirSessionOperation::CommandStatus(_) => "command_status",
WorkdirSessionOperation::CommandOutput(_) => "command_output",
WorkdirSessionOperation::CommandCancel(_) => "command_cancel",
}
}
@@ -18241,19 +18315,35 @@ async fn scoped_execute_current_worker_workdir_operation(
headers: HeaderMap,
Json(request): Json<WorkspaceWorkdirSessionOperationRequest>,
) -> std::result::Result<Json<WorkdirSessionOperationResult>, InternalWorkdirOperationError> {
let mut context = server_api::WorkdirOperationLogContext {
operation: workdir_operation_label(&request.operation),
workspace_id_hash: crate::worker_source::diagnostic_id_hash(&path.workspace_id),
attachment_alias_hash: crate::worker_source::diagnostic_id_hash(&request.target_workdir),
..Default::default()
};
let result = async {
context.stage = "workspace_scope";
validate_workspace_scope(&api, &path.workspace_id)?;
context.stage = "worker_identity";
let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
context.runtime_id_hash = Some(crate::worker_source::diagnostic_id_hash(&worker.runtime_id));
context.worker_id_hash = Some(crate::worker_source::diagnostic_id_hash(&worker.worker_id));
let target_workdir = request.target_workdir;
let result = match request.operation {
WorkdirSessionOperation::CommandStart(command) => {
let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await;
context.stage = "attachment_validation";
let link = validated_current_worker_attachment(&api, &worker, &target_workdir)?;
context.workdir_id_hash = Some(crate::worker_source::diagnostic_id_hash(&link.workdir_id));
context.stage = "session_open";
let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
context.stage = "provider_dispatch";
let provider_handle = source
.start_command(command)
.await
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?;
.map_err(workdir_operation_transport_error)?;
context.stage = "command_registration";
let registered_source = api
.workdir_sessions
.lock()
@@ -18279,31 +18369,31 @@ async fn scoped_execute_current_worker_workdir_operation(
}
WorkdirSessionOperation::CommandStatus(external_handle) => {
let (session, provider_handle) =
current_worker_command_session(&api, &worker, &target_workdir, &external_handle)?;
current_worker_command_session(&api, &worker, &target_workdir, &external_handle, &mut context)?;
session
.command_status(provider_handle)
.await
.map(WorkdirSessionOperationResult::CommandStatus)
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?
.map_err(workdir_operation_transport_error)?
}
WorkdirSessionOperation::CommandOutput(mut output) => {
let (session, provider_handle) =
current_worker_command_session(&api, &worker, &target_workdir, &output.handle)?;
current_worker_command_session(&api, &worker, &target_workdir, &output.handle, &mut context)?;
output.handle = provider_handle;
session
.command_output(output)
.await
.map(WorkdirSessionOperationResult::CommandOutput)
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?
.map_err(workdir_operation_transport_error)?
}
WorkdirSessionOperation::CommandCancel(external_handle) => {
let (session, provider_handle) =
current_worker_command_session(&api, &worker, &target_workdir, &external_handle)?;
current_worker_command_session(&api, &worker, &target_workdir, &external_handle, &mut context)?;
session
.cancel_command(provider_handle)
.await
.map(|()| WorkdirSessionOperationResult::CommandCancel)
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?
.map_err(workdir_operation_transport_error)?
}
operation @ (WorkdirSessionOperation::AuthorizeScope(_)
| WorkdirSessionOperation::ScopeRulesOverlap(_)
@@ -18320,14 +18410,23 @@ async fn scoped_execute_current_worker_workdir_operation(
| WorkdirSessionOperation::Grep(_)) => {
let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await;
context.stage = "attachment_validation";
let link = validated_current_worker_attachment(&api, &worker, &target_workdir)?;
context.workdir_id_hash = Some(crate::worker_source::diagnostic_id_hash(&link.workdir_id));
context.stage = "session_open";
let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
context.stage = "provider_dispatch";
execute_workdir_session_operation(&source, operation)
.await
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?
.map_err(workdir_operation_transport_error)?
}
};
Ok(Json(result))
}.await;
result.map_err(|error| InternalWorkdirOperationError::Contextual {
error: Box::new(error),
context,
})
}
fn current_worker_command_session(
@@ -18335,26 +18434,27 @@ fn current_worker_command_session(
worker: &RuntimeWorkerRef,
target_workdir: &str,
external_handle: &CommandHandle,
context: &mut server_api::WorkdirOperationLogContext,
) -> std::result::Result<(WorkdirSessionHandle, CommandHandle), InternalWorkdirOperationError> {
let _link = validated_current_worker_attachment(api, worker, target_workdir)?;
context.stage = "attachment_validation";
let link = validated_current_worker_attachment(api, worker, target_workdir)?;
context.workdir_id_hash = Some(crate::worker_source::diagnostic_id_hash(&link.workdir_id));
context.stage = "command_lookup";
let command = api
.workdir_sessions
.lock()
.expect("Workdir session registry lock poisoned")
.command(worker, target_workdir, external_handle)
.ok_or_else(|| {
InternalWorkdirOperationError::Provider(current_worker_workdir_operation_error(
worker,
InternalWorkdirOperationError::Provider(workdir_operation_transport_error(
workdir::WorkdirError::UnknownCommand(external_handle.0.clone()),
))
})?;
context.stage = "provider_dispatch";
Ok((command.source, command.provider_handle))
}
fn current_worker_workdir_operation_error(
_worker: &RuntimeWorkerRef,
error: workdir::WorkdirError,
) -> WorkdirTransportError {
fn workdir_operation_transport_error(error: workdir::WorkdirError) -> WorkdirTransportError {
WorkdirTransportError::from_workdir_error(&error)
}
@@ -23453,6 +23553,7 @@ async fn serve_external_workdir_provider(
Some(ExternalProviderCommand::Operation { operation_id, operation, response }) => {
if revoking {
let _ = response.send(Err(WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::Unavailable,
message: "External Workdir provider shutdown is in progress".to_string(),
}));
@@ -23460,6 +23561,7 @@ async fn serve_external_workdir_provider(
}
if pending.len() >= 16 {
let _ = response.send(Err(WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::Unavailable,
message: "External Workdir provider is at its in-flight operation limit".to_string(),
}));
@@ -23470,6 +23572,7 @@ async fn serve_external_workdir_provider(
Ok(operation) => operation,
Err(message) => {
let _ = response.send(Err(WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::InvalidRequest,
message,
}));
@@ -23526,6 +23629,7 @@ async fn serve_external_workdir_provider(
Err(error.into_transport_error())
}
ExternalWorkdirOperationOutcome::Cancelled => Err(WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::Unavailable,
message: "External Workdir operation was cancelled".to_string(),
}),
@@ -24007,6 +24111,9 @@ fn workdir_runtime_failure_code(error: &RuntimeRegistryError) -> String {
RuntimeRegistryError::UnknownRuntime(_) | RuntimeRegistryError::UnknownHost(_) => {
"runtime_unavailable".to_string()
}
RuntimeRegistryError::WorkdirSessionOpenFailed { .. } => {
"workdir_session_open_failed".to_string()
}
RuntimeRegistryError::RuntimeOperationFailed { code, .. } => code.clone(),
RuntimeRegistryError::InvalidIdentifier { .. } => "invalid_runtime".to_string(),
RuntimeRegistryError::UnknownWorker { .. } => "runtime_workdir_lookup_failed".to_string(),
@@ -24518,6 +24625,7 @@ fn runtime_reports_workdir_not_found(result: &crate::hosts::RuntimeWorkingDirect
fn classify_workdir_provider_error(error: &RuntimeRegistryError) -> (&'static str, bool) {
match error {
RuntimeRegistryError::WorkdirSessionOpenFailed { .. } => ("provider_unavailable", true),
RuntimeRegistryError::UnknownRuntime(_) => ("runtime_unavailable", true),
RuntimeRegistryError::RuntimeOperationFailed { code, .. } => {
let category =
@@ -32556,6 +32664,8 @@ struct ApiErrorLog {
kind: String,
message: String,
diagnostics: Vec<RuntimeDiagnostic>,
workdir_log: Option<server_api::WorkdirOperationLogContext>,
denial_reason: Option<workdir::WorkdirDenialReason>,
}
impl ApiErrorLog {
@@ -32568,9 +32678,11 @@ impl ApiErrorLog {
kind: format!("workdir_session_operation_{}", code.as_str()),
message: sanitize_backend_error(&error.message),
diagnostics: Vec::new(),
workdir_log: error.workdir_log.clone(),
denial_reason: error.denial_reason,
};
}
Self::from_repository_error(&server_api::RepositoryApiError::new(
let mut log = Self::from_repository_error(&server_api::RepositoryApiError::new(
status.as_u16(),
error.error.as_deref().unwrap_or_else(|| {
status
@@ -32579,11 +32691,25 @@ impl ApiErrorLog {
}),
&error.message,
error.diagnostics.clone().unwrap_or_default(),
))
));
log.workdir_log = error.workdir_log.clone();
log.denial_reason = error.denial_reason;
if log.workdir_log.is_some() {
// Existing public API details are unchanged; do not copy request-derived
// alias/handle text or session-open errors into operational diagnostics.
log.kind = format!("workdir_session_operation_api_{}", status.as_u16());
// This can also fail after a command was dispatched. Do not claim
// a pre-dispatch refusal or absence of effects in the safe message.
log.message = "Workdir operation failed".to_string();
log.diagnostics.clear();
}
log
}
fn from_repository_error(error: &server_api::RepositoryApiError) -> Self {
Self {
workdir_log: None,
denial_reason: None,
kind: error
.diagnostics
.first()
@@ -32616,6 +32742,11 @@ impl From<merge_request::MergeRequestError> for ApiError {
impl From<Error> for ApiError {
fn from(error: Error) -> Self {
let diagnostics = match &error {
Error::WorkdirSessionOpenFailed { transport, .. } => vec![RuntimeDiagnostic {
code: "workdir_session_open_failed".to_string(),
severity: HostDiagnosticSeverity::Error,
message: transport.message.clone(),
}],
Error::RuntimeOperationFailed { code, message, .. } => vec![RuntimeDiagnostic {
code: code.clone(),
severity: HostDiagnosticSeverity::Error,
@@ -32834,13 +32965,18 @@ fn api_error_status(error: &Error) -> StatusCode {
StatusCode::SERVICE_UNAVAILABLE
}
Error::WorkspaceSigningIdentity { .. } => StatusCode::SERVICE_UNAVAILABLE,
Error::RuntimeOperationFailed { .. } => StatusCode::BAD_GATEWAY,
Error::WorkdirSessionOpenFailed { .. } | Error::RuntimeOperationFailed { .. } => {
StatusCode::BAD_GATEWAY
}
_ => StatusCode::INTERNAL_SERVER_ERROR,
}
}
fn api_error_response_message(error: &Error) -> String {
match error {
Error::WorkdirSessionOpenFailed { transport, .. } => {
format!("workdir_session_open_failed: {}", transport.message)
}
Error::RuntimeOperationFailed { code, message, .. } => {
format!("{code}: {}", sanitize_backend_error(message))
}
@@ -32853,6 +32989,8 @@ impl IntoResponse for ApiError {
let status = api_error_status(&self.error);
let response_message = api_error_response_message(&self.error);
let log = ApiErrorLog {
workdir_log: None,
denial_reason: None,
kind: self
.diagnostics
.first()
@@ -32890,6 +33028,7 @@ mod tests {
mod drive_api_tests;
mod drive_tests;
mod drive_web_roundtrip_tests;
mod external_checkout_tests;
mod subject_spawn_tests;
mod subjektiv_jobs_tests;
mod ticket_evidence_tests;
@@ -33155,6 +33294,7 @@ mod tests {
assert!(matches!(operation, WorkdirSessionOperation::Read(_)));
response
.send(Err(WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::NotFound,
message: "logical file was not found".to_string(),
}))
@@ -35153,6 +35293,13 @@ mod tests {
#[derive(Clone, Default)]
struct WorkdirlessFixtureRuntime {
workdir_session_factory: Option<
Arc<
dyn Fn(&str) -> std::result::Result<WorkdirSessionHandle, workdir::WorkdirError>
+ Send
+ Sync,
>,
>,
input_requests: Arc<Mutex<Vec<WorkerInputRequest>>>,
workers: Arc<Mutex<Vec<InternalWorkerSummary>>>,
spawn_requests: Arc<Mutex<Vec<WorkerSpawnRequest>>>,
@@ -35291,6 +35438,27 @@ mod tests {
}
impl crate::hosts::WorkspaceWorkerRuntime for WorkdirlessFixtureRuntime {
fn open_workdir_session<'a>(
&'a self,
working_directory_id: &'a str,
_owner_worker_id: Option<&'a str>,
) -> std::pin::Pin<
Box<
dyn std::future::Future<
Output = std::result::Result<WorkdirSessionHandle, workdir::WorkdirError>,
> + Send
+ 'a,
>,
> {
Box::pin(async move {
self.workdir_session_factory.as_ref().ok_or_else(|| {
workdir::WorkdirError::Unavailable(
"Runtime does not expose Workdir operation sessions".to_string(),
)
})?(working_directory_id)
})
}
fn send_input(
&self,
worker_id: &str,
@@ -36446,6 +36614,8 @@ mod tests {
.parse::<Uri>()
.expect("valid URI");
let error = ApiErrorLog {
workdir_log: None,
denial_reason: None,
kind: "ticket_backend_error".to_string(),
message: "sqlite error: FOREIGN KEY constraint failed".to_string(),
diagnostics: vec![RuntimeDiagnostic {
@@ -36587,6 +36757,7 @@ mod tests {
),
] {
let response = InternalWorkdirOperationError::Provider(WorkdirTransportError {
denial_reason: None,
code,
message: "safe provider message".to_string(),
})
@@ -36606,14 +36777,11 @@ mod tests {
#[tokio::test]
async fn local_and_remote_validation_errors_share_workspace_classification() {
let worker = RuntimeWorkerRef::new("runtime", "worker");
let local = current_worker_workdir_operation_error(
&worker,
workdir::WorkdirError::InvalidGlob("[".to_string()),
);
let remote = current_worker_workdir_operation_error(
&worker,
let local =
workdir_operation_transport_error(workdir::WorkdirError::InvalidGlob("[".to_string()));
let remote = workdir_operation_transport_error(
WorkdirTransportError {
denial_reason: None,
code: workdir::http::WorkdirTransportErrorCode::InvalidRequest,
message: "Workdir operation request is invalid".to_string(),
}
@@ -36639,10 +36807,7 @@ mod tests {
path: PathBuf::from("/host/private/worktree/secret.txt"),
source: std::io::Error::other("token=secret"),
};
let public = current_worker_workdir_operation_error(
&RuntimeWorkerRef::new("runtime", "worker"),
error,
);
let public = workdir_operation_transport_error(error);
let response = InternalWorkdirOperationError::Provider(public).into_response();
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
let text = String::from_utf8(body.to_vec()).unwrap();
@@ -36651,10 +36816,7 @@ mod tests {
let decoded: WorkdirTransportError = serde_json::from_str(&text).unwrap();
assert_eq!(decoded.code, workdir::http::WorkdirTransportErrorCode::Io);
let remote = current_worker_workdir_operation_error(
&RuntimeWorkerRef::new("runtime", "worker"),
workdir::WorkdirError::OperationFailed,
);
let remote = workdir_operation_transport_error(workdir::WorkdirError::OperationFailed);
let response = InternalWorkdirOperationError::Provider(remote).into_response();
assert_eq!(response.status(), StatusCode::INTERNAL_SERVER_ERROR);
}
@@ -38254,6 +38416,7 @@ mod tests {
.uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"))
// Unit fixture stands below proof middleware, so inject its verified subject.
.extension(crate::worker_source::VerifiedRuntimeRequestSource {
token_id_hash: None,
runtime_id: fixture.controller.runtime_id.clone(),
worker_id: Some(fixture.controller.worker_id.clone()),
})
@@ -38577,6 +38740,7 @@ mod tests {
.method("POST")
.uri(format!("/api/w/{TEST_WORKSPACE_ID}/worker-control/workers"))
.extension(crate::worker_source::VerifiedRuntimeRequestSource {
token_id_hash: None,
runtime_id: controller.runtime_id.clone(),
worker_id: Some(controller.worker_id.clone()),
})
crates/workspace-server/src/server/tests/api_logging_tests.rs
+440 -11
View File
@@ -17,6 +17,12 @@ impl Write for CapturedApiLog {
}
async fn request_with_logs(app: Router, request: Request<Body>) -> (StatusCode, Value, Vec<Value>) {
let token_hash = request
.headers()
.get(worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER)
.and_then(|value| value.to_str().ok())
.and_then(|proof| worker_runtime::auth::decode_runtime_request_source_claims(proof).ok())
.map(|claims| crate::worker_source::diagnostic_id_hash(&claims.jti));
let logs = CapturedApiLog::default();
let writer = logs.clone();
let subscriber = tracing_subscriber::fmt()
@@ -32,18 +38,41 @@ async fn request_with_logs(app: Router, request: Request<Body>) -> (StatusCode,
.await
.unwrap();
let status = response.status();
assert_eq!(response.headers()[CONTENT_TYPE], "application/json");
let content_type = response.headers()[CONTENT_TYPE].clone();
let body = to_bytes(response.into_body(), usize::MAX).await.unwrap();
assert_eq!(
content_type,
"application/json",
"status={status}, body={}",
String::from_utf8_lossy(&body)
);
let body: Value = serde_json::from_slice(&body).unwrap();
let text = String::from_utf8(logs.0.lock().unwrap().clone()).unwrap();
for secret in ["query-secret", "bearer-secret", "body-secret"] {
for secret in [
"query-secret",
"bearer-secret",
"body-secret",
"/private/provider/path",
"/host/private",
"proof-secret",
"credential-secret",
"key-secret",
] {
assert!(!text.contains(secret), "log leaked request data: {text}");
}
let events = text
let events: Vec<Value> = text
.lines()
.map(|line| serde_json::from_str::<Value>(line).unwrap())
.filter(|event| event["event"] == "api_error")
.collect();
for event in &events {
if event.get("operation").is_some() && token_hash.is_some() {
assert_eq!(
event["request_token_id_hash"],
serde_json::to_value(&token_hash).unwrap()
);
}
}
(status, body, events)
}
@@ -278,17 +307,50 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
)
};
let app = build_router(fixture.api.clone());
for (code, status, message) in [
(Code::Denied, 403, "Workdir operation was denied"),
(Code::OutOfScope, 403, "Workdir path is out of scope"),
for (code, status, message, reason) in [
(Code::Denied, 403, "Workdir operation was denied", None),
(
Code::Denied,
403,
"Workdir operation was denied",
Some(workdir::WorkdirDenialReason::OsPermissionDenied),
),
(
Code::Denied,
403,
"Workdir operation was denied",
Some(workdir::WorkdirDenialReason::ReadOnlySession),
),
(
Code::OutOfScope,
403,
"Workdir path is out of scope",
Some(workdir::WorkdirDenialReason::PathOutOfScope),
),
(
Code::SymlinkOutOfScope,
403,
"Workdir symlink target is out of scope",
Some(workdir::WorkdirDenialReason::SymlinkTargetOutOfScope),
),
(
Code::ReadOnly,
403,
"Workdir path is read-only",
Some(workdir::WorkdirDenialReason::PathReadOnly),
),
(
Code::UnknownCommand,
404,
"Workdir command was not found",
None,
),
(
Code::Unavailable,
503,
"Workdir session is unavailable",
None,
),
(Code::ReadOnly, 403, "Workdir path is read-only"),
(Code::UnknownCommand, 404, "Workdir command was not found"),
(Code::Unavailable, 503, "Workdir session is unavailable"),
] {
let provider = async {
let Some(ExternalProviderCommand::Operation {
@@ -305,6 +367,7 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
));
response
.send(Err(WorkdirTransportError {
denial_reason: reason,
code,
message: "body-secret /private/provider/path".to_string(),
}))
@@ -324,6 +387,22 @@ async fn generated_workdir_provider_errors_preserve_response_and_log_details() {
assert_eq!(events[0]["status"], status);
assert_eq!(events[0]["method"], "POST");
assert_eq!(events[0]["path"], path);
assert_eq!(events[0]["operation"], "command_status");
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
assert_workdir_log_identity(
&events[0],
&fixture,
"checkout",
Some(&fixture.main_workdir_id),
);
assert_eq!(
events[0]["denial_reason"],
serde_json::to_value(reason).unwrap()
);
assert_eq!(
events[0]["request_token_id_hash"].as_str().unwrap().len(),
64
);
}
let provider = async {
let Some(ExternalProviderCommand::Operation { response, .. }) = commands.recv().await
@@ -374,8 +453,358 @@ async fn generated_workdir_api_errors_preserve_response_and_log_details() {
assert_eq!(status, StatusCode::FORBIDDEN);
assert_eq!(body, serde_json::to_value(expected).unwrap());
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["kind"], "forbidden");
assert_eq!(events[0]["message"], body["message"]);
assert_eq!(events[0]["kind"], "workdir_session_operation_api_403");
assert_eq!(events[0]["message"], "Workdir operation failed");
assert_eq!(events[0]["operation"], "command_status");
assert_eq!(events[0]["workdir_stage"], "worker_identity");
assert_eq!(events[0]["runtime_id_hash"], Value::Null);
assert_eq!(events[0]["worker_id_hash"], Value::Null);
assert_eq!(events[0]["workdir_id_hash"], Value::Null);
assert_eq!(events[0]["diagnostics"], "[]");
assert_eq!(events[0]["path"], path);
}
fn assert_workdir_log_identity(
event: &Value,
fixture: &ManualCoderAssignmentFixture,
alias: &str,
workdir_id: Option<&str>,
) {
let hash = crate::worker_source::diagnostic_id_hash;
assert_eq!(event["workspace_id_hash"], hash(TEST_WORKSPACE_ID));
assert_eq!(event["runtime_id_hash"], hash(&fixture.worker.runtime_id));
assert_eq!(event["worker_id_hash"], hash(&fixture.worker.worker_id));
assert_eq!(event["attachment_alias_hash"], hash(alias));
assert_eq!(
event["workdir_id_hash"],
serde_json::to_value(workdir_id.map(hash)).unwrap()
);
}
#[tokio::test]
async fn self_workdir_provider_refusals_log_operation_identity_without_request_text() {
use workdir::{WorkdirDenialReason as Reason, WorkdirSessionCapabilities as Caps};
let mut fixture = manual_worker_assignment_fixture().await;
let root = tempfile::tempdir().unwrap();
let provider_root = root.path().to_path_buf();
// Fresh sessions exercise the real opening/broker/dispatch path on each request.
fixture.runtime.workdir_session_factory = Some(Arc::new(move |id| {
Ok(Arc::new(workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new(id),
provider_root.clone(),
provider_root.clone(),
manifest::SharedScope::new(manifest::Scope::writable(&provider_root).unwrap()),
Caps::READ_ONLY,
)) as WorkdirSessionHandle)
}));
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 app = build_router(fixture.api.clone());
let secret = format!(
"body-secret credential-secret key-secret proof-secret{}",
"z".repeat(20_000)
);
let cases = [
(
json!({"operation": "read", "request": {"path": "/host/private/secret.txt", "offset": 0, "limit": 10, "max_bytes": 1024}}),
"read",
"out_of_scope",
Reason::PathOutOfScope,
),
(
json!({"operation": "write", "request": {"path": "secret.txt", "content": secret.as_bytes()}}),
"write",
"denied",
Reason::ScopedCapabilityDenied,
),
(
json!({"operation": "command_start", "request": {"command": secret, "timeout_secs": 1, "output_limit": 1024, "cwd": "", "tool_call_id": secret}}),
"command_start",
"denied",
Reason::ScopedCapabilityDenied,
),
(
// Scoped forwarding validates this READ_ONLY parent's capabilities
// before provider resolution. Local now supports scope resolution;
// denying write rules is not a missing-resolver failure.
json!({"operation": "authorize_scope", "request": {"path": "secret.txt", "permission": "write", "rules": [{"target": "", "permission": "write", "recursive": true}]}}),
"authorize_scope",
"denied",
Reason::ParentCapabilityDenied,
),
];
for (operation, label, code, reason) in cases {
let _: WorkdirSessionOperation =
serde_json::from_value(operation.clone()).expect("valid operation fixture");
let request = runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
"POST",
&format!("{path}?token=query-secret"),
serde_json::to_vec(&json!({"target_workdir": "checkout", "operation": operation}))
.unwrap(),
);
let (status, body, events) = request_with_logs(app.clone(), request).await;
assert_eq!(status, StatusCode::FORBIDDEN, "{body}");
assert_eq!(body["code"], code, "{body}");
assert!(
body.get("denial_reason").is_none(),
"internal diagnostics leaked: {body}"
);
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["operation"], label);
assert_eq!(events[0]["workdir_stage"], "provider_dispatch");
assert_eq!(
events[0]["denial_reason"],
serde_json::to_value(reason).unwrap()
);
assert_workdir_log_identity(
&events[0],
&fixture,
"checkout",
Some(&fixture.main_workdir_id),
);
assert_eq!(
events[0]["request_token_id_hash"].as_str().unwrap().len(),
64
);
assert!(
events[0].to_string().len() < 2000,
"unbounded log: {events:?}"
);
}
// Control: the same READ_ONLY provider can authorize read rules through
// scoped forwarding and native resolution. Keep a capability refusal
// distinct from an unavailable scope resolver; successful authorization
// must not emit an api_error or mutate the filesystem.
let read_scope = runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
"POST",
&format!("{path}?token=query-secret"),
serde_json::to_vec(&json!({
"target_workdir": "checkout",
"operation": {"operation": "authorize_scope", "request": {
"path": "", "permission": "read",
"rules": [{"target": "", "permission": "read", "recursive": true}],
}},
}))
.unwrap(),
);
let (status, body, events) = request_with_logs(app, read_scope).await;
assert_eq!(status, StatusCode::OK, "{body}");
assert_eq!(
body,
serde_json::to_value(WorkdirSessionOperationResult::AuthorizeScope).unwrap(),
);
assert!(
events.is_empty(),
"successful scope authorization logged as failure: {events:?}"
);
assert!(
!root.path().join("secret.txt").exists(),
"refusal changed filesystem"
);
}
#[tokio::test]
async fn self_workdir_attachment_rejection_hashes_oversized_alias_and_omits_unknown_workdir() {
let mut fixture = manual_worker_assignment_fixture().await;
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 alias = format!(
"body-secret /host/private credential-secret key-secret proof-secret{}",
"あ".repeat(10_000)
);
let request = runtime_source_request(
&identity,
Some(&fixture.worker.worker_id),
"POST",
&path,
serde_json::to_vec(&json!({
"target_workdir": alias,
"operation": {"operation": "command_status", "request": "body-secret"}
}))
.unwrap(),
);
let (status, body, events) =
request_with_logs(build_router(fixture.api.clone()), request).await;
assert_eq!(status, StatusCode::CONFLICT, "{body}");
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["workdir_stage"], "attachment_validation");
assert_eq!(events[0]["operation"], "command_status");
assert_workdir_log_identity(&events[0], &fixture, &alias, None);
assert_eq!(events[0]["denial_reason"], Value::Null);
assert!(
events[0].to_string().len() < 2000,
"unbounded log: {events:?}"
);
}
#[tokio::test]
async fn self_workdir_session_open_refusals_keep_reason_before_registry_conversion() {
use workdir::WorkdirDenialReason as Reason;
for (reason, os_error) in [
(Reason::ReadOnlySession, false),
(Reason::OsPermissionDenied, true),
] {
let mut fixture = manual_worker_assignment_fixture().await;
fixture.runtime.workdir_session_factory = Some(Arc::new(move |_| {
Err(if os_error {
workdir::WorkdirError::Io {
path: PathBuf::from("/host/private/key-secret"),
source: std::io::Error::from_raw_os_error(13),
}
} else {
workdir::WorkdirError::denied(reason, "body-secret credential-secret /host/private")
})
}));
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;
// Opening retained its existing Registry HTTP meaning, not an operation's 403.
assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}");
assert_eq!(
body["message"],
"workdir_session_open_failed: Workdir operation was denied"
);
assert!(body.get("denial_reason").is_none(), "{body}");
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["workdir_stage"], "session_open");
assert_eq!(events[0]["operation"], "read");
assert_eq!(
events[0]["denial_reason"],
serde_json::to_value(reason).unwrap()
);
assert_workdir_log_identity(
&events[0],
&fixture,
"checkout",
Some(&fixture.main_workdir_id),
);
assert!(!body.to_string().contains("body-secret"));
assert!(!body.to_string().contains("/host/private"));
}
}
#[tokio::test]
async fn self_workdir_post_start_registration_failure_does_not_claim_pre_dispatch_refusal() {
let mut fixture = manual_worker_assignment_fixture().await;
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 (sender, mut commands) = tokio::sync::mpsc::channel(1);
let connection = Arc::new(ExternalProviderConnection {
grant_id: "logging-grant".to_string(),
workdir_id: fixture.main_workdir_id.clone(),
provider_instance_id: "logging-provider".to_string(),
generation: 1,
expires_at: None,
capabilities: workdir::WorkdirSessionCapabilities::ALL,
admission: Arc::new(tokio::sync::Semaphore::new(1)),
shutdown_confirmed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
sender,
});
fixture.runtime.workdir_session_factory = Some(Arc::new(move |_| {
Ok(Arc::new(ExternalProviderWorkdirSession::new(
connection.clone(),
None,
)) as WorkdirSessionHandle)
}));
fixture
.api
.runtime
.register_or_replace(fixture.runtime.clone());
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": "command_start", "request": {
"command": "body-secret", "timeout_secs": 1, "output_limit": 1024,
"cwd": "", "spill_dir": null, "tool_call_id": "body-secret"
}}
}))
.unwrap(),
);
// Observe command dispatch, then lose only the fixture registry entry before
// completing start. No OS process is launched, no sleep or retry is needed.
let provider = async {
let Some(ExternalProviderCommand::Operation {
operation,
response,
..
}) = commands.recv().await
else {
panic!("expected provider start operation");
};
assert!(matches!(
operation,
WorkdirSessionOperation::CommandStart(_)
));
fixture
.api
.workdir_sessions
.lock()
.unwrap()
.remove_attachment(&fixture.worker)
.unwrap();
response
.send(Ok(WorkdirSessionOperationResult::CommandStart(
CommandHandle("fixture-provider-handle".into()),
)))
.unwrap();
};
let ((status, body, events), ()) = tokio::join!(
request_with_logs(build_router(fixture.api.clone()), request),
provider
);
assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}");
assert!(
body["message"]
.as_str()
.unwrap()
.starts_with("workdir_session_registration_failed:")
);
assert_eq!(events.len(), 1, "{events:?}");
assert_eq!(events[0]["operation"], "command_start");
assert_eq!(events[0]["workdir_stage"], "command_registration");
assert_eq!(events[0]["message"], "Workdir operation failed");
assert_eq!(events[0]["denial_reason"], Value::Null);
assert_workdir_log_identity(
&events[0],
&fixture,
"checkout",
Some(&fixture.main_workdir_id),
);
assert!(
commands.try_recv().is_err(),
"post-dispatch failure was retried"
);
}
crates/workspace-server/src/server/tests/drive_api_tests.rs
+1
View File
@@ -718,6 +718,7 @@ async fn drive_verified_upload_revoked_during_reception_cannot_commit_or_replay(
let app = generated_workspace_contract_router(ServerApiContractService::Workspace(api.clone()))
.layer(Extension(
crate::worker_source::VerifiedRuntimeRequestSource {
token_id_hash: None,
runtime_id: workers[0].runtime_id.clone(),
worker_id: Some(workers[0].worker_id.clone()),
},
crates/workspace-server/src/server/tests/external_checkout_tests.rs
+129
View File
@@ -0,0 +1,129 @@
//! External codec -> actual session dispatch -> mutation safety -> HTTP error.
//! Channel fixture only; no product process or real mutation.
use super::*;
use workdir::{
CheckoutOperation as Op, CheckoutRequest, WorkdirDenialReason as Reason, WorkdirPath,
WorkdirSession, WorkdirSessionCapability, external::ExternalWorkdirOperationOutcome,
http::WorkdirTransportErrorCode as Code,
};
#[tokio::test]
async fn external_checkout_diagnostics_preserve_mutation_outcome_and_read_failures() {
for code in [
Code::Unavailable,
Code::Transport,
Code::Denied,
Code::Conflict,
Code::OutcomeUnknown,
] {
for reason in [None, Some(Reason::ReadOnlySession)] {
for operation in [
Op::Read {
offset: 0,
limit: 1,
max_bytes: 10,
},
Op::Create {
path: WorkdirPath::new("new.txt").unwrap(),
content: b"fixture".to_vec(),
},
Op::Write {
content: b"fixture".to_vec(),
expected_hash: [0; 32],
},
Op::Edit {
old_string: "old".into(),
new_string: "new".into(),
replace_all: false,
expected_hash: [0; 32],
},
] {
let mutation = operation.capability() != WorkdirSessionCapability::Read;
let expected_code =
if mutation && matches!(code, Code::Unavailable | Code::Transport) {
Code::OutcomeUnknown
} else {
code
};
let (sender, mut commands) = tokio::sync::mpsc::channel(1);
let connection = Arc::new(ExternalProviderConnection {
grant_id: "fixture-grant".into(),
workdir_id: "fixture-workdir".into(),
provider_instance_id: "fixture-provider".into(),
generation: 1,
expires_at: None,
capabilities: workdir::WorkdirSessionCapabilities::ALL,
admission: Arc::new(tokio::sync::Semaphore::new(16)),
shutdown_confirmed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
sender,
});
let session = Arc::new(ExternalProviderWorkdirSession::new(connection, None));
let caller = tokio::spawn(async move {
session
.checkout_execute(CheckoutRequest {
target: WorkdirPath::root(),
validator: vec![1],
operation,
})
.await
});
let command =
tokio::time::timeout(std::time::Duration::from_secs(5), commands.recv())
.await
.unwrap()
.unwrap();
let ExternalProviderCommand::Operation {
operation,
response,
..
} = command
else {
panic!("expected dispatched operation");
};
assert!(matches!(
operation,
WorkdirSessionOperation::CheckoutExecute(_)
));
let failed = serde_json::from_value::<ExternalWorkdirOperationOutcome>(json!({
"outcome":"failed", "error":{"code":code, "denial_reason":reason},
}))
.unwrap();
let ExternalWorkdirOperationOutcome::Failed { error } = failed else {
panic!("expected fixture failure");
};
response.send(Err(error.into_transport_error())).unwrap();
let error = tokio::time::timeout(std::time::Duration::from_secs(5), caller)
.await
.unwrap()
.unwrap()
.unwrap_err();
assert!(
matches!(
commands.try_recv(),
Err(tokio::sync::mpsc::error::TryRecvError::Empty
| tokio::sync::mpsc::error::TryRecvError::Disconnected)
),
"must not retry"
);
let forwarded = WorkdirTransportError::from_workdir_error(&error);
assert_eq!(
forwarded.code, expected_code,
"{code:?}, {reason:?}, mutation={mutation}: {error:?}"
);
assert_eq!(error.denial_reason(), reason);
assert_eq!(forwarded.denial_reason, reason);
if expected_code == Code::OutcomeUnknown {
assert!(forwarded.message.contains("inspect before retrying"));
}
let response = InternalWorkdirOperationError::Provider(forwarded).into_response();
assert_eq!(response.status().as_u16(), expected_code.http_status());
let body = axum::body::to_bytes(response.into_body(), 2048)
.await
.unwrap();
let body: serde_json::Value = serde_json::from_slice(&body).unwrap();
assert_eq!(body["code"], expected_code.as_str());
assert!(body.get("denial_reason").is_none());
}
}
}
}
crates/workspace-server/src/worker_source/runtime_request.rs
+3
View File
@@ -12,6 +12,8 @@ use crate::store::ControlPlaneStore;
pub struct VerifiedRuntimeRequestSource {
pub runtime_id: String,
pub worker_id: Option<String>,
/// Existing proof JTI correlation, retained only as a fixed-length diagnostic fingerprint.
pub token_id_hash: Option<String>,
}
/// Safe diagnostics only: never stores proof, raw claims, keys or store errors.
@@ -268,6 +270,7 @@ pub(crate) async fn verify_runtime_request_source_proof_with_clock(
})?;
}
Ok(VerifiedRuntimeRequestSource {
token_id_hash: Some(diagnostic_id_hash(&claims.jti)),
runtime_id: claims.iss,
worker_id: claims.worker_id,
})