diff --git a/crates/protocol/src/lib.rs b/crates/protocol/src/lib.rs index 7f7f9a096..b8c386bf5 100644 --- a/crates/protocol/src/lib.rs +++ b/crates/protocol/src/lib.rs @@ -2,6 +2,7 @@ pub mod identity; pub mod invocation; +pub mod startup_failure; #[cfg(feature = "stream")] pub mod stream; pub mod subscription; diff --git a/crates/protocol/src/startup_failure.rs b/crates/protocol/src/startup_failure.rs new file mode 100644 index 000000000..8a40e48ee --- /dev/null +++ b/crates/protocol/src/startup_failure.rs @@ -0,0 +1,397 @@ +//! Startup diagnostics are observations, never lifecycle, ownership, or retry authority. +//! +//! Attachments contain a domain's explicitly projected error-contract envelope. Receivers +//! decode through the existing codec, and must project again for a different audience; +//! deserializing stored bytes never creates a disclosure capability. Unknown and corrupt +//! attachments remain readable alongside metadata instead of breaking its lifecycle fields. +use error_contract::{ + Audience, Budget, ContractError, DecodeFailure, ProjectedFailure, WireDetails, decode_failure, + encode_projected_failure, +}; +use serde::{Deserialize, Serialize}; + +/// A bounded, projected envelope at the Worker/Runtime or metadata boundary. +/// No constructor accepts Display text, arbitrary JSON fields, or an unprojected error. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(transparent)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct StartupFailureAttachment(String); + +impl std::fmt::Debug for StartupFailureAttachment { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("StartupFailureAttachment") + .finish_non_exhaustive() + } +} + +impl StartupFailureAttachment { + pub fn from_projected( + failure: &ProjectedFailure, + ) -> Result { + let encoded = encode_projected_failure(failure, Self::budget())?; + String::from_utf8(encoded.as_bytes().to_vec()) + .map(Self) + .map_err(|_| ContractError::Encoding) + } + + /// Missing attachments are represented by the enclosing Option, not a fabricated cause. + pub fn decode(&self) -> DecodeFailure { + decode_failure(self.0.as_bytes(), Self::budget()) + } + + /// Redisclosure requires a fresh domain/audience projection and current authorization. + /// Unsupported or corrupt stored bytes are never forwarded as public details. + pub fn project_for>( + &self, + authorization: &D::Authorization, + ) -> Result, ContractError> { + match self.decode::() { + DecodeFailure::Known(received) => { + error_contract::project_failure(received.facts(), authorization) + } + DecodeFailure::Unsupported(_) | DecodeFailure::Corrupt(_) => { + Err(ContractError::ProjectionUnavailable) + } + } + } + + fn budget() -> Budget { + Budget::new( + error_contract::MAX_FAILURE_BYTES, + error_contract::MAX_FAILURE_BYTES, + ) + } +} + +/// Storage encoding cannot be constructed by relabeling an internal-transfer attachment. +/// A loaded value remains untrusted evidence; publication still requires `project_for`. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(transparent)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct StoredStartupFailureAttachment(StartupFailureAttachment); + +impl StoredStartupFailureAttachment { + pub fn from_projected( + failure: &ProjectedFailure, + ) -> Result { + StartupFailureAttachment::from_projected(failure).map(Self) + } + pub fn decode(&self) -> DecodeFailure { + self.0.decode() + } + pub fn project_for>( + &self, + authorization: &D::Authorization, + ) -> Result, ContractError> { + self.0.project_for::(authorization) + } +} + +/// Owner-selected startup stage; independent of resource cleanup disposition. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupStage { + Preflight, + Metadata, + Reconstruction, + Recovery, + Binding, + Feature, + ActivationEffects, + Controller, +} + +/// Safe local Runtime cause when no more specific domain attachment was supplied. +/// The primary cause is selected where the failure occurs, not parsed from its diagnostic. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct RuntimeStartupCause { + pub stage: StartupStage, + pub reason: RuntimeStartupReason, +} + +/// Coarse local causes do not claim a domain authorization verdict. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum RuntimeStartupReason { + StartupFailed, + SegmentEmpty, + SystemPromptMissing, + MetadataPending, + MetadataMissing, + MetadataInactive, + ManifestMissing, + AllocationFailed, + SessionReadFailed, + MetadataReadFailed, + ProviderFailed, + AttributionFailed, + FeatureFailed, +} + +impl error_contract::FailureCause for RuntimeStartupCause { + fn identity(&self) -> error_contract::FailureIdentity { + error_contract::FailureIdentity { + domain: "worker_startup", + reason: "failed", + category: error_contract::Category::Internal, + details_version: 1, + } + } + fn validate(&self) -> Result<(), ContractError> { + Ok(()) + } +} +impl WireDetails for RuntimeStartupCause { + fn supports(domain: &str, reason: &str, version: u16) -> bool { + domain == "worker_startup" && reason == "failed" && version == 1 + } +} + +// This cause contains only an enum. Arbitrary provider messages, IO paths and credentials +// cannot enter its projection. Domain-specific causes must supply their own projector. +macro_rules! project_runtime { + ($($audience:ty),*) => {$( + impl error_contract::ProjectFor<$audience> for RuntimeStartupCause { + type Authorization = (); + type Details = Self; + fn disclose(&self, facts: &error_contract::FailureFacts, _: &()) + -> Result, ContractError> + { + let mut disclosure = error_contract::Disclosure::new(self.clone(), "Worker startup failed."); + // Only fixed owner-selected operation/stage labels are published here. + for context in &facts.context().operations { + if context.operation == "worker_startup" && context.stage == "runtime" { + disclosure.context.operations.push(context.clone()); + if let Some(last) = disclosure.context.operations.last_mut() { + last.operation_ref = None; + } + } + } + Ok(disclosure) + } + } + )*}; +} +project_runtime!( + error_contract::InternalTransfer, + error_contract::Storage, + error_contract::Browser +); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupFailurePhase { + Cleanup, + RuntimeRecordDelete, + RuntimeSnapshotSave, + RuntimeRemovalPublish, + /// Retaining create reconciliation observations in the canonical Worker record. + RuntimeReconciliationRecordSave, + /// Publishing a retained record to Runtime subscribers; not a durable write. + RuntimeReconciliationPublish, + MetadataSave, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupCleanupFailure { + Ownership, + Workdir, + Artifacts, + Allocation, + Feature, + Unsupported, +} + +/// A separate secondary fact, not a replacement primary cause or a causal-chain link. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupSecondaryFailure { + ProjectionUnavailable { + phase: StartupFailurePhase, + }, + Cleanup { + failure: StartupCleanupFailure, + }, + Persistence { + phase: StartupFailurePhase, + outcome: StartupPersistenceOutcome, + }, +} + +/// Persisted is evidence only for this attachment/record write, not execution completion. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupPersistenceOutcome { + Persisted, + NotSaved, + Conflict, + OutcomeUnknown, +} + +/// Independent resource observations from the existing execution owner. These diagnostics +/// never promote a result to RolledBack or authorize replay; the backend disposition does. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupResources { + None, + WorkerStartup, + Unproven, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupCleanup { + NotNeeded, + Required, + Confirmed, + Failed(StartupCleanupFailure), +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub struct StartupResourceFacts { + pub resources: StartupResources, + pub cleanup: StartupCleanup, + pub durable_effects_possible: bool, +} + +/// Path-free IO classification. Nonportable OS details remain at the resource owner. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))] +#[cfg_attr(feature = "typescript", derive(ts_rs::TS))] +pub enum StartupIoKind { + NotFound, + PermissionDenied, + AlreadyExists, + InvalidInput, + InvalidData, + Interrupted, + TimedOut, + WouldBlock, + UnexpectedEof, + Other, +} +impl From for StartupIoKind { + fn from(kind: std::io::ErrorKind) -> Self { + use std::io::ErrorKind; + match kind { + ErrorKind::NotFound => Self::NotFound, + ErrorKind::PermissionDenied => Self::PermissionDenied, + ErrorKind::AlreadyExists => Self::AlreadyExists, + ErrorKind::InvalidInput => Self::InvalidInput, + ErrorKind::InvalidData => Self::InvalidData, + ErrorKind::Interrupted => Self::Interrupted, + ErrorKind::TimedOut => Self::TimedOut, + ErrorKind::WouldBlock => Self::WouldBlock, + ErrorKind::UnexpectedEof => Self::UnexpectedEof, + _ => Self::Other, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use error_contract::{ + Browser, FailureFacts, InternalTransfer, OperationContext, Storage, project_failure, + }; + + #[test] + fn startup_projection_preserves_allowlisted_facts_without_secret_context() { + let mut facts = FailureFacts::new(RuntimeStartupCause { + stage: StartupStage::Binding, + reason: RuntimeStartupReason::AttributionFailed, + }) + .unwrap(); + let mut context = OperationContext::new("worker_startup", "runtime").unwrap(); + context.operation_ref = Some("TEST_ONLY_CREDENTIAL".into()); + facts.with_context(context).unwrap(); + facts + .with_context(OperationContext::new("/private/internal/path", "raw_http_body").unwrap()) + .unwrap(); + let transfer = project_failure::(&facts, &()).unwrap(); + let attachment = StartupFailureAttachment::from_projected(&transfer).unwrap(); + let storage_projection = attachment + .project_for::(&()) + .unwrap(); + let stored = StoredStartupFailureAttachment::from_projected(&storage_projection).unwrap(); + let browser = stored + .project_for::(&()) + .unwrap(); + assert_eq!(browser.details(), facts.cause()); + assert_eq!(browser.context().operations.len(), 1); + assert_eq!(browser.context().operations[0].operation_ref, None); + let public = + encode_projected_failure(&browser, StartupFailureAttachment::budget()).unwrap(); + let wire = std::str::from_utf8(public.as_bytes()).unwrap(); + for secret in [ + "TEST_ONLY_CREDENTIAL", + "/private/internal/path", + "raw_http_body", + ] { + assert!(!wire.contains(secret)); + assert!(!serde_json::to_string(&stored).unwrap().contains(secret)); + } + assert!(matches!( + stored.decode::(), + DecodeFailure::Known(_) + )); + } + + #[test] + fn unsupported_and_corrupt_startup_attachments_cannot_be_redisclosed() { + let facts = FailureFacts::new(RuntimeStartupCause { + stage: StartupStage::Controller, + reason: RuntimeStartupReason::StartupFailed, + }) + .unwrap(); + let projected = project_failure::(&facts, &()).unwrap(); + let attachment = StartupFailureAttachment::from_projected(&projected).unwrap(); + let future = + StartupFailureAttachment(attachment.0.replace("\"version\":1", "\"version\":999")); + assert!(matches!( + future.decode::(), + DecodeFailure::Unsupported(_) + )); + assert!( + future + .project_for::(&()) + .is_err() + ); + let corrupt = StartupFailureAttachment("{invalid TEST_ONLY_SECRET".into()); + assert!(matches!( + corrupt.decode::(), + DecodeFailure::Corrupt(_) + )); + assert!( + corrupt + .project_for::(&()) + .is_err() + ); + assert!(!format!("{corrupt:?}").contains("TEST_ONLY_SECRET")); + } +} diff --git a/crates/protocol/src/typescript.rs b/crates/protocol/src/typescript.rs index 1fc420916..45e645d6a 100644 --- a/crates/protocol/src/typescript.rs +++ b/crates/protocol/src/typescript.rs @@ -64,6 +64,19 @@ pub fn generated_protocol_types() -> String { // Large integer fields are JSON numbers on the wire and are emitted as TypeScript `number`.\n\n", ); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); + push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); push_decl::(&cfg, &mut output); diff --git a/crates/runtime-api/src/lib.rs b/crates/runtime-api/src/lib.rs index a205e7893..533b657ae 100644 --- a/crates/runtime-api/src/lib.rs +++ b/crates/runtime-api/src/lib.rs @@ -10,6 +10,13 @@ use protocol::{CompletionEntry, CompletionKind, Segment, WorkerId, WorkerStateSn use serde::{Deserialize, Serialize}; use workdir::workspace::{MaterializerKind, RuntimeWorkerRef, RuntimeWorkingDirectorySummary}; +pub use protocol::startup_failure::{ + RuntimeStartupCause, RuntimeStartupReason, StartupCleanup, StartupCleanupFailure, + StartupFailureAttachment, StartupFailurePhase, StartupIoKind, StartupPersistenceOutcome, + StartupResourceFacts, StartupResources, StartupSecondaryFailure, StartupStage, + StoredStartupFailureAttachment, +}; + pub const RUNTIME_API_VERSION: &str = "v1"; pub const RUNTIME_API_BASE_PATH: &str = "/v1"; /// Runtime transport classification for a materialized Repository source. diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index 0f057d242..d494c17f7 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -114,9 +114,11 @@ pub use uploaded_file::{ UploadedFileUploadContext, }; pub use worker_metadata::{ - CombinedStore, FsWorkerStore, SubjektivSessionAttributionState, WorkerActiveSegmentRef, - WorkerAggregateStore, WorkerMetadata, WorkerMetadataStore, WorkerPeer, WorkerReclaimedChild, - WorkerSpawnedChild, WorkerSpawnedScopeRule, WorkerStoreError, validate_worker_name, + CombinedStore, FsWorkerStore, SessionStartupFailure, StartupFailureBindingMismatch, + StartupFailureExpectation, StartupFailureRead, StartupFailureSaveResult, + SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerAggregateStore, WorkerMetadata, + WorkerMetadataStore, WorkerPeer, WorkerReclaimedChild, WorkerSpawnedChild, + WorkerSpawnedScopeRule, WorkerStoreError, validate_worker_name, }; pub use worker_session_store::WorkerSessionStore; diff --git a/crates/session-store/src/startup_failure_tests.rs b/crates/session-store/src/startup_failure_tests.rs new file mode 100644 index 000000000..d65c92b02 --- /dev/null +++ b/crates/session-store/src/startup_failure_tests.rs @@ -0,0 +1,727 @@ +//! Component coverage of startup attachments through real FsWorkerStore files, +//! reopen and process-local CAS. Only the directory-sync IO boundary is replaced +//! for the post-rename failure case; these tests do not claim OS-crash durability. +//! Selected by the normal session-store lib test target. + +use super::*; +use error_contract::{FailureFacts, OperationContext, Storage, project_failure}; +use protocol::startup_failure::{RuntimeStartupCause, RuntimeStartupReason, StartupStage}; + +fn attachment(stage: StartupStage) -> StoredStartupFailureAttachment { + let mut facts = FailureFacts::new(RuntimeStartupCause { + stage, + reason: RuntimeStartupReason::StartupFailed, + }) + .unwrap(); + facts + .with_context(OperationContext { + operation: "worker_startup".into(), + stage: "runtime".into(), + operation_ref: None, + }) + .unwrap(); + let projected = project_failure::(&facts, &()).unwrap(); + StoredStartupFailureAttachment::from_projected(&projected).unwrap() +} + +fn metadata() -> WorkerMetadata { + let session_id = crate::new_session_id(); + let mut metadata = WorkerMetadata::new( + "agent", + Some(WorkerActiveSegmentRef::pending_segment(session_id)), + ) + .with_workspace_id("workspace-one"); + metadata.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed { + session_id, + diagnostic: "legacy diagnostic is not a typed cause".into(), + }); + metadata.resolved_manifest_snapshot = Some(serde_json::json!({"retained": true})); + metadata.peers.push(WorkerPeer { + worker_name: "peer".into(), + }); + metadata +} + +fn bound_failure( + metadata: &WorkerMetadata, + attachment: &StoredStartupFailureAttachment, +) -> SessionStartupFailure { + SessionStartupFailure::from_expected( + &StartupFailureExpectation::from_metadata(metadata).unwrap(), + attachment, + ) +} + +fn assert_stored_cause(metadata: &WorkerMetadata, stage: StartupStage) { + match metadata.decode_startup_failure::() { + StartupFailureRead::Present(DecodeFailure::Known(received)) => { + assert_eq!(received.facts().cause().stage, stage); + assert_eq!( + received.facts().cause().reason, + RuntimeStartupReason::StartupFailed + ); + assert_eq!(received.facts().context().operations.len(), 1); + assert_eq!( + received.facts().context().operations[0].operation, + "worker_startup" + ); + } + _ => panic!("expected a known typed startup cause"), + } +} + +#[test] +fn startup_failure_cas_reopens_typed_cause_and_preserves_unrelated_metadata() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let cause = attachment(StartupStage::Binding); + let received = cause.clone(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Persisted + )); + assert_eq!(cause, received); + drop(store); + + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + assert_stored_cause(&restored, StartupStage::Binding); + let mut wanted = original; + wanted.startup_failure = Some(SessionStartupFailure::from_expected(&expected, &received)); + assert_eq!(restored, wanted); +} + +#[test] +fn startup_failure_wrapper_fingerprints_attribution_without_copying_raw_diagnostics() { + const CREDENTIAL: &str = "credential=TEST_ONLY_METADATA_SENTINEL"; + const PRIVATE_PATH: &str = "/private/test-only-attribution/session.json"; + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let mut original = metadata(); + original.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed { + session_id: original.active.as_ref().unwrap().session_id, + diagnostic: format!("{CREDENTIAL} {PRIVATE_PATH}"), + }); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Persisted + )); + drop(store); + + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + assert_eq!( + restored.subjektiv_session_attribution, + original.subjektiv_session_attribution + ); + // The original field still retains the diagnostic: absence from the wrapper + // must not be explained by removing or redacting the source metadata instead. + let retained_attribution = + serde_json::to_string(&restored.subjektiv_session_attribution).unwrap(); + assert!(retained_attribution.contains(CREDENTIAL)); + assert!(retained_attribution.contains(PRIVATE_PATH)); + let wrapper = restored.startup_failure.as_ref().unwrap(); + let wrapper_json = serde_json::to_string(wrapper).unwrap(); + assert!(!wrapper_json.contains(CREDENTIAL)); + assert!(!wrapper_json.contains(PRIVATE_PATH)); + let wrapper_value = serde_json::to_value(wrapper).unwrap(); + assert!(wrapper_value.get("attribution").is_none()); + let expected_fingerprint: [u8; 32] = Sha256::digest(retained_attribution.as_bytes()).into(); + assert_eq!(bound(wrapper).attribution_fingerprint, expected_fingerprint); + assert_stored_cause(&restored, StartupStage::Binding); +} + +#[test] +fn startup_failure_cas_rejects_changed_session_owner_attribution_and_cause_without_rewrite() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let cause = attachment(StartupStage::Binding); + let path = temp.path().join("agent/metadata.json"); + + // Each row changes one compared fact. Worker-name mismatch is an on-disk + // identity mismatch, not a write to a different key. + let changed = [ + { + let mut value = original.clone(); + value.active = Some(WorkerActiveSegmentRef::pending_segment( + crate::new_session_id(), + )); + value + }, + { + let mut value = original.clone(); + value.workspace_id = Some("new-owner".into()); + value + }, + { + let mut value = original.clone(); + value.worker_name = "new-worker".into(); + value + }, + { + let mut value = original.clone(); + value.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Confirmed { + session_id: expected.session_id, + subject_id: "new-subject".into(), + }); + value + }, + { + let mut value = original.clone(); + value.subjektiv_session_attribution = None; + value + }, + { + let mut value = original.clone(); + value.subjektiv_session_attribution = Some(SubjektivSessionAttributionState::Failed { + session_id: expected.session_id, + diagnostic: "new attribution observation".into(), + }); + value + }, + { + let mut value = original.clone(); + value.startup_failure = Some(bound_failure(&value, &attachment(StartupStage::Feature))); + value + }, + { + let mut value = original.clone(); + value.active = None; + value + }, + ]; + for value in changed { + fs::create_dir_all(path.parent().unwrap()).unwrap(); + // Retain noncanonical whitespace to detect even a no-op rewrite. + let bytes = format!(" {}\n", serde_json::to_string(&value).unwrap()); + fs::write(&path, &bytes).unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Conflict + )); + assert_eq!(fs::read(&path).unwrap(), bytes.as_bytes()); + assert_eq!(store.read_by_name("agent").unwrap().unwrap(), value); + } +} + +#[test] +fn startup_failure_replacement_requires_the_observed_existing_attachment() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let mut original = metadata(); + original.startup_failure = Some(bound_failure(&original, &attachment(StartupStage::Feature))); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let mut stale = expected.clone(); + stale.startup_failure = None; + let cause = attachment(StartupStage::Binding); + assert!(matches!( + store.compare_and_swap_startup_failure(&stale, &cause), + StartupFailureSaveResult::Conflict + )); + assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Persisted + )); + assert_stored_cause( + &store.read_by_name("agent").unwrap().unwrap(), + StartupStage::Binding, + ); +} + +#[test] +fn startup_failure_cas_missing_metadata_does_not_create_a_worker() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&metadata()).unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Conflict + )); + assert!(store.read_by_name("agent").unwrap().is_none()); + assert!(!temp.path().join("agent").exists()); +} + +#[test] +fn startup_failure_cas_allows_segment_movement_within_expected_session() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let active = + WorkerActiveSegmentRef::active_segment(expected.session_id, crate::new_segment_id()); + store + .set_active("agent", Some(active.clone()), None) + .unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Persisted + )); + let restored = store.read_by_name("agent").unwrap().unwrap(); + assert_eq!(restored.active, Some(active)); + assert_eq!( + restored.subjektiv_session_attribution, + original.subjektiv_session_attribution + ); + // A later Segment switch is not a Session/ownership change either. + store + .set_active( + "agent", + Some(WorkerActiveSegmentRef::active_segment( + expected.session_id, + crate::new_segment_id(), + )), + None, + ) + .unwrap(); + assert_stored_cause( + &store.read_by_name("agent").unwrap().unwrap(), + StartupStage::Binding, + ); +} + +#[test] +fn startup_failure_saved_binding_rejects_session_owner_and_attribution_changes_after_reopen() { + for case in 0..7 { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let cause = attachment(StartupStage::Binding); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Persisted + )); + let saved = store + .read_by_name("agent") + .unwrap() + .unwrap() + .startup_failure; + store + .update_by_name("agent", |value| match case { + 0 => { + value.active = Some(WorkerActiveSegmentRef::pending_segment( + crate::new_session_id(), + )) + } + 1 => value.workspace_id = Some("new-workspace-owner".into()), + 2 => { + value.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Confirmed { + session_id: expected.session_id, + subject_id: "new-subject-owner".into(), + }) + } + 3 => { + value.subjektiv_session_attribution = + Some(SubjektivSessionAttributionState::Failed { + session_id: expected.session_id, + diagnostic: "changed attribution observation".into(), + }) + } + 4 => value.active = None, + 5 => value.workspace_id = None, + 6 => value.subjektiv_session_attribution = None, + _ => unreachable!(), + }) + .unwrap(); + drop(store); + + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + // State changes retain the old cause as evidence; they do not rebind it + // to the new Session/owner and do not turn a mismatch into absence. + assert_eq!(restored.startup_failure, saved); + let path = temp.path().join("agent/metadata.json"); + let bytes_before_read = fs::read(&path).unwrap(); + match restored.decode_startup_failure::() { + StartupFailureRead::Mismatch(mismatch) => assert_eq!( + mismatch, + StartupFailureBindingMismatch { + worker_name: false, + session_id: matches!(case, 0 | 4), + workspace_id: matches!(case, 1 | 5), + attribution: matches!(case, 2 | 3 | 6), + } + ), + _ => panic!("saved cause must not decode for a changed binding (case {case})"), + } + assert_eq!(fs::read(&path).unwrap(), bytes_before_read); + assert!(matches!( + reopened.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Conflict + )); + } +} + +#[test] +fn startup_failure_copy_to_another_worker_keeps_original_binding() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Persisted + )); + let mut copied = store.read_by_name("agent").unwrap().unwrap(); + copied.worker_name = "replacement".into(); + store.write(&copied).unwrap(); + drop(store); + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("replacement").unwrap().unwrap(); + assert!(matches!( + restored.decode_startup_failure::(), + StartupFailureRead::Mismatch(StartupFailureBindingMismatch { + worker_name: true, + session_id: false, + workspace_id: false, + attribution: false, + }) + )); + assert_stored_cause( + &reopened.read_by_name("agent").unwrap().unwrap(), + StartupStage::Binding, + ); +} + +#[test] +fn startup_failure_cas_compares_existing_binding_even_when_attachment_is_unchanged() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let mut original = metadata(); + original.startup_failure = Some(bound_failure(&original, &attachment(StartupStage::Feature))); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let mut stale = expected.clone(); + bound_mut(stale.startup_failure.as_mut().unwrap()).session_id = crate::new_session_id(); + assert!(matches!( + store.compare_and_swap_startup_failure(&stale, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Conflict + )); + assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original); + let mut changed_fingerprint = expected.clone(); + bound_mut(changed_fingerprint.startup_failure.as_mut().unwrap()).attribution_fingerprint[0] ^= + 1; + assert!(matches!( + store.compare_and_swap_startup_failure( + &changed_fingerprint, + &attachment(StartupStage::Binding) + ), + StartupFailureSaveResult::Conflict + )); + assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original); +} + +#[test] +fn startup_failure_bound_to_absent_optional_owner_fields_reopens_as_present() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let mut original = metadata(); + original.workspace_id = None; + original.subjektiv_session_attribution = None; + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Binding)), + StartupFailureSaveResult::Persisted + )); + drop(store); + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + assert_stored_cause(&restored, StartupStage::Binding); + let expected_none_fingerprint: [u8; 32] = Sha256::digest(b"null").into(); + assert_eq!( + bound(restored.startup_failure.as_ref().unwrap()).attribution_fingerprint, + expected_none_fingerprint + ); +} + +#[test] +fn startup_failure_cas_independently_opened_stores_have_one_concurrent_winner() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let barrier = Arc::new(std::sync::Barrier::new(3)); + let handles = [StartupStage::Binding, StartupStage::Feature].map(|stage| { + let store = FsWorkerStore::new(temp.path()).unwrap(); + let expected = expected.clone(); + let barrier = barrier.clone(); + std::thread::spawn(move || { + barrier.wait(); + ( + stage, + store.compare_and_swap_startup_failure(&expected, &attachment(stage)), + ) + }) + }); + barrier.wait(); + let mut winner = None; + let mut conflicts = 0; + for handle in handles { + let (stage, result) = handle.join().unwrap(); + match result { + StartupFailureSaveResult::Persisted => { + assert!(winner.replace(stage).is_none()); + } + StartupFailureSaveResult::Conflict => conflicts += 1, + other => panic!("unexpected persistence result: {other:?}"), + } + } + assert_eq!(conflicts, 1); + drop(store); + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + assert_stored_cause( + &reopened.read_by_name("agent").unwrap().unwrap(), + winner.unwrap(), + ); +} + +#[test] +fn startup_failure_legacy_missing_unknown_and_corrupt_survive_reopen_with_attribution() { + let temp = tempfile::tempdir().unwrap(); + let original = metadata(); + let mut future: serde_json::Value = serde_json::from_str( + serde_json::to_value(attachment(StartupStage::Binding)) + .unwrap() + .as_str() + .unwrap(), + ) + .unwrap(); + future["version"] = serde_json::json!(65535); + let fixtures = [ + None, + Some(future.to_string()), + Some("not an envelope".to_owned()), + ]; + + for (case, encoded) in fixtures.into_iter().enumerate() { + let mut fixture = serde_json::to_value(&original).unwrap(); + assert!(fixture.get("startup_failure").is_none()); + if let Some(encoded) = encoded { + // The new field is always a binding wrapper. Unknown/corrupt fixtures + // change only its opaque codec payload, not the surrounding binding. + let mut bound = + serde_json::to_value(bound_failure(&original, &attachment(StartupStage::Binding))) + .unwrap(); + bound["attachment"] = serde_json::Value::String(encoded); + fixture["startup_failure"] = bound; + } + let path = temp.path().join("agent/metadata.json"); + fs::create_dir_all(path.parent().unwrap()).unwrap(); + fs::write(path, serde_json::to_vec(&fixture).unwrap()).unwrap(); + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + match ( + case, + restored.decode_startup_failure::(), + ) { + (0, StartupFailureRead::Missing) + | (1, StartupFailureRead::Present(DecodeFailure::Unsupported(_))) + | (2, StartupFailureRead::Present(DecodeFailure::Corrupt(_))) => {} + _ => panic!("legacy absence, unknown and corrupt must be distinguishable"), + } + assert_eq!( + restored.subjektiv_session_attribution, + original.subjektiv_session_attribution + ); + assert_eq!(restored.active, original.active); + assert_eq!(restored.workspace_id, original.workspace_id); + if case != 0 { + reopened + .update_by_name("agent", |value| { + value.workspace_id = Some("new-owner".into()) + }) + .unwrap(); + let changed = reopened.read_by_name("agent").unwrap().unwrap(); + assert_eq!(changed.startup_failure, restored.startup_failure); + assert!(matches!( + changed.decode_startup_failure::(), + StartupFailureRead::Mismatch(StartupFailureBindingMismatch { + workspace_id: true, + .. + }) + )); + } + } +} + +#[test] +fn startup_failure_save_read_error_is_not_saved_and_preserves_received_cause() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let path = temp.path().join("agent/metadata.json"); + fs::write(&path, b"invalid metadata").unwrap(); + let cause = attachment(StartupStage::Binding); + let received = cause.clone(); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::NotSaved(WorkerStoreError::Serde(_)) + )); + assert_eq!(cause, received); + assert_eq!(fs::read(path).unwrap(), b"invalid metadata"); +} + +#[test] +fn startup_failure_rename_error_is_not_saved_but_post_rename_sync_error_is_unknown() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let cause = attachment(StartupStage::Binding); + let received = cause.clone(); + + // Real IO rename failure before publication. The destination is an isolated + // directory rather than metadata.json, so no retained metadata is destroyed. + let blocked_path = temp.path().join("blocked/metadata.json"); + fs::create_dir_all(&blocked_path).unwrap(); + let result = + save_startup_failure_locked(&store, &store.update_lock, &expected, &cause, |value| { + write_metadata_atomically(value, &blocked_path) + }); + assert!(matches!( + result, + StartupFailureSaveResult::NotSaved(WorkerStoreError::Io(_)) + )); + assert_eq!(store.read_by_name("agent").unwrap().unwrap(), original); + assert_eq!( + fs::read_dir(blocked_path.parent().unwrap()) + .unwrap() + .count(), + 1 + ); + + // Production serialization + file sync + rename, replacing only directory + // sync with an IO failure. Reopen sees publication, but durability is unknown. + let path = store.metadata_path("agent").unwrap(); + let result = + save_startup_failure_locked(&store, &store.update_lock, &expected, &cause, |value| { + write_metadata_with_directory_sync(value, &path, |_| { + Err(std::io::Error::other("directory sync unavailable")) + }) + }); + assert!(matches!( + result, + StartupFailureSaveResult::OutcomeUnknown(WorkerStoreError::Io(_)) + )); + assert_eq!(cause, received); + drop(store); + let reopened = FsWorkerStore::new(temp.path()).unwrap(); + let restored = reopened.read_by_name("agent").unwrap().unwrap(); + assert_stored_cause(&restored, StartupStage::Binding); + let mut wanted = original; + wanted.startup_failure = Some(SessionStartupFailure::from_expected(&expected, &received)); + assert_eq!(restored, wanted); + // An uncertain attempt is not replayed against its stale expected cause. + assert!(matches!( + reopened.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Conflict + )); +} + +#[test] +fn startup_failure_aggregate_and_combined_stores_use_atomic_cas() { + let temp = tempfile::tempdir().unwrap(); + let aggregate = WorkerAggregateStore::new(temp.path().join("aggregate"), "agent").unwrap(); + let original = metadata(); + aggregate.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let store = CombinedStore::new((), aggregate); + let cause = attachment(StartupStage::Binding); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &cause), + StartupFailureSaveResult::Persisted + )); + assert!(matches!( + store.compare_and_swap_startup_failure(&expected, &attachment(StartupStage::Feature)), + StartupFailureSaveResult::Conflict + )); + let reopened = + WorkerAggregateStore::open_read_only(temp.path().join("aggregate"), "agent").unwrap(); + assert_stored_cause( + &reopened.read_read_only().unwrap().unwrap(), + StartupStage::Binding, + ); +} + +fn bound(failure: &SessionStartupFailure) -> &StartupFailureBinding { + match failure { + SessionStartupFailure::Bound(binding) => binding, + _ => panic!("expected current binding"), + } +} +fn bound_mut(failure: &mut SessionStartupFailure) -> &mut StartupFailureBinding { + match failure { + SessionStartupFailure::Bound(binding) => binding, + _ => panic!("expected current binding"), + } +} + +#[test] +fn startup_binding_unknown_version_and_malformed_value_do_not_destroy_lifecycle_metadata() { + let temp = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(temp.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let mut wire = serde_json::to_value(&original).unwrap(); + for (value, unsupported) in [ + ( + serde_json::json!({"version": 900, "future_binding": true}), + true, + ), + ( + serde_json::json!({"version": 1, "attachment": {"wrong_type": true}}), + false, + ), + (serde_json::json!("invalid binding"), false), + (serde_json::json!([1, 2]), false), + ] { + wire["startup_failure"] = value.clone(); + std::fs::write( + temp.path().join("agent/metadata.json"), + serde_json::to_vec(&wire).unwrap(), + ) + .unwrap(); + let reopened = FsWorkerStore::new(temp.path()) + .unwrap() + .read_by_name("agent") + .unwrap() + .unwrap(); + assert_eq!(reopened.active, original.active); + assert_eq!( + reopened.subjektiv_session_attribution, + original.subjektiv_session_attribution + ); + assert_eq!( + reopened.resolved_manifest_snapshot, + original.resolved_manifest_snapshot + ); + match reopened.decode_startup_failure::() { + StartupFailureRead::UnsupportedBindingVersion(900) if unsupported => {} + StartupFailureRead::CorruptBinding if !unsupported => {} + _ => panic!("binding diagnosis lost"), + } + assert_eq!( + serde_json::to_value(&reopened).unwrap()["startup_failure"], + value + ); + } +} diff --git a/crates/session-store/src/worker_metadata.rs b/crates/session-store/src/worker_metadata.rs index 997419ba7..bfa5b00d4 100644 --- a/crates/session-store/src/worker_metadata.rs +++ b/crates/session-store/src/worker_metadata.rs @@ -16,7 +16,10 @@ //! execution data, not proof of liveness. use crate::{SegmentId, SessionId}; +use error_contract::{DecodeFailure, WireDetails}; +use protocol::startup_failure::StoredStartupFailureAttachment; use serde::{Deserialize, Serialize}; +use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::fs; use std::path::{Path, PathBuf}; @@ -181,6 +184,168 @@ pub struct WorkerMetadata { /// absence is also the legacy representation and never grants visibility. #[serde(default, skip_serializing_if = "Option::is_none")] pub subjektiv_session_attribution: Option, + /// Projected startup cause bound to its Session and ownership observation. + /// Pointer/ownership changes retain this evidence, but the read accessor + /// rejects a stale binding before decoding. This is not execution state. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_failure: Option, +} + +/// Opaque cause and the exact local binding under which it was saved. +/// +/// Only the CAS save path constructs this binding from its matched expectation. +/// Attribution is fingerprinted rather than copied: its existing diagnostics may +/// contain credentials or private paths and must not enter this new attachment. +/// The fingerprint is an equality fence, not encryption or ownership authority. +/// Deserialization merely restores evidence, not current ownership or authority; +/// consumers read it through `WorkerMetadata::decode_startup_failure`. +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(untagged)] +pub enum SessionStartupFailure { + Bound(StartupFailureBinding), + /// Malformed or future binding representations remain alongside lifecycle metadata. + Undecodable(serde_json::Value), +} + +impl std::fmt::Debug for SessionStartupFailure { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SessionStartupFailure") + .finish_non_exhaustive() + } +} + +#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StartupFailureBinding { + version: u16, + worker_name: String, + session_id: SessionId, + workspace_id: Option, + attribution_fingerprint: [u8; 32], + attachment: StoredStartupFailureAttachment, +} + +// The compact JSON serialization of this typed Option is the fingerprint input: +// None is `null`; Some retains every enum field in its declared Serialize order. +// There are no maps whose iteration order could vary. Changing that representation +// requires a new binding version. Version 1 uses SHA-256 over precisely these bytes. +// This enum and SessionId serialize infallibly +// to JSON; no caller-controlled Serialize implementation participates here. +fn attribution_fingerprint(attribution: &Option) -> [u8; 32] { + let encoded = serde_json::to_vec(attribution).expect("typed attribution serializes to JSON"); + Sha256::digest(encoded).into() +} + +impl SessionStartupFailure { + fn from_expected( + expected: &StartupFailureExpectation, + attachment: &StoredStartupFailureAttachment, + ) -> Self { + Self::Bound(StartupFailureBinding { + version: 1, + worker_name: expected.worker_name.clone(), + session_id: expected.session_id, + workspace_id: expected.workspace_id.clone(), + attribution_fingerprint: attribution_fingerprint(&expected.attribution), + attachment: attachment.clone(), + }) + } +} + +impl StartupFailureBinding { + fn binding_mismatch(&self, metadata: &WorkerMetadata) -> Option { + let mismatch = StartupFailureBindingMismatch { + worker_name: self.worker_name != metadata.worker_name, + session_id: Some(self.session_id) + != metadata.active.as_ref().map(|active| active.session_id), + workspace_id: self.workspace_id != metadata.workspace_id, + attribution: self.attribution_fingerprint + != attribution_fingerprint(&metadata.subjektiv_session_attribution), + }; + (mismatch.worker_name + || mismatch.session_id + || mismatch.workspace_id + || mismatch.attribution) + .then_some(mismatch) + } +} + +/// Each `true` field identifies a binding that differs from current metadata. +/// Values and legacy diagnostic text are not copied into the diagnosis. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct StartupFailureBindingMismatch { + pub worker_name: bool, + /// Also true when there is no longer an active Session. + pub session_id: bool, + pub workspace_id: bool, + pub attribution: bool, +} + +/// A current binding must match before the consumer's typed codec runs. +/// `Present` is only cause evidence, never execution or retry authorization. +pub enum StartupFailureRead { + Missing, + /// Binding encoding is versioned independently of the domain error envelope. + UnsupportedBindingVersion(u16), + CorruptBinding, + /// Retained cause belongs to a different Session/ownership observation. + /// No cause decoding is attempted, even for unknown or corrupt attachments. + Mismatch(StartupFailureBindingMismatch), + /// Known, unsupported and corrupt causes remain distinct in the codec. + Present(DecodeFailure), +} + +/// The observed ownership and cause values for one startup-cause write. +/// +/// Segment movement within the same Session does not invalidate this expectation. +/// Attribution is compared in full, including legacy diagnostics, never inferred +/// from those diagnostics. The existing cause comparison includes its entire saved +/// binding, not just attachment bytes. A cause write changes neither attribution +/// nor ownership; it builds the new binding from this matched expectation. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct StartupFailureExpectation { + pub worker_name: String, + pub workspace_id: Option, + pub session_id: SessionId, + pub attribution: Option, + pub startup_failure: Option, +} + +impl StartupFailureExpectation { + /// Capture an expectation only when metadata identifies an active Session. + pub fn from_metadata(metadata: &WorkerMetadata) -> Option { + Some(Self { + worker_name: metadata.worker_name.clone(), + workspace_id: metadata.workspace_id.clone(), + session_id: metadata.active.as_ref()?.session_id, + attribution: metadata.subjektiv_session_attribution.clone(), + startup_failure: metadata.startup_failure.clone(), + }) + } + + /// Compare the complete observation without adopting newer ownership or causes. + pub fn matches(&self, metadata: &WorkerMetadata) -> bool { + metadata.worker_name == self.worker_name + && metadata.workspace_id == self.workspace_id + && metadata.active.as_ref().map(|active| active.session_id) == Some(self.session_id) + && metadata.subjektiv_session_attribution == self.attribution + && metadata.startup_failure == self.startup_failure + } +} + +/// Local persistence evidence, never a replacement for the received startup cause. +#[derive(Debug)] +pub enum StartupFailureSaveResult { + /// The attachment and containing directory were synced successfully. + Persisted, + /// The observed Session, ownership, attribution, or cause no longer matches. + /// No write was attempted. Missing metadata is also a conflict, not creation. + Conflict, + /// This attempt did not publish metadata. The caller still owns its cause. + NotSaved(WorkerStoreError), + /// Publication may have happened; reconcile retained metadata before retrying. + /// For the filesystem stores this means rename succeeded but directory sync failed. + OutcomeUnknown(WorkerStoreError), } impl WorkerMetadata { @@ -196,9 +361,44 @@ impl WorkerMetadata { peers: Vec::new(), resolved_manifest_snapshot: None, subjektiv_session_attribution: None, + startup_failure: None, } } + /// Absence is legacy/missing evidence, not a cause inferred from diagnostics. + /// A retained cause whose Session or ownership observation no longer matches + /// yields `Mismatch`, never `Missing` or a decoded cause for the new owner. + /// Segment movement within the same Session does not invalidate the binding. + /// Matched attachments use only the consumer's typed codec, preserving known, + /// unsupported and corrupt results. No result authorizes execution or retry. + pub fn decode_startup_failure(&self) -> StartupFailureRead { + let Some(failure) = &self.startup_failure else { + return StartupFailureRead::Missing; + }; + #[derive(Deserialize)] + struct BindingVersion { + version: u16, + } + let failure = match failure { + SessionStartupFailure::Bound(binding) if binding.version == 1 => binding, + SessionStartupFailure::Bound(binding) => { + return StartupFailureRead::UnsupportedBindingVersion(binding.version); + } + SessionStartupFailure::Undecodable(raw) => { + return match serde_json::from_value::(raw.clone()) { + Ok(header) if header.version != 1 => { + StartupFailureRead::UnsupportedBindingVersion(header.version) + } + _ => StartupFailureRead::CorruptBinding, + }; + } + }; + if let Some(mismatch) = failure.binding_mismatch(self) { + return StartupFailureRead::Mismatch(mismatch); + } + StartupFailureRead::Present(failure.attachment.decode()) + } + pub fn with_workspace_root(mut self, workspace_root: PathBuf) -> Self { self.workspace_root = Some(workspace_root); self @@ -265,6 +465,25 @@ pub trait WorkerMetadataStore: Send + Sync { Ok(matched) } + /// Persist a received cause only against the exact observed Session and owner. + /// + /// Implementations must serialize this comparison and publication with their + /// other read/modify/write operations. The filesystem stores use the same + /// process-local mutex as `update_by_name`; this is not a cross-process lock. + /// Uncoordinated direct writes are outside that guarantee. A default backend + /// refuses rather than offering a non-atomic read/write approximation. + /// The borrowed attachment is never consumed or changed on any result. + fn compare_and_swap_startup_failure( + &self, + _expected: &StartupFailureExpectation, + _failure: &StoredStartupFailureAttachment, + ) -> StartupFailureSaveResult { + StartupFailureSaveResult::NotSaved(WorkerStoreError::Io(std::io::Error::new( + std::io::ErrorKind::Unsupported, + "atomic startup failure persistence is unsupported by this backend", + ))) + } + /// Set the active pointer while preserving spawned children, workspace ownership, and manifest snapshot. fn set_active( &self, @@ -404,6 +623,90 @@ pub trait WorkerMetadataStore: Send + Sync { } } +/// Whether an atomic metadata replacement reached its publication point. +struct MetadataWriteFailure { + published: bool, + error: WorkerStoreError, +} + +fn write_metadata_atomically( + metadata: &WorkerMetadata, + path: &Path, +) -> Result<(), MetadataWriteFailure> { + write_metadata_with_directory_sync(metadata, path, |parent| { + std::fs::File::open(parent)?.sync_all() + }) +} + +// Keep the directory durability boundary explicit: a failure here follows rename, +// unlike failures writing/syncing the temporary file. Tests replace only this IO +// boundary, retaining the real serialization, temporary file and rename path. +fn write_metadata_with_directory_sync( + metadata: &WorkerMetadata, + path: &Path, + sync_directory: impl FnOnce(&Path) -> std::io::Result<()>, +) -> Result<(), MetadataWriteFailure> { + let mut published = false; + let mut temp = None; + let result = (|| -> Result<(), WorkerStoreError> { + use std::io::Write; + let mut content = serde_json::to_vec_pretty(metadata)?; + content.push(b'\n'); + let parent = path.parent().expect("metadata path has parent"); + fs::create_dir_all(parent)?; + let temporary_path = parent.join(format!( + ".metadata.json.tmp-{}-{}", + std::process::id(), + uuid::Uuid::now_v7() + )); + let mut file = std::fs::OpenOptions::new() + .write(true) + .create_new(true) + .open(&temporary_path)?; + temp = Some(temporary_path.clone()); + file.write_all(&content)?; + file.sync_all()?; + drop(file); + fs::rename(&temporary_path, path)?; + published = true; + sync_directory(parent)?; + Ok(()) + })(); + if result.is_err() { + if let Some(temp) = temp { + let _ = fs::remove_file(temp); + } + } + result.map_err(|error| MetadataWriteFailure { published, error }) +} + +fn save_startup_failure_locked( + store: &S, + update_lock: &Mutex<()>, + expected: &StartupFailureExpectation, + failure: &StoredStartupFailureAttachment, + publish: impl FnOnce(&WorkerMetadata) -> Result<(), MetadataWriteFailure>, +) -> StartupFailureSaveResult { + let _guard = update_lock.lock().expect("metadata update lock poisoned"); + let mut metadata = match store.read_by_name(&expected.worker_name) { + Ok(Some(metadata)) if expected.matches(&metadata) => metadata, + Ok(_) => return StartupFailureSaveResult::Conflict, + Err(error) => return StartupFailureSaveResult::NotSaved(error), + }; + metadata.startup_failure = Some(SessionStartupFailure::from_expected(expected, failure)); + match publish(&metadata) { + Ok(()) => StartupFailureSaveResult::Persisted, + Err(MetadataWriteFailure { + published: false, + error, + }) => StartupFailureSaveResult::NotSaved(error), + Err(MetadataWriteFailure { + published: true, + error, + }) => StartupFailureSaveResult::OutcomeUnknown(error), + } +} + /// Metadata store for one canonical Worker aggregate. /// /// The Worker key is fixed by Runtime authority and metadata lives directly at @@ -489,31 +792,7 @@ impl WorkerAggregateStore { impl WorkerMetadataStore for WorkerAggregateStore { fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> { self.validate_name(&metadata.worker_name)?; - let mut content = serde_json::to_vec_pretty(metadata)?; - content.push(b'\n'); - let path = self.metadata_path(); - let temp = self.root.join(format!( - ".metadata.json.tmp-{}-{}", - std::process::id(), - uuid::Uuid::now_v7() - )); - let result = (|| -> Result<(), WorkerStoreError> { - use std::io::Write; - let mut file = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp)?; - file.write_all(&content)?; - file.sync_all()?; - drop(file); - fs::rename(&temp, &path)?; - std::fs::File::open(&self.root)?.sync_all()?; - Ok(()) - })(); - if result.is_err() { - let _ = fs::remove_file(temp); - } - result + write_metadata_atomically(metadata, &self.metadata_path()).map_err(|failure| failure.error) } fn read_by_name(&self, worker_name: &str) -> Result, WorkerStoreError> { @@ -553,6 +832,16 @@ impl WorkerMetadataStore for WorkerAggregateStore { Ok(metadata) } + fn compare_and_swap_startup_failure( + &self, + expected: &StartupFailureExpectation, + failure: &StoredStartupFailureAttachment, + ) -> StartupFailureSaveResult { + save_startup_failure_locked(self, &self.update_lock, expected, failure, |metadata| { + write_metadata_atomically(metadata, &self.metadata_path()) + }) + } + fn compare_and_swap_active( &self, worker_name: &str, @@ -627,32 +916,7 @@ impl FsWorkerStore { impl WorkerMetadataStore for FsWorkerStore { fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> { let path = self.metadata_path(&metadata.worker_name)?; - let mut content = serde_json::to_vec_pretty(metadata)?; - content.push(b'\n'); - let parent = path.parent().expect("metadata path has parent"); - fs::create_dir_all(parent)?; - let temp = parent.join(format!( - ".metadata.json.tmp-{}-{}", - std::process::id(), - uuid::Uuid::now_v7() - )); - let result = (|| -> Result<(), WorkerStoreError> { - use std::io::Write; - let mut file = std::fs::OpenOptions::new() - .write(true) - .create_new(true) - .open(&temp)?; - file.write_all(&content)?; - file.sync_all()?; - drop(file); - fs::rename(&temp, &path)?; - std::fs::File::open(parent)?.sync_all()?; - Ok(()) - })(); - if result.is_err() { - let _ = fs::remove_file(temp); - } - result + write_metadata_atomically(metadata, &path).map_err(|failure| failure.error) } fn read_by_name(&self, worker_name: &str) -> Result, WorkerStoreError> { @@ -685,6 +949,20 @@ impl WorkerMetadataStore for FsWorkerStore { Ok(metadata) } + fn compare_and_swap_startup_failure( + &self, + expected: &StartupFailureExpectation, + failure: &StoredStartupFailureAttachment, + ) -> StartupFailureSaveResult { + let path = match self.metadata_path(&expected.worker_name) { + Ok(path) => path, + Err(error) => return StartupFailureSaveResult::NotSaved(error), + }; + save_startup_failure_locked(self, &self.update_lock, expected, failure, |metadata| { + write_metadata_atomically(metadata, &path) + }) + } + fn compare_and_swap_active( &self, worker_name: &str, @@ -1007,6 +1285,15 @@ where self.worker_metadata_store .update_by_name(worker_name, update) } + fn compare_and_swap_startup_failure( + &self, + expected: &StartupFailureExpectation, + failure: &StoredStartupFailureAttachment, + ) -> StartupFailureSaveResult { + self.worker_metadata_store + .compare_and_swap_startup_failure(expected, failure) + } + fn compare_and_swap_active( &self, worker_name: &str, @@ -1512,3 +1799,7 @@ mod tests { assert_eq!(winners, 1); } } + +#[cfg(test)] +#[path = "startup_failure_tests.rs"] +mod startup_failure_tests; diff --git a/crates/worker-runtime/src/error.rs b/crates/worker-runtime/src/error.rs index 433c6ec54..5b6e4e5af 100644 --- a/crates/worker-runtime/src/error.rs +++ b/crates/worker-runtime/src/error.rs @@ -147,3 +147,14 @@ pub enum RuntimeError { #[error("runtime state lock was poisoned")] StatePoisoned, } + +impl RuntimeError { + /// The original execution result, including primary facts and independent secondary + /// failures. Display is presentation only and must not be parsed to recover causes. + pub fn execution_result(&self) -> Option<&WorkerExecutionResult> { + match self { + Self::WorkerExecutionRejected { result, .. } => Some(result), + _ => None, + } + } +} diff --git a/crates/worker-runtime/src/execution.rs b/crates/worker-runtime/src/execution.rs index 72c22e46e..bb19a307c 100644 --- a/crates/worker-runtime/src/execution.rs +++ b/crates/worker-runtime/src/execution.rs @@ -94,6 +94,17 @@ pub struct WorkerExecutionResult { pub submission: Option, #[serde(default, skip_serializing_if = "Option::is_none")] pub notification: Option, + /// Projected primary facts. Never stop, cleanup, or retry evidence. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_failure: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub secondary_failures: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_resources: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_failure_persistence: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub startup_io_kind: Option, } /// Backend result class for a Worker execution operation. @@ -116,6 +127,11 @@ impl WorkerExecutionResult { message: None, submission: None, notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -141,6 +157,11 @@ impl WorkerExecutionResult { disposition, }), notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -157,6 +178,11 @@ impl WorkerExecutionResult { notification: Some(WorkerNotificationAck { notification_request_id: notification_request_id.into(), }), + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -168,6 +194,11 @@ impl WorkerExecutionResult { message: Some(message.into()), submission: None, notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -179,6 +210,11 @@ impl WorkerExecutionResult { message: Some(message.into()), submission: None, notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -190,6 +226,11 @@ impl WorkerExecutionResult { message: Some(message.into()), submission: None, notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } @@ -201,9 +242,37 @@ impl WorkerExecutionResult { message: Some(message.into()), submission: None, notification: None, + startup_failure: None, + secondary_failures: Vec::new(), + startup_resources: None, + startup_failure_persistence: None, + startup_io_kind: None, } } + pub fn with_startup_failure( + mut self, + failure: protocol::startup_failure::StartupFailureAttachment, + ) -> Self { + self.startup_failure = Some(failure); + self + } + + pub fn startup_failure( + &self, + ) -> Option> { + self.startup_failure + .as_ref() + .map(|failure| failure.decode()) + } + + pub fn push_secondary_failure( + &mut self, + failure: protocol::startup_failure::StartupSecondaryFailure, + ) { + self.secondary_failures.push(failure); + } + pub fn is_accepted(&self) -> bool { self.outcome == WorkerExecutionOutcome::Accepted } diff --git a/crates/worker-runtime/src/runtime.rs b/crates/worker-runtime/src/runtime.rs index 32930beb4..3c57b93c0 100644 --- a/crates/worker-runtime/src/runtime.rs +++ b/crates/worker-runtime/src/runtime.rs @@ -47,6 +47,9 @@ use crate::retention::{ FsWorkerRetentionProvider, WorkerRetentionExecutionRequest, WorkerRetentionExecutionResult, WorkerRetentionInventory, WorkerRetentionInventorySnapshot, WorkerRetentionProvider, }; +use protocol::startup_failure::{ + StartupFailurePhase, StartupPersistenceOutcome, StartupSecondaryFailure, +}; use protocol::subscription::{ EventSubscriptionSelector, SubscriptionEventPayload, SubscriptionSnapshot, SubscriptionValidationError, SubscriptionWorkdirId, SubscriptionWorker, SubscriptionWorkerId, @@ -1522,10 +1525,41 @@ impl Runtime { worker_state, workdir_attachments, } => (worker_state, workdir_attachments), - WorkerExecutionSpawnResult::Rejected(result) - | WorkerExecutionSpawnResult::RolledBack(result) - | WorkerExecutionSpawnResult::Errored(result) => { - self.rollback_failed_create(&worker_ref)?; + WorkerExecutionSpawnResult::Rejected(mut result) + | WorkerExecutionSpawnResult::RolledBack(mut result) + | WorkerExecutionSpawnResult::Errored(mut result) => { + if let Err((error, phase, outcome)) = + self.rollback_failed_create_with_phase(&worker_ref) + { + tracing::error!( + worker_id = %worker_ref.worker_id, + ?phase, + %error, + "failed to roll back Worker creation" + ); + // A later failure must not make a confirmed record deletion + // indistinguishable from a failed deletion with unknown effects. + if matches!( + phase, + StartupFailurePhase::RuntimeSnapshotSave + | StartupFailurePhase::RuntimeRemovalPublish + ) { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeRecordDelete, + outcome: StartupPersistenceOutcome::Persisted, + }); + } + if phase == StartupFailurePhase::RuntimeRemovalPublish { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeSnapshotSave, + outcome: StartupPersistenceOutcome::Persisted, + }); + } + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase, + outcome, + }); + } return Err(RuntimeError::WorkerExecutionRejected { worker_id: worker_ref.worker_id.clone(), operation: result.operation, @@ -1535,7 +1569,7 @@ impl Runtime { }); } WorkerExecutionSpawnResult::ReconciliationRequired { - result, + mut result, worker_state, workdir_attachments, } => { @@ -1548,12 +1582,33 @@ impl Runtime { } else { self.retain_create_reconciliation_pending(&worker_ref, workdir_attachments) }; - if let Err(error) = retain_result { + if let Err((phase, error)) = retain_result { + let outcome = if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) + { + StartupPersistenceOutcome::OutcomeUnknown + } else { + StartupPersistenceOutcome::NotSaved + }; + if phase == StartupFailurePhase::RuntimeReconciliationPublish { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::Persisted, + }); + } + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase, + outcome, + }); tracing::error!( worker_id = %worker_ref.worker_id, %error, "failed to persist Worker create reconciliation evidence" ); + } else { + result.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeReconciliationRecordSave, + outcome: StartupPersistenceOutcome::Persisted, + }); } return Err(RuntimeError::WorkerExecutionRejected { worker_id: worker_ref.worker_id.clone(), @@ -3811,9 +3866,33 @@ impl Runtime { } fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> { - let mut state = self.lock()?; + self.rollback_failed_create_with_phase(worker_ref) + .map_err(|(error, _, _)| error) + } + + // The legacy caller still receives its RuntimeError. Spawn failures instead + // retain their primary result and attach the completed/uncertain rollback phases. + fn rollback_failed_create_with_phase( + &self, + worker_ref: &WorkerRef, + ) -> Result<(), (RuntimeError, StartupFailurePhase, StartupPersistenceOutcome)> { + let mut state = self.lock().map_err(|error| { + ( + error, + StartupFailurePhase::RuntimeRecordDelete, + StartupPersistenceOutcome::NotSaved, + ) + })?; if state.workers.contains_key(&worker_ref.worker_id) { - state.delete_worker_record(&worker_ref.worker_id)?; + state + .delete_worker_record(&worker_ref.worker_id) + .map_err(|error| { + ( + error, + StartupFailurePhase::RuntimeRecordDelete, + StartupPersistenceOutcome::OutcomeUnknown, + ) + })?; let record = state .workers .remove(&worker_ref.worker_id) @@ -3826,8 +3905,22 @@ impl Runtime { state .observation_events .retain(|event| event.worker_ref != *worker_ref); - state.persist_runtime_snapshot()?; - state.publish_worker_removed(worker_ref.worker_id, workspace_id.as_deref())?; + state.persist_runtime_snapshot().map_err(|error| { + ( + error, + StartupFailurePhase::RuntimeSnapshotSave, + StartupPersistenceOutcome::OutcomeUnknown, + ) + })?; + state + .publish_worker_removed(worker_ref.worker_id, workspace_id.as_deref()) + .map_err(|error| { + ( + error, + StartupFailurePhase::RuntimeRemovalPublish, + StartupPersistenceOutcome::OutcomeUnknown, + ) + })?; } Ok(()) } @@ -5121,9 +5214,10 @@ impl Runtime { worker_state: protocol::WorkerStateSnapshot, workdir_attachments: Vec, - ) -> Result<(), RuntimeError> { - let mut state = self.lock()?; - let mut candidate = state.worker(worker_ref)?.clone(); + ) -> Result<(), (StartupFailurePhase, RuntimeError)> { + let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error); + let mut state = self.lock().map_err(record_error)?; + let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone(); candidate.restore_guard.rotate(); candidate.execution_metadata_available = true; @@ -5132,9 +5226,13 @@ impl Runtime { candidate.apply_worker_state(&worker_state); candidate.restore_intent = WorkerRestoreIntent::Automatic; candidate.workdir_attachments = workdir_attachments; - state.persist_worker_record(&candidate)?; + state + .persist_worker_record(&candidate) + .map_err(record_error)?; state.workers.insert(worker_ref.worker_id, candidate); - state.publish_worker_upsert(worker_ref.worker_id)?; + state + .publish_worker_upsert(worker_ref.worker_id) + .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?; Ok(()) } @@ -5156,9 +5254,10 @@ impl Runtime { &self, worker_ref: &WorkerRef, workdir_attachments: Vec, - ) -> Result<(), RuntimeError> { - let mut state = self.lock()?; - let mut candidate = state.worker(worker_ref)?.clone(); + ) -> Result<(), (StartupFailurePhase, RuntimeError)> { + let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error); + let mut state = self.lock().map_err(record_error)?; + let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone(); candidate.restore_guard.rotate(); candidate.execution_metadata_available = false; @@ -5174,8 +5273,12 @@ impl Runtime { logical_workdir_attachments_from_statuses(&workdir_attachments, claims); candidate.workdir_attachments = workdir_attachments; state.workers.insert(worker_ref.worker_id, candidate); - state.persist_worker(&worker_ref.worker_id)?; - state.publish_worker_upsert(worker_ref.worker_id)?; + state + .persist_worker(&worker_ref.worker_id) + .map_err(record_error)?; + state + .publish_worker_upsert(worker_ref.worker_id) + .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?; Ok(()) } @@ -8257,6 +8360,225 @@ mod tests { use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Barrier, Condvar, Mutex}; + #[cfg(feature = "fs-store")] + mod create_rollback_tests { + //! Connect the real create/rollback path to FsRuntimeStore and reopen its disk state. + //! Only execution startup is a fixture. Filesystem topology faults are injected after + //! the initial record is durable; existing write hooks cover Worker writes, not these + //! delete/snapshot boundaries. Run with the normal worker-runtime library tests. + use super::*; + use error_contract::{ + DecodeFailure, FailureFacts, InternalTransfer, OperationContext, project_failure, + }; + use protocol::startup_failure::{ + RuntimeStartupCause, StartupCleanup, StartupCleanupFailure, StartupFailureAttachment, + StartupResourceFacts, StartupResources, StartupStage, + }; + + #[derive(Clone, Copy)] + enum Fault { + None, + RecordDelete, + SnapshotSave, + } + + fn primary_failure() -> WorkerExecutionResult { + let mut facts = FailureFacts::new(RuntimeStartupCause { + reason: protocol::startup_failure::RuntimeStartupReason::StartupFailed, + stage: StartupStage::Controller, + }) + .unwrap(); + facts + .with_context(OperationContext::new("worker_startup", "runtime").unwrap()) + .unwrap(); + let projected = project_failure::(&facts, &()).unwrap(); + let attachment = StartupFailureAttachment::from_projected(&projected).unwrap(); + let mut result = WorkerExecutionResult::errored( + WorkerExecutionOperation::Spawn, + "primary startup failure", + ) + .with_startup_failure(attachment) + .with_worker_state(protocol::WorkerStatus::Stopped.into()); + result.push_secondary_failure(StartupSecondaryFailure::Cleanup { + failure: StartupCleanupFailure::Workdir, + }); + result.startup_resources = Some(StartupResourceFacts { + resources: StartupResources::Unproven, + cleanup: StartupCleanup::Failed(StartupCleanupFailure::Workdir), + durable_effects_possible: true, + }); + result + } + + include!("runtime/create_reconciliation_tests.rs"); + + fn check_rollback(fault: Fault) { + let primary = primary_failure(); + for spawn_result in [ + WorkerExecutionSpawnResult::Rejected(primary.clone()), + WorkerExecutionSpawnResult::RolledBack(primary.clone()), + WorkerExecutionSpawnResult::Errored(primary.clone()), + ] { + let root = tempfile::tempdir().unwrap(); + let store_root = root.path().join("runtime"); + let options = + FsRuntimeStoreOptions::new(&store_root).with_runtime_id("create-rollback"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let request = task_request("failed create rollback"); + let worker_ref = WorkerRef::new(request.worker_id); + let worker_dir = store_root + .join("workers") + .join(request.worker_id.to_string()); + let runtime_path = store_root.join("runtime.json"); + let saved_snapshot = root.path().join("saved-runtime.json"); + let record_fault_path = worker_dir.join("metadata.json"); + *backend.spawn_result.lock().unwrap() = Some(spawn_result); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let worker_dir = worker_dir.clone(); + let runtime_path = runtime_path.clone(); + let saved_snapshot = saved_snapshot.clone(); + let record_fault_path = record_fault_path.clone(); + move || { + assert!(worker_dir.join("worker.json").is_file()); + match fault { + Fault::None => {} + Fault::RecordDelete => { + // Reading a directory as metadata fails before aggregate removal. + std::fs::create_dir(record_fault_path).unwrap(); + } + Fault::SnapshotSave => { + std::fs::rename(&runtime_path, saved_snapshot).unwrap(); + std::fs::create_dir(runtime_path).unwrap(); + } + } + } + })); + + let error = runtime.create_worker(request).unwrap_err(); + let RuntimeError::WorkerExecutionRejected { + worker_id, + operation, + outcome, + message, + result, + } = error + else { + panic!("rollback must not replace the primary execution result: {error:?}"); + }; + assert_eq!(worker_id, worker_ref.worker_id); + assert_eq!(operation, primary.operation); + assert_eq!(outcome, primary.outcome); + assert_eq!(message, primary.message_or_default()); + let mut expected = primary.clone(); + match fault { + Fault::None => {} + Fault::RecordDelete => { + expected.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeRecordDelete, + outcome: StartupPersistenceOutcome::OutcomeUnknown, + }); + } + Fault::SnapshotSave => { + expected.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeRecordDelete, + outcome: StartupPersistenceOutcome::Persisted, + }); + expected.push_secondary_failure(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::RuntimeSnapshotSave, + outcome: StartupPersistenceOutcome::OutcomeUnknown, + }); + } + } + assert_eq!( + result, expected, + "primary facts and existing cleanup must survive" + ); + let Some(DecodeFailure::Known(decoded)) = + result.startup_failure::() + else { + panic!("primary typed cause must remain decodable"); + }; + assert_eq!(decoded.facts().cause().stage, StartupStage::Controller); + assert_eq!( + decoded.facts().context().operations, + vec![OperationContext::new("worker_startup", "runtime").unwrap()] + ); + + let record_retained = matches!(fault, Fault::RecordDelete); + assert_eq!(worker_dir.join("worker.json").is_file(), record_retained); + assert_eq!(runtime.worker_detail(&worker_ref).is_ok(), record_retained); + match fault { + Fault::RecordDelete => { + let persisted = runtime_store(&runtime) + .read_worker_record(&worker_ref) + .unwrap(); + assert_eq!(persisted.worker_ref, worker_ref); + assert!( + runtime_store(&runtime) + .load_runtime_state() + .unwrap() + .workers + .contains_key(&worker_ref.worker_id) + ); + // Remove only the injected obstruction, never the retained record. + std::fs::remove_dir(record_fault_path).unwrap(); + } + Fault::SnapshotSave => { + assert!(runtime_path.is_dir()); + assert!(runtime_store(&runtime).load_runtime_state().is_err()); + } + Fault::None => { + assert!( + !runtime_store(&runtime) + .load_runtime_state() + .unwrap() + .workers + .contains_key(&worker_ref.worker_id) + ); + } + } + drop(runtime); + drop(backend); + if matches!(fault, Fault::SnapshotSave) { + assert!(Runtime::with_fs_store(options.clone()).is_err()); + // Repair the injected obstruction with the last durable snapshot, + // not a new snapshot synthesized from the in-memory rollback. + std::fs::remove_dir(runtime_path.clone()).unwrap(); + std::fs::rename(saved_snapshot, runtime_path).unwrap(); + } + let reopened = Runtime::with_fs_store(options).unwrap(); + assert_eq!(reopened.worker_detail(&worker_ref).is_ok(), record_retained); + assert_eq!( + runtime_store(&reopened) + .load_runtime_state() + .unwrap() + .workers + .contains_key(&worker_ref.worker_id), + record_retained + ); + } + } + + #[test] + fn create_rollback_success_preserves_primary_and_existing_cleanup() { + check_rollback(Fault::None); + } + + #[test] + fn create_rollback_record_delete_failure_preserves_primary_and_retained_record() { + check_rollback(Fault::RecordDelete); + } + + #[test] + fn create_rollback_snapshot_failure_preserves_primary_and_confirmed_record_deletion() { + check_rollback(Fault::SnapshotSave); + } + } + #[test] fn worker_create_failure_fields_only_include_sanitized_execution_detail() { let invalid_request = RuntimeError::InvalidRequest("private-token".to_string()); @@ -9239,6 +9561,8 @@ mod tests { recovery_absent: AtomicBool, recovery_checks: std::sync::atomic::AtomicUsize, spawn_result: Mutex>, + #[cfg(feature = "fs-store")] + before_spawn_result: Mutex>>, job_bindings: Mutex>>, dispatch_result: Mutex>, stop_result: Mutex>, @@ -9446,6 +9770,10 @@ mod tests { .lock() .unwrap() .push(request.request.backend_job.clone()); + #[cfg(feature = "fs-store")] + if let Some(before_result) = self.before_spawn_result.lock().unwrap().take() { + before_result(); + } if let Some(result) = self.spawn_result.lock().unwrap().take() { return result; } diff --git a/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs b/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs new file mode 100644 index 000000000..d58f6eecc --- /dev/null +++ b/crates/worker-runtime/src/runtime/create_reconciliation_tests.rs @@ -0,0 +1,113 @@ +mod create_reconciliation_tests { + //! The real create retention and FsRuntimeStore write/reopen path; only spawn + //! disposition and the existing deterministic atomic-write hook are substituted. + use super::*; + use crate::fs_store::AtomicWriteFault; + + fn check_reconciliation(fault: AtomicWriteFault) { + for has_state in [false, true] { + let root = tempfile::tempdir().unwrap(); + let options = FsRuntimeStoreOptions::new(root.path().join("runtime")) + .with_runtime_id("create-reconciliation"); + let backend = Arc::new(TestExecutionBackend::default()); + let runtime = + Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone()) + .unwrap(); + runtime.store_config_bundle(test_bundle()).unwrap(); + let request = task_request("reconciliation failure"); + let worker_ref = WorkerRef::new(request.worker_id); + let primary = primary_failure(); + *backend.spawn_result.lock().unwrap() = + Some(WorkerExecutionSpawnResult::ReconciliationRequired { + result: primary.clone(), + worker_state: has_state.then(|| protocol::WorkerStatus::Stopped.into()), + workdir_attachments: Vec::new(), + }); + let initial = Arc::new(Mutex::new(None)); + *backend.before_spawn_result.lock().unwrap() = Some(Box::new({ + let store = runtime_store(&runtime); + let initial = initial.clone(); + let worker_ref = worker_ref.clone(); + move || { + *initial.lock().unwrap() = Some(store.read_worker_record(&worker_ref).unwrap()); + store.fail_next_worker_write(fault); + } + })); + let RuntimeError::WorkerExecutionRejected { result, .. } = + runtime.create_worker(request).unwrap_err() + else { + panic!("retention failure must not replace primary result") + }; + assert_eq!(result.startup_failure, primary.startup_failure); + assert_eq!(result.startup_resources, primary.startup_resources); + assert_eq!(result.outcome, primary.outcome); + assert!( + result + .secondary_failures + .contains(&primary.secondary_failures[0]) + ); + let outcome = match fault { + AtomicWriteFault::BeforeRename => StartupPersistenceOutcome::NotSaved, + AtomicWriteFault::AfterRename => StartupPersistenceOutcome::OutcomeUnknown, + AtomicWriteFault::None => StartupPersistenceOutcome::Persisted, + }; + assert!(result.secondary_failures.iter().any(|failure| matches!(failure, StartupSecondaryFailure::Persistence { phase: StartupFailurePhase::RuntimeReconciliationRecordSave, outcome: observed } if *observed == outcome)), "retention persistence must be explicit: {result:?}"); + let disk = runtime_store(&runtime) + .read_worker_record(&worker_ref) + .unwrap(); + let initial = initial.lock().unwrap().take().unwrap(); + if fault == AtomicWriteFault::BeforeRename { + assert_eq!(disk.execution_state, initial.execution_state); + assert_eq!( + disk.restore_guard.observation_token, + initial.restore_guard.observation_token + ); + } else { + assert_ne!( + disk.restore_guard.observation_token, + initial.restore_guard.observation_token + ); + if has_state { + assert!( + matches!(disk.execution_state, crate::fs_store::PersistedWorkerExecutionState::Available(ref execution) if execution.binding.is_some()) + ); + } else { + assert_eq!( + disk.execution_state, + crate::fs_store::PersistedWorkerExecutionState::Unavailable + ); + } + } + // A pre-rename failure left the initial record; an uncertain response + // here followed a real rename and is reconciled from disk, not guessed. + let on_disk = std::fs::read( + root.path() + .join("runtime/workers") + .join(worker_ref.worker_id.to_string()) + .join("worker.json"), + ) + .unwrap(); + drop(runtime); + drop(backend); + let reopened = Runtime::with_fs_store(options).unwrap(); + let reread = runtime_store(&reopened) + .read_worker_record(&worker_ref) + .unwrap(); + assert_eq!(reread.execution_state, disk.execution_state); + assert!(reopened.worker_detail(&worker_ref).is_ok()); + assert!(!on_disk.is_empty()); + } + } + #[test] + fn create_reconciliation_write_failure_retains_primary_and_unsaved_evidence() { + check_reconciliation(AtomicWriteFault::BeforeRename); + } + #[test] + fn create_reconciliation_unknown_write_retains_primary_and_reopens_actual_record() { + check_reconciliation(AtomicWriteFault::AfterRename); + } + #[test] + fn create_reconciliation_success_retains_primary_and_durable_evidence() { + check_reconciliation(AtomicWriteFault::None); + } +} diff --git a/crates/worker-runtime/src/subject_delete_create_tests.rs b/crates/worker-runtime/src/subject_delete_create_tests.rs index 57642940a..f92687a47 100644 --- a/crates/worker-runtime/src/subject_delete_create_tests.rs +++ b/crates/worker-runtime/src/subject_delete_create_tests.rs @@ -15,7 +15,7 @@ impl RuntimeWorkerFactory for SubjectDeleteDeferredCreate { async fn spawn_controller( &self, _: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { self.entered.send(()).unwrap(); self.release.notified().await; Err("create failed before any Controller".into()) diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 87c066099..09b93f923 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -71,7 +71,7 @@ fn rollback_materialized_spawn_workdirs( materializer: &dyn WorkingDirectoryMaterializer, bindings: &BTreeMap, workdir_ids: &[String], - failure_context: &str, + failure_context: &WorkerExecutionResult, ) -> Result<(), WorkerExecutionSpawnResult> { let mut cleanup_failures = Vec::new(); let mut uncertain_attachments = Vec::new(); @@ -93,13 +93,15 @@ fn rollback_materialized_spawn_workdirs( return Ok(()); } Err(WorkerExecutionSpawnResult::ReconciliationRequired { - result: WorkerExecutionResult::errored( - WorkerExecutionOperation::Spawn, - format!( - "{failure_context}; Workdir cleanup could not be proven: {}", - cleanup_failures.join("; ") - ), - ), + result: { + let mut result = failure_context.clone(); + result.push_secondary_failure( + protocol::startup_failure::StartupSecondaryFailure::Cleanup { + failure: protocol::startup_failure::StartupCleanupFailure::Workdir, + }, + ); + result + }, worker_state: None, workdir_attachments: uncertain_attachments, }) @@ -165,6 +167,8 @@ fn subscribe_worker_protocol_session(handle: &WorkerHandle) -> WorkerProtocolSes worker::ipc::protocol_session::subscribe_worker_protocol_session(handle) } +mod startup_failure; + const DEFAULT_BACKEND_ID: &str = "worker-crate"; const RUNTIME_TASK_TIMEOUT: Duration = Duration::from_secs(10); const SPAWN_RESTORE_TASK_TIMEOUT: Duration = Duration::from_secs(60); @@ -262,9 +266,81 @@ pub struct RuntimeRestoreFailure { io_kind: Option, cleanup_failure: Option, durable_effects_possible: bool, + primary: protocol::startup_failure::StartupFailureAttachment, + secondary: Vec, + legacy_diagnostic: Option, + metadata_save: Option, } impl RuntimeRestoreFailure { + fn fallback_primary( + stage: RuntimeRestoreStage, + ) -> protocol::startup_failure::StartupFailureAttachment { + worker::startup_failure::runtime_startup_attachment( + protocol::startup_failure::RuntimeStartupCause { + stage: stage.into(), + reason: protocol::startup_failure::RuntimeStartupReason::StartupFailed, + }, + ) + .expect("fixed Runtime startup projection fits the error-contract budget") + } + + pub fn with_startup_failure( + mut self, + primary: protocol::startup_failure::StartupFailureAttachment, + ) -> Self { + self.primary = primary; + self + } + + pub fn startup_failure(&self) -> &protocol::startup_failure::StartupFailureAttachment { + &self.primary + } + + pub fn io_kind(&self) -> Option { + self.io_kind + } + + fn from_worker(stage: RuntimeRestoreStage, error: &WorkerError) -> Self { + let mut failure = Self::started(stage, restore_worker_cause(error)); + failure.primary = error + .startup_failure_at(stage.into()) + .expect("fixed Worker startup projection fits the error-contract budget"); + failure.secondary = error.startup_secondary_failures().to_vec(); + failure.metadata_save = error.startup_metadata_save(); + failure + } + + pub fn execution_result(&self, operation: WorkerExecutionOperation) -> WorkerExecutionResult { + use protocol::startup_failure::*; + let mut result = WorkerExecutionResult::errored(operation, self.to_string()) + .with_startup_failure(self.primary.clone()); + result.secondary_failures = self.secondary.clone(); + result.startup_failure_persistence = self.metadata_save; + result.startup_io_kind = self.io_kind.map(Into::into); + // Retain the first cleanup failure even if a later retry completed cleanup. + if let Some(failure) = self.cleanup_failure { + result.push_secondary_failure(StartupSecondaryFailure::Cleanup { + failure: failure.into(), + }); + } + result.startup_resources = Some(StartupResourceFacts { + resources: match self.resources { + RuntimeRestoreResources::None => StartupResources::None, + RuntimeRestoreResources::WorkerStartup => StartupResources::WorkerStartup, + RuntimeRestoreResources::Unproven => StartupResources::Unproven, + }, + cleanup: match self.cleanup { + RuntimeRestoreCleanup::NotNeeded => StartupCleanup::NotNeeded, + RuntimeRestoreCleanup::Required => StartupCleanup::Required, + RuntimeRestoreCleanup::Confirmed => StartupCleanup::Confirmed, + RuntimeRestoreCleanup::Failed(failure) => StartupCleanup::Failed(failure.into()), + }, + durable_effects_possible: self.durable_effects_possible, + }); + result + } + pub fn refused(stage: RuntimeRestoreStage, cause: &'static str) -> Self { Self { stage, @@ -274,6 +350,10 @@ impl RuntimeRestoreFailure { io_kind: None, cleanup_failure: None, durable_effects_possible: false, + primary: Self::fallback_primary(stage), + secondary: Vec::new(), + legacy_diagnostic: None, + metadata_save: None, } } @@ -286,6 +366,10 @@ impl RuntimeRestoreFailure { io_kind: None, cleanup_failure: None, durable_effects_possible: false, + primary: Self::fallback_primary(stage), + secondary: Vec::new(), + legacy_diagnostic: None, + metadata_save: None, } } @@ -298,6 +382,10 @@ impl RuntimeRestoreFailure { io_kind: None, cleanup_failure: None, durable_effects_possible: false, + primary: Self::fallback_primary(stage), + secondary: Vec::new(), + legacy_diagnostic: None, + metadata_save: None, } } @@ -311,7 +399,13 @@ impl RuntimeRestoreFailure { impl std::fmt::Display for RuntimeRestoreFailure { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - let cause: String = self.cause.chars().take(160).collect(); + let cause: String = self + .legacy_diagnostic + .as_deref() + .unwrap_or(self.cause) + .chars() + .take(160) + .collect(); write!( f, "restore {:?}: {}; io={:?}; resources={:?}; cleanup={:?}; prior_cleanup={:?}", @@ -323,6 +417,7 @@ impl std::fmt::Display for RuntimeRestoreFailure { fn restore_worker_cause(error: &WorkerError) -> &'static str { match error { WorkerError::ConstructionCleanup { source, .. } + | WorkerError::StartupFailure { source, .. } | WorkerError::RestoreActivationEffects { source } => restore_worker_cause(source), WorkerError::SegmentEmpty { .. } => "saved segment empty", WorkerError::SegmentSystemPromptMissing { .. } => "saved segment system prompt missing", @@ -368,14 +463,13 @@ async fn prepare_restore_operation( request: WorkerExecutionRestoreRequest, ) -> Result, RuntimeRestoreFailure> { let prior_effects = request.source_repair_possible; - factory.preflight_restore(&request).await.map_err(|_| { - let mut failure = RuntimeRestoreFailure::refused( - RuntimeRestoreStage::Preflight, - "restore preflight refused", - ); - failure.durable_effects_possible = prior_effects; - failure - })?; + factory + .preflight_restore(&request) + .await + .map_err(|mut failure| { + failure.durable_effects_possible = prior_effects; + failure + })?; let repaired = factory .recover_restore_source(&request) .await @@ -479,12 +573,12 @@ pub trait RuntimeWorkerFactory: Send + Sync + 'static { async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, - ) -> Result; + ) -> Result; async fn preflight_restore( &self, _request: &WorkerExecutionRestoreRequest, - ) -> Result<(), String> { + ) -> Result<(), RuntimeRestoreFailure> { Ok(()) } @@ -907,7 +1001,9 @@ impl ProfileRuntimeWorkerFactory { request: &WorkerExecutionRestoreRequest, error: &WorkerError, ) { - if let WorkerError::RestoreActivationEffects { source } = error { + if let WorkerError::RestoreActivationEffects { source } + | WorkerError::StartupFailure { source, .. } = error + { self.retain_construction_failure(request, source); return; } @@ -2008,7 +2104,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { let worker_name = Self::runtime_worker_name(&request); let profile = Self::runtime_profile(&request); let has_local_filesystem = request.workdir_attachments.len() == 1; @@ -2128,13 +2224,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { Some(factory) => bootstrap.with_model_client(factory()), None => bootstrap, }; - let mut prepared = bootstrap.prepare().await.map_err(|error| match error { - WorkerBootstrapError::Worker(source) => { - format!("failed to create Worker from profile: {source}") - } - WorkerBootstrapError::Controller { source, .. } => { - format!("failed to prepare Worker controller: {source}") - } + let mut prepared = bootstrap.prepare().await.map_err(|error| { + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Reconstruction, &error) })?; let worker = prepared.worker_mut(); if let Some(snapshot) = &request.request.plugin_execution { @@ -2160,12 +2251,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .map_err(str::to_string)?; } validate_worker_memory_settings(worker.manifest(), &request.request)?; - worker - .finalize_subjektiv_session_attribution( - request.request.subjektiv_attached, - SubjektivSessionAttributionLifecycle::NewSession, - ) - .map_err(|error| format!("finalize Worker Session attribution: {error}"))?; + startup_failure::finalize_startup_binding(worker, request.request.subjektiv_attached)?; let workdir_capabilities = request .workdir_attachments .keys() @@ -2206,19 +2292,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { return Err(format!( "install Flow transition feature failed: {:?}", report.reports - )); + ) + .into()); } } let workspace_client = worker.workspace_client_handle(); - let started = prepared.start().await.map_err(|error| match error { - WorkerBootstrapError::Worker(source) => { - format!("failed to prepare Worker before controller start: {source}") - } - WorkerBootstrapError::Controller { source, .. } => format!( - "failed to spawn Worker controller in {}: {source}", - run_dir.display() - ), + let started = prepared.start().await.map_err(|error| { + RuntimeRestoreFailure::from_bootstrap(RuntimeRestoreStage::Controller, &error) })?; let (handle, shutdown_rx, controller_task) = (started.handle, started.shutdown, started.controller_task); @@ -2244,79 +2325,90 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { async fn preflight_restore( &self, request: &WorkerExecutionRestoreRequest, - ) -> Result<(), String> { - #[cfg(not(feature = "http-server"))] - RuntimeWorkspaceBackendRef::from_worker_request( - &request.request, - self.runtime_id.as_deref(), - )?; - let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref); - let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?; - if !worker_aggregate_dir.is_dir() { - return Err("Persisted Worker aggregate metadata is unavailable".to_string()); - } - if !worker_aggregate_dir.join("session").is_dir() { - return Err("Persisted Worker Session metadata is unavailable".to_string()); - } - let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name) - .map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?; - let metadata = metadata_store - .read_by_name(&worker_name) - .map_err(|error| format!("failed to read Worker metadata: {error}"))? - .ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?; - let (manifest, loader) = Self::manifest_for_restore(&metadata, &request.request)?; - if let Some(active) = metadata.active.as_ref() - && let Some(segment_id) = active.segment_id - { - // Preflight must not create lock files or migrate retained sources. - // Actual restore reacquires the live guard before any publication. - self.session_retention()? - .ensure_live(active.session_id) - .map_err(|error| format!("Worker Session source is unavailable: {error}"))?; - let session_store = - WorkerSessionStore::open_read_only(worker_aggregate_dir.join("session")) - .map_err(|error| format!("failed to read Worker Session store: {error}"))?; - if !session_store - .exists(active.session_id, segment_id) - .map_err(|error| format!("failed to inspect Worker Session segment: {error}"))? + ) -> Result<(), RuntimeRestoreFailure> { + let result: Result<(), String> = async { + #[cfg(not(feature = "http-server"))] + RuntimeWorkspaceBackendRef::from_worker_request( + &request.request, + self.runtime_id.as_deref(), + )?; + let worker_name = Self::runtime_worker_name_for_ref(&request.worker_ref); + let worker_aggregate_dir = self.worker_aggregate_dir(&request.worker_ref)?; + if !worker_aggregate_dir.is_dir() { + return Err("Persisted Worker aggregate metadata is unavailable".to_string()); + } + if !worker_aggregate_dir.join("session").is_dir() { + return Err("Persisted Worker Session metadata is unavailable".to_string()); + } + let metadata_store = WorkerAggregateStore::new(&worker_aggregate_dir, &worker_name) + .map_err(|error| format!("failed to read Worker aggregate metadata: {error}"))?; + let metadata = metadata_store + .read_by_name(&worker_name) + .map_err(|error| format!("failed to read Worker metadata: {error}"))? + .ok_or_else(|| "Persisted Worker metadata is unavailable".to_string())?; + let (manifest, loader) = Self::manifest_for_restore(&metadata, &request.request)?; + if let Some(active) = metadata.active.as_ref() + && let Some(segment_id) = active.segment_id { - return Err("Persisted Worker Session segment is unavailable".to_string()); + // Preflight must not create lock files or migrate retained sources. + // Actual restore reacquires the live guard before any publication. + self.session_retention()? + .ensure_live(active.session_id) + .map_err(|error| format!("Worker Session source is unavailable: {error}"))?; + let session_store = + WorkerSessionStore::open_read_only(worker_aggregate_dir.join("session")) + .map_err(|error| format!("failed to read Worker Session store: {error}"))?; + if !session_store + .exists(active.session_id, segment_id) + .map_err(|error| format!("failed to inspect Worker Session segment: {error}"))? + { + return Err("Persisted Worker Session segment is unavailable".to_string()); + } + session_store + .read_all(active.session_id, segment_id) + .map_err(|error| format!("failed to read Worker Session segment: {error}"))?; } - session_store - .read_all(active.session_id, segment_id) - .map_err(|error| format!("failed to read Worker Session segment: {error}"))?; - } - if let Some(api) = request.request.workspace_api.as_ref() - && metadata - .active - .as_ref() - .is_some_and(|active| active.segment_id.is_none()) - { - let bundle = request.config_bundle.as_ref().ok_or_else(|| { - "Pending Workspace Worker restore requires operation-owned launch material" - .to_string() - })?; - crate::config_bundle::validate_config_bundle(bundle) - .map_err(|error| format!("invalid pending Worker launch material: {error}"))?; - if bundle.metadata.workspace_id != api.workspace_id { - return Err(format!( - "Workspace Prompt projection scope mismatch: expected {}, got {}", - api.workspace_id, bundle.metadata.workspace_id - )); + if let Some(api) = request.request.workspace_api.as_ref() + && metadata + .active + .as_ref() + .is_some_and(|active| active.segment_id.is_none()) + { + let bundle = request.config_bundle.as_ref().ok_or_else(|| { + "Pending Workspace Worker restore requires operation-owned launch material" + .to_string() + })?; + crate::config_bundle::validate_config_bundle(bundle) + .map_err(|error| format!("invalid pending Worker launch material: {error}"))?; + if bundle.metadata.workspace_id != api.workspace_id { + return Err(format!( + "Workspace Prompt projection scope mismatch: expected {}, got {}", + api.workspace_id, bundle.metadata.workspace_id + )); + } + let prompt_catalog = bundle.prompt_catalog.as_ref().ok_or_else(|| { + "pending Workspace Worker restore requires a saved Workspace Prompt projection" + .to_string() + })?; + // Validate the exact launch catalog without publishing it into the + // current projection cache or allocating live Worker resources. + worker::SystemPromptTemplate::parse( + &manifest.engine.instruction, + loader.with_effective_catalog(prompt_catalog.clone()), + ) + .map_err(|error| format!("invalid pending Worker launch Prompt: {error}"))?; } - let prompt_catalog = bundle.prompt_catalog.as_ref().ok_or_else(|| { - "pending Workspace Worker restore requires a saved Workspace Prompt projection" - .to_string() - })?; - // Validate the exact launch catalog without publishing it into the - // current projection cache or allocating live Worker resources. - worker::SystemPromptTemplate::parse( - &manifest.engine.instruction, - loader.with_effective_catalog(prompt_catalog.clone()), - ) - .map_err(|error| format!("invalid pending Worker launch Prompt: {error}"))?; + Ok(()) } - Ok(()) + .await; + result.map_err(|diagnostic| { + let mut failure = RuntimeRestoreFailure::refused( + RuntimeRestoreStage::Preflight, + "restore preflight refused", + ); + failure.legacy_diagnostic = Some(diagnostic); + failure + }) } async fn recover_restore_source( @@ -2347,12 +2439,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { self.runtime_id.as_deref(), ) .map_err(|cause| RuntimeRestoreFailure::refused(RuntimeRestoreStage::Preflight, cause))?; - self.preflight_restore(&request).await.map_err(|_| { - RuntimeRestoreFailure::refused( - RuntimeRestoreStage::Preflight, - "restore preflight refused", - ) - })?; + self.preflight_restore(&request).await?; #[cfg(test)] self.restore_faults .executions @@ -2455,7 +2542,10 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .await .map_err(|error| { self.retain_construction_failure(&request, &error); - restore_worker_cause(&error) + let failure = RuntimeRestoreFailure::from_worker(stage, &error); + let cause = failure.cause; + preparation_failure = Some(failure); + cause })?; self.failed_restore_resources .lock() @@ -2728,7 +2818,8 @@ impl RuntimeRestoreCandidate for ProfileRestoreCandidate { Some(source.kind()), ), }; - let mut failure = RuntimeRestoreFailure::started(stage, cause); + let mut failure = RuntimeRestoreFailure::from_bootstrap(stage, &error); + failure.cause = cause; failure.durable_effects_possible = true; failure.io_kind = io_kind; failure @@ -2870,6 +2961,7 @@ impl ProfileRuntimeWorkerFactory { ) -> Result { let mut stage = RuntimeRestoreStage::Binding; let mut io_kind = None; + let mut primary_failure = None; let result: Result = async { let feature_tasks = candidate .retain_startup_feature_cleanup() @@ -2998,6 +3090,7 @@ 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) } WorkerBootstrapError::Controller { source, .. } => { @@ -3028,7 +3121,8 @@ impl ProfileRuntimeWorkerFactory { } .await; result.map_err(|cause| { - let mut failure = RuntimeRestoreFailure::started(stage, cause); + let mut failure = + primary_failure.unwrap_or_else(|| RuntimeRestoreFailure::started(stage, cause)); failure.io_kind = io_kind; failure }) @@ -3744,10 +3838,16 @@ where let factory = Arc::clone(&self.factory); let request = request.clone(); - self.run_on_adapter_runtime(async move { factory.preflight_restore(&request).await }) + let outcome = self + .run_on_adapter_runtime(async move { Ok(factory.preflight_restore(&request).await) }) .map_err(|message| { WorkerExecutionResult::rejected(WorkerExecutionOperation::Restore, message) })?; + outcome.map_err(|failure| { + let mut result = failure.execution_result(WorkerExecutionOperation::Restore); + result.outcome = crate::execution::WorkerExecutionOutcome::Rejected; + result + })?; Ok(()) } @@ -5418,10 +5518,15 @@ where let worker_ref = request.worker_ref.clone(); let spawn_result = self .run_cancellable_on_adapter_runtime(self.spawn_restore_timeout, async move { - factory.spawn_controller(request).await + Ok(factory.spawn_controller(request).await) }); let controller = match spawn_result { - Ok(controller) => controller, + Ok(Ok(controller)) => controller, + Ok(Err(failure)) => { + return WorkerExecutionSpawnResult::Errored( + failure.execution_result(WorkerExecutionOperation::Spawn), + ); + } Err(message) => { return WorkerExecutionSpawnResult::Errored( WorkerExecutionResult::errored( @@ -5457,7 +5562,10 @@ where materializer.as_ref(), &request.workdir_attachments, &rollback_workdirs, - &message, + &WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + message.clone(), + ), ) { return result; } @@ -5479,7 +5587,10 @@ where materializer.as_ref(), &request.workdir_attachments, &rollback_workdirs, - &message, + &WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + message.clone(), + ), ) { return result; } @@ -5497,7 +5608,10 @@ where materializer.as_ref(), &request.workdir_attachments, &rollback_workdirs, - &message, + &WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + message.clone(), + ), ) { return result; } @@ -5521,7 +5635,10 @@ where materializer.as_ref(), &request.workdir_attachments, &rollback_workdirs, - &message, + &WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + message.clone(), + ), ) { return result; } @@ -5539,17 +5656,32 @@ where let worker_ref = request.worker_ref.clone(); let spawn_result = self .run_cancellable_on_adapter_runtime(self.spawn_restore_timeout, async move { - factory.spawn_controller(request).await + Ok(factory.spawn_controller(request).await) }); let controller = match spawn_result { - Ok(controller) => controller, + Ok(Ok(controller)) => controller, + Ok(Err(failure)) => { + let result = failure.execution_result(WorkerExecutionOperation::Spawn); + if let Err(cleanup) = rollback_materialized_spawn_workdirs( + materializer.as_ref(), + &workdir_attachments, + &rollback_workdirs, + &result, + ) { + return cleanup; + } + return WorkerExecutionSpawnResult::Errored(result); + } Err(message) => { if let Err(result) = rollback_materialized_spawn_workdirs( materializer.as_ref(), &workdir_attachments, &rollback_workdirs, - &message, + &WorkerExecutionResult::rejected( + WorkerExecutionOperation::Spawn, + message.clone(), + ), ) { return result; } @@ -5663,8 +5795,7 @@ where if *completed_id != operation_id { return None; } - let result = - WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, failure.to_string()); + let result = failure.execution_result(WorkerExecutionOperation::Restore); Some(if failure.cleanup == RuntimeRestoreCleanup::NotNeeded { WorkerExecutionSpawnResult::Rejected(result) } else { @@ -5935,7 +6066,7 @@ where }; if let Err(message) = self.wait_pending_restore(&pending, self.spawn_restore_timeout) { return WorkerExecutionSpawnResult::ReconciliationRequired { - result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message), + result: pending.failure_result(message), worker_state: None, workdir_attachments: Vec::new(), }; @@ -5946,10 +6077,7 @@ where .and_then(|()| self.wait_pending_restore(&pending, self.spawn_restore_timeout)) { return WorkerExecutionSpawnResult::ReconciliationRequired { - result: WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - message, - ), + result: pending.failure_result(message), worker_state: None, workdir_attachments: Vec::new(), }; @@ -5962,10 +6090,15 @@ where .as_ref() { return WorkerExecutionSpawnResult::ReconciliationRequired { - result: WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - failure.to_string(), - ), + result: { + let mut result = pending.failure_result(failure.to_string()); + result.push_secondary_failure( + protocol::startup_failure::StartupSecondaryFailure::Cleanup { + failure: (*failure).into(), + }, + ); + result + }, worker_state: None, workdir_attachments: pending.request.previous_workdir_attachments.clone(), }; @@ -5982,13 +6115,15 @@ where workdir_attachments: pending.request.previous_workdir_attachments.clone(), }; } - let message = match activated.as_ref() { - Some(Err(failure)) => failure.to_string(), - _ => "restore activation is owned by the same operation; reconcile activation" - .to_string(), + let result = match activated.as_ref() { + Some(Err(failure)) => failure.execution_result(WorkerExecutionOperation::Restore), + _ => WorkerExecutionResult::errored( + WorkerExecutionOperation::Restore, + "restore activation is owned by the same operation; reconcile activation", + ), }; return WorkerExecutionSpawnResult::ReconciliationRequired { - result: WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message), + result, worker_state: None, workdir_attachments: pending.request.previous_workdir_attachments.clone(), }; @@ -6010,10 +6145,7 @@ where match retained.as_ref() { Some(Ok(candidate)) => candidate.initial_state(), Some(Err(failure)) => { - let result = WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - failure.to_string(), - ); + let result = failure.execution_result(WorkerExecutionOperation::Restore); let cleanup = failure.cleanup; if failure.durable_effects_possible { return WorkerExecutionSpawnResult::ReconciliationRequired { @@ -6224,10 +6356,7 @@ where .insert(worker_ref.clone(), (operation_id, failure.clone())); self.remove_pending_restore(worker_ref)?; Ok(WorkerExecutionSpawnResult::RolledBack( - WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - failure.to_string(), - ), + failure.execution_result(WorkerExecutionOperation::Restore), )) })(); reclaimed.unwrap_or_else( @@ -6360,10 +6489,7 @@ where .insert(worker_ref.clone(), (operation_id, failure.clone())); self.remove_pending_restore(worker_ref)?; Ok(WorkerExecutionSpawnResult::RolledBack( - WorkerExecutionResult::errored( - WorkerExecutionOperation::Restore, - failure.to_string(), - ), + failure.execution_result(WorkerExecutionOperation::Restore), )) })(); discarded.unwrap_or_else( @@ -7034,6 +7160,7 @@ mod tests { use worker::PreparedWorker; include!("subject_delete_backend_tests.rs"); + include!("worker_backend/startup_binding_tests.rs"); #[cfg(feature = "fs-store")] include!("subject_delete_create_tests.rs"); @@ -8876,7 +9003,11 @@ mod tests { let before = persisted_files(&runtime_store); let error = factory.preflight_restore(&request).await.unwrap_err(); assert!( - error.contains("Pending Workspace Worker restore requires"), + error + .legacy_diagnostic + .as_deref() + .unwrap_or_default() + .contains("Pending Workspace Worker restore requires"), "{error}" ); assert_eq!(persisted_files(&runtime_store), before); @@ -8936,7 +9067,14 @@ mod tests { (missing_instruction, "invalid pending Worker launch Prompt"), ] { let error = factory.preflight_restore(&invalid).await.unwrap_err(); - assert!(error.contains(expected), "{error}"); + assert!( + error + .legacy_diagnostic + .as_deref() + .unwrap_or_default() + .contains(expected), + "{error}" + ); let error = match factory.restore_controller(invalid).await { Ok(controller) => { shutdown_profile_controller(controller).await; @@ -9075,7 +9213,11 @@ mod tests { .await .unwrap_err(); assert!( - error.contains("subjektiv settings snapshot mismatch"), + error + .legacy_diagnostic + .as_deref() + .unwrap_or_default() + .contains("subjektiv settings snapshot mismatch"), "{error}" ); let error = match factory.restore_controller(wrong_settings).await { @@ -10053,7 +10195,7 @@ mod tests { async fn spawn_controller( &self, _request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { let _drop = FactoryFutureDrop(self.dropped.clone()); self.started.send(()).unwrap(); std::future::pending().await @@ -10233,7 +10375,7 @@ mod tests { async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { let manifest = WorkerManifest::from_toml( r#" [worker] @@ -10664,8 +10806,8 @@ mod tests { async fn spawn_controller( &self, _request: WorkerExecutionSpawnRequest, - ) -> Result { - Err("spawn failed".to_string()) + ) -> Result { + Err("spawn failed".into()) } async fn restore_controller( @@ -10701,7 +10843,7 @@ mod tests { async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { self.inner.spawn_controller(request).await } @@ -11021,7 +11163,7 @@ mod tests { async fn spawn_controller( &self, _request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { panic!("stop must not spawn a Controller") } async fn restore_controller( @@ -11077,11 +11219,14 @@ mod tests { async fn spawn_controller( &self, _: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { panic!("reclaim must not spawn") } - async fn preflight_restore(&self, _: &WorkerExecutionRestoreRequest) -> Result<(), String> { + async fn preflight_restore( + &self, + _: &WorkerExecutionRestoreRequest, + ) -> Result<(), RuntimeRestoreFailure> { panic!("reclaim must not preflight preparation") } @@ -11461,7 +11606,7 @@ mod tests { async fn spawn_controller( &self, request: WorkerExecutionSpawnRequest, - ) -> Result { + ) -> Result { self.inner.spawn_controller(request).await } async fn restore_controller( diff --git a/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs new file mode 100644 index 000000000..1a532fc26 --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/startup_binding_tests.rs @@ -0,0 +1,253 @@ +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. + use super::*; + use protocol::startup_failure::{RuntimeStartupCause, StartupPersistenceOutcome}; + use session_store::worker_metadata::StartupFailureRead; + use worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse}; + + #[derive(Debug, Clone, Copy)] + enum Response { + Rejected, + Lost, + BeforeSession, + AfterSession, + BeforeOwner, + AfterOwner, + AfterAttribution, + } + struct AttributionClient { + response: Response, + store: WorkerAggregateStore, + name: String, + } + impl std::fmt::Debug for AttributionClient { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("AttributionClient") + } + } + impl WorkspaceClient for AttributionClient { + fn workspace_id(&self) -> Option<&str> { + Some("startup-binding-workspace") + } + fn kind(&self) -> &str { + "startup-binding-fixture" + } + fn is_available(&self) -> bool { + true + } + fn execute(&self, _: WorkspaceRequest) -> Result { + match self.response { + Response::AfterSession => self + .store + .update_by_name(&self.name, |metadata| { + metadata.active = Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )); + metadata.subjektiv_session_attribution = None; + }) + .unwrap(), + Response::AfterOwner => self + .store + .update_by_name(&self.name, |metadata| { + metadata.workspace_id = Some("replacement-owner".into()); + }) + .unwrap(), + Response::AfterAttribution => self + .store + .update_by_name(&self.name, |metadata| { + metadata.subjektiv_session_attribution = + Some(session_store::SubjektivSessionAttributionState::Confirmed { + session_id: metadata.active.as_ref().unwrap().session_id, + subject_id: "replacement-subject".into(), + }); + }) + .unwrap(), + _ => self.store.read_by_name(&self.name).unwrap().unwrap(), + }; + if matches!(self.response, Response::Lost) { + return Err(WorkspaceClientError::Unavailable( + "response deliberately lost".into(), + )); + } + Ok(WorkspaceResponse { + status: 403, + body: "synthetic rejection with private token /internal/path".into(), + }) + } + } + async fn check_binding(response: Response) { + require_restore_test_environment(); + let root = tempfile::tempdir().unwrap(); + let name = "startup-binding"; + let store = WorkerAggregateStore::new(root.path(), name).unwrap(); + let session_store = WorkerSessionStore::new(root.path().join("session")).unwrap(); + let client = Arc::new(AttributionClient { + response, + store: store.clone(), + name: name.into(), + }); + let manifest = WorkerManifest::from_toml( + r#" + [worker] + name = "startup-binding" + [model] + scheme = "anthropic" + model_id = "unused" + auth = { kind = "none" } + [engine] + max_tokens = 100 + [scope] + allow = [] + "#, + ) + .unwrap(); + let llm: Box = Box::new(MockClient::new(Vec::new())); + let mut worker = Worker::new( + manifest, + Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(llm), + CombinedStore::new(session_store, store.clone()), + WorkerWorkspaceContext::with_client( + Some(WorkspaceId::new("startup-binding-workspace").unwrap()), + client, + ), + WorkerFilesystemAuthority::None, + Scope::empty(), + ) + .await + .unwrap(); + worker.enable_worker_metadata_write_through().unwrap(); + let origin = worker.session_id(); + match response { + Response::BeforeSession => store + .update_by_name(name, |metadata| { + metadata.active = Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )); + }) + .unwrap(), + Response::BeforeOwner => store + .update_by_name(name, |metadata| { + metadata.workspace_id = Some("replacement-owner".into()); + }) + .unwrap(), + _ => store.read_by_name(name).unwrap().unwrap(), + }; + let failure = startup_failure::finalize_startup_binding(&worker, true) + .err() + .expect("failure must remain a failure"); + let result = failure.execution_result(WorkerExecutionOperation::Spawn); + let primary = result.startup_failure::().unwrap(); + assert!(matches!(primary, error_contract::DecodeFailure::Known(_))); + drop(worker); + drop(store); + let reopened = WorkerAggregateStore::new(root.path(), name).unwrap(); + let metadata = reopened.read_by_name(name).unwrap().unwrap(); + match response { + Response::Rejected | Response::Lost => { + assert_eq!( + result.startup_failure_persistence, + Some(StartupPersistenceOutcome::Persisted) + ); + let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) = + 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 + ); + assert_eq!(metadata.active.unwrap().session_id, origin); + let saved = serde_json::to_string(&metadata.startup_failure).unwrap(); + assert!(!saved.contains("private token")); + assert!(!saved.contains("/internal/path")); + assert_eq!( + metadata + .subjektiv_session_attribution + .as_ref() + .unwrap() + .session_id(), + origin + ); + assert_eq!( + matches!( + metadata.subjektiv_session_attribution, + Some( + session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. } + ) + ), + matches!(response, Response::Lost) + ); + } + _ => { + assert!( + matches!( + metadata.decode_startup_failure::(), + StartupFailureRead::Missing + ), + "replacement Session/owner must not acquire stale cause" + ); + assert_ne!( + result.startup_failure_persistence, + Some(StartupPersistenceOutcome::Persisted) + ); + if matches!(response, Response::BeforeSession | Response::AfterSession) { + assert_ne!(metadata.active.unwrap().session_id, origin); + } + if matches!( + response, + Response::BeforeOwner | Response::BeforeSession | Response::AfterSession + ) { + assert!(metadata.subjektiv_session_attribution.is_none()); + } + if matches!(response, Response::AfterOwner) { + assert!(matches!( + metadata.subjektiv_session_attribution, + Some( + session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. } + ) + )); + } + if matches!(response, Response::BeforeOwner | Response::AfterOwner) { + assert_eq!(metadata.workspace_id.as_deref(), Some("replacement-owner")); + } + if matches!(response, Response::AfterAttribution) { + assert!(matches!( + metadata.subjektiv_session_attribution, + Some(session_store::SubjektivSessionAttributionState::Confirmed { .. }) + )); + } + } + } + } + #[tokio::test] + async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() { + check_binding(Response::Rejected).await; + } + #[tokio::test] + async fn factory_unknown_response_saves_cause_without_claiming_attribution_success() { + check_binding(Response::Lost).await; + } + #[tokio::test] + async fn factory_session_replaced_before_capture_rejects_stale_cause() { + check_binding(Response::BeforeSession).await; + } + #[tokio::test] + async fn factory_session_replaced_after_capture_rejects_stale_cause() { + check_binding(Response::AfterSession).await; + } + #[tokio::test] + async fn factory_owner_replaced_before_capture_rejects_stale_cause() { + check_binding(Response::BeforeOwner).await; + } + #[tokio::test] + async fn factory_owner_replaced_after_capture_rejects_stale_cause() { + check_binding(Response::AfterOwner).await; + } + #[tokio::test] + async fn factory_attribution_replaced_after_capture_preserves_new_attribution() { + check_binding(Response::AfterAttribution).await; + } +} diff --git a/crates/worker-runtime/src/worker_backend/startup_failure.rs b/crates/worker-runtime/src/worker_backend/startup_failure.rs new file mode 100644 index 000000000..8f70f3bde --- /dev/null +++ b/crates/worker-runtime/src/worker_backend/startup_failure.rs @@ -0,0 +1,301 @@ +//! Conversion preserves owner-selected facts. The string-only legacy seams carry +//! no domain verdict; they retain their diagnostic privately and receive a coarse +//! Runtime fallback, never a cause guessed from that diagnostic. +use super::*; + +impl From for protocol::startup_failure::StartupStage { + fn from(stage: RuntimeRestoreStage) -> Self { + match stage { + RuntimeRestoreStage::Preflight => Self::Preflight, + RuntimeRestoreStage::Metadata => Self::Metadata, + RuntimeRestoreStage::Reconstruction => Self::Reconstruction, + RuntimeRestoreStage::Recovery => Self::Recovery, + RuntimeRestoreStage::Binding => Self::Binding, + RuntimeRestoreStage::Feature => Self::Feature, + RuntimeRestoreStage::ActivationEffects => Self::ActivationEffects, + RuntimeRestoreStage::Controller => Self::Controller, + } + } +} + +impl From for protocol::startup_failure::StartupCleanupFailure { + fn from(failure: RuntimeRestoreCleanupFailure) -> Self { + match failure { + RuntimeRestoreCleanupFailure::Ownership => Self::Ownership, + RuntimeRestoreCleanupFailure::Workdir => Self::Workdir, + RuntimeRestoreCleanupFailure::Artifacts => Self::Artifacts, + RuntimeRestoreCleanupFailure::Allocation => Self::Allocation, + RuntimeRestoreCleanupFailure::Feature => Self::Feature, + RuntimeRestoreCleanupFailure::Unsupported => Self::Unsupported, + } + } +} + +impl RuntimeRestoreFailure { + pub(super) fn from_bootstrap(stage: RuntimeRestoreStage, error: &WorkerBootstrapError) -> Self { + match error { + WorkerBootstrapError::Worker(source) => Self::from_worker(stage, source), + WorkerBootstrapError::Controller { source, .. } => { + let mut failure = Self::started( + RuntimeRestoreStage::Controller, + "Worker controller startup failed", + ); + failure.io_kind = Some(source.kind()); + failure + } + } + } +} + +impl From for RuntimeRestoreFailure { + fn from(error: session_store::WorkerStoreError) -> Self { + Self::from_worker( + RuntimeRestoreStage::Metadata, + &WorkerError::WorkerStore(error), + ) + } +} + +impl From for RuntimeRestoreFailure { + fn from(diagnostic: String) -> Self { + let mut failure = + Self::unproven(RuntimeRestoreStage::Reconstruction, "Worker factory failed"); + failure.legacy_diagnostic = Some(diagnostic); + failure + } +} +impl From<&'static str> for RuntimeRestoreFailure { + fn from(diagnostic: &'static str) -> Self { + diagnostic.to_string().into() + } +} + +impl PendingRuntimeRestore { + /// Adapter waiting/cleanup errors supplement, rather than stringify and replace, + /// any primary result already received from this exact operation's owner. + pub(super) fn failure_result(&self, message: String) -> WorkerExecutionResult { + let failure = self + .result + .lock() + .ok() + .and_then(|saved| { + saved + .as_ref() + .and_then(|result| result.as_ref().err()) + .cloned() + }) + .or_else(|| { + self.activated.lock().ok().and_then(|saved| { + saved + .as_ref() + .and_then(|result| result.as_ref().err()) + .cloned() + }) + }); + let mut result = failure + .map(|failure| failure.execution_result(WorkerExecutionOperation::Restore)) + .unwrap_or_else(|| { + WorkerExecutionResult::errored(WorkerExecutionOperation::Restore, message.clone()) + }); + result.message = Some(message); + result + } +} + +pub(super) fn finalize_startup_binding( + worker: &Worker, St>, + attached: bool, +) -> Result<(), RuntimeRestoreFailure> +where + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, +{ + worker + .finalize_subjektiv_session_attribution_retaining_startup_failure( + attached, + SubjektivSessionAttributionLifecycle::NewSession, + ) + .map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error)) +} + +#[cfg(test)] +mod tests { + //! Connect the typed factory seam to the real Tokio adapter and protocol codec. + //! The external startup action is a deterministic failure, not a real domain + //! authorization decision. Resource disposition remains the existing backend result. + use super::*; + use crate::execution::WorkerExecutionContext; + use crate::identity::WorkerId; + use protocol::startup_failure::*; + + struct TypedFailingFactory; + fn primary() -> StartupFailureAttachment { + worker::startup_failure::runtime_startup_attachment(RuntimeStartupCause { + stage: StartupStage::Feature, + reason: RuntimeStartupReason::FeatureFailed, + }) + .unwrap() + } + fn failure() -> RuntimeRestoreFailure { + let mut failure = RuntimeRestoreFailure::unproven( + RuntimeRestoreStage::Controller, + "factory failed after receiving primary", + ) + .with_startup_failure(primary()); + failure.cleanup = RuntimeRestoreCleanup::Failed(RuntimeRestoreCleanupFailure::Workdir); + failure.cleanup_failure = Some(RuntimeRestoreCleanupFailure::Workdir); + failure.io_kind = Some(std::io::ErrorKind::PermissionDenied); + failure + } + #[async_trait] + impl RuntimeWorkerFactory for TypedFailingFactory { + async fn spawn_controller( + &self, + _: WorkerExecutionSpawnRequest, + ) -> Result { + Err(failure()) + } + async fn preflight_restore( + &self, + _: &WorkerExecutionRestoreRequest, + ) -> Result<(), RuntimeRestoreFailure> { + Err(failure()) + } + async fn prepare_restore( + &self, + _: WorkerExecutionRestoreRequest, + ) -> Result, RuntimeRestoreFailure> { + panic!("typed preflight rejection must not prepare live work") + } + } + + fn request() -> WorkerExecutionSpawnRequest { + let worker_ref = WorkerRef::new(WorkerId::now_v7()); + // No factory reads this archive: it is explicit launch input, not a retained approval. + let archive = crate::profile_archive::ProfileSourceArchive::build( + crate::profile_archive::ProfileSourceArchiveInput { + id: "profile-source-archive:typed-startup".into(), + entrypoints: BTreeMap::from([( + "builtin:companion".into(), + "profiles/default.dcdl".into(), + )]), + imports: BTreeMap::new(), + sources: BTreeMap::from([("profiles/default.dcdl".into(), "{}".into())]), + }, + ) + .unwrap(); + let request = crate::catalog::CreateWorkerRequest { + worker_id: worker_ref.worker_id, + create_fingerprint: "typed-startup-failure".into(), + profile: crate::catalog::ProfileSelector::Builtin("builtin:companion".into()), + display_name: None, + profile_source: crate::catalog::ProfileSourceArchiveSource::Embedded { archive }, + config_bundle: None, + initial_input: None, + workdir_attachment_requests: Vec::new(), + workdir_attachments: Vec::new(), + worker_observation_enabled: false, + worker_observation_grants: Vec::new(), + workspace_api: None, + memory_settings: None, + plugin_execution: None, + subjektiv_attached: false, + backend_job: None, + }; + #[cfg(feature = "ws-server")] + let context = WorkerExecutionContext::new( + worker_ref.clone(), + Arc::new(|_, _| unreachable!("failed startup must not publish a Controller")), + ); + #[cfg(not(feature = "ws-server"))] + let context = WorkerExecutionContext::new(worker_ref.clone()); + WorkerExecutionSpawnRequest { + worker_ref, + request, + workspace_scope: None, + context, + workdir_attachments: BTreeMap::new(), + config_bundle: None, + } + } + + fn assert_primary(result: &WorkerExecutionResult) { + assert_eq!(result.startup_failure.as_ref(), Some(&primary())); + let Some(error_contract::DecodeFailure::Known(received)) = + result.startup_failure::() + else { + panic!("typed primary missing") + }; + assert_eq!(received.facts().cause().stage, StartupStage::Feature); + assert_eq!( + received.facts().cause().reason, + RuntimeStartupReason::FeatureFailed + ); + assert_eq!( + received.facts().context().operations[0].operation, + "worker_startup" + ); + assert!( + result + .secondary_failures + .contains(&StartupSecondaryFailure::Cleanup { + failure: StartupCleanupFailure::Workdir + }) + ); + assert_eq!( + result.startup_io_kind, + Some(StartupIoKind::PermissionDenied) + ); + assert!(!result.is_accepted()); + } + + #[test] + fn typed_factory_failure_survives_adapter_and_protocol_without_changing_disposition() { + let backend = WorkerRuntimeExecutionBackend::new(TypedFailingFactory).unwrap(); + let WorkerExecutionSpawnResult::Errored(result) = backend.spawn_worker(request()) else { + panic!("cause attachment must not reclassify the existing spawn result") + }; + assert_primary(&result); + let transferred: WorkerExecutionResult = + serde_json::from_slice(&serde_json::to_vec(&result).unwrap()).unwrap(); + assert_eq!(transferred, result); + assert_primary(&transferred); + assert_eq!( + transferred.startup_resources.unwrap().cleanup, + StartupCleanup::Failed(StartupCleanupFailure::Workdir) + ); + } + + #[test] + fn typed_preflight_failure_survives_adapter_without_starting_live_preparation() { + let backend = WorkerRuntimeExecutionBackend::new(TypedFailingFactory).unwrap(); + let spawn = request(); + let request = WorkerExecutionRestoreRequest { + source_repair_possible: false, + operation_id: crate::execution::WorkerLifecycleOperationId::new(), + worker_ref: spawn.worker_ref, + request: spawn.request, + workspace_scope: None, + context: spawn.context, + previous_workdir_attachments: Vec::new(), + logical_workdir_attachments: Vec::new(), + workdir_attachments: BTreeMap::new(), + deferred_workdir_sessions: BTreeMap::new(), + config_bundle: None, + }; + let result = backend.preflight_restore(&request).unwrap_err(); + assert_eq!( + result.outcome, + crate::execution::WorkerExecutionOutcome::Rejected + ); + assert_primary(&result); + let failure = backend + .run_on_adapter_runtime(async move { + Ok(prepare_restore_operation(&TypedFailingFactory, request) + .await + .err() + .unwrap()) + }) + .unwrap(); + assert_eq!(failure.startup_failure(), &primary()); + } +} diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index e8052500b..119c62d36 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -34,6 +34,7 @@ mod subjektiv_metadata; pub use subject_release::SubjectReleaseReservation; pub mod skill; pub mod spawn; +pub mod startup_failure; pub mod subjektiv; pub mod wip; diff --git a/crates/worker/src/startup_failure.rs b/crates/worker/src/startup_failure.rs new file mode 100644 index 000000000..dc856231a --- /dev/null +++ b/crates/worker/src/startup_failure.rs @@ -0,0 +1,352 @@ +//! Typed startup diagnostics across Worker ownership wrappers. Domain causes are +//! projected at their origin; fallback Runtime facts are selected from variants, +//! never inferred from Display or legacy attribution diagnostics. +use crate::WorkerError; +use error_contract::{FailureFacts, InternalTransfer, OperationContext, project_failure}; +use protocol::startup_failure::{ + RuntimeStartupCause, RuntimeStartupReason, StartupFailureAttachment, StartupSecondaryFailure, + StartupStage, +}; + +impl WorkerError { + pub fn with_startup_failure(self, failure: StartupFailureAttachment) -> Self { + Self::StartupFailure { + source: Box::new(self), + failure, + secondary: Vec::new(), + metadata_save: None, + } + } + + pub fn startup_failure(&self) -> Option<&StartupFailureAttachment> { + match self { + Self::StartupFailure { failure, .. } => Some(failure), + Self::ConstructionCleanup { source, .. } + | Self::RestoreActivationEffects { source } => source.startup_failure(), + _ => None, + } + } + + pub fn startup_secondary_failures(&self) -> &[StartupSecondaryFailure] { + match self { + Self::StartupFailure { secondary, .. } => secondary, + Self::ConstructionCleanup { source, .. } + | Self::RestoreActivationEffects { source } => source.startup_secondary_failures(), + _ => &[], + } + } + + pub fn startup_metadata_save( + &self, + ) -> Option { + match self { + Self::StartupFailure { metadata_save, .. } => *metadata_save, + Self::ConstructionCleanup { source, .. } + | Self::RestoreActivationEffects { source } => source.startup_metadata_save(), + _ => None, + } + } + + /// Preserve the original attachment when an outer resource owner adds context. + pub fn startup_failure_at( + &self, + stage: StartupStage, + ) -> Result { + if let Some(failure) = self.startup_failure() { + return Ok(failure.clone()); + } + let reason = match self { + Self::ConstructionCleanup { source, .. } + | Self::RestoreActivationEffects { source } => return source.startup_failure_at(stage), + Self::SegmentEmpty { .. } => RuntimeStartupReason::SegmentEmpty, + Self::SegmentSystemPromptMissing { .. } => RuntimeStartupReason::SystemPromptMissing, + Self::WorkerMetadataPending { .. } => RuntimeStartupReason::MetadataPending, + Self::WorkerMetadataMissing { .. } => RuntimeStartupReason::MetadataMissing, + Self::WorkerMetadataInactive { .. } => RuntimeStartupReason::MetadataInactive, + Self::WorkerMetadataManifestSnapshotMissing { .. } => { + RuntimeStartupReason::ManifestMissing + } + Self::ScopeLock(_) => RuntimeStartupReason::AllocationFailed, + Self::Store(_) => RuntimeStartupReason::SessionReadFailed, + Self::WorkerStore(_) => RuntimeStartupReason::MetadataReadFailed, + Self::Provider(_) => RuntimeStartupReason::ProviderFailed, + Self::SubjektivSessionAttribution { .. } => RuntimeStartupReason::AttributionFailed, + Self::FeatureInstall(_) | Self::FeatureLifecycle(_) => { + RuntimeStartupReason::FeatureFailed + } + _ => RuntimeStartupReason::StartupFailed, + }; + runtime_startup_attachment(RuntimeStartupCause { stage, reason }) + } +} + +/// The local fallback accepts only enums, so provider strings and internal paths +/// cannot accidentally enter storage or public cause details. +pub fn runtime_startup_attachment( + cause: RuntimeStartupCause, +) -> Result { + let mut facts = FailureFacts::new(cause)?; + facts.with_context(OperationContext::new("worker_startup", "runtime")?)?; + let projected = project_failure::(&facts, &())?; + StartupFailureAttachment::from_projected(&projected) +} + +/// Bind a caller-owned cause to the exact pre-observed Session/attribution. Saving +/// never replaces the original error, changes ownership, or issues retry authority. +pub fn persist_startup_failure( + store: &S, + expected: &session_store::worker_metadata::StartupFailureExpectation, + source: WorkerError, + stage: StartupStage, + authorization: &D::Authorization, +) -> WorkerError +where + S: session_store::WorkerMetadataStore, + D: error_contract::WireDetails + error_contract::ProjectFor, +{ + use protocol::startup_failure::{StartupFailurePhase, StartupPersistenceOutcome}; + use session_store::worker_metadata::StartupFailureSaveResult; + let failure = source + .startup_failure_at(stage) + .expect("fixed Worker startup projection fits the error-contract budget"); + let storage = failure + .project_for::(authorization) + .and_then(|projected| { + protocol::startup_failure::StoredStartupFailureAttachment::from_projected(&projected) + }); + let outcome = match storage + .as_ref() + .map(|stored| store.compare_and_swap_startup_failure(expected, stored)) + { + Ok(StartupFailureSaveResult::Persisted) => StartupPersistenceOutcome::Persisted, + Ok(StartupFailureSaveResult::Conflict) => StartupPersistenceOutcome::Conflict, + Ok(StartupFailureSaveResult::NotSaved(_)) => StartupPersistenceOutcome::NotSaved, + Ok(StartupFailureSaveResult::OutcomeUnknown(_)) => { + StartupPersistenceOutcome::OutcomeUnknown + } + Err(_) => StartupPersistenceOutcome::NotSaved, + }; + let mut secondary = source.startup_secondary_failures().to_vec(); + if storage.is_err() { + secondary.push(StartupSecondaryFailure::ProjectionUnavailable { + phase: StartupFailurePhase::MetadataSave, + }); + } + if outcome != StartupPersistenceOutcome::Persisted { + secondary.push(StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome, + }); + } + WorkerError::StartupFailure { + source: Box::new(source), + failure, + secondary, + metadata_save: Some(outcome), + } +} + +#[cfg(test)] +mod tests { + //! Connect Worker error retention, projection and metadata CAS to the real filesystem store. + //! One wrapper deliberately loses the save response after the real CAS commits; it does not + //! replace persistence or claim a real OS crash. Run in the normal worker lib test target. + use super::*; + use protocol::startup_failure::{ + StartupFailurePhase, StartupPersistenceOutcome, StoredStartupFailureAttachment, + }; + use session_store::worker_metadata::{ + StartupFailureExpectation, StartupFailureRead, StartupFailureSaveResult, + }; + use session_store::{ + FsWorkerStore, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, + WorkerStoreError, + }; + + struct LostSaveResponse(FsWorkerStore); + impl WorkerMetadataStore for LostSaveResponse { + fn write(&self, metadata: &WorkerMetadata) -> Result<(), WorkerStoreError> { + self.0.write(metadata) + } + fn read_by_name(&self, name: &str) -> Result, WorkerStoreError> { + self.0.read_by_name(name) + } + fn list_names(&self) -> Result, WorkerStoreError> { + self.0.list_names() + } + fn delete_by_name(&self, name: &str) -> Result<(), WorkerStoreError> { + self.0.delete_by_name(name) + } + fn compare_and_swap_startup_failure( + &self, + expected: &StartupFailureExpectation, + failure: &StoredStartupFailureAttachment, + ) -> StartupFailureSaveResult { + match self.0.compare_and_swap_startup_failure(expected, failure) { + StartupFailureSaveResult::Persisted => StartupFailureSaveResult::OutcomeUnknown( + WorkerStoreError::Io(std::io::Error::other("TEST_ONLY lost response")), + ), + other => other, + } + } + } + + fn metadata() -> WorkerMetadata { + WorkerMetadata::new( + "startup", + Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )), + ) + .with_workspace_id("workspace") + } + + fn error() -> WorkerError { + WorkerError::FeatureInstall( + "TEST_ONLY_CREDENTIAL /private/internal/path raw_http_body".into(), + ) + } + + #[test] + fn startup_error_save_reopens_same_cause_without_replacing_primary_or_disclosing_source() { + let dir = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(dir.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let source = error(); + let primary = source.startup_failure_at(StartupStage::Feature).unwrap(); + let retained = persist_startup_failure::<_, RuntimeStartupCause>( + &store, + &expected, + source, + StartupStage::Feature, + &(), + ); + assert_eq!(retained.startup_failure(), Some(&primary)); + assert_eq!( + retained.startup_metadata_save(), + Some(StartupPersistenceOutcome::Persisted) + ); + let reopened = FsWorkerStore::new(dir.path()) + .unwrap() + .read_by_name("startup") + .unwrap() + .unwrap(); + assert!(matches!( + reopened.decode_startup_failure::(), + StartupFailureRead::Present(error_contract::DecodeFailure::Known(_)) + )); + for secret in [ + "TEST_ONLY_CREDENTIAL", + "/private/internal/path", + "raw_http_body", + ] { + assert!( + !serde_json::to_string(&reopened.startup_failure) + .unwrap() + .contains(secret) + ); + } + assert_eq!(reopened.active, original.active); + assert_eq!( + reopened.subjektiv_session_attribution, + original.subjektiv_session_attribution + ); + } + + #[test] + fn startup_error_unknown_save_keeps_primary_and_requires_reconciliation_of_real_commit() { + let dir = tempfile::tempdir().unwrap(); + let store = LostSaveResponse(FsWorkerStore::new(dir.path()).unwrap()); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + let source = error(); + let primary = source.startup_failure_at(StartupStage::Feature).unwrap(); + let retained = persist_startup_failure::<_, RuntimeStartupCause>( + &store, + &expected, + source, + StartupStage::Feature, + &(), + ); + assert_eq!(retained.startup_failure(), Some(&primary)); + assert_eq!( + retained.startup_metadata_save(), + Some(StartupPersistenceOutcome::OutcomeUnknown) + ); + assert!(retained.startup_secondary_failures().contains( + &StartupSecondaryFailure::Persistence { + phase: StartupFailurePhase::MetadataSave, + outcome: StartupPersistenceOutcome::OutcomeUnknown, + } + )); + let reopened = FsWorkerStore::new(dir.path()) + .unwrap() + .read_by_name("startup") + .unwrap() + .unwrap(); + assert!(matches!( + reopened.decode_startup_failure::(), + StartupFailureRead::Present(error_contract::DecodeFailure::Known(_)) + )); + // The original observation must not be retried as a fresh write after an unknown response. + assert!(matches!( + store.compare_and_swap_startup_failure( + &expected, + &StoredStartupFailureAttachment::from_projected( + &primary + .project_for::(&()) + .unwrap() + ) + .unwrap() + ), + StartupFailureSaveResult::Conflict + )); + } + + #[test] + fn startup_error_save_failure_and_conflict_preserve_original_primary() { + for conflict in [true, false] { + let dir = tempfile::tempdir().unwrap(); + let store = FsWorkerStore::new(dir.path()).unwrap(); + let original = metadata(); + store.write(&original).unwrap(); + let expected = StartupFailureExpectation::from_metadata(&original).unwrap(); + if conflict { + store + .set_active( + "startup", + Some(WorkerActiveSegmentRef::pending_segment( + session_store::new_session_id(), + )), + None, + ) + .unwrap(); + } else { + let path = dir.path().join("startup/metadata.json"); + std::fs::remove_file(&path).unwrap(); + std::fs::create_dir(&path).unwrap(); // Real read failure before a save can be attempted. + } + let source = error(); + let primary = source.startup_failure_at(StartupStage::Feature).unwrap(); + let retained = persist_startup_failure::<_, RuntimeStartupCause>( + &store, + &expected, + source, + StartupStage::Feature, + &(), + ); + assert_eq!(retained.startup_failure(), Some(&primary)); + assert_eq!( + retained.startup_metadata_save(), + Some(if conflict { + StartupPersistenceOutcome::Conflict + } else { + StartupPersistenceOutcome::NotSaved + }) + ); + } + } +} diff --git a/crates/worker/src/subjektiv_metadata.rs b/crates/worker/src/subjektiv_metadata.rs index e24b4c926..198bba479 100644 --- a/crates/worker/src/subjektiv_metadata.rs +++ b/crates/worker/src/subjektiv_metadata.rs @@ -3,8 +3,7 @@ use crate::metadata_lifecycle::SessionMetadataLifecycle; use session_store::{ - SessionId, SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, - WorkerMetadataStore, + SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore, }; pub(crate) struct SessionAttributionMetadata; @@ -25,18 +24,13 @@ impl SessionMetadataLifecycle for SessionAttributionMetadata { pub(crate) fn compare_and_swap_attribution( store: &St, - worker_name: &str, - session_id: SessionId, - expected: Option<&SubjektivSessionAttributionState>, + expected: &mut session_store::worker_metadata::StartupFailureExpectation, replacement: SubjektivSessionAttributionState, ) -> Result<(), crate::WorkerError> { let mut matched = false; - store.update_by_name(worker_name, |metadata| { - if metadata.active.as_ref().map(|active| active.session_id) == Some(session_id) - && metadata.subjektiv_session_attribution.as_ref() == expected - && replacement.session_id() == session_id - { - metadata.subjektiv_session_attribution = Some(replacement); + store.update_by_name(&expected.worker_name, |metadata| { + if expected.matches(metadata) && replacement.session_id() == expected.session_id { + metadata.subjektiv_session_attribution = Some(replacement.clone()); matched = true; } })?; @@ -47,6 +41,8 @@ pub(crate) fn compare_and_swap_attribution( .into(), }); } + // Advance only after the exact owner/Session CAS returned confirmed success. + expected.attribution = Some(replacement); Ok(()) } diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 5c8004dd7..707752550 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -9698,6 +9698,48 @@ where &self, subjektiv_attached: bool, lifecycle: SubjektivSessionAttributionLifecycle, + ) -> Result<(), WorkerError> { + self.finalize_subjektiv_session_attribution_observed( + subjektiv_attached, + lifecycle, + &mut None, + ) + } + + /// Retain a failed startup against its constructed Session and Workspace. Only + /// this operation's successful CAS transitions advance the attribution expectation; + /// a reread never adopts a replacement owner or another operation's attribution. + pub fn finalize_subjektiv_session_attribution_retaining_startup_failure( + &self, + subjektiv_attached: bool, + lifecycle: SubjektivSessionAttributionLifecycle, + ) -> Result<(), WorkerError> { + let mut expected = None; + self.finalize_subjektiv_session_attribution_observed( + subjektiv_attached, + lifecycle, + &mut expected, + ) + .map_err(|error| match expected { + Some(expected) => crate::startup_failure::persist_startup_failure::< + _, + protocol::startup_failure::RuntimeStartupCause, + >( + &self.store, + &expected, + error, + protocol::startup_failure::StartupStage::Binding, + &(), + ), + None => error, + }) + } + + fn finalize_subjektiv_session_attribution_observed( + &self, + subjektiv_attached: bool, + lifecycle: SubjektivSessionAttributionLifecycle, + observation: &mut Option, ) -> Result<(), WorkerError> { let _admission = self.segment_state.mutation_authority.enter()?; let worker_name = self.manifest.worker.name.as_str(); @@ -9726,6 +9768,31 @@ where }); } + if metadata.worker_name != worker_name + || metadata.workspace_id.as_deref() != self.workspace_id().map(|id| id.as_str()) + { + return Err(WorkerError::SubjektivSessionAttribution { + state: "owner_mismatch", + message: "Worker metadata does not match constructed Workspace ownership".into(), + }); + } + *observation = + session_store::worker_metadata::StartupFailureExpectation::from_metadata(&metadata); + let mut persist_attribution = + |previous: Option<&SubjektivSessionAttributionState>, + next: SubjektivSessionAttributionState| { + let expected = observation + .as_mut() + .expect("validated active Session has an expectation"); + if expected.attribution.as_ref() != previous { + return Err(WorkerError::SubjektivSessionAttribution { + state: "attribution_conflict", + message: "startup attribution expectation changed".into(), + }); + } + crate::subjektiv_metadata::compare_and_swap_attribution(&self.store, expected, next) + }; + if self .backend_job() .is_some_and(crate::BackendJobExecutionBinding::has_subjektiv_job_grant) @@ -9783,7 +9850,7 @@ where create_if_missing, initial_diagnostic.to_string(), ); - self.persist_subjektiv_session_attribution( + persist_attribution( metadata.subjektiv_session_attribution.as_ref(), pending_attribution.clone(), )?; @@ -9819,7 +9886,7 @@ where message.clone(), ) }; - self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; + persist_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing && rejected { "failed" @@ -9844,7 +9911,7 @@ where let message = bounded_attribution_diagnostic(format!( "Session attribution transport outcome is unknown: {error}" )); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9874,7 +9941,7 @@ where diagnostic: message.clone(), } }; - self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?; + persist_attribution(Some(&pending_attribution), state)?; return Err(WorkerError::SubjektivSessionAttribution { state: if create_if_missing { "failed" @@ -9891,7 +9958,7 @@ where let message = bounded_attribution_diagnostic(format!( "Session attribution response outcome is unknown: {error}" )); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9914,7 +9981,7 @@ where let message = "Session attribution response did not confirm the requested immutable identity" .to_string(); - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), incomplete_subjektiv_attribution_state( session_id, @@ -9927,7 +9994,7 @@ where message, }); } - self.persist_subjektiv_session_attribution( + persist_attribution( Some(&pending_attribution), SubjektivSessionAttributionState::Confirmed { session_id, @@ -9972,20 +10039,6 @@ where Ok(()) } - fn persist_subjektiv_session_attribution( - &self, - expected: Option<&SubjektivSessionAttributionState>, - state: SubjektivSessionAttributionState, - ) -> Result<(), WorkerError> { - crate::subjektiv_metadata::compare_and_swap_attribution( - &self.store, - &self.manifest.worker.name, - self.session_id(), - expected, - state, - ) - } - /// Build an in-process Internal Worker without machine-wide allocation or durable Worker metadata. pub(crate) async fn from_internal_manifest_with_context( manifest: WorkerManifest, @@ -11457,6 +11510,16 @@ fn compact_failure_category(error: &WorkerError) -> CompactFailureCategory { #[derive(Debug, thiserror::Error)] pub enum WorkerError { + /// The domain-projected primary error survives wrappers and metadata-save failures. + /// This attachment is not execution cleanup or lifecycle evidence. + #[error("{source}")] + StartupFailure { + #[source] + source: Box, + failure: protocol::startup_failure::StartupFailureAttachment, + secondary: Vec, + metadata_save: Option, + }, /// A construction or activation step failed after acquiring resources. Retain /// these handles until explicit cleanup succeeds. Consuming constructors drop /// the Worker before returning; in-place activation leaves its owner responsible diff --git a/crates/workspace-server/src/server/tests/restored_parent_control_tests.rs b/crates/workspace-server/src/server/tests/restored_parent_control_tests.rs index fc8a4cf28..22db11e1c 100644 --- a/crates/workspace-server/src/server/tests/restored_parent_control_tests.rs +++ b/crates/workspace-server/src/server/tests/restored_parent_control_tests.rs @@ -66,7 +66,10 @@ impl RuntimeWorkerFactory for CapturingFactory { async fn spawn_controller( &self, mut request: worker_runtime::execution::WorkerExecutionSpawnRequest, - ) -> std::result::Result { + ) -> std::result::Result< + RuntimeWorkerController, + worker_runtime::worker_backend::RuntimeRestoreFailure, + > { let archive = worker_runtime::profile_archive::ProfileSourceArchive::build( worker_runtime::profile_archive::ProfileSourceArchiveInput { id: "profile-source-archive:restored-parent".into(), @@ -93,7 +96,7 @@ impl RuntimeWorkerFactory for CapturingFactory { async fn preflight_restore( &self, request: &worker_runtime::execution::WorkerExecutionRestoreRequest, - ) -> std::result::Result<(), String> { + ) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> { self.inner.preflight_restore(request).await } async fn recover_restore_source( 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 7c2172d49..d008b6667 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 @@ -1002,7 +1002,10 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for RestoreAuthFactory async fn spawn_controller( &self, mut request: worker_runtime::execution::WorkerExecutionSpawnRequest, - ) -> std::result::Result { + ) -> std::result::Result< + worker_runtime::worker_backend::RuntimeWorkerController, + worker_runtime::worker_backend::RuntimeRestoreFailure, + > { if request.request.backend_job.is_some() { return self.inner.spawn_controller(request).await; } @@ -1040,7 +1043,7 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for RestoreAuthFactory async fn preflight_restore( &self, request: &worker_runtime::execution::WorkerExecutionRestoreRequest, - ) -> std::result::Result<(), String> { + ) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> { self.inner.preflight_restore(request).await } diff --git a/crates/workspace-server/src/server_assignment_attention_tests.rs b/crates/workspace-server/src/server_assignment_attention_tests.rs index 4cbaf37de..686255bc3 100644 --- a/crates/workspace-server/src/server_assignment_attention_tests.rs +++ b/crates/workspace-server/src/server_assignment_attention_tests.rs @@ -730,9 +730,13 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for AttentionWorkerFac async fn spawn_controller( &self, request: worker_runtime::execution::WorkerExecutionSpawnRequest, - ) -> std::result::Result { + ) -> std::result::Result< + worker_runtime::worker_backend::RuntimeWorkerController, + worker_runtime::worker_backend::RuntimeRestoreFailure, + > { self.controller(request.worker_ref.worker_id.to_string(), false) .await + .map_err(Into::into) } async fn prepare_restore( &self, diff --git a/crates/workspace-server/src/server_assignment_delivery_tests.rs b/crates/workspace-server/src/server_assignment_delivery_tests.rs index 32d457350..2fab20c84 100644 --- a/crates/workspace-server/src/server_assignment_delivery_tests.rs +++ b/crates/workspace-server/src/server_assignment_delivery_tests.rs @@ -858,14 +858,19 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for DeliveryWorkerFact async fn spawn_controller( &self, request: delivery_execution::WorkerExecutionSpawnRequest, - ) -> std::result::Result { + ) -> std::result::Result< + worker_runtime::worker_backend::RuntimeWorkerController, + worker_runtime::worker_backend::RuntimeRestoreFailure, + > { let worker_id = request.worker_ref.worker_id.to_string(); let manifest = Self::manifest( &worker_id, &request.request.profile, request.config_bundle.as_ref(), )?; - self.controller(worker_id, false, manifest).await + self.controller(worker_id, false, manifest) + .await + .map_err(Into::into) } async fn prepare_restore( &self, diff --git a/crates/workspace-server/src/server_workspace_config_wip_tests.rs b/crates/workspace-server/src/server_workspace_config_wip_tests.rs index d0ba52e17..58ab4d2f8 100644 --- a/crates/workspace-server/src/server_workspace_config_wip_tests.rs +++ b/crates/workspace-server/src/server_workspace_config_wip_tests.rs @@ -494,7 +494,9 @@ async fn workspace_config_apply_model_union_cases_and_composite_persist_evaluate // Subsequent calls use the Client's returned root validator, without reinspection or // model-supplied preconditions. All four variants share the same public contract. fixture - .apply_observed(json!([{"$case":"update","value":{"path":"notes/edit.txt","content":"updated"}}])) + .apply_observed( + json!([{"$case":"update","value":{"path":"notes/edit.txt","content":"updated"}}]), + ) .await; fixture .apply_observed( @@ -541,7 +543,8 @@ async fn workspace_config_apply_model_union_cases_and_composite_persist_evaluate ); assert_eq!( fragment.content, - "{\n common = {\n language = \"TYPED_BATCH\";\n };\n}\n" ); + "{\n common = {\n language = \"TYPED_BATCH\";\n };\n}\n" + ); assert_eq!( state .snapshot @@ -804,10 +807,7 @@ async fn workspace_config_apply_cached_contract_cannot_bypass_readonly_or_revoca let error = config_call(&fixture.wip, "/workspace-config", "apply_changes", json!({"changes":[ {"$case":"create","value":{"path":"notes/denied.txt","content":"denied","content_type":"text"}} ]})).await.unwrap_err(); - assert!( - !error.to_string().contains("outcome unknown"), - "{error}" - ); + assert!(!error.to_string().contains("outcome unknown"), "{error}"); assert_eq!(fixture.state(), before); assert_eq!(fixture.commits(), commits); if !revoke { @@ -845,10 +845,7 @@ async fn workspace_config_apply_known_rejection_differs_from_lost_response_witho let error = config_call(&fixture.wip, "/workspace-config", "apply_changes", json!({"changes":[ {"$case":"create","value":{"path":"notes/unknown.txt","content":"committed once","content_type":"text"}} ]})).await.unwrap_err(); - assert!( - error.to_string().contains("outcome unknown"), - "{error}" - ); + assert!(error.to_string().contains("outcome unknown"), "{error}"); let state = fixture.state(); assert_eq!( state @@ -1165,7 +1162,10 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for ConfigCaptureFacto async fn spawn_controller( &self, mut request: worker_runtime::execution::WorkerExecutionSpawnRequest, - ) -> std::result::Result { + ) -> std::result::Result< + worker_runtime::worker_backend::RuntimeWorkerController, + worker_runtime::worker_backend::RuntimeRestoreFailure, + > { // No LLM request is issued; explicit unauthenticated test model avoids // depending on developer credentials during production bootstrap. let archive = worker_runtime::profile_archive::ProfileSourceArchive::build(worker_runtime::profile_archive::ProfileSourceArchiveInput { @@ -1188,7 +1188,7 @@ impl worker_runtime::worker_backend::RuntimeWorkerFactory for ConfigCaptureFacto async fn preflight_restore( &self, request: &worker_runtime::execution::WorkerExecutionRestoreRequest, - ) -> std::result::Result<(), String> { + ) -> std::result::Result<(), worker_runtime::worker_backend::RuntimeRestoreFailure> { self.inner.preflight_restore(request).await } diff --git a/web/workspace/src/lib/generated/protocol.ts b/web/workspace/src/lib/generated/protocol.ts index 8b252d488..6fb931d91 100644 --- a/web/workspace/src/lib/generated/protocol.ts +++ b/web/workspace/src/lib/generated/protocol.ts @@ -2,6 +2,32 @@ // Do not edit by hand. Rust DTO authority lives in `crates/protocol`. // Large integer fields are JSON numbers on the wire and are emitted as TypeScript `number`. +export type StartupFailureAttachment = string; + +export type StoredStartupFailureAttachment = StartupFailureAttachment; + +export type StartupStage = "preflight" | "metadata" | "reconstruction" | "recovery" | "binding" | "feature" | "activation_effects" | "controller"; + +export type RuntimeStartupReason = "startup_failed" | "segment_empty" | "system_prompt_missing" | "metadata_pending" | "metadata_missing" | "metadata_inactive" | "manifest_missing" | "allocation_failed" | "session_read_failed" | "metadata_read_failed" | "provider_failed" | "attribution_failed" | "feature_failed"; + +export type RuntimeStartupCause = { stage: StartupStage, reason: RuntimeStartupReason, }; + +export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "runtime_reconciliation_record_save" | "runtime_reconciliation_publish" | "metadata_save"; + +export type StartupCleanupFailure = "ownership" | "workdir" | "artifacts" | "allocation" | "feature" | "unsupported"; + +export type StartupPersistenceOutcome = "persisted" | "not_saved" | "conflict" | "outcome_unknown"; + +export type StartupSecondaryFailure = { "kind": "projection_unavailable", phase: StartupFailurePhase, } | { "kind": "cleanup", failure: StartupCleanupFailure, } | { "kind": "persistence", phase: StartupFailurePhase, outcome: StartupPersistenceOutcome, }; + +export type StartupResources = "none" | "worker_startup" | "unproven"; + +export type StartupCleanup = "not_needed" | "required" | "confirmed" | { "failed": StartupCleanupFailure }; + +export type StartupResourceFacts = { resources: StartupResources, cleanup: StartupCleanup, durable_effects_possible: boolean, }; + +export type StartupIoKind = "not_found" | "permission_denied" | "already_exists" | "invalid_input" | "invalid_data" | "interrupted" | "timed_out" | "would_block" | "unexpected_eof" | "other"; + export type AlertLevel = "warn" | "error"; export type AlertPosition = { "kind": "start" } | { "kind": "after_entry", entry_id: string, } | { "kind": "after_tool_call", call_id: string, } | { "kind": "after_in_flight", block_id: number, };