From 7873f4359ab62c83107591afd1a8d9f1311bd26b Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 15:50:30 +0900 Subject: [PATCH] feat: transfer typed startup causes over Runtime HTTP with real registration fixture --- crates/runtime-api/src/lib.rs | 16 + crates/worker-runtime/src/http_server.rs | 39 +- .../src/http_server/runtime_management_api.rs | 5 +- crates/worker/src/startup_failure.rs | 14 + crates/workspace-server/src/hosts.rs | 43 +- crates/workspace-server/src/server.rs | 50 +++ .../session_registration_connected_tests.rs | 369 ++++++++++++++++++ .../tests/subject_authorization_tests.rs | 14 + 8 files changed, 543 insertions(+), 7 deletions(-) create mode 100644 crates/workspace-server/src/server/tests/session_registration_connected_tests.rs diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index 533b657ae..cdb819085 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -105,10 +105,25 @@ pub struct RuntimeApiError { status: u16, } +/// Transient, allowlisted startup observations copied from the execution owner. +/// This is not an operation result, lifecycle record, or permission to retry. +#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] +#[serde(deny_unknown_fields)] +pub struct WorkerStartupFailure { + pub primary: StartupFailureAttachment, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub secondary: Vec, + pub resources: Option, + pub persistence: Option, + pub io_kind: Option, +} + #[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)] pub struct RuntimeApiErrorDetail { pub code: String, pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup: Option, } impl RuntimeApiError { @@ -117,6 +132,7 @@ impl RuntimeApiError { error: RuntimeApiErrorDetail { code: code.into(), message: message.into(), + startup: None, }, status, } diff --git a/crates/worker-runtime/src/http_server.rs b/crates/worker-runtime/src/http_server.rs index d591008a4..5d04e4c6d 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -542,6 +542,8 @@ pub struct RuntimeHttpErrorResponse { pub struct RuntimeHttpErrorDetail { pub code: String, pub message: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup: Option, } #[cfg(feature = "ws-server")] @@ -2325,6 +2327,7 @@ struct RuntimeHttpRestError { status: StatusCode, code: String, message: String, + startup: Option, } impl RuntimeHttpRestError { @@ -2333,13 +2336,28 @@ impl RuntimeHttpRestError { status, code: code.into(), message: message.into(), + startup: None, } } fn runtime(error: RuntimeError) -> Self { let status = status_for_runtime_error(&error); let code = code_for_runtime_error(&error); - Self::new(status, code, error.to_string()) + let startup = match &error { + RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result), + _ => None, + }; + let message = if startup.is_some() { + "Worker startup failed".to_owned() + } else { + error.to_string() + }; + Self { + status, + code, + message, + startup, + } } fn working_directory_join(_error: tokio::task::JoinError) -> Self { @@ -2375,12 +2393,31 @@ impl IntoResponse for RuntimeHttpRestError { error: RuntimeHttpErrorDetail { code: self.code, message: self.message, + startup: self.startup, }, }; (self.status, Json(body)).into_response() } } +/// Explicit transfer mapping, not another execution/lifecycle result. HTTP request +/// authentication has already established this internal Runtime caller's scope. +fn startup_transfer( + result: &crate::execution::WorkerExecutionResult, +) -> Option { + let primary = worker::startup_failure::reproject_internal_startup_failure( + result.startup_failure.as_ref()?, + ) + .ok()?; + Some(runtime_api::WorkerStartupFailure { + primary, + secondary: result.secondary_failures.clone(), + resources: result.startup_resources, + persistence: result.startup_failure_persistence, + io_kind: result.startup_io_kind, + }) +} + fn status_for_runtime_error(error: &RuntimeError) -> StatusCode { match error { RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => { diff --git a/crates/worker-runtime/src/http_server/runtime_management_api.rs b/crates/worker-runtime/src/http_server/runtime_management_api.rs index 860a8f266..cac820371 100644 --- a/crates/worker-runtime/src/http_server/runtime_management_api.rs +++ b/crates/worker-runtime/src/http_server/runtime_management_api.rs @@ -224,7 +224,10 @@ where } fn api_error(error: RuntimeHttpRestError) -> runtime_api::RuntimeApiError { - runtime_api::RuntimeApiError::new(error.status.as_u16(), error.code, error.message) + let mut response = + runtime_api::RuntimeApiError::new(error.status.as_u16(), error.code, error.message); + response.error.startup = error.startup; + response } impl runtime_api::RuntimeApi for RuntimeManagementApi { diff --git a/crates/worker/src/startup_failure.rs b/crates/worker/src/startup_failure.rs index 2c48c7138..b0c4c59fa 100644 --- a/crates/worker/src/startup_failure.rs +++ b/crates/worker/src/startup_failure.rs @@ -106,6 +106,20 @@ pub fn runtime_startup_attachment( StartupFailureAttachment::from_projected(&projected) } +/// The internal HTTP/Host transfer admits only implemented startup domains. +/// Unknown/corrupt envelopes cannot be blindly forwarded or classified from text. +pub fn reproject_internal_startup_failure( + failure: &StartupFailureAttachment, +) -> Result { + if let Ok(projected) = failure + .project_for::(&()) + { + return StartupFailureAttachment::from_projected(&projected); + } + let projected = failure.project_for::(&())?; + StartupFailureAttachment::from_projected(&projected) +} + /// Dispatch by the domain attested in process, not by parsing attachment fields. pub(crate) fn persist_binding_startup_failure( store: &S, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 7f4d9a847..9975b5f45 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -117,6 +117,9 @@ where #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct RuntimeDiagnostic { + /// Internal startup observations; public consumers must separately authorize projection. + #[serde(skip)] + pub startup: Option, pub code: String, pub severity: HostDiagnosticSeverity, pub message: String, @@ -6649,7 +6652,7 @@ fn sanitize_embedded_execution_detail(summary: &str, message: &str) -> String { } fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnostic { - match error { + let mut projection = match error { EmbeddedRuntimeError::RestoreAdmissionBusy { .. } => diagnostic( "restore_admission_busy", HostDiagnosticSeverity::Warning, @@ -6767,7 +6770,19 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti HostDiagnosticSeverity::Error, "Embedded Runtime state is unavailable".to_string(), ), + }; + if let EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } = error { + projection.startup = result.startup_failure.as_ref().and_then(|primary| { + internal_startup_transfer(runtime_api::WorkerStartupFailure { + primary: primary.clone(), + secondary: result.secondary_failures.clone(), + resources: result.startup_resources, + persistence: result.startup_failure_persistence, + io_kind: result.startup_io_kind, + }) + }); } + projection } fn host_id_for_embedded_workspace(workspace_id: &str) -> String { @@ -6846,6 +6861,7 @@ fn runtime_api_diagnostic( ) -> RuntimeDiagnostic { match error { runtime_api::client_support::ClientError::Public { status, error } => { + let startup = error.error.startup.and_then(internal_startup_transfer); let remote_code = error.error.code; let message = sanitize_remote_runtime_message(&remote_code, &error.error.message); let (code, severity) = match status { @@ -6862,13 +6878,15 @@ fn runtime_api_diagnostic( }, ), }; - diagnostic( + let mut projection = diagnostic( code, severity, format!( "Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}" ), - ) + ); + projection.startup = startup; + projection } runtime_api::client_support::ClientError::Failure(failure) => { let code = match failure { @@ -6945,11 +6963,25 @@ fn remote_http_status_diagnostic( } _ => (remote_code, HostDiagnosticSeverity::Warning), }; - diagnostic( + let mut projection = diagnostic( code, severity, format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {remote_message}"), - ) + ); + projection.startup = error + .and_then(|error| error.error.startup) + .and_then(internal_startup_transfer); + projection +} + +/// Treat transferred details as untrusted observations and reproject through the +/// same domain allowlist; HTTP status and text never supply a missing cause. +fn internal_startup_transfer( + mut startup: runtime_api::WorkerStartupFailure, +) -> Option { + startup.primary = + worker::startup_failure::reproject_internal_startup_failure(&startup.primary).ok()?; + Some(startup) } fn diagnostic( @@ -6958,6 +6990,7 @@ fn diagnostic( message: impl Into, ) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.into(), severity, message: message.into(), diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ccfa2c8b1..d79a839ff 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -8612,6 +8612,7 @@ fn worker_input_result_to_api( // distinguishable without collapsing it into a definite refusal or replaying input. if result.disposition == WorkerInputDisposition::Unknown { result.diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_input_outcome_unknown".into(), severity: HostDiagnosticSeverity::Warning, message: "Worker input delivery outcome is unknown; inspect the current session before retrying".into(), @@ -27020,6 +27021,7 @@ fn skill_api_error(error: skills::SkillError) -> ApiError { message: format!("unknown Skill `{name}`"), }, vec![RuntimeDiagnostic { + startup: None, code: "skill_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: format!("unknown Skill `{name}`"), @@ -30426,6 +30428,7 @@ fn prepare_workspace_workdir_create( .to_string(), ), vec![RuntimeDiagnostic { + startup: None, code: "default_runtime_not_configured".to_string(), severity: HostDiagnosticSeverity::Error, message: "Workspace default Runtime is not configured".to_string(), @@ -30580,6 +30583,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( Error::UnknownRuntime(reserved.resolved_runtime_id), vec![RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "Selected Runtime is not available".to_string(), @@ -30603,6 +30607,7 @@ async fn create_workspace_working_directory( message: "Selected Runtime is not connected".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "Selected Runtime is not available".to_string(), @@ -30659,6 +30664,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( error.into_error(), vec![RuntimeDiagnostic { + startup: None, code: failure_code, severity: HostDiagnosticSeverity::Error, message: "Selected Runtime could not inspect the reserved Workdir".to_string(), @@ -30697,6 +30703,7 @@ async fn create_workspace_working_directory( return Err(ApiError::with_diagnostics( error.into_error(), vec![RuntimeDiagnostic { + startup: None, code: failure_code, severity: HostDiagnosticSeverity::Error, message: "Selected Runtime rejected Workdir creation".to_string(), @@ -30736,6 +30743,7 @@ async fn create_workspace_working_directory( diagnostics.insert( 0, RuntimeDiagnostic { + startup: None, code: failure_code.clone(), severity: HostDiagnosticSeverity::Error, message: "Runtime did not create the reserved Workdir".to_string(), @@ -31490,6 +31498,7 @@ fn build_runtime_cleanup_plan( let mut diagnostics = error.diagnostics; if diagnostics.is_empty() { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "workspace_cleanup_runtime_observation_unavailable".to_string(), severity: HostDiagnosticSeverity::Warning, message: sanitize_backend_error(&error.error.to_string()), @@ -35599,6 +35608,7 @@ fn reject_workdir_for_embedded_runtime(runtime_id: &str, has_workdir: bool) -> A message: "The embedded Runtime does not accept working directories".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "embedded_worker_workdir_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message: "Choose a non-embedded Runtime for workspace-file Workers; embedded Workers are no-workdir Workspace-API workers.".to_string(), @@ -35689,6 +35699,7 @@ fn finalize_worker_spawn_stage( .map(|assignment| assignment.operation_id.as_str()) .unwrap_or("none"); let mut diagnostics = vec![RuntimeDiagnostic { + startup: None, code: format!("worker_spawn_finalize_{}_failed", stage.as_str()), severity: HostDiagnosticSeverity::Error, message: format!( @@ -35704,6 +35715,7 @@ fn finalize_worker_spawn_stage( diagnostics.extend(compensation_errors); if !compensation_failed { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_spawn_compensated".to_string(), severity: HostDiagnosticSeverity::Info, message: format!( @@ -36039,6 +36051,7 @@ fn lifecycle_failure_detail( fn spawn_compensation_diagnostic(code: &str, message: String) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -37044,6 +37057,7 @@ fn project_workspace_worker_with_attachments( summary.state = "unavailable".to_string(); summary.worker_state = None; summary.diagnostics.push(RuntimeDiagnostic { + startup: None, code: "worker_singleton_fenced".to_string(), severity: HostDiagnosticSeverity::Warning, message: "This historical singleton generation is fenced by Backend ownership." @@ -37749,6 +37763,7 @@ fn working_directory_summaries( } if let Some(grant_id) = legacy_external_workdir_grant_id(api, &record)? { diagnostics.push(RuntimeDiagnostic { + startup: None, code: "external_workdir_legacy_permissions".to_string(), severity: HostDiagnosticSeverity::Warning, message: format!( @@ -37955,6 +37970,7 @@ fn worker_summary_from_registry(record: &WorkerRegistryRecord) -> InternalWorker }, workdir_attachments: Vec::new(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "backend_worker_missing".to_string(), severity: HostDiagnosticSeverity::Info, message: @@ -37999,6 +38015,7 @@ fn worker_summary_from_projection( Vec::new() } else { vec![RuntimeDiagnostic { + startup: None, code: "worker_observation_unavailable".to_string(), severity: HostDiagnosticSeverity::Info, message: "The Worker remains in the Backend catalog, but its Runtime observation is currently unavailable.".to_string(), @@ -38645,6 +38662,7 @@ fn validate_working_directory_claims_for_browser( message: "working directory relative_cwd must stay inside the Runtime-owned working directory".to_string(), }, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_relative_cwd_invalid".to_string(), severity: HostDiagnosticSeverity::Error, message: "relative_cwd must be a relative path without parent traversal".to_string(), @@ -39239,6 +39257,7 @@ fn settings_diagnostic( message: impl Into, ) -> RuntimeDiagnostic { RuntimeDiagnostic { + startup: None, code: code.into(), severity, message: message.into(), @@ -39260,6 +39279,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_not_configured".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39277,6 +39297,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_provider_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39292,6 +39313,7 @@ fn repository_lookup(result: std::result::Result) - message: message.clone(), }, vec![RuntimeDiagnostic { + startup: None, code: "repository_evidence_invalid".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39452,6 +39474,7 @@ fn workspace_id_mismatch_error() -> ApiError { ApiError::with_diagnostics( Error::WorkspaceIdMismatch, vec![RuntimeDiagnostic { + startup: None, code: "workspace_id_mismatch".to_string(), severity: HostDiagnosticSeverity::Error, message, @@ -39561,6 +39584,7 @@ impl ApiErrorLog { .diagnostics .iter() .map(|diagnostic| RuntimeDiagnostic { + startup: None, code: diagnostic.code.clone(), severity: match diagnostic.severity { server_api::DiagnosticSeverity::Info => HostDiagnosticSeverity::Info, @@ -39598,6 +39622,7 @@ impl From for ApiError { Error::DbCapacityExceeded | Error::DbExecutorUnavailable | Error::DbOperationOutcomeUnknown => vec![RuntimeDiagnostic { + startup: None, code: match &error { Error::DbCapacityExceeded => "db_capacity_exceeded", Error::DbExecutorUnavailable => "db_executor_unavailable", @@ -39608,21 +39633,25 @@ impl From for ApiError { message: error.to_string(), }], Error::WorkdirSessionOpenFailed { transport, .. } => vec![RuntimeDiagnostic { + startup: None, code: "workdir_session_open_failed".to_string(), severity: HostDiagnosticSeverity::Error, message: transport.message.clone(), }], Error::RuntimeOperationFailed { code, message, .. } => vec![RuntimeDiagnostic { + startup: None, code: code.clone(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), }], Error::WorkspaceSigningIdentity { code, message } => vec![RuntimeDiagnostic { + startup: None, code: code.clone(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), }], Error::Ticket(ticket_error) => vec![RuntimeDiagnostic { + startup: None, code: match ticket_error { ticket::TicketError::NotFound(_) => "ticket_not_found", ticket::TicketError::Ambiguous { .. } => "ticket_ambiguous", @@ -39669,6 +39698,7 @@ impl From for ApiError { subjektiv_diagnostic_code(message) .map(|code| { vec![RuntimeDiagnostic { + startup: None, code: code.to_string(), severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(message), @@ -43074,6 +43104,7 @@ mod tests { worker: None, acceptance_evidence: Vec::new(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "fixture_runtime_rejected".to_string(), severity: HostDiagnosticSeverity::Error, message: "fixture rejected Worker create".to_string(), @@ -44981,6 +45012,7 @@ mod tests { kind: "ticket_backend_error".to_string(), message: "sqlite error: FOREIGN KEY constraint failed".to_string(), diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "ticket_backend_error".to_string(), severity: HostDiagnosticSeverity::Error, message: "FOREIGN KEY constraint failed".to_string(), @@ -51633,6 +51665,7 @@ mod tests { #[test] fn unrecoverable_pending_workspace_restore_is_typed_for_replacement() { let diagnostics = [RuntimeDiagnostic { + startup: None, code: "embedded_worker_execution_rejected".to_string(), severity: HostDiagnosticSeverity::Error, message: "Restore Errored: pending Workspace Worker restore requires operation-owned launch material; generic restore must not reconstruct it from current Workspace config".to_string(), @@ -51644,6 +51677,7 @@ mod tests { #[test] fn workdir_runtime_miss_uses_exact_typed_code() { let typed_not_found = [RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "missing".to_string(), @@ -51655,6 +51689,7 @@ mod tests { assert!(workdir_runtime_miss_is_not_found(&typed_not_found)); let unrelated = [RuntimeDiagnostic { + startup: None, code: "some_other_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "not a typed workdir miss".to_string(), @@ -63358,6 +63393,7 @@ mod tests { &api, deleted, &[RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Warning, message: "missing".to_string(), @@ -63384,6 +63420,7 @@ mod tests { &api, unknown, &[RuntimeDiagnostic { + startup: None, code: "runtime_unavailable".to_string(), severity: HostDiagnosticSeverity::Warning, message: "temporarily unavailable".to_string(), @@ -65080,6 +65117,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".into(), severity: HostDiagnosticSeverity::Warning, message: "authoritatively absent".into(), @@ -65131,6 +65169,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_resource_busy".into(), severity: HostDiagnosticSeverity::Warning, message: "busy".into(), @@ -65360,6 +65399,7 @@ mod tests { let (missing_operation, _) = reserve_removal_fixture(&api, "provider-not-found"); let not_found = RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: "missing".to_string(), @@ -65395,6 +65435,7 @@ mod tests { InternalWorkerOperationState::Accepted, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_provider_timeout".to_string(), severity: HostDiagnosticSeverity::Error, message: "timeout".to_string(), @@ -65511,6 +65552,7 @@ mod tests { InternalWorkerOperationState::Unsupported, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_unsupported".to_string(), severity: HostDiagnosticSeverity::Error, message: "unsupported".to_string(), @@ -65571,6 +65613,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: code.into(), severity: HostDiagnosticSeverity::Error, message: "secret /private/path".into(), @@ -65602,6 +65645,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_mount_present".into(), severity: HostDiagnosticSeverity::Error, message: "/private/mount credentials must not escape".into(), @@ -65752,6 +65796,7 @@ mod tests { InternalWorkerOperationState::Rejected, None, vec![RuntimeDiagnostic { + startup: None, code: "working_directory_cleanup_permission_denied".into(), severity: HostDiagnosticSeverity::Error, message: "permission denied /private/path".into(), @@ -65938,6 +65983,7 @@ mod tests { state: InternalWorkerOperationState::Rejected, working_directory: None, diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "working_directory_not_found".to_string(), severity: HostDiagnosticSeverity::Error, message: "missing".to_string(), @@ -65949,6 +65995,7 @@ mod tests { state: InternalWorkerOperationState::Rejected, working_directory: None, diagnostics: vec![RuntimeDiagnostic { + startup: None, code: "working_directory_provider_timeout".to_string(), severity: HostDiagnosticSeverity::Error, message: "timeout".to_string(), @@ -67641,11 +67688,13 @@ mod tests { fn provider_rejection_classification_prefers_stable_error_diagnostic() { let diagnostics = vec![ RuntimeDiagnostic { + startup: None, code: "runtime_capacity_warning".to_string(), severity: HostDiagnosticSeverity::Warning, message: "capacity is low".to_string(), }, RuntimeDiagnostic { + startup: None, code: "runtime_capacity_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "capacity is exhausted".to_string(), @@ -67705,6 +67754,7 @@ mod tests { }) .unwrap(); let diagnostics = vec![RuntimeDiagnostic { + startup: None, code: "runtime_capacity_unavailable".to_string(), severity: HostDiagnosticSeverity::Error, message: "capacity is exhausted".to_string(), diff --git a/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs b/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs new file mode 100644 index 000000000..8ad590fb5 --- /dev/null +++ b/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs @@ -0,0 +1,369 @@ +//! Focused T840 connection: real Subject policy -> signed Session POST -> real +//! ProfileRuntimeWorkerFactory startup -> RuntimeHTTP/Host diagnostic -> saved +//! Worker metadata. Only the unused external model is replaced. No restore, +//! operation replay, singleton-release, or Browser rendering contract is asserted. +use super::*; +use protocol::session_failure::{ + SessionPrincipal, SessionRegistrationCause, session_registration_attachment, +}; +use session_store::WorkerMetadataStore as _; +use std::sync::atomic::{AtomicUsize, Ordering}; +use worker_runtime::worker_backend::RuntimeWorkerFactory as _; + +#[derive(Clone, Default)] +struct UnusedModel(Arc); + +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for UnusedModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + _: agen::llm_client::Request, + ) -> std::result::Result + { + self.0.fetch_add(1, Ordering::SeqCst); + panic!("Session registration does not require a model turn") + } +} + +struct RegistrationFixture { + connected: ConnectedWorkerHarness, + subject: String, + callbacks: Arc>>, + model: UnusedModel, +} + +impl RegistrationFixture { + async fn new(revoke_on_registration: bool) -> Self { + // Armed only after real Subject creation with its normal default policy. + // The callback is a deterministic synchronization point, not a fake Host + // rejection: the original signed request and real handler remain intact. + let revoke = Arc::new(Mutex::new(None::<(WorkspaceApi, String, PathBuf)>)); + let hook = revoke.clone(); + let callbacks = Arc::new(Mutex::new(Vec::new())); + let observed = callbacks.clone(); + let model = UnusedModel::default(); + let connected = ConnectedWorkerHarness::start_with_layer(model.clone(), move |router| { + router.layer(axum::middleware::from_fn( + move |request: axum::extract::Request, next: axum::middleware::Next| { + let hook = hook.clone(); + let observed = observed.clone(); + async move { + if request.method() != axum::http::Method::POST + || request.uri().path() + != format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/sessions") + { + return next.run(request).await; + } + assert!(request.headers().contains_key( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER + )); + let armed = hook.lock().unwrap().take(); + let pin_root = armed.as_ref().map(|(_, _, root)| root.clone()); + if let Some((api, subject, _)) = armed { + let store = open_subjektiv_store(&api).unwrap(); + let policy = store.subject_permissions(&subject).unwrap().unwrap(); + assert!( + policy + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + store + .set_subject_permissions( + &subject, + &SubjectPermissionsUpdate { + operation_id: "session-registration-revoke-at-callback" + .into(), + expected_change_id: Some(policy.change_id), + execution: policy + .execution + .into_iter() + .filter(|p| p.operation != "session.record") + .collect(), + delegation: policy.delegation, + }, + ) + .unwrap(); + } + let (parts, body) = request.into_parts(); + let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); + let input: Value = serde_json::from_slice(&bytes).unwrap(); + if let Some(root) = pin_root { + let request: server_api::SubjektivRecordSessionRequest = serde_json::from_slice(&bytes).unwrap(); + let session = request.session_id.parse().unwrap(); + let retention = worker_runtime::retention::runtime_session_retention(&root); + retention.pin(session, session_store::retention::SessionRetentionPin(uuid::Uuid::now_v7()), || { + let entries = std::fs::read_dir(root.join("workers"))?; + for entry in entries { + let entry = entry?; + let name = format!("worker-runtime-{}", entry.file_name().to_string_lossy()); + let store = session_store::WorkerAggregateStore::open_read_only(&entry.path(), &name) + .map_err(|error| session_store::StoreError::Io(std::io::Error::other(error)))?; + if store.read_by_name(&name).unwrap().and_then(|m| m.active).is_some_and(|a| a.session_id == session) { return Ok(true); } + } + Ok(false) + }).unwrap(); + } + let response = next + .run(Request::from_parts(parts, Body::from(bytes))) + .await; + observed.lock().unwrap().push((response.status(), input)); + response + } + }, + )) + }) + .await; + let subject = open_subjektiv_store(&connected.api) + .unwrap() + .create_subject( + crate::subjektiv::SubjectRole::new("session-registration-connected").unwrap(), + ) + .unwrap() + .id; + let initial = open_subjektiv_store(&connected.api) + .unwrap() + .subject_permissions(&subject) + .unwrap() + .unwrap(); + assert!( + initial + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + if revoke_on_registration { + *revoke.lock().unwrap() = Some(( + connected.api.clone(), + subject.clone(), + connected.runtime_root.clone(), + )); + } + Self { + connected, + subject, + callbacks, + model, + } + } + + fn request(&self) -> CreateWorkspaceWorkerRequest { + CreateWorkspaceWorkerRequest { + runtime_id: CONNECTED_RUNTIME.into(), + display_name: "Connected Session registration".into(), + singleton_key: None, + profile: Some("builtin:companion".into()), + ticket_assignment: None, + initial_submit: vec![], + workdir_attachments: vec![], + feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest { + subjektiv: Some(server_api::SubjektivWorkerConnectionRequest { + subject_id: self.subject.clone(), + }), + }, + control_operation_id: None, + execution_grant: None, + } + } + + fn assert_callback(&self, expected: StatusCode) -> String { + let callbacks = self.callbacks.lock().unwrap(); + assert!(!callbacks.is_empty(), "real signed startup callback"); + assert!(callbacks.iter().all(|(status, body)| *status == expected + && body["session_id"] == callbacks[0].1["session_id"])); + let (status, input) = &callbacks[0]; + assert_eq!(*status, expected); + assert_eq!(input["create_if_missing"], true); + assert_eq!(self.model.0.load(Ordering::SeqCst), 0); + input["session_id"].as_str().unwrap().to_owned() + } +} + +// This deliberately inspects only persisted serde shape, not rendered error text +// or an error-contract dependency. The expected value is a typed live attachment. +fn contains_saved_value(saved: &Value, expected: &Value) -> bool { + saved == expected + || match saved { + Value::Object(fields) => fields + .values() + .any(|value| contains_saved_value(value, expected)), + Value::Array(values) => values + .iter() + .any(|value| contains_saved_value(value, expected)), + _ => false, + } +} + +fn read_metadata(root: &std::path::Path, worker: &str) -> Value { + // WorkerAggregateStore owns metadata.json; worker.json is the Runtime + // catalog. Open a new file handle on each read. + let path = root.join("workers").join(worker).join("metadata.json"); + serde_json::from_reader(std::fs::File::open(path).unwrap()).unwrap() +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn real_session_record_denial_survives_runtime_host_and_metadata_reopen() { + let fixture = RegistrationFixture::new(true).await; + // The real create preflight sees normal defaults. Revocation happens only + // after the actual factory has issued its signed Session-registration POST. + let error = scoped_create_workspace_worker( + State(fixture.connected.api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(fixture.request()), + ) + .await + .unwrap_err(); + fixture.assert_callback(StatusCode::FORBIDDEN); + let store = open_subjektiv_store(&fixture.connected.api).unwrap(); + assert!( + store + .subject_session_attributions(&fixture.subject) + .unwrap() + .is_empty() + ); + assert!( + !store + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap() + .execution + .iter() + .any(|p| p.operation == "session.record") + ); + + let failures = fixture.connected.runtime_startup_failures.lock().unwrap(); + assert_eq!( + failures.len(), + 1, + "actual RuntimeHTTP startup error envelope" + ); + let wire = &failures[0]; + let expected_primary = session_registration_attachment(SessionRegistrationCause::denied( + SessionPrincipal::Subject, + )) + .unwrap(); + assert_eq!(wire.primary, expected_primary); + assert_eq!( + wire.persistence, + Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted) + ); + let host = error + .diagnostics + .iter() + .filter_map(|d| d.startup.as_ref()) + .find(|startup| startup.primary == wire.primary) + .expect("RemoteWorkerRuntime must preserve the typed startup primary in Host diagnostics"); + assert_eq!(host.primary, wire.primary); + assert!( + host.secondary.iter().any(|failure| matches!( + failure, + protocol::startup_failure::StartupSecondaryFailure::Persistence { + phase: protocol::startup_failure::StartupFailurePhase::RuntimeRecordDelete, + .. + } + )), + "a real Session retention pin rejects outer Runtime deletion without hiding the primary" + ); + let expected = serde_json::to_value(&wire.primary).unwrap(); + assert!( + expected.is_string(), + "the shared attachment contains a codec envelope" + ); + + // A fresh fixture has exactly one attempted Worker. Discover its canonical + // saved aggregate, without making any assumptions about retained ownership + // or public catalog visibility after a failed create. + let workers = std::fs::read_dir(fixture.connected.runtime_root.join("workers")) + .unwrap() + .map(|entry| entry.unwrap().file_name().into_string().unwrap()) + .collect::>(); + assert_eq!( + workers.len(), + 1, + "factory attempt must have real saved metadata" + ); + let saved = read_metadata(&fixture.connected.runtime_root, &workers[0]); + assert!( + contains_saved_value(&saved, &expected), + "saved metadata must retain the original typed Session registration denial" + ); + let reopened = read_metadata(&fixture.connected.runtime_root, &workers[0]); + assert_eq!( + reopened, saved, + "read-only reopen, not a startup operation retry" + ); + assert!(contains_saved_value(&reopened, &expected)); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn permitted_session_registration_reaches_real_factory_and_saved_confirmation() { + let fixture = RegistrationFixture::new(false).await; + let initial = open_subjektiv_store(&fixture.connected.api) + .unwrap() + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap(); + let Json(created) = scoped_create_workspace_worker( + State(fixture.connected.api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(fixture.request()), + ) + .await + .unwrap(); + let session_id = fixture.assert_callback(StatusCode::OK); + let store = open_subjektiv_store(&fixture.connected.api).unwrap(); + assert_eq!( + store + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap(), + initial + ); + let attributions = store + .subject_session_attributions(&fixture.subject) + .unwrap(); + assert_eq!(attributions.len(), 1); + assert_eq!(attributions[0].session_id, session_id); + assert_eq!(attributions[0].worker_id, created.worker_id); + assert_eq!(attributions[0].runtime_id, CONNECTED_RUNTIME); + assert!( + fixture + .connected + .runtime_startup_failures + .lock() + .unwrap() + .is_empty() + ); + + let name = format!("worker-runtime-{}", created.worker_id); + let metadata = session_store::WorkerAggregateStore::open_read_only( + &fixture + .connected + .runtime_root + .join("workers") + .join(&created.worker_id), + &name, + ) + .unwrap(); + let saved = metadata.read_by_name(&name).unwrap().unwrap(); + assert!(matches!( + saved.subjektiv_session_attribution, + Some(session_store::SubjektivSessionAttributionState::Confirmed { .. }) + )); + let source = RuntimeWorkerRef::new(created.runtime_id, created.worker_id); + fixture + .connected + .factory + .retained_session_snapshot(&worker_runtime::identity::WorkerRef::try_from(&source).unwrap()) + .expect("real factory retained the permitted Worker Session"); +} diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index ab95afe7a..514fed0de 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs @@ -9,6 +9,8 @@ use super::*; mod chat_delegation_connected; #[path = "chat_tests.rs"] mod chat_tests; +#[path = "session_registration_connected_tests.rs"] +mod session_registration_connected; #[path = "subject_body_cleanup_connected_tests.rs"] mod subject_body_cleanup_connected; #[path = "subject_body_connected_tests.rs"] @@ -2737,6 +2739,7 @@ pub(super) struct ConnectedWorkerHarness { materializer: worker_runtime::working_directory::RuntimeGitMaterializer, authorizations: Arc>>, runtime_errors: Arc>>, + runtime_startup_failures: Arc>>, _backend_server: AuthorizationCallbackServer, _runtime_server: ConnectedRuntimeServer, _dir: tempfile::TempDir, @@ -2942,6 +2945,8 @@ impl ConnectedWorkerHarness { .unwrap(); let runtime_errors = Arc::new(Mutex::new(Vec::::new())); let captured_errors = runtime_errors.clone(); + let runtime_startup_failures = Arc::new(Mutex::new(Vec::new())); + let captured_startup_failures = runtime_startup_failures.clone(); let runtime_server = ConnectedRuntimeServer::start( runtime_listener, worker_runtime::http_server::runtime_http_router_with_workspace_auth( @@ -2955,6 +2960,7 @@ impl ConnectedWorkerHarness { .layer(axum::middleware::from_fn( move |request: axum::extract::Request, next: axum::middleware::Next| { let errors = captured_errors.clone(); + let startup_failures = captured_startup_failures.clone(); async move { let response = next.run(request).await; if !response.status().is_client_error() @@ -2965,6 +2971,13 @@ impl ConnectedWorkerHarness { let (parts, body) = response.into_parts(); let bytes = to_bytes(body, 2 * 1024 * 1024).await.unwrap(); let body: Value = serde_json::from_slice(&bytes).unwrap(); + if let Some(startup) = body["error"].get("startup").filter(|v| !v.is_null()) + { + startup_failures.lock().unwrap().push( + serde_json::from_value(startup.clone()) + .expect("RuntimeHTTP structured startup failure"), + ); + } errors.lock().unwrap().push( body["error"]["message"] .as_str() @@ -3045,6 +3058,7 @@ impl ConnectedWorkerHarness { materializer, authorizations, runtime_errors, + runtime_startup_failures, _backend_server: backend_server, _runtime_server: runtime_server, _dir: dir,