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

feat: transfer typed startup causes over Runtime HTTP with real registration fixture

Browse Source
This commit is contained in:
Hare
2026-11-04 15:50:30 +09:00
parent c4a34cf0cc
commit 7873f4359a
8 changed files with 543 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/runtime-api/src/lib.rs
+16
View File
@@ -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<StartupSecondaryFailure>,
pub resources: Option<StartupResourceFacts>,
pub persistence: Option<StartupPersistenceOutcome>,
pub io_kind: Option<StartupIoKind>,
}
#[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<WorkerStartupFailure>,
}
impl RuntimeApiError {
@@ -117,6 +132,7 @@ impl RuntimeApiError {
error: RuntimeApiErrorDetail {
code: code.into(),
message: message.into(),
startup: None,
},
status,
}
crates/worker-runtime/src/http_server.rs
+38 -1
View File
@@ -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<runtime_api::WorkerStartupFailure>,
}
#[cfg(feature = "ws-server")]
@@ -2325,6 +2327,7 @@ struct RuntimeHttpRestError {
status: StatusCode,
code: String,
message: String,
startup: Option<runtime_api::WorkerStartupFailure>,
}
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<runtime_api::WorkerStartupFailure> {
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 { .. } => {
crates/worker-runtime/src/http_server/runtime_management_api.rs
+4 -1
View File
@@ -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 {
crates/worker/src/startup_failure.rs
+14
View File
@@ -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<StartupFailureAttachment, error_contract::ContractError> {
if let Ok(projected) = failure
.project_for::<InternalTransfer, protocol::session_failure::SessionRegistrationCause>(&())
{
return StartupFailureAttachment::from_projected(&projected);
}
let projected = failure.project_for::<InternalTransfer, RuntimeStartupCause>(&())?;
StartupFailureAttachment::from_projected(&projected)
}
/// Dispatch by the domain attested in process, not by parsing attachment fields.
pub(crate) fn persist_binding_startup_failure<S: session_store::WorkerMetadataStore>(
store: &S,
crates/workspace-server/src/hosts.rs
+38 -5
View File
@@ -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<runtime_api::WorkerStartupFailure>,
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<runtime_api::WorkerStartupFailure> {
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<String>,
) -> RuntimeDiagnostic {
RuntimeDiagnostic {
startup: None,
code: code.into(),
severity,
message: message.into(),
crates/workspace-server/src/server.rs
+50
View File
@@ -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<T>(
.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<T>(
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<String>,
) -> RuntimeDiagnostic {
RuntimeDiagnostic {
startup: None,
code: code.into(),
severity,
message: message.into(),
@@ -39260,6 +39279,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
message: message.clone(),
},
vec![RuntimeDiagnostic {
startup: None,
code: "repository_not_configured".to_string(),
severity: HostDiagnosticSeverity::Error,
message,
@@ -39277,6 +39297,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
message: message.clone(),
},
vec![RuntimeDiagnostic {
startup: None,
code: "repository_provider_unsupported".to_string(),
severity: HostDiagnosticSeverity::Error,
message,
@@ -39292,6 +39313,7 @@ fn repository_lookup<T>(result: std::result::Result<T, RepositoryLookupError>) -
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<Error> 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<Error> 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<Error> 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(),
crates/workspace-server/src/server/tests/session_registration_connected_tests.rs
+369
View File
@@ -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<AtomicUsize>);
#[async_trait::async_trait]
impl agen::llm_client::LlmClient for UnusedModel {
fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
Box::new(self.clone())
}
async fn stream(
&self,
_: agen::llm_client::Request,
) -> std::result::Result<agen::llm_client::client::ResponseStream, agen::llm_client::ClientError>
{
self.0.fetch_add(1, Ordering::SeqCst);
panic!("Session registration does not require a model turn")
}
}
struct RegistrationFixture {
connected: ConnectedWorkerHarness,
subject: String,
callbacks: Arc<Mutex<Vec<(StatusCode, Value)>>>,
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::<Vec<_>>();
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");
}
crates/workspace-server/src/server/tests/subject_authorization_tests.rs
+14
View File
@@ -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<Mutex<Vec<(Value, StatusCode)>>>,
runtime_errors: Arc<Mutex<Vec<String>>>,
runtime_startup_failures: Arc<Mutex<Vec<runtime_api::WorkerStartupFailure>>>,
_backend_server: AuthorizationCallbackServer,
_runtime_server: ConnectedRuntimeServer,
_dir: tempfile::TempDir,
@@ -2942,6 +2945,8 @@ impl ConnectedWorkerHarness {
.unwrap();
let runtime_errors = Arc::new(Mutex::new(Vec::<String>::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,