From 9fadd1bcaffcf26784f475aed9731f2053f1f534 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 4 Nov 2026 19:06:41 +0900 Subject: [PATCH] fix: retain local Session binding causes and factory sources safely --- Cargo.lock | 1 + crates/standalone/Cargo.toml | 1 + crates/standalone/src/subjektiv.rs | 36 ++++- crates/standalone/src/subjektiv/journal.rs | 51 ++++--- crates/standalone/src/subjektiv/tests.rs | 139 ++++++++++++++++++ crates/worker-runtime/src/worker_backend.rs | 57 +++++-- .../worker_backend/startup_binding_tests.rs | 86 ++++++++++- .../src/worker_backend/startup_failure.rs | 14 +- crates/workspace-server/src/hosts.rs | 119 ++++++++++++++- 9 files changed, 462 insertions(+), 42 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index ffec5d235..edb7fced8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5040,6 +5040,7 @@ dependencies = [ "agen", "async-trait", "client", + "error-contract", "feature-storage", "fs4", "futures", diff --git a/crates/standalone/Cargo.toml b/crates/standalone/Cargo.toml index 599d4a565..a72a9f308 100644 --- a/crates/standalone/Cargo.toml +++ b/crates/standalone/Cargo.toml @@ -28,6 +28,7 @@ uuid = { workspace = true, features = ["v7"] } worker.workspace = true [dev-dependencies] +error-contract.workspace = true async-trait.workspace = true futures.workspace = true tempfile.workspace = true diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 82a47b836..99543af5f 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -275,6 +275,8 @@ impl StandaloneSubjects { writer, self_weak: weak.clone(), journal_dispatch: Mutex::new(false), + #[cfg(test)] + registration_before_attribution: Mutex::new(None), language: self.memory_language.clone(), catalog: self, binding: binding.clone(), @@ -373,6 +375,8 @@ pub(crate) struct LocalSubjectHost { writer: subjektiv::journal::JournalWriterBinding, self_weak: std::sync::Weak, journal_dispatch: Mutex, + #[cfg(test)] + registration_before_attribution: Mutex>>, catalog: StandaloneSubjects, binding: StandaloneSubjectBinding, worker_id: protocol::WorkerId, @@ -467,7 +471,11 @@ impl LocalSubjectHost { context: &SubjektivHostContext, create: bool, ) -> Result { - self.body(context)?; + self.registration_authority(context)?; + #[cfg(test)] + if let Some(checkpoint) = self.registration_before_attribution.lock().unwrap().take() { + checkpoint(); + } let store = &self.catalog.store; if let Some(record) = store .session_attribution(&context.session_id) @@ -484,7 +492,7 @@ impl LocalSubjectHost { } store .bind_journal_session(&self.writer, record.clone()) - .map_err(journal::journal_error)?; + .map_err(registration_journal_error)?; return Ok(record); } if !create { @@ -507,11 +515,25 @@ impl LocalSubjectHost { .map_err(registration_domain_error)?; store .bind_journal_session(&self.writer, record.clone()) - .map_err(journal::journal_error)?; + .map_err(registration_journal_error)?; Ok(record) } } +fn registration_journal_error(error: subjektiv::journal::JournalError) -> SubjektivHostError { + let cause = match &error { + subjektiv::journal::JournalError::PermissionDenied(_) => { + SessionRegistrationCause::binding_mismatch() + } + subjektiv::journal::JournalError::Subject( + subjektiv::SubjektivError::SessionSubjectConflict { .. } + | subjektiv::SubjektivError::SessionAttributionConflict { .. }, + ) => SessionRegistrationCause::binding_mismatch(), + _ => SessionRegistrationCause::outcome_unknown(), + }; + SubjektivHostError::session_registration(cause, error) +} + fn registration_domain_error(error: subjektiv::SubjektivError) -> SubjektivHostError { let cause = match &error { subjektiv::SubjektivError::SessionSubjectConflict { .. } @@ -868,7 +890,11 @@ impl SubjektivHost for LocalSubjectHost { // An explicit operator connection issues this local body capability. // Revalidate its live writer on every access/supply, not Feature enablement. // Job decisions and Backend conversations are not local body capabilities. - self.body(context)?; + if operation == "session.record" { + self.registration_authority(context)?; + } else { + self.body(context)?; + } match operation { "session.record" | "memory.resident.read" @@ -1095,7 +1121,7 @@ impl SubjektivHost for LocalSubjectHost { ) -> Result { self.registration_authority(context)?; let attribution = self.attribution(context, create_if_missing)?; - self.reserve_journal_session(context)?; + self.reserve_journal_session(context, &attribution)?; Ok(SubjektivRecordSessionResponse { subject_id: attribution.subject_id, session_id: attribution.session_id, diff --git a/crates/standalone/src/subjektiv/journal.rs b/crates/standalone/src/subjektiv/journal.rs index 85bbe26f1..67699f11a 100644 --- a/crates/standalone/src/subjektiv/journal.rs +++ b/crates/standalone/src/subjektiv/journal.rs @@ -220,6 +220,12 @@ pub(super) fn journal_error(error: JournalError) -> SubjektivHostError { other => SubjektivHostError::Unavailable(other.to_string()), } } +fn registration_reservation_error( + error: impl std::error::Error + Send + Sync + 'static, +) -> SubjektivHostError { + SubjektivHostError::session_registration(SessionRegistrationCause::outcome_unknown(), error) +} + fn source_error(error: session_store::StoreError) -> SubjektivHostError { SubjektivHostError::Unavailable(error.to_string()) } @@ -368,45 +374,56 @@ impl LocalSubjectHost { pub(super) fn reserve_journal_session( &self, context: &SubjektivHostContext, + attribution: &SubjectSessionAttribution, ) -> Result<(), SubjektivHostError> { - self.body(context)?; - if self.journal_session_erased(&context.session_id)? { + self.registration_authority(context)?; + let journal_context = + JournalHostContext::validated_session(&self.catalog.store, attribution.clone()) + .map_err(registration_journal_error)?; + if self + .catalog + .store + .journal_session_erased(&journal_context) + .map_err(registration_journal_error)? + { return Ok(()); } - let session: session_store::SessionId = context - .session_id - .parse() - .map_err(|_| SubjektivHostError::InvalidArgument("invalid Session".into()))?; + let session: session_store::SessionId = context.session_id.parse().map_err(|error| { + SubjektivHostError::session_registration( + SessionRegistrationCause::invalid_response(), + error, + ) + })?; let root = self .catalog .state_root .join(self.worker_id.to_string()) .join("sessions"); - reject_symlink_components(&root) - .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?; + reject_symlink_components(&root).map_err(registration_reservation_error)?; let source_store = session_store::FsStore::new_with_retention( &root, crate::store::session_retention(&self.catalog.state_root), ) - .map_err(source_error)?; + .map_err(registration_reservation_error)?; { let _guard = source_store .retention() .lock_live(session) - .map_err(source_error)?; + .map_err(registration_reservation_error)?; let path = root.join(session.to_string()); match fs::create_dir(&path) { Ok(()) => {} Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => { - let metadata = fs::symlink_metadata(&path) - .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?; + let metadata = + fs::symlink_metadata(&path).map_err(registration_reservation_error)?; if !metadata.is_dir() || metadata.file_type().is_symlink() { - return Err(SubjektivHostError::InvalidArgument( - "Session retention source is not an owned directory".into(), - )); + return Err(registration_reservation_error(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Session retention source is not an owned directory", + ))); } } - Err(error) => return Err(SubjektivHostError::Unavailable(error.to_string())), + Err(error) => return Err(registration_reservation_error(error)), } } // Establish retention at attribution, before the first Run can publish bytes. This @@ -416,7 +433,7 @@ impl LocalSubjectHost { session, session_store::retention::SessionRetentionPin(session), ) - .map_err(source_error) + .map_err(registration_reservation_error) } pub(super) fn admit_journal( diff --git a/crates/standalone/src/subjektiv/tests.rs b/crates/standalone/src/subjektiv/tests.rs index f4c69a29e..4f7a7fdfd 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -324,3 +324,142 @@ fn session_registration_uses_real_capability_and_attribution_decisions() { SessionRegistrationCause::denied(SessionPrincipal::Subject), ); } + +#[derive(Clone)] +struct RegistrationUnusedModel; +#[async_trait::async_trait] +impl agen::llm_client::LlmClient for RegistrationUnusedModel { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + async fn stream( + &self, + _: agen::llm_client::Request, + ) -> Result { + panic!("Session registration must not invoke a model") + } +} + +#[tokio::test] +async fn local_writer_revocation_during_registration_retains_known_cause_and_real_journal_source() { + use session_store::{CombinedStore, FsStore, FsWorkerStore, WorkerMetadataStore}; + use worker::{ + WorkerBootstrap, WorkerBootstrapLayout, WorkerFilesystemAuthority, WorkerWorkspaceContext, + }; + // The direct non-Browser caller uses the real bootstrap, local capability and + // transactional Journal store. Only the unused model and revocation timing are substituted. + for existing in [false, true] { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("state"); + let catalog = StandaloneSubjects::open(&root).unwrap(); + let subject = catalog.create("registration-race", None).unwrap(); + let id = protocol::WorkerId::now_v7(); + let mut connection = catalog.connect(&subject.id, id).unwrap(); + let port = connection.host.clone(); + let name = format!("standalone-{id}"); + let sessions = FsStore::new_with_retention( + root.join(id.to_string()).join("sessions"), + crate::store::session_retention(&root), + ) + .unwrap(); + let metadata_root = temp.path().join("metadata"); + let metadata = FsWorkerStore::new(&metadata_root).unwrap(); + let mut manifest = manifest::WorkerManifest::from_toml(&format!("[worker]\nname='{name}'\n[model]\nscheme='anthropic'\nmodel_id='unused'\n[engine]\n[scope]\nallow=[]")).unwrap(); + manifest.feature.subjektiv.profile.enabled = true; + manifest.feature.subjektiv.profile.extraction.enabled = false; + let mut prepared = WorkerBootstrap::new( + manifest, + CombinedStore::new(sessions, metadata.clone()), + worker::PromptCatalogSource::builtins_only(), + WorkerWorkspaceContext::no_workspace(), + WorkerFilesystemAuthority::None, + WorkerBootstrapLayout::Direct { + runtime_base: temp.path().join("runtime"), + }, + ) + .with_model_client(RegistrationUnusedModel) + .prepare() + .await + .unwrap(); + let worker = prepared.worker_mut(); + worker.enable_worker_metadata_write_through().unwrap(); + worker.bind_subjektiv_host(Some(port.clone())); + let session = worker.session_id(); + let context = SubjektivHostContext { + worker_id: name.clone(), + session_id: session.to_string(), + }; + if existing { + port.record_session(&context, true).unwrap(); + } + let revoke_store = port.catalog.store.clone(); + let writer = port.writer.clone(); + *port.registration_before_attribution.lock().unwrap() = Some(Box::new(move || { + // This actual release follows the last preflight, before transactional bind. + revoke_store.release_journal_writer(&writer).unwrap(); + })); + let error = worker + .finalize_subjektiv_session_attribution_retaining_startup_failure( + true, + worker::SubjektivSessionAttributionLifecycle::NewSession, + ) + .unwrap_err(); + assert_eq!( + error.session_registration_cause(), + Some(SessionRegistrationCause::binding_mismatch()) + ); + let mut source = Some(&error as &(dyn std::error::Error + 'static)); + let mut journal_source = false; + while let Some(current) = source { + if let Some(journal) = current.downcast_ref::() { + assert!(matches!( + journal, + subjektiv::journal::JournalError::PermissionDenied(_) + )); + journal_source = true; + } + source = current.source(); + } + assert!( + journal_source, + "original store refusal must survive the Worker wrapper" + ); + assert_eq!( + error.startup_metadata_save(), + Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted) + ); + drop(prepared); + drop(metadata); + let reopened = FsWorkerStore::new(metadata_root) + .unwrap() + .read_by_name(&name) + .unwrap() + .unwrap(); + let session_store::worker_metadata::StartupFailureRead::Present( + error_contract::DecodeFailure::Known(cause), + ) = reopened.decode_startup_failure::() + else { + panic!("typed cause must really reopen") + }; + assert_eq!( + *cause.facts().cause(), + SessionRegistrationCause::binding_mismatch() + ); + assert!( + port.catalog + .store + .session_attribution(&session.to_string()) + .unwrap() + .is_some(), + "committed attribution is independent from denied writer binding" + ); + assert!( + !port + .catalog + .store + .journal_writer_active(&port.writer) + .unwrap() + ); + connection.close().unwrap(); + } +} diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 09b93f923..16ed5bffb 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -257,7 +257,9 @@ enum RuntimeRestoreCleanup { /// Diagnostics are fixed owner-selected causes, not arbitrary provider messages, /// paths, configuration, or credentials. Cleanup cannot overwrite the first cause. -#[derive(Debug, Clone, PartialEq, Eq)] +/// The original in-process source is privately shared across clones; it is neither +/// formatted into diagnostics nor included in the wire/storage projection. +#[derive(Clone)] pub struct RuntimeRestoreFailure { pub stage: RuntimeRestoreStage, pub resources: RuntimeRestoreResources, @@ -270,6 +272,7 @@ pub struct RuntimeRestoreFailure { secondary: Vec, legacy_diagnostic: Option, metadata_save: Option, + source: Option>, } impl RuntimeRestoreFailure { @@ -301,7 +304,13 @@ impl RuntimeRestoreFailure { self.io_kind } - fn from_worker(stage: RuntimeRestoreStage, error: &WorkerError) -> Self { + fn from_worker(stage: RuntimeRestoreStage, error: WorkerError) -> Self { + let mut failure = Self::from_worker_ref(stage, &error); + failure.source = Some(Arc::new(error)); + failure + } + + fn from_worker_ref(stage: RuntimeRestoreStage, error: &WorkerError) -> Self { let mut failure = Self::started(stage, restore_worker_cause(error)); failure.primary = error .startup_failure_at(stage.into()) @@ -354,6 +363,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -370,6 +380,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -386,6 +397,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -397,6 +409,25 @@ impl RuntimeRestoreFailure { } } +impl std::fmt::Debug for RuntimeRestoreFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("RuntimeRestoreFailure") + .field("stage", &self.stage) + .field("resources", &self.resources) + .field("cause", &self.cause) + .field("cleanup", &self.cleanup) + .field("io_kind", &self.io_kind) + .field("cleanup_failure", &self.cleanup_failure) + .finish_non_exhaustive() + } +} + +impl std::error::Error for RuntimeRestoreFailure { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + self.source.as_deref().map(|source| source as _) + } +} + impl std::fmt::Display for RuntimeRestoreFailure { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let cause: String = self @@ -2225,7 +2256,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { None => bootstrap, }; let mut prepared = bootstrap.prepare().await.map_err(|error| { - RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error) + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, error) })?; let worker = prepared.worker_mut(); if let Some(snapshot) = &request.request.plugin_execution { @@ -2299,7 +2330,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { let workspace_client = worker.workspace_client_handle(); let started = prepared.start().await.map_err(|error| { - RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error) + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, error) })?; let (handle, shutdown_rx, controller_task) = (started.handle, started.shutdown, started.controller_task); @@ -2542,7 +2573,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .await .map_err(|error| { self.retain_construction_failure(&request, &error); - let failure = RuntimeRestoreFailure::from_worker(stage, &error); + let failure = RuntimeRestoreFailure::from_worker(stage, error); let cause = failure.cause; preparation_failure = Some(failure); cause @@ -2818,7 +2849,7 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate { Some(source.kind()), ), }; - let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, &error); + let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, error); failure.cause = cause; failure.durable_effects_possible = true; failure.io_kind = io_kind; @@ -3090,11 +3121,17 @@ impl ProfileRuntimeWorkerFactory { WorkerBootstrapError::Worker(error) => { stage = RuntimeRestoreStage::Reconstruction; self.retain_construction_failure(request, &error); - primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, &error)); - restore_worker_cause(&error) + let cause = restore_worker_cause(&error); + primary_failure = Some(RuntimeRestoreFailure::from_worker(stage, error)); + cause } - WorkerBootstrapError::Controller { source, .. } => { - io_kind = Some(source.kind()); + error @ WorkerBootstrapError::Controller { .. } => { + let failure = RuntimeRestoreFailure::from_bootstrap( + RuntimeRestoreStage::Controller, + error, + ); + io_kind = failure.io_kind; + primary_failure = Some(failure); "restored Worker preparation failed" } })?; diff --git a/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs index 0053e596c..783e17e15 100644 --- a/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs +++ b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs @@ -1,7 +1,8 @@ mod startup_binding_tests { //! Exercise the factory's production binding step with a constructed Worker, //! real aggregate/Session stores and reopen. Only the Workspace response is - //! synthetic; this suite does not test domain authorization or network transport. + //! synthetic in the binding cases; the direct internal Profile factory case uses + //! a refused real loopback transport. Domain authorization is tested by Server. use super::*; use protocol::session_failure::{ SessionPrincipal, SessionRegistrationCause, session_registration_attachment, @@ -154,6 +155,21 @@ mod startup_binding_tests { let failure = startup_failure::finalize_startup_binding(&worker, true) .err() .expect("failure must remain a failure"); + if matches!(response, Response::Lost | Response::Rejected) { + let mut source = std::error::Error::source(&failure); + let mut worker = false; + let mut host = false; + let mut client = false; + while let Some(current) = source { + worker |= current.is::(); + host |= current.is::(); + client |= current.is::(); + source = current.source(); + } + assert!(worker && host, "common factory binding step retains in-process domain sources"); + assert_eq!(client, matches!(response, Response::Lost)); + assert!(!format!("{failure:?} {failure}").contains("response deliberately lost")); + } let result = failure.execution_result(WorkerExecutionOperation::Spawn); if matches!(response, Response::BeforeSession | Response::BeforeOwner) { let primary = result.startup_failure::().unwrap(); @@ -260,6 +276,74 @@ mod startup_binding_tests { } } } + #[cfg(feature = "http-server")] + #[tokio::test] + #[serial_test::serial(worker_allocation)] + async fn internal_profile_factory_transport_failure_retains_source_chain_and_reopened_cause() { + require_restore_test_environment(); + let root = tempfile::tempdir().unwrap(); + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let endpoint = format!("http://{}", listener.local_addr().unwrap()); + drop(listener); + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + let mut request = create_request("internal factory source"); + request.worker_id = worker_ref.worker_id; + request.workspace_api = Some(WorkspaceApiRef { workspace_id: "adapter-test".into(), base_url: endpoint.clone() }); + request.subjektiv_attached = true; + request.memory_settings = Some(manifest::WorkspaceMemorySettingsSnapshot { workspace_id: "adapter-test".into(), language: "English".into() }); + let archive = crate::profile_archive::ProfileSourceArchive::build_evaluated_profile( + "profile-source-archive:internal-source".into(), "builtin:companion".into(), + serde_json::json!({"slug":"internal-source", "description":"internal source test", "scope":"workspace_read", + "model":{"scheme":"anthropic", "model_id":"unused", "auth":{"kind":"none"}}, + "engine":{"max_tokens":100}, "feature":{"subjektiv":{"enabled":true,"extraction":{"enabled":false}}}}) + ).unwrap(); + let mut bundle = test_bundle(); + bundle.profile_source_archive = Some(archive.clone()); + let bundle = bundle.with_computed_digest(); + request.profile_source = crate::catalog::ProfileSourceArchiveSource::Embedded { archive }; + request.config_bundle = Some(ConfigBundleRef { id: bundle.metadata.id.clone(), digest: bundle.metadata.digest.clone() }); + let request = WorkerExecutionSpawnRequest { + worker_ref: worker_ref.clone(), request, + workspace_scope: None, + context: test_execution_context(worker_ref.clone()), + workdir_attachments: BTreeMap::new(), config_bundle: Some(bundle), + }; + // Direct internal caller, not a Browser handler or Runtime result decoder. + // The real transport is refused at an unused loopback socket; only the model is replaced. + let factory = ProfileRuntimeWorkerFactory::new(root.path()) + .with_runtime_id("internal-source-runtime") + .with_runtime_store_dir(root.path().join("runtime")) + .with_spawn_model_client_factory(|| Box::new(MockClient::new(vec![]))); + let failure = match factory.spawn_controller(request).await { + Err(failure) => failure, + Ok(_) => panic!("disconnected Backend must not create a Controller"), + }; + assert_eq!(failure.stage, RuntimeRestoreStage::Binding); + let mut source = std::error::Error::source(&failure); + let mut worker = false; let mut host = false; let mut transport = false; + while let Some(current) = source { + worker |= current.is::(); + host |= current.is::(); + if let Some(error) = current.downcast_ref::() { + assert!(matches!(error, WorkspaceClientError::Request(_))); + transport = true; + } + source = current.source(); + } + assert!(worker && host && transport, "factory must retain original typed in-process source"); + let result = failure.execution_result(WorkerExecutionOperation::Spawn); + let Some(error_contract::DecodeFailure::Known(cause)) = result.startup_failure::() else { panic!("Session cause must remain typed") }; + assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown()); + assert_eq!(result.startup_failure_persistence, Some(StartupPersistenceOutcome::Persisted)); + let aggregate = root.path().join("runtime/workers").join(worker_ref.worker_id.to_string()); + let name = ProfileRuntimeWorkerFactory::runtime_worker_name_for_ref(&worker_ref); + let metadata = WorkerAggregateStore::new(aggregate, &name).unwrap().read_by_name(&name).unwrap().unwrap(); + let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = metadata.decode_startup_failure::() else { panic!("saved factory cause must reopen") }; + assert_eq!(*cause.facts().cause(), SessionRegistrationCause::outcome_unknown()); + assert!(!format!("{failure:?} {failure}").contains(&endpoint)); + assert!(!serde_json::to_string(&result).unwrap().contains(&endpoint)); + } + #[tokio::test] async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() { check_binding(Response::Rejected).await; diff --git a/crates/worker-runtime/src/worker_backend/startup_failure.rs b/crates/worker-runtime/src/worker_backend/startup_failure.rs index 8f70f3bde..1b01db12a 100644 --- a/crates/worker-runtime/src/worker_backend/startup_failure.rs +++ b/crates/worker-runtime/src/worker_backend/startup_failure.rs @@ -32,9 +32,9 @@ impl From for protocol::startup_failure::StartupCl } impl RuntimeRestoreFailure { - pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: &WorkerBootstrapError) -> Self { - match error { - WorkerBootstrapError::Worker(source) => Self::from_worker(stage, source), + pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: WorkerBootstrapError) -> Self { + let mut failure = match &error { + WorkerBootstrapError::Worker(source) => Self::from_worker_ref(stage, source), WorkerBootstrapError::Controller { source, .. } => { let mut failure = Self::started( RuntimeRestoreStage::Controller, @@ -43,7 +43,9 @@ impl RuntimeRestoreFailure { failure.io_kind = Some(source.kind()); failure } - } + }; + failure.source = Some(Arc::new(error)); + failure } } @@ -51,7 +53,7 @@ impl From for RuntimeRestoreFailure { fn from(error: session_store::WorkerStoreError) -> Self { Self::from_worker( RuntimeRestoreStage::Metadata, - &WorkerError::WorkerStore(error), + WorkerError::WorkerStore(error), ) } } @@ -114,7 +116,7 @@ where attached, SubjektivSessionAttributionLifecycle::NewSession, ) - .map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error)) + .map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, error)) } #[cfg(test)] diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 58e5871de..a89cff9ed 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6867,7 +6867,10 @@ 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); + if let Some(startup) = error.error.startup { + return remote_startup_diagnostic(runtime_id, status, startup); + } + let startup = None; let remote_code = error.error.code; let message = sanitize_remote_runtime_message(&remote_code, &error.error.message); let (code, severity) = match status { @@ -6946,6 +6949,9 @@ fn remote_http_status_diagnostic( .ok() .filter(|_| body.len() <= MAX_RUNTIME_PING_RESPONSE_BYTES) .and_then(|_| serde_json::from_slice::(&body).ok()); + if let Some(startup) = error.as_ref().and_then(|error| error.error.startup.clone()) { + return remote_startup_diagnostic(runtime_id, status, startup); + } let remote_code = error .as_ref() .map(|error| error.error.code.as_str()) @@ -6980,6 +6986,29 @@ fn remote_http_status_diagnostic( projection } +fn remote_startup_diagnostic( + runtime_id: &str, + status: StatusCode, + startup: runtime_api::WorkerStartupFailure, +) -> RuntimeDiagnostic { + let startup = internal_startup_transfer(startup); + let message = startup.as_ref().map_or("Worker startup failed", |startup| { + worker::startup_failure::internal_startup_failure_summary(&startup.primary) + }); + let code = if matches!(status, StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN) { + "remote_runtime_auth_failed" + } else { + "remote_runtime_startup_failed" + }; + let mut projection = diagnostic( + code, + HostDiagnosticSeverity::Error, + format!("Remote Runtime '{runtime_id}' rejected request (HTTP {status}): {message}"), + ); + projection.startup = startup; + 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( @@ -7219,7 +7248,7 @@ mod tests { let mut wire = runtime_api::RuntimeApiError::new( 500, "worker_execution_rejected", - "unused remote prose", + "TEST_ONLY_CREDENTIAL private-subject-id raw_http_body", ); wire.error.startup = embedded.startup.clone(); let received = serde_json::from_str(&serde_json::to_string(&wire).unwrap()).unwrap(); @@ -7237,8 +7266,14 @@ mod tests { public.get("startup").is_none(), "no blind public/storage attachment forwarding" ); - for secret in ["TEST_ONLY_CREDENTIAL", "/private/path", "raw_http_body"] { + for secret in [ + "TEST_ONLY_CREDENTIAL", + "private-subject-id", + "/private/path", + "raw_http_body", + ] { assert!(!serde_json::to_string(&embedded).unwrap().contains(secret)); + assert!(!serde_json::to_string(&remote).unwrap().contains(secret)); } let legacy = runtime_api_diagnostic( @@ -7257,6 +7292,84 @@ mod tests { assert!(internal_startup_transfer(corrupt).is_none()); } + #[test] + fn remote_startup_receivers_discard_untrusted_prose_and_code_even_without_paths() { + use protocol::session_failure::{ + SessionRegistrationCause, session_registration_attachment, + }; + let valid = runtime_api::WorkerStartupFailure { + primary: session_registration_attachment(SessionRegistrationCause::binding_mismatch()) + .unwrap(), + secondary: vec![], + persistence: None, + resources: None, + io_kind: None, + }; + for corrupt in [false, true] { + for status in [StatusCode::FORBIDDEN, StatusCode::INTERNAL_SERVER_ERROR] { + let mut startup = valid.clone(); + if corrupt { + startup.primary = serde_json::from_value(json!("corrupt attachment")).unwrap(); + } + let mut wire = runtime_api::RuntimeApiError::new( + status.as_u16(), + "PRIVATE_CODE_SENTINEL", + "TEST_ONLY_CREDENTIAL private-subject-id raw_http_body", + ); + wire.error.startup = Some(startup); + let body = serde_json::to_string(&wire).unwrap(); + let typed = runtime_api_diagnostic( + "remote", + runtime_api::client_support::ClientError::Public { + status, + error: serde_json::from_str(&body).unwrap(), + }, + ); + let (url, server) = serve_mock_http(vec![mock_response( + "GET", + "/", + false, + status.as_u16(), + body, + )]); + let response = reqwest::blocking::Client::builder() + .timeout(std::time::Duration::from_secs(5)) + .build() + .unwrap() + .get(url) + .send() + .unwrap(); + let http = remote_http_status_diagnostic("remote", status, response); + server.join().unwrap(); + for diagnostic in [typed, http] { + assert_eq!(diagnostic.startup.is_some(), !corrupt); + assert_eq!(diagnostic.message.contains("session.record"), !corrupt); + assert_eq!( + diagnostic.code, + if status == StatusCode::FORBIDDEN { + "remote_runtime_auth_failed" + } else { + "remote_runtime_startup_failed" + } + ); + let public = serde_json::to_string(&diagnostic).unwrap(); + for secret in [ + "PRIVATE_CODE_SENTINEL", + "TEST_ONLY_CREDENTIAL", + "private-subject-id", + "raw_http_body", + "corrupt attachment", + ] { + assert!( + !public.contains(secret), + "received prose/code must never bypass startup projection: {public}" + ); + } + } + } + } + } + #[tokio::test] async fn async_workdir_authorization_uses_store_capacity_without_bearer_fallback() { use crate::db_executor::{DbExecutor, DbExecutorLimits};