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

fix: retain structured Workdir refusal diagnostics across provider boundaries

Browse Source
This commit is contained in:
Hare
2026-10-09 22:23:22 +09:00
parent e8828cd1b5
commit f9d70e2ae2
28 changed files with 2053 additions and 198 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
+212 -63
View File
@@ -11458,6 +11458,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()
@@ -11465,22 +11474,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))
}
@@ -11722,6 +11727,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();
@@ -11742,11 +11751,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(""),
@@ -17224,9 +17245,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"),
}
}
@@ -18217,6 +18239,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 {
@@ -18239,21 +18288,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",
}
}
@@ -18263,19 +18326,34 @@ 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)?;
let registered_source = api
.workdir_sessions
.lock()
@@ -18301,31 +18379,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(_)
@@ -18342,14 +18420,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(
@@ -18357,26 +18444,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)
}
@@ -23469,6 +23557,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(),
}));
@@ -23476,6 +23565,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(),
}));
@@ -23486,6 +23576,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,
}));
@@ -23542,6 +23633,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(),
}),
@@ -24023,6 +24115,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(),
@@ -24534,6 +24629,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 =
@@ -32572,6 +32668,8 @@ struct ApiErrorLog {
kind: String,
message: String,
diagnostics: Vec<RuntimeDiagnostic>,
workdir_log: Option<server_api::WorkdirOperationLogContext>,
denial_reason: Option<workdir::WorkdirDenialReason>,
}
impl ApiErrorLog {
@@ -32584,9 +32682,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
@@ -32595,11 +32695,23 @@ 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());
log.message = "Workdir operation was rejected before provider dispatch".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()
@@ -32632,6 +32744,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,
@@ -32850,13 +32967,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))
}
@@ -32869,6 +32991,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()
@@ -33169,6 +33293,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(),
}))
@@ -35167,6 +35292,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>>>,
@@ -35304,6 +35436,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,
@@ -36453,6 +36606,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 {
@@ -36594,6 +36749,7 @@ mod tests {
),
] {
let response = InternalWorkdirOperationError::Provider(WorkdirTransportError {
denial_reason: None,
code,
message: "safe provider message".to_string(),
})
@@ -36613,14 +36769,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(),
}
@@ -36646,10 +36799,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();
@@ -36658,10 +36808,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);
}
@@ -38261,6 +38408,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()),
})
@@ -38584,6 +38732,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
+312 -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,230 @@ 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 was rejected before provider dispatch"
);
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,
),
(
json!({"operation": "authorize_scope", "request": {"path": "secret.txt", "permission": "write", "rules": [{"target": "", "permission": "write", "recursive": true}]}}),
"authorize_scope",
"denied",
Reason::ScopeResolutionUnavailable,
),
];
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:?}"
);
}
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"));
}
}
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/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.
@@ -242,6 +244,7 @@ pub async fn verify_runtime_request_source_proof_with_store(
})?;
}
Ok(VerifiedRuntimeRequestSource {
token_id_hash: Some(diagnostic_id_hash(&claims.jti)),
runtime_id: claims.iss,
worker_id: claims.worker_id,
})