diff --git a/Cargo.lock b/Cargo.lock index ac342bd1b..e288d73ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5040,6 +5040,7 @@ dependencies = [ "agen", "async-trait", "client", + "error-contract", "feature-storage", "fs4", "futures", @@ -7579,6 +7580,7 @@ dependencies = [ "content-history", "content-history-adapters", "content-history-store", + "error-contract", "feature-permissions", "feature-storage", "flate2", 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 01e40d7ae..daa1bdc49 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -276,6 +276,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(), @@ -374,6 +376,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, @@ -468,7 +472,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) @@ -485,7 +493,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 { @@ -508,11 +516,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 { .. } @@ -984,7 +1006,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" @@ -1211,7 +1237,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 e5cd1c404..1197b2f7f 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 5c1d70c1f..420b36aee 100644 --- a/crates/standalone/src/subjektiv/tests.rs +++ b/crates/standalone/src/subjektiv/tests.rs @@ -767,3 +767,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/http_server.rs b/crates/worker-runtime/src/http_server.rs index 0a4729380..9126f3c94 100644 --- a/crates/worker-runtime/src/http_server.rs +++ b/crates/worker-runtime/src/http_server.rs @@ -2347,8 +2347,8 @@ impl RuntimeHttpRestError { RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result), _ => None, }; - let message = if startup.is_some() { - "Worker startup failed".to_owned() + let message = if let Some(startup) = &startup { + worker::startup_failure::internal_startup_failure_summary(&startup.primary).to_owned() } else { error.to_string() }; diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index beba4493a..40d47e8ba 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -271,7 +271,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, @@ -284,6 +286,7 @@ pub struct RuntimeRestoreFailure { secondary: Vec, legacy_diagnostic: Option, metadata_save: Option, + source: Option>, } impl RuntimeRestoreFailure { @@ -315,7 +318,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()) @@ -368,6 +377,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -384,6 +394,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -400,6 +411,7 @@ impl RuntimeRestoreFailure { secondary: Vec::new(), legacy_diagnostic: None, metadata_save: None, + source: None, } } @@ -411,6 +423,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 @@ -2352,7 +2383,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { if let WorkerBootstrapError::Worker(source) = &error { self.retain_spawn_construction_failure(&request.worker_ref, source); } - RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error) + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, error) })?; let worker = prepared.worker_mut(); { @@ -2449,7 +2480,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { if let WorkerBootstrapError::Worker(source) = &error { self.retain_spawn_construction_failure(&request.worker_ref, source); } - RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error) + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, error) })?; self.failed_spawn_resources .lock() @@ -2696,7 +2727,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 @@ -2972,7 +3003,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; @@ -3244,11 +3275,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/restore_recovery.rs b/crates/worker-runtime/src/worker_backend/restore_recovery.rs index 4c751e872..db5c8c85c 100644 --- a/crates/worker-runtime/src/worker_backend/restore_recovery.rs +++ b/crates/worker-runtime/src/worker_backend/restore_recovery.rs @@ -116,7 +116,7 @@ impl ProfileRuntimeWorkerFactory { .await; let mut worker = worker_result.map_err(|error| { self.retain_construction_failure(request, &error); - cause(restore_worker_cause(&error)) + RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Recovery, error) })?; let tasks = worker.retain_startup_feature_cleanup(); let sessions = worker.workdir_sessions(); @@ -141,11 +141,13 @@ impl ProfileRuntimeWorkerFactory { .map_err(|_| cause("recovery Backend Job binding failed"))?; } worker - .finalize_subjektiv_session_attribution( + .finalize_subjektiv_session_attribution_retaining_startup_failure( request.request.subjektiv_attached, SubjektivSessionAttributionLifecycle::RestoredSession, ) - .map_err(|error| cause(restore_worker_cause(&error)))?; + .map_err(|error| { + RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Recovery, error) + })?; if pending { worker::WorkerController::recover_pending_head(&mut worker, &root) .await @@ -161,21 +163,38 @@ impl ProfileRuntimeWorkerFactory { tasks .shutdown() .await - .map_err(|_| cause("recovery Feature settlement failed"))?; + .map_err(|_| RuntimeRestoreCleanupFailure::Feature)?; sessions .close_all() .await - .map_err(|_| cause("recovery Workdir close failed"))?; + .map_err(|_| RuntimeRestoreCleanupFailure::Workdir)?; if let Some(allocation) = allocation { allocation .release() - .map_err(|_| cause("recovery allocation release failed"))?; + .map_err(|_| RuntimeRestoreCleanupFailure::Allocation)?; } - Ok::<_, RuntimeRestoreFailure>(()) + Ok::<_, RuntimeRestoreCleanupFailure>(()) } .await; - result?; - cleanup?; + match (result, cleanup) { + (Err(mut primary), cleanup) => { + primary.cleanup = match cleanup { + Ok(()) => RuntimeRestoreCleanup::Confirmed, + Err(failure) => { + primary.cleanup_failure = Some(failure); + RuntimeRestoreCleanup::Failed(failure) + } + }; + return Err(primary); + } + (Ok(()), Err(cleanup)) => { + let mut failure = cause("restore recovery cleanup failed"); + failure.cleanup_failure = Some(cleanup); + failure.cleanup = RuntimeRestoreCleanup::Failed(cleanup); + return Err(failure); + } + (Ok(()), Ok(())) => {} + } self.failed_restore_resources .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) 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 1a532fc26..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,8 +1,12 @@ 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, + }; use protocol::startup_failure::{RuntimeStartupCause, StartupPersistenceOutcome}; use session_store::worker_metadata::StartupFailureRead; use worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; @@ -67,13 +71,27 @@ mod startup_binding_tests { _ => self.store.read_by_name(&self.name).unwrap().unwrap(), }; if matches!(self.response, Response::Lost) { - return Err(WorkspaceClientError::Unavailable( + return Err(WorkspaceClientError::Request( "response deliberately lost".into(), )); } Ok(WorkspaceResponse { status: 403, - body: "synthetic rejection with private token /internal/path".into(), + body: serde_json::to_string( + &server_api::RepositoryApiError::new( + 403, + "PermissionDenied", + "synthetic rejection with private token /internal/path", + vec![], + ) + .with_failure( + session_registration_attachment(SessionRegistrationCause::denied( + SessionPrincipal::Subject, + )) + .unwrap(), + ), + ) + .unwrap(), }) } } @@ -137,9 +155,41 @@ 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); - let primary = result.startup_failure::().unwrap(); - assert!(matches!(primary, error_contract::DecodeFailure::Known(_))); + if matches!(response, Response::BeforeSession | Response::BeforeOwner) { + let primary = result.startup_failure::().unwrap(); + assert!(matches!(primary, error_contract::DecodeFailure::Known(_))); + } else { + let error_contract::DecodeFailure::Known(primary) = result + .startup_failure::() + .unwrap() + else { + panic!("a Session decision must not be relabeled as a generic Runtime failure"); + }; + assert_eq!( + *primary.facts().cause(), + if matches!(response, Response::Lost) { + SessionRegistrationCause::outcome_unknown() + } else { + SessionRegistrationCause::denied(SessionPrincipal::Subject) + } + ); + } drop(worker); drop(store); let reopened = WorkerAggregateStore::new(root.path(), name).unwrap(); @@ -151,13 +201,17 @@ mod startup_binding_tests { Some(StartupPersistenceOutcome::Persisted) ); let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = - metadata.decode_startup_failure::() + metadata.decode_startup_failure::() else { panic!("same-Session failure must actually save a typed cause") }; assert_eq!( - cause.facts().cause().reason, - protocol::startup_failure::RuntimeStartupReason::AttributionFailed + *cause.facts().cause(), + if matches!(response, Response::Lost) { + SessionRegistrationCause::outcome_unknown() + } else { + SessionRegistrationCause::denied(SessionPrincipal::Subject) + } ); assert_eq!(metadata.active.unwrap().session_id, origin); let saved = serde_json::to_string(&metadata.startup_failure).unwrap(); @@ -222,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/worker/src/startup_failure.rs b/crates/worker/src/startup_failure.rs index b0c4c59fa..9e76a56cf 100644 --- a/crates/worker/src/startup_failure.rs +++ b/crates/worker/src/startup_failure.rs @@ -120,6 +120,20 @@ pub fn reproject_internal_startup_failure( StartupFailureAttachment::from_projected(&projected) } +/// Fixed, identifier-free summary for an already authenticated internal caller. +/// Keep the required operation actionable without exposing the source's Display. +/// This neither projects a Browser cause nor chooses an HTTP status. +pub fn internal_startup_failure_summary(failure: &StartupFailureAttachment) -> &'static str { + if failure + .project_for::(&()) + .is_ok() + { + "Worker startup failed while registering Session (session.record)" + } else { + "Worker startup failed" + } +} + /// Dispatch by the domain attested in process, not by parsing attachment fields. pub(crate) fn persist_binding_startup_failure( store: &S, @@ -249,6 +263,36 @@ mod tests { WorkerStoreError, }; + #[test] + fn internal_summary_retains_only_attested_required_operation() { + use protocol::session_failure::{SessionPrincipal, SessionRegistrationCause}; + let session = protocol::session_failure::session_registration_attachment( + SessionRegistrationCause::denied(SessionPrincipal::Subject), + ) + .unwrap(); + assert_eq!( + internal_startup_failure_summary(&session), + "Worker startup failed while registering Session (session.record)" + ); + let runtime = runtime_startup_attachment(RuntimeStartupCause { + stage: StartupStage::Feature, + reason: RuntimeStartupReason::FeatureFailed, + }) + .unwrap(); + assert_eq!( + internal_startup_failure_summary(&runtime), + "Worker startup failed" + ); + let corrupt = serde_json::from_value(serde_json::json!( + "session.record denied TEST_ONLY_CREDENTIAL /private/path raw_http_body" + )) + .unwrap(); + assert_eq!( + internal_startup_failure_summary(&corrupt), + "Worker startup failed" + ); + } + struct LostSaveResponse(FsWorkerStore); impl WorkerMetadataStore for LostSaveResponse { fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> { diff --git a/crates/workspace-server/Cargo.toml b/crates/workspace-server/Cargo.toml index 388ef0447..4f14a0de4 100644 --- a/crates/workspace-server/Cargo.toml +++ b/crates/workspace-server/Cargo.toml @@ -70,6 +70,7 @@ webauthn-rs = { workspace = true, features = ["conditional-ui"] } webauthn-rs-proto = "0.5.5" [dev-dependencies] +error-contract.workspace = true rusqlite = { workspace = true, features = ["hooks"] } feature-storage = { workspace = true, features = ["test-support"] } native-tls = "0.2" diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index 3674e36f9..b341c617b 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -6950,8 +6950,11 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti ), }; if let EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } = error { - if result.startup_failure.is_some() { - projection.message = "Embedded Runtime Worker startup failed".into(); + if let Some(primary) = &result.startup_failure { + projection.message = format!( + "Embedded Runtime: {}", + worker::startup_failure::internal_startup_failure_summary(primary) + ); } projection.startup = result.startup_failure.as_ref().and_then(|primary| { internal_startup_transfer(runtime_api::WorkerStartupFailure { @@ -7042,7 +7045,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 { @@ -7121,6 +7127,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()) @@ -7155,6 +7164,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( @@ -7394,7 +7426,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(); @@ -7412,8 +7444,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( @@ -7432,6 +7470,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}; diff --git a/crates/workspace-server/src/server/subjektiv_journal.rs b/crates/workspace-server/src/server/subjektiv_journal.rs index da0db6cd1..9158014ec 100644 --- a/crates/workspace-server/src/server/subjektiv_journal.rs +++ b/crates/workspace-server/src/server/subjektiv_journal.rs @@ -423,10 +423,14 @@ pub(super) fn bind_host_session_for_registration( .config_store .subject_release_pending(api.workspace_id(), &worker)? { - return Err(Error::RepositoryConflict( - "Subject release intent fences Controller publication".into(), - ) - .into()); + // The current body is explicitly fenced from publication, not an + // unobserved callback outcome. Keep the conflict separate from its cause. + return Err(session_registration::Judgment::SessionRecord.error( + Error::RepositoryConflict( + "Subject release intent fences Controller publication".into(), + ), + protocol::session_failure::SessionRegistrationCause::binding_mismatch(), + )); } let store = store(api)?; if store 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 index 7dbb1a929..8faf4f226 100644 --- a/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/session_registration_connected_tests.rs @@ -3,6 +3,7 @@ //! Worker metadata. Only the unused external model is replaced. No restore, //! operation replay, singleton-release, or Browser rendering contract is asserted. use super::*; +use api_macros::HttpError as _; use protocol::session_failure::{ SessionPrincipal, SessionRegistrationCause, session_registration_attachment, }; @@ -29,15 +30,29 @@ impl agen::llm_client::LlmClient for UnusedModel { } } +type ReleaseFence = ( + crate::store::WorkerSingletonLease, + subjektiv::journal::JournalWriterBinding, +); + struct RegistrationFixture { connected: ConnectedWorkerHarness, subject: String, callbacks: Arc>>, model: UnusedModel, + release_fence: Arc>>, +} + +#[derive(Clone, Copy)] +enum RegistrationFault { + Allow, + Deny, + DenyWithMetadataReadFailure, + ReleasePending, } impl RegistrationFixture { - async fn new(revoke_on_registration: bool) -> Self { + async fn new(fault: RegistrationFault) -> 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. @@ -46,11 +61,14 @@ impl RegistrationFixture { let callbacks = Arc::new(Mutex::new(Vec::new())); let observed = callbacks.clone(); let model = UnusedModel::default(); + let release_fence = Arc::new(Mutex::new(None)); + let reserved = release_fence.clone(); 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(); + let reserved = reserved.clone(); async move { if request.method() != axum::http::Method::POST || request.uri().path() @@ -63,6 +81,9 @@ impl RegistrationFixture { )); let armed = hook.lock().unwrap().take(); let pin_root = armed.as_ref().map(|(_, _, root)| root.clone()); + let metadata_fault_root = pin_root.clone().filter(|_| { + matches!(fault, RegistrationFault::DenyWithMetadataReadFailure) + }); if let Some((api, subject, _)) = armed { let store = open_subjektiv_store(&api).unwrap(); let policy = store.subject_permissions(&subject).unwrap().unwrap(); @@ -72,22 +93,39 @@ impl RegistrationFixture { .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(); + if matches!(fault, RegistrationFault::ReleasePending) { + let key = crate::subjektiv::subject_worker_singleton_key(&subject).unwrap(); + let lease = api.store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key) + .unwrap().unwrap(); + let writer = store.claim_journal_writer( + &subject, &lease.worker.runtime_id, &lease.worker.worker_id, + ).unwrap(); + api.config_store.reserve_subject_release( + TEST_WORKSPACE_ID, &lease, &writer.binding_id, + "registration-release-fence", + ).unwrap(); + assert!(store.journal_writer_active(&writer).unwrap()); + assert_eq!(api.store.current_worker_singleton_owner(TEST_WORKSPACE_ID, &key).unwrap(), Some(lease.clone())); + *reserved.lock().unwrap() = Some((lease, writer)); + } else { + 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(); @@ -111,6 +149,19 @@ impl RegistrationFixture { let response = next .run(Request::from_parts(parts, Body::from(bytes))) .await; + if let Some(root) = metadata_fault_root { + assert_eq!(response.status(), StatusCode::FORBIDDEN); + // The real denial has already been decided. Replace only this + // fixture's metadata file with a directory before Worker save: + // a real read error, not a fake authorization/store response. + let workers = std::fs::read_dir(root.join("workers")) + .unwrap().map(|entry| entry.unwrap().path()).collect::>(); + assert_eq!(workers.len(), 1); + let metadata = workers[0].join("metadata.json"); + assert!(metadata.is_file()); + std::fs::remove_file(&metadata).unwrap(); + std::fs::create_dir(&metadata).unwrap(); + } observed.lock().unwrap().push((response.status(), input)); response } @@ -136,7 +187,7 @@ impl RegistrationFixture { .iter() .any(|p| p.operation == "session.record") ); - if revoke_on_registration { + if !matches!(fault, RegistrationFault::Allow) { *revoke.lock().unwrap() = Some(( connected.api.clone(), subject.clone(), @@ -148,6 +199,7 @@ impl RegistrationFixture { subject, callbacks, model, + release_fence, } } @@ -296,7 +348,7 @@ fn assert_registration_cause(body: Value, cause: SessionRegistrationCause) { #[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; + let fixture = RegistrationFixture::new(RegistrationFault::Deny).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( @@ -391,9 +443,222 @@ async fn real_session_record_denial_survives_runtime_host_and_metadata_reopen() assert!(contains_saved_value(&reopened, &expected)); } +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn real_session_record_denial_and_metadata_save_failure_remain_separate() { + use protocol::startup_failure::{ + StartupFailurePhase, StartupPersistenceOutcome, StartupSecondaryFailure, + }; + let fixture = RegistrationFixture::new(RegistrationFault::DenyWithMetadataReadFailure).await; + 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 failures = fixture.connected.runtime_startup_failures.lock().unwrap(); + assert_eq!( + failures.len(), + 1, + "one real startup attempt, not an operation replay" + ); + let wire = &failures[0]; + assert_eq!( + wire.primary, + session_registration_attachment(SessionRegistrationCause::denied( + SessionPrincipal::Subject + ),) + .unwrap() + ); + assert_eq!(wire.persistence, Some(StartupPersistenceOutcome::NotSaved)); + assert!(wire.secondary.iter().any(|failure| matches!( + failure, + StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome: StartupPersistenceOutcome::NotSaved + } + ))); + let host = error + .diagnostics + .iter() + .filter_map(|d| d.startup.as_ref()) + .find(|startup| startup.primary == wire.primary) + .unwrap(); + assert_eq!( + host, wire, + "Remote Host preserves the refusal and independent save failure" + ); + assert!( + open_subjektiv_store(&fixture.connected.api) + .unwrap() + .subject_session_attributions(&fixture.subject) + .unwrap() + .is_empty() + ); + let workers = std::fs::read_dir(fixture.connected.runtime_root.join("workers")) + .unwrap() + .map(|entry| entry.unwrap().path()) + .collect::>(); + assert_eq!(workers.len(), 1); + let name = format!( + "worker-runtime-{}", + workers[0].file_name().unwrap().to_str().unwrap() + ); + let reopened = session_store::WorkerAggregateStore::open_read_only(&workers[0], &name).unwrap(); + assert!( + reopened.read_by_name(&name).is_err(), + "the real read failure remains, not a false saved cause" + ); +} + +#[tokio::test] +async fn pending_release_binding_refusal_retains_original_conflict_source() { + let f = Fixture::new().await; + let store = open_subjektiv_store(&f.api).unwrap(); + let lease = f + .api + .store + .require_current_worker_singleton_owner(TEST_WORKSPACE_ID, &f.source) + .unwrap() + .unwrap(); + let writer = store + .claim_journal_writer(&f.subject, &f.source.runtime_id, &f.source.worker_id) + .unwrap(); + f.api + .config_store + .reserve_subject_release( + TEST_WORKSPACE_ID, + &lease, + &writer.binding_id, + "registration-source-release", + ) + .unwrap(); + let session = session_store::new_session_id().to_string(); + let attr = crate::subjektiv::SubjectSessionAttribution::new( + &f.subject, + &f.source.runtime_id, + &f.source.worker_id, + &session, + ) + .unwrap(); + let error = session_registration::bind_host_session(&f.api, attr).unwrap_err(); + assert_eq!( + error.session_registration_cause, + Some(SessionRegistrationCause::binding_mismatch()) + ); + assert!(matches!( + std::error::Error::source(&error) + .unwrap() + .downcast_ref::(), + Some(Error::RepositoryConflict(_)) + )); + assert_eq!( + error.into_repository_api_error().status_code(), + StatusCode::CONFLICT.as_u16() + ); + assert!(store.session_attribution(&session).unwrap().is_none()); + // A conflict without this explicit decision must not acquire a denial. + let unrelated = session_registration::Judgment::SessionRecord + .binding_rejection(Error::RepositoryConflict("an unrelated conflict".into()).into()); + assert!(unrelated.session_registration_cause.is_none()); +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 6)] +async fn release_fence_refusal_survives_signed_registration_and_metadata_reopen() { + let fixture = RegistrationFixture::new(RegistrationFault::ReleasePending).await; + let api = &fixture.connected.api; + let store = open_subjektiv_store(api).unwrap(); + let policy = store + .subject_permissions(&fixture.subject) + .unwrap() + .unwrap(); + let error = scoped_create_workspace_worker( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.into(), + }), + HeaderMap::new(), + Json(fixture.request()), + ) + .await + .unwrap_err(); + let session = fixture.assert_callback(StatusCode::CONFLICT); + let (lease, writer) = fixture.release_fence.lock().unwrap().clone().unwrap(); + assert_eq!( + store.subject_permissions(&fixture.subject).unwrap(), + Some(policy) + ); + assert!(store.session_attribution(&session).unwrap().is_none()); + assert!(store.journal_writer_active(&writer).unwrap()); + api.config_store + .require_subject_release_intent( + TEST_WORKSPACE_ID, + &lease, + &writer.binding_id, + "registration-release-fence", + ) + .unwrap(); + assert!( + api.config_store + .subject_release_pending(TEST_WORKSPACE_ID, &lease.worker) + .unwrap() + ); + assert!( + !api.config_store + .subject_release_proven( + TEST_WORKSPACE_ID, + &lease, + &writer.binding_id, + "registration-release-fence" + ) + .unwrap() + ); + + let failures = fixture.connected.runtime_startup_failures.lock().unwrap(); + assert_eq!(failures.len(), 1, "one real factory startup, not replay"); + let wire = &failures[0]; + let expected = + session_registration_attachment(SessionRegistrationCause::binding_mismatch()).unwrap(); + assert_eq!(wire.primary, expected); + assert_eq!( + wire.persistence, + Some(protocol::startup_failure::StartupPersistenceOutcome::Persisted) + ); + let host = error + .diagnostics + .iter() + .filter_map(|d| d.startup.as_ref()) + .find(|failure| failure.primary == expected) + .expect("Remote Host retains known refusal"); + 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, + .. + } + )), + "retention cleanup failure is separate from refusal" + ); + let saved = read_metadata(&fixture.connected.runtime_root, &lease.worker.worker_id); + assert!(contains_saved_value( + &saved, + &serde_json::to_value(&expected).unwrap() + )); + assert_eq!( + read_metadata(&fixture.connected.runtime_root, &lease.worker.worker_id), + saved + ); +} + #[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 fixture = RegistrationFixture::new(RegistrationFault::Allow).await; let initial = open_subjektiv_store(&fixture.connected.api) .unwrap() .subject_permissions(&fixture.subject) diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs index 48ce064b8..e43c2037b 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs @@ -2366,9 +2366,21 @@ async fn connected_explicit_repair(scenario: ConnectedRepair) { assert_eq!(probe.activations.load(SeqCst), 0); assert_eq!(probe.publications.load(SeqCst), 0); assert_eq!(fixture.model.calls.load(SeqCst), model_before); - assert!( - matches!(metadata.read_by_name(&name).unwrap().unwrap().subjektiv_session_attribution, - Some(session_store::SubjektivSessionAttributionState::Failed { session_id: id, .. }) if id == session_id) + let saved_failure = metadata.read_by_name(&name).unwrap().unwrap(); + // Trust rejection occurs before the Session domain handler. An unsigned + // legacy 401 attests neither a Session refusal nor absence of earlier effects. + assert!(matches!(saved_failure.subjektiv_session_attribution, + Some(session_store::SubjektivSessionAttributionState::OutcomeUnknown { session_id: id, .. }) if id == session_id)); + let session_store::worker_metadata::StartupFailureRead::Present( + error_contract::DecodeFailure::Known(cause), + ) = saved_failure + .decode_startup_failure::() + else { + panic!("recovery must retain its typed Session unknown cause") + }; + assert_eq!( + *cause.facts().cause(), + protocol::session_failure::SessionRegistrationCause::outcome_unknown() ); let denied = record(); assert_eq!(