fix(startup): retain postconnection creation facts and safe observations

Cargo.lock

@@ -4334,6 +4334,7 @@ dependencies = [
  "serde",
  "serde_json",
  "session-store",
+ "sha2 0.11.0",
  "tokio",
  "workdir",
 ]

crates/runtime-api/Cargo.toml

@@ -12,6 +12,7 @@ plugin-plan = { path = "../plugin-plan" }
 protocol = { workspace = true, features = ["json-schema"] }
 error-contract = { workspace = true, features = ["json-schema"] }
 schemars.workspace = true
+sha2.workspace = true
 serde_json.workspace = true
 session-store.workspace = true
 serde = { workspace = true, features = ["derive"] }

crates/runtime-api/src/startup.rs

@@ -658,6 +658,22 @@ pub struct StartupCorrelation(
 );
 
 impl StartupCorrelation {
+    /// A shared path-free lookup token for an owner-bound Runtime/Worker pair.
+    /// This is not the create fingerprint, operation key, or replay authority.
+    pub fn for_worker(runtime_id: &str, worker_id: &str) -> Self {
+        use sha2::{Digest, Sha256};
+        let mut hash = Sha256::new();
+        hash.update(runtime_id.as_bytes());
+        hash.update([0]);
+        hash.update(worker_id.as_bytes());
+        let digest = hash
+            .finalize()
+            .iter()
+            .map(|byte| format!("{byte:02x}"))
+            .collect::<String>();
+        Self(format!("worker-{digest}"))
+    }
+
     pub fn new(value: impl Into<String>) -> Result<Self, ContractError> {
         let value = value.into();
         if value.is_empty()

crates/worker-runtime/src/error.rs

@@ -63,6 +63,15 @@ pub enum RuntimeError {
         result: WorkerExecutionResult,
     },
 
+    /// Creation failed after connection; the original error remains internal source
+    /// evidence. Display deliberately never renders it or the startup attachment.
+    #[error("worker creation failed")]
+    WorkerCreateFailed {
+        failure: Box<runtime_api::startup::WorkerCreateFailure>,
+        #[source]
+        source: Box<RuntimeError>,
+    },
+
     #[error("limit {requested} exceeds maximum {max}")]
     LimitTooLarge { requested: usize, max: usize },
 
@@ -149,11 +158,26 @@ pub enum RuntimeError {
 }
 
 impl RuntimeError {
+    /// Only an owning create boundary or a Spawn result can identify a creation
+    /// failure. Independent Input errors retain their ordinary API contract.
+    pub fn create_failure(&self) -> Option<runtime_api::startup::WorkerCreateFailure> {
+        match self {
+            Self::WorkerCreateFailed { failure, .. } => Some((**failure).clone()),
+            Self::WorkerExecutionRejected { result, .. } => {
+                crate::execution::startup_create_failure(result)
+                    .ok()
+                    .flatten()
+            }
+            _ => None,
+        }
+    }
+
     /// 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),
+            Self::WorkerCreateFailed { source, .. } => source.execution_result(),
             _ => None,
         }
     }

crates/worker-runtime/src/http_server.rs

@@ -2347,14 +2347,7 @@ impl RuntimeHttpRestError {
     fn runtime(error: RuntimeError) -> Self {
         let status = status_for_runtime_error(&error);
         let code = code_for_runtime_error(&error);
-        let create_failure = match &error {
+        let create_failure = error.create_failure();
-            RuntimeError::WorkerExecutionRejected { result, .. } => {
-                crate::execution::startup_create_failure(result)
-                    .ok()
-                    .flatten()
-            }
-            _ => None,
-        };
         let startup = match &error {
             RuntimeError::WorkerExecutionRejected { result, .. } => startup_transfer(result),
             _ => None,
@@ -2440,6 +2433,7 @@ fn startup_transfer(
 
 fn status_for_runtime_error(error: &RuntimeError) -> StatusCode {
     match error {
+        RuntimeError::WorkerCreateFailed { .. } => StatusCode::INTERNAL_SERVER_ERROR,
         RuntimeError::WorkerNotFound { .. } | RuntimeError::ConfigBundleMissing { .. } => {
             StatusCode::NOT_FOUND
         }
@@ -2549,6 +2543,7 @@ fn code_for_runtime_error(error: &RuntimeError) -> String {
         RuntimeError::ExecutionBackendUnavailable { .. } => {
             "execution_backend_unavailable".to_string()
         }
+        RuntimeError::WorkerCreateFailed { .. } => "worker_create_failed".to_string(),
         RuntimeError::WorkerExecutionRejected { .. } => "worker_execution_rejected".to_string(),
         RuntimeError::WorkspaceOwnerMismatch { .. } => "workspace_owner_mismatch".to_string(),
         RuntimeError::LimitTooLarge { .. } => "limit_too_large".to_string(),

crates/worker-runtime/src/lib.rs

@@ -31,6 +31,7 @@ mod session_material;
 mod session_public;
 mod session_retention;
 pub mod ssh_host_key_probe;
+mod startup_observation;
 pub mod worker_backend;
 pub mod worker_source;
 pub mod working_directory;

crates/worker-runtime/src/runtime.rs

@@ -56,6 +56,11 @@ use protocol::subscription::{
     SubscriptionWorkerState, SubscriptionWorkerWorkdirAttachment,
 };
 use protocol::{Event, Method};
+use runtime_api::startup::{
+    CompensationOutcome, CreationAcceptance, ExecutionOutcome, FinalizeFacts, FinalizeStage,
+    StartupBoundaryCause, StartupBoundaryReason, StartupCause, StartupCorrelation,
+    StartupOperationFacts, WorkerCreateFailure,
+};
 use server_api::WorkerRestoreState;
 use std::collections::BTreeMap;
 #[cfg(feature = "ws-server")]
@@ -1381,7 +1386,34 @@ impl Runtime {
         let worker_id = request.worker_id;
         let workspace_id = scope.map(|scope| scope.workspace_id.as_str());
         let started_at = std::time::Instant::now();
-        let result = self.create_worker_with_workspace_inner(request, scope);
+        let result = self
+            .create_worker_with_workspace_inner(request, scope)
+            .map_err(|error| {
+                // Use only the host-bound local identity, never received correlation.
+                // A poisoned/unbound Runtime cannot invent correlation authority.
+                let correlation = self.lock().ok().and_then(|state| {
+                    state.runtime_identity.as_deref().map(|runtime_id| {
+                        StartupCorrelation::for_worker(runtime_id, &worker_id.to_string())
+                    })
+                });
+                match (error.create_failure(), correlation) {
+                    (Some(failure), Some(correlation)) => {
+                        let mut operation = failure.operation().clone();
+                        operation.correlation = Some(correlation);
+                        let failure = Box::new(failure.with_operation(operation));
+                        match error {
+                            RuntimeError::WorkerCreateFailed { source, .. } => {
+                                RuntimeError::WorkerCreateFailed { failure, source }
+                            }
+                            source => RuntimeError::WorkerCreateFailed {
+                                failure,
+                                source: Box::new(source),
+                            },
+                        }
+                    }
+                    _ => error,
+                }
+            });
         if let Err(error) = &result {
             write_runtime_worker_create_failure(
                 worker_id,
@@ -1528,13 +1560,13 @@ impl Runtime {
             WorkerExecutionSpawnResult::Rejected(mut result)
             | WorkerExecutionSpawnResult::RolledBack(mut result)
             | WorkerExecutionSpawnResult::Errored(mut result) => {
-                if let Err((error, phase, outcome)) =
+                if let Err((_error, phase, outcome)) =
                     self.rollback_failed_create_with_phase(&worker_ref)
                 {
                     tracing::error!(
                         worker_id = %worker_ref.worker_id,
                         ?phase,
-                        %error,
+                        ?outcome,
                         "failed to roll back Worker creation"
                     );
                     // A later failure must not make a confirmed record deletion
@@ -1601,7 +1633,8 @@ impl Runtime {
                     });
                     tracing::error!(
                         worker_id = %worker_ref.worker_id,
-                        %error,
+                        ?phase,
+                        ?outcome,
                         "failed to persist Worker create reconciliation evidence"
                     );
                 } else {
@@ -1620,80 +1653,119 @@ impl Runtime {
             }
         };
 
-        if let Some(mut initial_input) = {
+        // From Connected onward this owner, not the independent Input API,
-            let state = self.lock()?;
+        // supplies creation context. Later cleanup must not replace the source.
-            state
+        let mut execution = ExecutionOutcome::Unknown;
-                .worker(&worker_ref)?
+        let mut stage = FinalizeStage::InitialInput;
-                .request
+        let mut reason = StartupBoundaryReason::FinalizeFailed;
-                .as_ref()
+        let mut persistence_phase = None;
-                .and_then(|request| request.initial_input.clone())
+        let mut input_result = None;
-        } {
+        let result = (|| {
-            let expected_submission_id = initial_input
+            let initial_input = {
-                .submission_request_id
+                let state = self.lock()?;
-                .clone()
+                state
-                .filter(|request_id| !request_id.trim().is_empty())
+                    .worker(&worker_ref)?
-                .unwrap_or_else(|| Uuid::now_v7().to_string());
+                    .request
-            initial_input.submission_request_id = Some(expected_submission_id.clone());
+                    .as_ref()
-            let dispatch_result = backend.dispatch_input(&worker_ref, initial_input.clone());
+                    .and_then(|request| request.initial_input.clone())
-            if !dispatch_result.is_accepted() {
-                self.cleanup_connected_failed_create(&backend, &worker_ref)?;
-                return Err(RuntimeError::WorkerExecutionRejected {
-                    worker_id: worker_ref.worker_id.clone(),
-                    operation: dispatch_result.operation,
-                    outcome: dispatch_result.outcome,
-                    message: dispatch_result.message_or_default(),
-                    result: dispatch_result,
-                });
-            }
-            let has_durable_acceptance = dispatch_result
-                .submission
-                .as_ref()
-                .is_some_and(|ack| ack.submission_request_id == expected_submission_id);
-            if !has_durable_acceptance {
-                self.cleanup_connected_failed_create(&backend, &worker_ref)?;
-                let result = WorkerExecutionResult::rejected(
-                    WorkerExecutionOperation::Input,
-                    "execution backend accepted initial input without a durable submission acknowledgement",
-                );
-                return Err(RuntimeError::WorkerExecutionRejected {
-                    worker_id: worker_ref.worker_id.clone(),
-                    operation: result.operation,
-                    outcome: result.outcome,
-                    message: result.message_or_default(),
-                    result,
-                });
-            }
-            let detail = match self.commit_created_worker(
-                &worker_ref,
-                initial_worker_state.clone(),
-                workdir_attachments,
-                dispatch_result,
-            ) {
-                Ok(detail) => detail,
-                Err(error) => {
-                    self.cleanup_connected_failed_create(&backend, &worker_ref)?;
-                    return Err(error);
-                }
             };
-            if let Err(error) = self.record_input_observation(&worker_ref, initial_input) {
+            if let Some(mut initial_input) = initial_input {
-                self.cleanup_connected_failed_create(&backend, &worker_ref)?;
+                let expected_submission_id = initial_input
-                return Err(error);
+                    .submission_request_id
-            }
+                    .clone()
-            Ok(detail)
+                    .filter(|request_id| !request_id.trim().is_empty())
-        } else {
+                    .unwrap_or_else(|| Uuid::now_v7().to_string());
-            match self.commit_created_worker(
+                initial_input.submission_request_id = Some(expected_submission_id.clone());
-                &worker_ref,
+                let dispatch_result = backend.dispatch_input(&worker_ref, initial_input.clone());
-                initial_worker_state,
+                let has_durable_acceptance = dispatch_result
-                workdir_attachments,
+                    .submission
-                WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn),
+                    .as_ref()
-            ) {
+                    .is_some_and(|ack| ack.submission_request_id == expected_submission_id);
-                Ok(detail) => Ok(detail),
+                input_result = Some(dispatch_result.clone());
-                Err(error) => {
+                if !dispatch_result.is_accepted() {
-                    self.cleanup_connected_failed_create(&backend, &worker_ref)?;
+                    // Confirmed rejection is failed initial execution. A lost
-                    Err(error)
+                    // acknowledgement does not establish non-execution.
+                    execution = match dispatch_result.outcome {
+                        crate::execution::WorkerExecutionOutcome::Errored
+                            if !has_durable_acceptance =>
+                        {
+                            ExecutionOutcome::Unknown
+                        }
+                        _ => ExecutionOutcome::Errored,
+                    };
+                    return Err(RuntimeError::WorkerExecutionRejected {
+                        worker_id: worker_ref.worker_id,
+                        operation: dispatch_result.operation,
+                        outcome: dispatch_result.outcome,
+                        message: dispatch_result.message_or_default(),
+                        result: dispatch_result,
+                    });
                 }
+                if !has_durable_acceptance {
+                    reason = StartupBoundaryReason::InvalidResponse;
+                    return Err(RuntimeError::WorkerExecutionRejected {
+                        worker_id: worker_ref.worker_id,
+                        operation: dispatch_result.operation,
+                        outcome: dispatch_result.outcome,
+                        message: "initial input has no matching durable acknowledgement".into(),
+                        result: dispatch_result,
+                    });
+                }
+                execution = ExecutionOutcome::Accepted;
+                stage = FinalizeStage::Publication;
+                let detail = self.commit_created_worker(
+                    &worker_ref,
+                    initial_worker_state,
+                    workdir_attachments,
+                    dispatch_result,
+                    &mut persistence_phase,
+                )?;
+                self.record_input_observation(&worker_ref, initial_input)?;
+                Ok(detail)
+            } else {
+                execution = ExecutionOutcome::Accepted;
+                stage = FinalizeStage::Publication;
+                self.commit_created_worker(
+                    &worker_ref,
+                    initial_worker_state,
+                    workdir_attachments,
+                    WorkerExecutionResult::accepted(WorkerExecutionOperation::Spawn),
+                    &mut persistence_phase,
+                )
             }
-        }
+        })();
+        result.map_err(|source| {
+            let mut startup = connected_create_startup(&source, input_result.as_ref(), reason);
+            if let Some(phase) = persistence_phase {
+                push_create_secondary(
+                    &mut startup,
+                    StartupSecondaryFailure::Persistence {
+                        phase,
+                        outcome: create_persistence_outcome(&source),
+                    },
+                );
+            }
+            let compensation =
+                self.cleanup_connected_failed_create(&backend, &worker_ref, &mut startup);
+            let failure = WorkerCreateFailure::new(
+                startup,
+                StartupOperationFacts {
+                    acceptance: CreationAcceptance::Accepted,
+                    execution,
+                    finalize: Some(FinalizeFacts {
+                        stage,
+                        compensation,
+                    }),
+                    backend_compensation: None,
+                    correlation: None,
+                },
+            )
+            .expect("connected create facts have bounded secondary observations");
+            RuntimeError::WorkerCreateFailed {
+                failure: Box::new(failure),
+                source: Box::new(source),
+            }
+        })
     }
 
     /// List Workers known to this Runtime.
@@ -2267,7 +2339,7 @@ impl Runtime {
                 Some(RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::ReconciliationRequired,
                     "worker_restore_canonical_write_unknown",
-                    sanitize_worker_create_failure_message(&error.to_string()),
+                    sanitize_worker_restore_failure_message(&error.to_string()),
                 )),
                 None,
             ));
@@ -2322,7 +2394,7 @@ impl Runtime {
                         Some(RuntimeWorkerRestoreResult::failed(
                             WorkerRestoreState::ReconciliationRequired,
                             "worker_restore_authority_changed",
-                            sanitize_worker_create_failure_message(&error.to_string()),
+                            sanitize_worker_restore_failure_message(&error.to_string()),
                         )),
                         None,
                     ));
@@ -2432,7 +2504,7 @@ impl Runtime {
                             Some(RuntimeWorkerRestoreResult::failed(
                                 WorkerRestoreState::ReconciliationRequired,
                                 "worker_restore_admission_commit_unknown",
-                                sanitize_worker_create_failure_message(&error.to_string()),
+                                sanitize_worker_restore_failure_message(&error.to_string()),
                             )),
                             None,
                         ));
@@ -2665,7 +2737,7 @@ impl Runtime {
                         return Ok(RuntimeWorkerRestoreResult::failed(
                             WorkerRestoreState::ReconciliationRequired,
                             "worker_restore_saved_binding_not_ready",
-                            sanitize_worker_create_failure_message(&message),
+                            sanitize_worker_restore_failure_message(&message),
                         ));
                     }
                     let result = RuntimeWorkerRestoreResult::accepted(detail);
@@ -2813,7 +2885,7 @@ impl Runtime {
                             WorkerRestoreState::Rejected
                         },
                         "worker_restore_preparation_rejected",
-                        sanitize_worker_create_failure_message(&error.to_string()),
+                        sanitize_worker_restore_failure_message(&error.to_string()),
                     ));
                 }
             };
@@ -2922,7 +2994,7 @@ impl Runtime {
                         return Ok(RuntimeWorkerRestoreResult::failed(
                             WorkerRestoreState::ReconciliationRequired,
                             "worker_restore_observation_activation_pending",
-                            sanitize_worker_create_failure_message(&error.to_string()),
+                            sanitize_worker_restore_failure_message(&error.to_string()),
                         ));
                     }
                 }
@@ -2933,7 +3005,7 @@ impl Runtime {
             return Ok(RuntimeWorkerRestoreResult::failed(
                 WorkerRestoreState::Rejected,
                 "worker_restore_preflight_rejected",
-                sanitize_worker_create_failure_message(&result.message_or_default()),
+                sanitize_worker_restore_failure_message(&result.message_or_default()),
             ));
         }
 
@@ -3001,7 +3073,7 @@ impl Runtime {
                             return Ok(RuntimeWorkerRestoreResult::failed(
                                 WorkerRestoreState::ReconciliationRequired,
                                 "worker_restore_activation_pending",
-                                sanitize_worker_create_failure_message(&error),
+                                sanitize_worker_restore_failure_message(&error),
                             ));
                         }
                         #[cfg(feature = "restore-checkpoints")]
@@ -3021,7 +3093,7 @@ impl Runtime {
                             return Ok(RuntimeWorkerRestoreResult::failed(
                                 WorkerRestoreState::ReconciliationRequired,
                                 "worker_restore_observation_activation_pending",
-                                sanitize_worker_create_failure_message(&error.to_string()),
+                                sanitize_worker_restore_failure_message(&error.to_string()),
                             ));
                         }
                         let worker = self.lock()?.worker(worker_ref)?.detail();
@@ -3030,7 +3102,7 @@ impl Runtime {
                             return Ok(RuntimeWorkerRestoreResult::failed(
                                 WorkerRestoreState::ReconciliationRequired,
                                 "worker_restore_receipt_commit_pending",
-                                sanitize_worker_create_failure_message(&error.to_string()),
+                                sanitize_worker_restore_failure_message(&error.to_string()),
                             ));
                         }
                         Ok(result)
@@ -3057,7 +3129,7 @@ impl Runtime {
                                 let receipt = RuntimeWorkerRestoreResult::failed(
                                     WorkerRestoreState::RolledBack,
                                     "worker_restore_decision_not_committed",
-                                    sanitize_worker_create_failure_message(&error.to_string()),
+                                    sanitize_worker_restore_failure_message(&error.to_string()),
                                 );
                                 if self.settle_restore_rollback(worker_ref, &receipt).is_ok() {
                                     return Ok(receipt);
@@ -3084,7 +3156,7 @@ impl Runtime {
                         Ok(RuntimeWorkerRestoreResult::failed(
                             WorkerRestoreState::ReconciliationRequired,
                             "worker_restore_commit_reconciliation_required",
-                            sanitize_worker_create_failure_message(&error.to_string()),
+                            sanitize_worker_restore_failure_message(&error.to_string()),
                         ))
                     }
                 }
@@ -3094,14 +3166,14 @@ impl Runtime {
                     return Ok(RuntimeWorkerRestoreResult::failed(
                         WorkerRestoreState::ReconciliationRequired,
                         "worker_restore_reconciliation_rejected",
-                        sanitize_worker_create_failure_message(&result.message_or_default()),
+                        sanitize_worker_restore_failure_message(&result.message_or_default()),
                     ));
                 }
                 self.discard_restore_candidate_context(worker_ref, &candidate_context)?;
                 let receipt = RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::Rejected,
                     "worker_restore_rejected",
-                    sanitize_worker_create_failure_message(&result.message_or_default()),
+                    sanitize_worker_restore_failure_message(&result.message_or_default()),
                 );
                 if let Err(error) = self.settle_restore_rollback(worker_ref, &receipt) {
                     tracing::error!(
@@ -3121,7 +3193,7 @@ impl Runtime {
                 let receipt = RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::RolledBack,
                     "worker_restore_rolled_back",
-                    sanitize_worker_create_failure_message(&_result.message_or_default()),
+                    sanitize_worker_restore_failure_message(&_result.message_or_default()),
                 );
                 self.discard_restore_candidate_context(worker_ref, &candidate_context)?;
                 #[cfg(feature = "fs-store")]
@@ -3174,7 +3246,7 @@ impl Runtime {
                 Ok(RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::ReconciliationRequired,
                     "worker_restore_reconciliation_required",
-                    sanitize_worker_create_failure_message(&format!(
+                    sanitize_worker_restore_failure_message(&format!(
                         "Worker restore requires reconciliation ({:?}): {}",
                         result.outcome,
                         result.message_or_default(),
@@ -3194,7 +3266,7 @@ impl Runtime {
                 Ok(RuntimeWorkerRestoreResult::failed(
                     WorkerRestoreState::ReconciliationRequired,
                     "worker_restore_backend_outcome_unknown",
-                    sanitize_worker_create_failure_message(&_result.message_or_default()),
+                    sanitize_worker_restore_failure_message(&_result.message_or_default()),
                 ))
             }
         }
@@ -4007,6 +4079,7 @@ impl Runtime {
         initial_worker_state: protocol::WorkerStateSnapshot,
         workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
         result: WorkerExecutionResult,
+        persistence_phase: &mut Option<StartupFailurePhase>,
     ) -> Result<WorkerDetail, RuntimeError> {
         let mut state = self.lock()?;
         let detail = {
@@ -4030,41 +4103,155 @@ impl Runtime {
             worker.workdir_attachments = workdir_attachments;
             worker.detail()
         };
+        *persistence_phase = Some(StartupFailurePhase::RuntimeSnapshotSave);
         state.persist_runtime_snapshot()?;
+        *persistence_phase = Some(StartupFailurePhase::RuntimeReconciliationRecordSave);
         state.persist_worker(&worker_ref.worker_id)?;
+        *persistence_phase = None;
         state.publish_worker_upsert(worker_ref.worker_id)?;
         Ok(detail)
     }
 
+    /// Observe compensation separately from the original create failure. This
+    /// performs the same single stop and cleanup as before; no receipt lookup,
+    /// retry, reexecution or new durable lifecycle operation is introduced here.
     fn cleanup_connected_failed_create(
         &self,
         backend: &WorkerExecutionBackendRef,
         worker_ref: &WorkerRef,
-    ) -> Result<(), RuntimeError> {
+        startup: &mut runtime_api::WorkerStartupFailure,
+    ) -> CompensationOutcome {
+        use protocol::startup_failure::{
+            StartupCleanup, StartupCleanupFailure, StartupResourceFacts, StartupResources,
+        };
         let operation_id = WorkerLifecycleOperationId::new();
         let stop_result = backend.stop_worker_operation(WorkerExecutionStopRequest {
             operation_id,
             worker_ref: worker_ref.clone(),
         });
-        if stop_result.is_accepted()
+        for secondary in &stop_result.secondary_failures {
+            push_create_secondary(startup, secondary.clone());
+        }
+        if startup.io_kind.is_none() {
+            startup.io_kind = stop_result.startup_io_kind;
+        }
+        // Preserve the original owner's resource facts, if present. In their
+        // absence, connection establishes startup resources but not their release.
+        let had_resources = startup.resources.is_some() || stop_result.startup_resources.is_some();
+        startup
+            .resources
+            .get_or_insert(
+                stop_result
+                    .startup_resources
+                    .unwrap_or(StartupResourceFacts {
+                        resources: StartupResources::WorkerStartup,
+                        cleanup: StartupCleanup::Required,
+                        durable_effects_possible: true,
+                    }),
+            );
+        if let Some(outcome) = stop_result.startup_failure_persistence {
+            push_create_secondary(
+                startup,
+                StartupSecondaryFailure::Persistence {
+                    phase: StartupFailurePhase::MetadataSave,
+                    outcome,
+                },
+            );
+        }
+        let terminated = stop_result.is_accepted()
             && stop_result.termination.as_ref().is_some_and(|termination| {
                 termination.confirms_scope_shutdown(worker_ref, operation_id)
-            })
+            });
-        {
+        if terminated {
-            return self.rollback_failed_create(worker_ref);
+            if !had_resources {
+                // Confirms scope shutdown only, not removal of Workdir contents
+                // or undoing already accepted initial-input effects.
+                startup.resources.as_mut().unwrap().cleanup = StartupCleanup::Confirmed;
+            }
+            match self.rollback_failed_create_with_phase(worker_ref) {
+                Ok(()) => CompensationOutcome::Completed,
+                Err((error, phase, outcome)) => {
+                    if matches!(
+                        phase,
+                        StartupFailurePhase::RuntimeSnapshotSave
+                            | StartupFailurePhase::RuntimeRemovalPublish
+                    ) {
+                        push_create_secondary(
+                            startup,
+                            StartupSecondaryFailure::Persistence {
+                                phase: StartupFailurePhase::RuntimeRecordDelete,
+                                outcome: StartupPersistenceOutcome::Persisted,
+                            },
+                        );
+                    }
+                    if phase == StartupFailurePhase::RuntimeRemovalPublish {
+                        push_create_secondary(
+                            startup,
+                            StartupSecondaryFailure::Persistence {
+                                phase: StartupFailurePhase::RuntimeSnapshotSave,
+                                outcome: StartupPersistenceOutcome::Persisted,
+                            },
+                        );
+                    }
+                    record_create_persistence_failure(startup, &error, phase, outcome);
+                    if outcome == StartupPersistenceOutcome::OutcomeUnknown {
+                        CompensationOutcome::Unknown
+                    } else {
+                        CompensationOutcome::Residual
+                    }
+                }
+            }
+        } else {
+            let failed = !stop_result.is_accepted()
+                && stop_result.outcome != crate::execution::WorkerExecutionOutcome::Errored;
+            let feature_failed = stop_result.termination.as_ref().is_some_and(|receipt| {
+                receipt.worker_ref == *worker_ref
+                    && receipt.stop_operation_id == operation_id
+                    && !receipt.feature_failures.is_empty()
+            });
+            if failed || feature_failed {
+                let failure = if feature_failed {
+                    StartupCleanupFailure::Feature
+                } else if stop_result.outcome
+                    == crate::execution::WorkerExecutionOutcome::Unsupported
+                {
+                    StartupCleanupFailure::Unsupported
+                } else {
+                    StartupCleanupFailure::Ownership
+                };
+                push_create_secondary(startup, StartupSecondaryFailure::Cleanup { failure });
+                if !had_resources {
+                    startup.resources.as_mut().unwrap().cleanup = StartupCleanup::Failed(failure);
+                }
+            }
+            let mut phase = StartupFailurePhase::RuntimeSnapshotSave;
+            let retained = (|| {
+                let mut state = self.lock()?;
+                let record = state.worker_mut(worker_ref)?;
+                record.worker_state = stop_result.worker_state.clone();
+                state.persist_runtime_snapshot()?;
+                phase = StartupFailurePhase::RuntimeReconciliationRecordSave;
+                state.persist_worker(&worker_ref.worker_id)
+            })();
+            match retained {
+                Ok(()) => push_create_secondary(
+                    startup,
+                    StartupSecondaryFailure::Persistence {
+                        phase: StartupFailurePhase::RuntimeReconciliationRecordSave,
+                        outcome: StartupPersistenceOutcome::Persisted,
+                    },
+                ),
+                Err(error) => {
+                    let outcome = create_persistence_outcome(&error);
+                    record_create_persistence_failure(startup, &error, phase, outcome);
+                }
+            }
+            if failed || feature_failed {
+                CompensationOutcome::Residual
+            } else {
+                CompensationOutcome::Unknown
+            }
         }
-        let mut state = self.lock()?;
-        let record = state.worker_mut(worker_ref)?;
-
-        record.worker_state = stop_result.worker_state.clone();
-        state.persist_runtime_snapshot()?;
-        state.persist_worker(&worker_ref.worker_id)?;
-        Ok(())
-    }
-
-    fn rollback_failed_create(&self, worker_ref: &WorkerRef) -> Result<(), RuntimeError> {
-        self.rollback_failed_create_with_phase(worker_ref)
-            .map_err(|(error, _, _)| error)
     }
 
     // The legacy caller still receives its RuntimeError. Spawn failures instead
@@ -5110,7 +5297,7 @@ impl Runtime {
         let mut state = self.lock()?;
         let diagnostic_id = state.next_diagnostic_id;
         state.next_diagnostic_id = state.next_diagnostic_id.saturating_add(1);
-        let message = sanitize_worker_create_failure_message(
+        let message = sanitize_worker_restore_failure_message(
             result
                 .message
                 .as_deref()
@@ -7148,7 +7335,7 @@ impl RuntimeState {
         result: &WorkerExecutionResult,
         receipt: &RuntimeWorkerRestoreResult,
     ) -> Result<(), RuntimeError> {
-        let message = sanitize_worker_create_failure_message(&result.message_or_default());
+        let message = sanitize_worker_restore_failure_message(&result.message_or_default());
         let diagnostic_id = self.next_diagnostic_id;
         let workspace_id = self
             .workers
@@ -8381,25 +8568,90 @@ fn validate_create_worker_request(request: &CreateWorkerRequest) -> Result<(), R
     Ok(())
 }
 
-fn worker_execution_failure_code(result: &WorkerExecutionResult) -> &'static str {
+/// Keep the original result's typed facts even when publication is the failing
-    let message = result.message.as_deref().unwrap_or_default();
+/// boundary. No message/path parsing is used to choose the fallback cause.
-    if message.contains("timed out waiting for durable Worker Submit acceptance") {
+fn connected_create_startup(
-        "input_commit_timeout"
+    source: &RuntimeError,
-    } else if message.contains("worker adapter task did not complete within") {
+    result: Option<&WorkerExecutionResult>,
-        "adapter_task_timeout"
+    reason: StartupBoundaryReason,
-    } else if message.contains("failed to send Worker method") {
+) -> runtime_api::WorkerStartupFailure {
-        "worker_method_send_failed"
+    let fallback = WorkerCreateFailure::from_cause(
-    } else if message.contains("worker rejected Submit") {
+        StartupCause::Boundary(StartupBoundaryCause { reason }),
-        "worker_submit_rejected"
+        StartupOperationFacts {
-    } else if message.contains("before durable acceptance") {
+            acceptance: CreationAcceptance::Accepted,
-        "worker_failed_before_input_commit"
+            execution: ExecutionOutcome::Unknown,
-    } else if message.contains("event stream closed") {
+            finalize: None,
-        "worker_event_stream_closed"
+            backend_compensation: None,
-    } else {
+            correlation: None,
-        "worker_execution_failed"
+        },
+    )
+    .expect("closed startup boundary cause is valid");
+    let (mut startup, _) = fallback.into_parts();
+    if let Some(result) = result {
+        if let Some(primary) = &result.startup_failure {
+            startup.primary = primary.clone();
+        }
+        for secondary in &result.secondary_failures {
+            push_create_secondary(&mut startup, secondary.clone());
+        }
+        startup.resources = result.startup_resources;
+        startup.persistence = result.startup_failure_persistence;
+        startup.io_kind = result.startup_io_kind;
+    }
+    if startup.io_kind.is_none() {
+        startup.io_kind = create_io_kind(source);
+    }
+    startup
+}
+
+fn create_io_kind(error: &RuntimeError) -> Option<protocol::startup_failure::StartupIoKind> {
+    match error {
+        RuntimeError::StoreIo { source, .. }
+        | RuntimeError::StoreCommitOutcomeUnknown { source, .. } => Some(source.kind().into()),
+        _ => None,
     }
 }
 
+fn create_persistence_outcome(error: &RuntimeError) -> StartupPersistenceOutcome {
+    if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. }) {
+        StartupPersistenceOutcome::OutcomeUnknown
+    } else {
+        StartupPersistenceOutcome::NotSaved
+    }
+}
+
+fn push_create_secondary(
+    startup: &mut runtime_api::WorkerStartupFailure,
+    secondary: StartupSecondaryFailure,
+) {
+    // The create DTO has the core diagnostic bound; this is an observation,
+    // not another durable ledger. The original error remains boxed as source.
+    if startup.secondary.len() < error_contract::MAX_DIAGNOSTICS {
+        startup.secondary.push(secondary);
+    }
+}
+
+fn record_create_persistence_failure(
+    startup: &mut runtime_api::WorkerStartupFailure,
+    error: &RuntimeError,
+    phase: StartupFailurePhase,
+    outcome: StartupPersistenceOutcome,
+) {
+    push_create_secondary(
+        startup,
+        StartupSecondaryFailure::Persistence { phase, outcome },
+    );
+    if startup.io_kind.is_none() {
+        startup.io_kind = create_io_kind(error);
+    }
+    tracing::error!(
+        ?phase,
+        ?outcome,
+        "Worker create compensation persistence failed"
+    );
+}
+
+#[cfg(test)]
 fn runtime_worker_create_failure_fields(
     error: &RuntimeError,
 ) -> (&'static str, Option<String>, Option<String>) {
@@ -8425,6 +8677,7 @@ fn runtime_worker_create_failure_fields(
         RuntimeError::ExecutionBackendUnavailable { .. } => {
             ("execution_backend_unavailable", None, None)
         }
+        RuntimeError::WorkerCreateFailed { .. } => ("worker_create_failed", None, None),
         RuntimeError::WorkerExecutionRejected {
             operation, outcome, ..
         } => (
@@ -8454,9 +8707,11 @@ fn runtime_worker_create_failure_fields(
     }
 }
 
-const WORKER_CREATE_FAILURE_MESSAGE_LIMIT: usize = 512;
+// Legacy Restore diagnostic presentation only. This is not a create failure
+// projector or telemetry allowlist; creation uses startup_observation instead.
+const WORKER_RESTORE_FAILURE_MESSAGE_LIMIT: usize = 512;
 
-fn sanitize_worker_create_failure_message(message: &str) -> String {
+fn sanitize_worker_restore_failure_message(message: &str) -> String {
     let mut redact_next = false;
     let detail = message
         .split_whitespace()
@@ -8496,10 +8751,10 @@ fn sanitize_worker_create_failure_message(message: &str) -> String {
         })
         .collect::<Vec<_>>()
         .join(" ");
-    let truncated = detail.chars().count() > WORKER_CREATE_FAILURE_MESSAGE_LIMIT;
+    let truncated = detail.chars().count() > WORKER_RESTORE_FAILURE_MESSAGE_LIMIT;
     let mut detail = detail
         .chars()
-        .take(WORKER_CREATE_FAILURE_MESSAGE_LIMIT)
+        .take(WORKER_RESTORE_FAILURE_MESSAGE_LIMIT)
         .collect::<String>();
     if truncated {
         detail.push('…');
@@ -8507,43 +8762,13 @@ fn sanitize_worker_create_failure_message(message: &str) -> String {
     detail
 }
 
-fn runtime_worker_create_failure_message(error: &RuntimeError) -> Option<String> {
-    match error {
-        RuntimeError::WorkerExecutionRejected { message, .. } => {
-            Some(sanitize_worker_create_failure_message(message))
-        }
-        _ => None,
-    }
-}
-
 fn write_runtime_worker_create_failure(
     worker_id: WorkerId,
     workspace_id: Option<&str>,
     error: &RuntimeError,
     elapsed: std::time::Duration,
 ) {
-    let (error_kind, operation, outcome) = runtime_worker_create_failure_fields(error);
+    crate::startup_observation::write_create_failure(worker_id, workspace_id, error, elapsed);
-    let execution_failure_code = match error {
-        RuntimeError::WorkerExecutionRejected { result, .. } => {
-            worker_execution_failure_code(result)
-        }
-        _ => "",
-    };
-    let execution_failure_message = runtime_worker_create_failure_message(error);
-    tracing::error!(
-        target: "yoi::worker_create",
-        event = "worker_create_failed",
-        component = "runtime",
-        workspace_id = workspace_id.unwrap_or(""),
-        worker_id = %worker_id,
-        error_kind,
-        operation = operation.as_deref().unwrap_or(""),
-        outcome = outcome.as_deref().unwrap_or(""),
-        execution_failure_code,
-        execution_failure_message = execution_failure_message.as_deref().unwrap_or(""),
-        duration_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
-        "Worker creation failed"
-    );
 }
 
 fn validate_create_workspace_scope(
@@ -8692,6 +8917,11 @@ mod tests {
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::sync::{Arc, Barrier, Condvar, Mutex};
 
+    mod create_failure_tests {
+        use super::*;
+        include!("runtime/create_failure_tests.rs");
+    }
+
     #[cfg(feature = "fs-store")]
     mod create_rollback_tests {
         //! Connect the real create/rollback path to FsRuntimeStore and reopen its disk state.
@@ -8911,57 +9141,6 @@ mod tests {
         }
     }
 
-    #[test]
-    fn worker_create_failure_fields_only_include_sanitized_execution_detail() {
-        let invalid_request = RuntimeError::InvalidRequest("private-token".to_string());
-        let (error_kind, operation, outcome) =
-            runtime_worker_create_failure_fields(&invalid_request);
-        assert_eq!(error_kind, "invalid_request");
-        assert_eq!(operation, None);
-        assert_eq!(outcome, None);
-        assert_eq!(
-            runtime_worker_create_failure_message(&invalid_request),
-            None
-        );
-
-        let result = WorkerExecutionResult::errored(
-            WorkerExecutionOperation::Spawn,
-            "failed to spawn controller in /private/path: File exists; token super-secret",
-        );
-        let error = RuntimeError::WorkerExecutionRejected {
-            worker_id: WorkerId::now_v7(),
-            operation: result.operation,
-            outcome: result.outcome,
-            message: result.message_or_default(),
-            result,
-        };
-        assert_eq!(
-            runtime_worker_create_failure_message(&error).as_deref(),
-            Some("failed to spawn controller in [redacted] File exists; token [redacted]")
-        );
-
-        assert_eq!(
-            sanitize_worker_create_failure_message("backend private-token failed"),
-            "backend [redacted] failed"
-        );
-
-        let truncated = sanitize_worker_create_failure_message(&"x".repeat(513));
-        assert_eq!(truncated.chars().count(), 513);
-        assert!(truncated.ends_with('…'));
-    }
-
-    #[test]
-    fn worker_create_failure_code_classifies_durable_acceptance_timeout() {
-        let timeout = WorkerExecutionResult::errored(
-            WorkerExecutionOperation::Input,
-            "timed out waiting for durable Worker Submit acceptance; private-token",
-        );
-        assert_eq!(
-            worker_execution_failure_code(&timeout),
-            "input_commit_timeout"
-        );
-    }
-
     fn test_command() -> protocol::WorkerCommandEnvelope {
         protocol::WorkerCommandEnvelope { command_id: 1 }
     }
@@ -9950,7 +10129,6 @@ mod tests {
         recovery_absent: AtomicBool,
         recovery_checks: std::sync::atomic::AtomicUsize,
         spawn_result: Mutex<Option<WorkerExecutionSpawnResult>>,
-        #[cfg(feature = "fs-store")]
         before_spawn_result: Mutex<Option<Box<dyn FnOnce() + Send>>>,
         job_bindings: Mutex<Vec<Option<crate::catalog::BackendJobExecutionBinding>>>,
         dispatch_result: Mutex<Option<WorkerExecutionResult>>,
@@ -10197,7 +10375,6 @@ 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();
             }
@@ -15503,13 +15680,7 @@ mod tests {
 
         let error = runtime.create_worker(request).unwrap_err();
 
-        assert!(matches!(
+        create_failure_tests::assert_ack_failure(&error);
-            error,
-            RuntimeError::WorkerExecutionRejected {
-                outcome: crate::execution::WorkerExecutionOutcome::Rejected,
-                ..
-            }
-        ));
         assert!(runtime.list_workers().unwrap().is_empty());
     }
 
@@ -15524,13 +15695,7 @@ mod tests {
 
         let error = runtime.create_worker(request).unwrap_err();
 
-        assert!(matches!(
+        create_failure_tests::assert_ack_failure(&error);
-            error,
-            RuntimeError::WorkerExecutionRejected {
-                outcome: crate::execution::WorkerExecutionOutcome::Rejected,
-                ..
-            }
-        ));
         assert!(runtime.list_workers().unwrap().is_empty());
     }
 
@@ -17160,7 +17325,7 @@ mod tests {
         );
         assert!(!diagnostic.message.contains("credential-value"));
         assert!(!diagnostic.message.contains("/private/path"));
-        assert!(diagnostic.message.chars().count() <= WORKER_CREATE_FAILURE_MESSAGE_LIMIT + 1);
+        assert!(diagnostic.message.chars().count() <= WORKER_RESTORE_FAILURE_MESSAGE_LIMIT + 1);
         let before = std::fs::read(
             dir.path()
                 .join("runtime/workers")
@@ -17346,7 +17511,7 @@ mod tests {
             assert!(!message.contains("/private"), "{message}");
             assert!(!message.contains("restore-token"), "{message}");
             assert!(!message.contains("hidden-value"), "{message}");
-            assert!(message.chars().count() <= WORKER_CREATE_FAILURE_MESSAGE_LIMIT + 1);
+            assert!(message.chars().count() <= WORKER_RESTORE_FAILURE_MESSAGE_LIMIT + 1);
             assert!(
                 runtime
                     .lock()

crates/worker-runtime/src/runtime/create_failure_tests.rs

@@ -0,0 +1,626 @@
+// Registered by runtime.rs's normal library test module. Execution responses are
+// fixtures; creation, persistence, projection and compensation use the real owner.
+use protocol::startup_failure::{
+    RuntimeStartupCause, RuntimeStartupReason, StartupCleanup, StartupCleanupFailure,
+    StartupIoKind, StartupResourceFacts, StartupResources, StartupStage,
+};
+
+fn boundary_reason(failure: &WorkerCreateFailure) -> StartupBoundaryReason {
+    let projected = failure
+        .startup()
+        .primary
+        .project_for::<error_contract::InternalTransfer, StartupCause>(&())
+        .unwrap();
+    let StartupCause::Boundary(cause) = projected.details() else {
+        panic!("expected a closed boundary cause");
+    };
+    cause.reason
+}
+
+pub(super) fn assert_ack_failure(error: &RuntimeError) {
+    assert!(matches!(error, RuntimeError::WorkerCreateFailed { .. }));
+    let failure = error.create_failure().unwrap();
+    assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+    assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown);
+    assert_eq!(
+        failure.operation().finalize,
+        Some(FinalizeFacts {
+            stage: FinalizeStage::InitialInput,
+            compensation: CompensationOutcome::Completed,
+        })
+    );
+    assert_eq!(
+        boundary_reason(&failure),
+        StartupBoundaryReason::InvalidResponse
+    );
+    // The missing/foreign receipt is not rewritten into a confirmed rejection.
+    assert!(error.execution_result().unwrap().is_accepted());
+    assert_eq!(error.to_string(), "worker creation failed");
+}
+
+fn rejected_input() -> WorkerExecutionResult {
+    let attachment = worker::startup_failure::runtime_startup_attachment(RuntimeStartupCause {
+        stage: StartupStage::Controller,
+        reason: RuntimeStartupReason::StartupFailed,
+    })
+    .unwrap();
+    let mut result = WorkerExecutionResult::rejected(
+        WorkerExecutionOperation::Input,
+        "TEST_ONLY_SECRET original rejection /private/owner/path",
+    )
+    .with_startup_failure(attachment);
+    result
+        .secondary_failures
+        .push(StartupSecondaryFailure::Cleanup {
+            failure: StartupCleanupFailure::Workdir,
+        });
+    result.startup_resources = Some(StartupResourceFacts {
+        resources: StartupResources::Unproven,
+        cleanup: StartupCleanup::Failed(StartupCleanupFailure::Workdir),
+        durable_effects_possible: true,
+    });
+    result.startup_failure_persistence = Some(StartupPersistenceOutcome::NotSaved);
+    result.startup_io_kind = Some(StartupIoKind::PermissionDenied);
+    result
+}
+
+#[derive(Clone, Copy, Debug)]
+enum DispatchFault {
+    DroppedAck,
+    MismatchedAck,
+    ConfirmedRejection,
+    LostResponse,
+}
+
+#[derive(Clone, Copy, Debug)]
+enum CleanupFault {
+    None,
+    MissingTermination,
+    Rejected,
+    LostResponse,
+    MismatchedTermination,
+}
+
+#[test]
+fn connected_create_dispatch_and_cleanup_fault_matrix() {
+    for dispatch in [
+        DispatchFault::DroppedAck,
+        DispatchFault::MismatchedAck,
+        DispatchFault::ConfirmedRejection,
+        DispatchFault::LostResponse,
+    ] {
+        for cleanup in [
+            CleanupFault::None,
+            CleanupFault::MissingTermination,
+            CleanupFault::Rejected,
+            CleanupFault::LostResponse,
+            CleanupFault::MismatchedTermination,
+        ] {
+            let (runtime, backend) = runtime_and_backend();
+            let original = match dispatch {
+                DispatchFault::DroppedAck => {
+                    WorkerExecutionResult::accepted(WorkerExecutionOperation::Input)
+                }
+                DispatchFault::MismatchedAck => {
+                    backend.preserve_submission_acknowledgement_id();
+                    WorkerExecutionResult::accepted_submission(
+                        WorkerExecutionOperation::Input,
+                        "foreign-request",
+                        "foreign-submission",
+                        protocol::SubmissionDisposition::Started,
+                    )
+                }
+                DispatchFault::ConfirmedRejection => rejected_input(),
+                DispatchFault::LostResponse => WorkerExecutionResult::errored(
+                    WorkerExecutionOperation::Input,
+                    "response lost; TEST_ONLY_SECRET",
+                ),
+            };
+            backend.set_dispatch_result(original.clone());
+            let compensation = match cleanup {
+                CleanupFault::None => CompensationOutcome::Completed,
+                CleanupFault::MissingTermination => {
+                    backend.set_stop_result(WorkerExecutionResult::accepted(
+                        WorkerExecutionOperation::Stop,
+                    ));
+                    CompensationOutcome::Unknown
+                }
+                CleanupFault::Rejected => {
+                    backend.set_stop_result(WorkerExecutionResult::rejected(
+                        WorkerExecutionOperation::Stop,
+                        "TEST_ONLY_SECRET stop rejected",
+                    ));
+                    CompensationOutcome::Residual
+                }
+                CleanupFault::LostResponse => {
+                    backend.set_stop_result(WorkerExecutionResult::errored(
+                        WorkerExecutionOperation::Stop,
+                        "TEST_ONLY_SECRET stop response lost",
+                    ));
+                    CompensationOutcome::Unknown
+                }
+                CleanupFault::MismatchedTermination => {
+                    let mut stop = WorkerExecutionResult::accepted(WorkerExecutionOperation::Stop);
+                    stop.termination = Some(crate::execution::WorkerExecutionTermination {
+                        scope: crate::execution::WorkerExecutionTerminationScope::Controller,
+                        worker_ref: WorkerRef::new(WorkerId::now_v7()),
+                        stop_operation_id: WorkerLifecycleOperationId::new(),
+                        execution_id: Uuid::now_v7(),
+                        shutdown_command_id: None,
+                        submissions: Vec::new(),
+                        feature_failures: Vec::new(),
+                    });
+                    backend.set_stop_result(stop);
+                    CompensationOutcome::Unknown
+                }
+            };
+            let mut request = task_request("connected create fault matrix");
+            request.initial_input = Some(WorkerInput::user("start once"));
+            let error = runtime.create_worker(request).unwrap_err();
+            let failure = error.create_failure().unwrap();
+            assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+            assert_eq!(
+                failure.operation().execution,
+                if matches!(dispatch, DispatchFault::ConfirmedRejection) {
+                    ExecutionOutcome::Errored
+                } else {
+                    ExecutionOutcome::Unknown
+                },
+                "{dispatch:?}/{cleanup:?}"
+            );
+            assert_eq!(
+                failure.operation().finalize,
+                Some(FinalizeFacts {
+                    stage: FinalizeStage::InitialInput,
+                    compensation,
+                })
+            );
+            assert_eq!(failure.operation().backend_compensation, None);
+            assert_eq!(failure.operation().correlation, None);
+            assert_eq!(error.execution_result(), Some(&original));
+            assert_eq!(error.to_string(), "worker creation failed");
+            if let Some(primary) = original.startup_failure.as_ref() {
+                assert_eq!(&failure.startup().primary, primary);
+                assert!(
+                    failure
+                        .startup()
+                        .secondary
+                        .starts_with(&original.secondary_failures)
+                );
+                assert_eq!(failure.startup().resources, original.startup_resources);
+                assert_eq!(
+                    failure.startup().persistence,
+                    original.startup_failure_persistence
+                );
+                assert_eq!(failure.startup().io_kind, original.startup_io_kind);
+                assert!(
+                    std::error::Error::source(&error)
+                        .unwrap()
+                        .to_string()
+                        .contains("TEST_ONLY_SECRET")
+                );
+            } else {
+                assert_eq!(
+                    boundary_reason(&failure),
+                    if matches!(dispatch, DispatchFault::LostResponse) {
+                        StartupBoundaryReason::FinalizeFailed
+                    } else {
+                        StartupBoundaryReason::InvalidResponse
+                    }
+                );
+                let resources = failure.startup().resources.unwrap();
+                assert!(resources.durable_effects_possible);
+                assert_eq!(
+                    resources.cleanup,
+                    match cleanup {
+                        CleanupFault::None => StartupCleanup::Confirmed,
+                        CleanupFault::Rejected =>
+                            StartupCleanup::Failed(StartupCleanupFailure::Ownership),
+                        _ => StartupCleanup::Required,
+                    }
+                );
+            }
+            assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+            assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1);
+            assert!(
+                backend
+                    .submission_receipt_queries
+                    .lock()
+                    .unwrap()
+                    .is_empty()
+            );
+            assert_eq!(
+                runtime.list_workers().unwrap().is_empty(),
+                matches!(cleanup, CleanupFault::None)
+            );
+        }
+    }
+}
+
+#[test]
+fn independent_input_error_is_not_a_create_failure() {
+    let (runtime, backend) = runtime_and_backend();
+    let worker = runtime
+        .create_worker(task_request("independent input"))
+        .unwrap();
+    let original = rejected_input();
+    backend.set_dispatch_result(original.clone());
+    let error = runtime
+        .send_input(&worker.worker_ref, WorkerInput::user("ordinary input"))
+        .unwrap_err();
+    assert!(matches!(
+        &error,
+        RuntimeError::WorkerExecutionRejected { .. }
+    ));
+    assert_eq!(error.create_failure(), None);
+    assert_eq!(error.execution_result(), Some(&original));
+    assert_eq!(*backend.stop_count.lock().unwrap(), 0);
+}
+
+#[test]
+fn connected_create_initial_input_read_failure_keeps_create_context() {
+    let (runtime, backend) = runtime_and_backend();
+    let mut request = task_request("initial input read failure");
+    request.initial_input = Some(WorkerInput::user("never dispatched"));
+    let worker_id = request.worker_id;
+    *backend.before_spawn_result.lock().unwrap() = Some(Box::new({
+        let runtime = runtime.clone();
+        move || {
+            runtime.lock().unwrap().workers.remove(&worker_id);
+        }
+    }));
+    let error = runtime.create_worker(request).unwrap_err();
+    let RuntimeError::WorkerCreateFailed { failure, source } = error else {
+        panic!("post-Connected read failure must remain a create failure");
+    };
+    assert!(matches!(*source, RuntimeError::WorkerNotFound { .. }));
+    assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+    assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown);
+    assert_eq!(
+        failure.operation().finalize.unwrap().stage,
+        FinalizeStage::InitialInput
+    );
+    assert_eq!(
+        boundary_reason(&failure),
+        StartupBoundaryReason::FinalizeFailed
+    );
+    assert!(backend.dispatched_inputs.lock().unwrap().is_empty());
+    assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+}
+
+#[test]
+fn connected_create_publication_failure_preserves_known_input_acceptance() {
+    let (runtime, backend) = runtime_and_backend();
+    let mut request = task_request("publication fails after known acceptance");
+    request.initial_input = Some(WorkerInput::user("accepted once"));
+    let worker_ref = WorkerRef::new(request.worker_id);
+    // Substitute only Connected data. The invalid backend Workdir identifier
+    // reaches the real subscription projection after the owner commit.
+    backend
+        .contexts
+        .lock()
+        .unwrap()
+        .insert(request.worker_id, runtime.execution_context(worker_ref));
+    *backend.spawn_result.lock().unwrap() = Some(WorkerExecutionSpawnResult::Connected {
+        worker_state: protocol::WorkerStatus::Idle.into(),
+        workdir_attachments: vec![WorkingDirectoryAttachmentStatus {
+            alias: workdir::WorkdirAttachmentAlias::new("checkout").unwrap(),
+            working_directory: serde_json::from_value(serde_json::json!({
+                "summary": {
+                    "working_directory_id": "",
+                    "repository_id": "repo",
+                    "materializer_kind": "runtime_git_clone",
+                    "status": "active"
+                }
+            }))
+            .unwrap(),
+        }],
+    });
+    let error = runtime.create_worker(request).unwrap_err();
+    let failure = error.create_failure().unwrap();
+    assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+    assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted);
+    assert_eq!(
+        failure.operation().finalize,
+        Some(FinalizeFacts {
+            stage: FinalizeStage::Publication,
+            compensation: CompensationOutcome::Completed,
+        })
+    );
+    assert_eq!(
+        boundary_reason(&failure),
+        StartupBoundaryReason::FinalizeFailed
+    );
+    let RuntimeError::WorkerCreateFailed { source, .. } = error else {
+        unreachable!()
+    };
+    assert!(matches!(*source, RuntimeError::InvalidRequest(_)));
+    assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1);
+    assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+}
+
+#[test]
+fn create_correlation_uses_bound_runtime_identity_for_connected_and_spawn_failures() {
+    for connected in [false, true] {
+        let (runtime, backend) = runtime_and_backend();
+        runtime.bind_runtime_identity("runtime-identity").unwrap();
+        let mut request = task_request("safe correlation");
+        let worker_id = request.worker_id;
+        if connected {
+            request.initial_input = Some(WorkerInput::user("start once"));
+            backend.set_dispatch_result(rejected_input());
+        } else {
+            let mut result = rejected_input();
+            result.operation = WorkerExecutionOperation::Spawn;
+            *backend.spawn_result.lock().unwrap() =
+                Some(WorkerExecutionSpawnResult::Rejected(result));
+        }
+        let error = runtime.create_worker(request).unwrap_err();
+        assert_eq!(
+            error.create_failure().unwrap().operation().correlation,
+            Some(StartupCorrelation::for_worker(
+                "runtime-identity",
+                &worker_id.to_string()
+            ))
+        );
+        assert_eq!(error.to_string(), "worker creation failed");
+    }
+}
+
+#[cfg(feature = "fs-store")]
+#[test]
+fn connected_create_commit_write_faults_preserve_acceptance_and_io_facts() {
+    use crate::fs_store::AtomicWriteFault;
+    for fault in [
+        AtomicWriteFault::BeforeRename,
+        AtomicWriteFault::AfterRename,
+    ] {
+        for has_initial_input in [false, true] {
+            let root = tempfile::tempdir().unwrap();
+            let options = FsRuntimeStoreOptions::new(root.path().join("runtime"))
+                .with_runtime_id("create-commit-fault");
+            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();
+            *backend.before_spawn_result.lock().unwrap() = Some(Box::new({
+                let runtime = runtime.clone();
+                move || {
+                    runtime
+                        .lock()
+                        .unwrap()
+                        .fs_store()
+                        .unwrap()
+                        .fail_next_worker_write(fault)
+                }
+            }));
+            let mut request = task_request("create commit fault");
+            if has_initial_input {
+                request.initial_input = Some(WorkerInput::user("accepted once"));
+            }
+            let worker_id = request.worker_id;
+            let error = runtime.create_worker(request).unwrap_err();
+            let failure = error.create_failure().unwrap();
+            assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+            assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted);
+            assert_eq!(
+                failure.operation().finalize,
+                Some(FinalizeFacts {
+                    stage: FinalizeStage::Publication,
+                    compensation: CompensationOutcome::Completed,
+                })
+            );
+            assert_eq!(
+                boundary_reason(&failure),
+                StartupBoundaryReason::FinalizeFailed
+            );
+            assert_eq!(failure.startup().io_kind, Some(StartupIoKind::Other));
+            assert_eq!(failure.startup().persistence, None);
+            assert!(
+                failure
+                    .startup()
+                    .secondary
+                    .contains(&StartupSecondaryFailure::Persistence {
+                        phase: StartupFailurePhase::RuntimeReconciliationRecordSave,
+                        outcome: if fault == AtomicWriteFault::AfterRename {
+                            StartupPersistenceOutcome::OutcomeUnknown
+                        } else {
+                            StartupPersistenceOutcome::NotSaved
+                        },
+                    })
+            );
+            let RuntimeError::WorkerCreateFailed { source, .. } = error else {
+                unreachable!()
+            };
+            assert!(matches!(
+                *source,
+                RuntimeError::StoreIo { .. } | RuntimeError::StoreCommitOutcomeUnknown { .. }
+            ));
+            assert_eq!(
+                backend.dispatched_inputs.lock().unwrap().len(),
+                usize::from(has_initial_input)
+            );
+            assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+            assert!(
+                !options
+                    .root
+                    .join("workers")
+                    .join(worker_id.to_string())
+                    .exists()
+            );
+            drop(runtime);
+            drop(backend);
+            let reopened = Runtime::with_fs_store(options).unwrap();
+            assert!(reopened.list_workers().unwrap().is_empty());
+        }
+    }
+}
+
+#[cfg(feature = "fs-store")]
+#[test]
+fn connected_create_cleanup_persistence_faults_never_replace_primary() {
+    for delete_fault in [false, true] {
+        let root = tempfile::tempdir().unwrap();
+        let store_root = root.path().join("runtime");
+        let options =
+            FsRuntimeStoreOptions::new(&store_root).with_runtime_id("create-cleanup-fault");
+        let backend = Arc::new(TestExecutionBackend::default());
+        let runtime =
+            Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap();
+        runtime.store_config_bundle(test_bundle()).unwrap();
+        let mut request = task_request("create cleanup persistence fault");
+        request.initial_input = Some(WorkerInput::user("rejected once"));
+        let worker_dir = store_root
+            .join("workers")
+            .join(request.worker_id.to_string());
+        let runtime_path = store_root.join("runtime.json");
+        *backend.before_spawn_result.lock().unwrap() = Some(Box::new(move || {
+            if delete_fault {
+                std::fs::create_dir(worker_dir.join("metadata.json")).unwrap();
+            } else {
+                std::fs::rename(&runtime_path, runtime_path.with_extension("saved")).unwrap();
+                std::fs::create_dir(&runtime_path).unwrap();
+            }
+        }));
+        let original = rejected_input();
+        backend.set_dispatch_result(original.clone());
+        let error = runtime.create_worker(request).unwrap_err();
+        let failure = error.create_failure().unwrap();
+        assert_eq!(error.execution_result(), Some(&original));
+        assert_eq!(failure.startup().primary, original.startup_failure.unwrap());
+        assert_eq!(failure.startup().io_kind, original.startup_io_kind);
+        assert_eq!(failure.operation().execution, ExecutionOutcome::Errored);
+        assert_eq!(
+            failure.operation().finalize.unwrap().compensation,
+            CompensationOutcome::Unknown
+        );
+        assert!(
+            failure
+                .startup()
+                .secondary
+                .contains(&StartupSecondaryFailure::Persistence {
+                    phase: if delete_fault {
+                        StartupFailurePhase::RuntimeRecordDelete
+                    } else {
+                        StartupFailurePhase::RuntimeSnapshotSave
+                    },
+                    outcome: StartupPersistenceOutcome::OutcomeUnknown,
+                })
+        );
+        if !delete_fault {
+            assert!(
+                failure
+                    .startup()
+                    .secondary
+                    .contains(&StartupSecondaryFailure::Persistence {
+                        phase: StartupFailurePhase::RuntimeRecordDelete,
+                        outcome: StartupPersistenceOutcome::Persisted,
+                    })
+            );
+        }
+        assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+        assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1);
+    }
+}
+
+#[test]
+fn connected_create_retains_cleanup_owner_resource_and_metadata_observations() {
+    let (runtime, backend) = runtime_and_backend();
+    backend.set_dispatch_result(WorkerExecutionResult::accepted(
+        WorkerExecutionOperation::Input,
+    ));
+    let mut stop =
+        WorkerExecutionResult::errored(WorkerExecutionOperation::Stop, "TEST_ONLY_SECRET");
+    let resources = StartupResourceFacts {
+        resources: StartupResources::Unproven,
+        cleanup: StartupCleanup::Failed(StartupCleanupFailure::Feature),
+        durable_effects_possible: true,
+    };
+    stop.startup_resources = Some(resources);
+    stop.startup_io_kind = Some(StartupIoKind::TimedOut);
+    stop.startup_failure_persistence = Some(StartupPersistenceOutcome::OutcomeUnknown);
+    backend.set_stop_result(stop);
+    let mut request = task_request("cleanup owner observations");
+    request.initial_input = Some(WorkerInput::user("start once"));
+    let error = runtime.create_worker(request).unwrap_err();
+    let failure = error.create_failure().unwrap();
+    assert_eq!(failure.startup().resources, Some(resources));
+    assert_eq!(failure.startup().io_kind, Some(StartupIoKind::TimedOut));
+    assert_eq!(failure.startup().persistence, None);
+    assert!(
+        failure
+            .startup()
+            .secondary
+            .contains(&StartupSecondaryFailure::Persistence {
+                phase: StartupFailurePhase::MetadataSave,
+                outcome: StartupPersistenceOutcome::OutcomeUnknown,
+            })
+    );
+    assert_eq!(
+        failure.operation().finalize.unwrap().compensation,
+        CompensationOutcome::Unknown
+    );
+    assert_eq!(
+        boundary_reason(&failure),
+        StartupBoundaryReason::InvalidResponse
+    );
+    assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+}
+
+#[cfg(feature = "fs-store")]
+#[test]
+fn connected_create_unknown_cleanup_retention_write_failure_preserves_primary() {
+    use crate::fs_store::AtomicWriteFault;
+    let root = tempfile::tempdir().unwrap();
+    let options = FsRuntimeStoreOptions::new(root.path().join("runtime"))
+        .with_runtime_id("create-retain-fault");
+    let backend = Arc::new(TestExecutionBackend::default());
+    let runtime = Runtime::with_fs_store_and_execution_backend(options, backend.clone()).unwrap();
+    runtime.store_config_bundle(test_bundle()).unwrap();
+    *backend.before_spawn_result.lock().unwrap() = Some(Box::new({
+        let runtime = runtime.clone();
+        move || {
+            runtime
+                .lock()
+                .unwrap()
+                .fs_store()
+                .unwrap()
+                .fail_next_worker_write(AtomicWriteFault::BeforeRename)
+        }
+    }));
+    let original = rejected_input();
+    backend.set_dispatch_result(original.clone());
+    // Accepted stop without a termination receipt cannot authorize deletion.
+    backend.set_stop_result(WorkerExecutionResult::accepted(
+        WorkerExecutionOperation::Stop,
+    ));
+    let mut request = task_request("unknown cleanup retention persistence failure");
+    request.initial_input = Some(WorkerInput::user("rejected once"));
+    let error = runtime.create_worker(request).unwrap_err();
+    let failure = error.create_failure().unwrap();
+    assert_eq!(error.execution_result(), Some(&original));
+    assert_eq!(
+        failure.operation().finalize.unwrap().compensation,
+        CompensationOutcome::Unknown
+    );
+    assert!(
+        failure
+            .startup()
+            .secondary
+            .contains(&StartupSecondaryFailure::Persistence {
+                phase: StartupFailurePhase::RuntimeReconciliationRecordSave,
+                outcome: StartupPersistenceOutcome::NotSaved,
+            })
+    );
+    assert_eq!(runtime.list_workers().unwrap().len(), 1);
+    assert_eq!(*backend.stop_count.lock().unwrap(), 1);
+    assert_eq!(backend.dispatched_inputs.lock().unwrap().len(), 1);
+    assert!(
+        backend
+            .submission_receipt_queries
+            .lock()
+            .unwrap()
+            .is_empty()
+    );
+}

crates/worker-runtime/src/runtime/restore_continuation.rs

@@ -479,7 +479,7 @@ fn run(
             }
             Ok(Step::Waiting) | Ok(Step::Contended) => None,
             Ok(Step::Retry(reason)) => Some(reason),
-            Err(error) => Some(sanitize_worker_create_failure_message(&error.to_string())),
+            Err(error) => Some(sanitize_worker_restore_failure_message(&error.to_string())),
         };
         if let Some(reason) = &reason {
             repeated = repeated.saturating_add(1);

crates/worker-runtime/src/startup_observation.rs

@@ -0,0 +1,205 @@
+//! The Runtime creation responsibility sink records only projected typed fields.
+//! Source/provider strings are never safe merely because a blacklist did not match.
+//! Wrappers/projectors do not write events; a failing sink cannot change the result.
+use crate::{error::RuntimeError, identity::WorkerId};
+
+pub(crate) fn write_create_failure(
+    worker_id: WorkerId,
+    workspace_id: Option<&str>,
+    error: &RuntimeError,
+    elapsed: std::time::Duration,
+) {
+    let failure = error.create_failure();
+    let projected = failure
+        .as_ref()
+        .and_then(|failure| failure.telemetry().ok());
+    let fields = projected.as_ref().map(error_contract::observation_fields);
+    let operation = failure.as_ref().map(|failure| failure.operation());
+    tracing::error!(
+        target: "yoi::worker_create",
+        event = "worker_create_failed",
+        component = "runtime",
+        workspace_id = workspace_id.unwrap_or(""),
+        worker_id = %worker_id,
+        domain = fields.as_ref().map_or("startup_boundary", |fields| fields.domain),
+        reason = fields.as_ref().map_or("observation_unavailable", |fields| fields.reason),
+        operation = fields.as_ref().and_then(|fields| fields.operation),
+        stage = fields.as_ref().and_then(|fields| fields.stage),
+        correlation = operation.and_then(|operation| operation.correlation.as_ref()).map(|value| value.as_str()),
+        acceptance = ?operation.map(|operation| operation.acceptance),
+        execution = ?operation.map(|operation| operation.execution),
+        finalize = ?operation.and_then(|operation| operation.finalize),
+        persistence = ?failure.as_ref().and_then(|failure| failure.startup().persistence),
+        resources = ?failure.as_ref().and_then(|failure| failure.startup().resources),
+        duration_ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX),
+        "Worker creation returned a typed failure"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::catalog::{CreateWorkerRequest, ProfileSelector, ProfileSourceArchiveSource};
+    use crate::execution::{
+        WorkerExecutionBackend, WorkerExecutionOperation, WorkerExecutionResult,
+        WorkerExecutionSpawnRequest, WorkerExecutionSpawnResult,
+    };
+    use crate::profile_archive::{ProfileSourceArchive, ProfileSourceArchiveInput};
+    use crate::{Runtime, RuntimeOptions};
+    use runtime_api::startup::*;
+    use std::collections::BTreeMap;
+    use std::sync::{Arc, Mutex};
+    use tracing_subscriber::prelude::*;
+
+    const CANARIES: &[&str] = &[
+        "aQ7xV2zN9mL5",
+        "/private/startup/provider",
+        "Bearer q3fV8aJ6",
+        "token=K7sM2pQ4",
+    ];
+    struct FailingBackend;
+    impl WorkerExecutionBackend for FailingBackend {
+        fn backend_id(&self) -> &str {
+            "log-owner-fixture"
+        }
+        fn dispatch_input(
+            &self,
+            _: &crate::identity::WorkerRef,
+            _: crate::interaction::WorkerInput,
+        ) -> WorkerExecutionResult {
+            panic!("rejected creation never dispatches input")
+        }
+        fn spawn_worker(&self, _: WorkerExecutionSpawnRequest) -> WorkerExecutionSpawnResult {
+            let result =
+                WorkerExecutionResult::errored(WorkerExecutionOperation::Spawn, CANARIES.join(" "))
+                    .with_startup_failure(
+                        worker::startup_failure::runtime_startup_attachment(
+                            runtime_api::RuntimeStartupCause {
+                                stage: runtime_api::StartupStage::Controller,
+                                reason: runtime_api::RuntimeStartupReason::StartupFailed,
+                            },
+                        )
+                        .unwrap(),
+                    );
+            WorkerExecutionSpawnResult::Errored(result)
+        }
+    }
+    #[derive(Clone, Default)]
+    struct Events(Arc<Mutex<Vec<BTreeMap<String, String>>>>);
+    struct Fields(BTreeMap<String, String>);
+    impl tracing::field::Visit for Fields {
+        fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
+            self.0.insert(field.name().into(), value.into());
+        }
+        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+            self.0.insert(field.name().into(), format!("{value:?}"));
+        }
+    }
+    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for Events {
+        fn on_event(
+            &self,
+            event: &tracing::Event<'_>,
+            _: tracing_subscriber::layer::Context<'_, S>,
+        ) {
+            if event.metadata().target() == "yoi::worker_create" {
+                let mut fields = Fields(BTreeMap::new());
+                event.record(&mut fields);
+                self.0.lock().unwrap().push(fields.0);
+            }
+        }
+    }
+    fn request() -> CreateWorkerRequest {
+        let archive = ProfileSourceArchive::build(ProfileSourceArchiveInput {
+            id: "log-profile".into(),
+            entrypoints: BTreeMap::from([("builtin:coder".into(), "profiles/coder.dcdl".into())]),
+            imports: BTreeMap::new(),
+            sources: BTreeMap::from([("profiles/coder.dcdl".into(), "{}".into())]),
+        })
+        .unwrap();
+        CreateWorkerRequest {
+            worker_id: WorkerId::now_v7(),
+            create_fingerprint: "log-create".into(),
+            profile: ProfileSelector::Builtin("builtin:coder".into()),
+            display_name: None,
+            profile_source: 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,
+        }
+    }
+    #[test]
+    fn real_runtime_creation_sink_omits_unlabelled_provider_canaries_and_matches_typed_result() {
+        let runtime =
+            Runtime::with_execution_backend(RuntimeOptions::default(), Arc::new(FailingBackend))
+                .unwrap();
+        runtime
+            .bind_runtime_identity("runtime-log-fixture")
+            .unwrap();
+        let request = request();
+        let worker_id = request.worker_id;
+        let events = Events::default();
+        let subscriber = tracing_subscriber::registry().with(events.clone());
+        let error =
+            tracing::subscriber::with_default(subscriber, || runtime.create_worker(request))
+                .unwrap_err();
+        let failure = error
+            .create_failure()
+            .expect("creation owner must retain a typed cause");
+        let wire = failure
+            .problem(
+                StartupBoundary::Runtime,
+                &SessionFailureVisibility {
+                    can_view_cause: true,
+                },
+                "",
+            )
+            .unwrap();
+        let encoded = failure.encode_storage().unwrap();
+        let DecodeStoredWorkerCreateFailure::Known(saved) =
+            decode_stored_worker_create_failure(&encoded)
+        else {
+            panic!("known storage");
+        };
+        assert_eq!(saved.public_facts(), failure.public_facts());
+        let captured = events.0.lock().unwrap();
+        assert_eq!(captured.len(), 1);
+        let event = &captured[0];
+        let projection = failure.telemetry().unwrap();
+        let fields = error_contract::observation_fields(&projection);
+        assert_eq!(event["domain"], fields.domain);
+        assert_eq!(event["reason"], fields.reason);
+        assert_eq!(
+            event["correlation"],
+            StartupCorrelation::for_worker("runtime-log-fixture", &worker_id.to_string()).as_str()
+        );
+        for canary in CANARIES {
+            assert!(!format!("{captured:?}").contains(canary));
+            assert!(!String::from_utf8_lossy(wire.body()).contains(canary));
+            assert!(!encoded.contains(canary));
+        }
+        assert!(!event.contains_key("execution_failure_message"));
+    }
+    #[test]
+    fn unsupported_failure_logs_only_fixed_fallback_not_source_text() {
+        let error = RuntimeError::InvalidRequest(CANARIES.join(" "));
+        let events = Events::default();
+        let subscriber = tracing_subscriber::registry().with(events.clone());
+        tracing::subscriber::with_default(subscriber, || {
+            write_create_failure(WorkerId::now_v7(), None, &error, std::time::Duration::ZERO)
+        });
+        let captured = events.0.lock().unwrap();
+        assert_eq!(captured.len(), 1);
+        assert_eq!(captured[0]["reason"], "observation_unavailable");
+        for canary in CANARIES {
+            assert!(!format!("{captured:?}").contains(canary));
+        }
+    }
+}

crates/workspace-server/src/backend_job_upgrade_tests.rs

@@ -1476,6 +1476,21 @@ fn captured_history_is_accepted_at_each_real_retained_prefix_and_cannot_be_forge
     let dir = tempfile::tempdir().unwrap();
     let conn = legacy_fixture_to_85(&dir.path().join("prefixes.sqlite"), 85);
     let original = pre_content_values(&conn, false);
+    let mut expected_receipts = rows(
+        &conn,
+        "SELECT version,name FROM __yoi_schema_migrations ORDER BY version",
+    );
+    expected_receipts.extend(
+        store_owner::MIGRATIONS
+            .iter()
+            .filter(|migration| migration.version > 85)
+            .map(|migration| {
+                vec![
+                    Value::Integer(migration.version),
+                    Value::Text(migration.name.into()),
+                ]
+            }),
+    );
     for migration in store_owner::MIGRATIONS
         .iter()
         .filter(|migration| migration.version >= 86 && migration.version <= 122)
@@ -1496,10 +1511,13 @@ fn captured_history_is_accepted_at_each_real_retained_prefix_and_cannot_be_forge
         &conn,
         "SELECT version,name FROM __yoi_schema_migrations ORDER BY version",
     );
+    assert_eq!(receipts, expected_receipts);
     assert_eq!(
-        receipts.last().unwrap(),
+        store_owner::current_schema_version(&conn).unwrap(),
-        &vec![Value::Integer(123), Value::Text(MIGRATION_NAME.into())]
+        store_owner::LATEST_SCHEMA_VERSION
     );
+    super::super::storage_formats::verify_history(&conn, store_owner::LATEST_SCHEMA_VERSION)
+        .unwrap();
     assert_eq!(original, pre_content_values(&conn, true));
     assert!(conn.execute_batch("INSERT INTO backend_job_pre_content_history SELECT * FROM backend_job_pre_content_history LIMIT 1").is_err());
 }

crates/workspace-server/src/hosts.rs

@@ -3316,14 +3316,7 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
                 diagnostics,
             },
             Err(err) => {
-                let failure = match &err {
+                let failure = err.create_failure();
-                    EmbeddedRuntimeError::WorkerExecutionRejected { result, .. } => {
-                        worker_runtime::execution::startup_create_failure(result)
-                            .ok()
-                            .flatten()
-                    }
-                    _ => None,
-                };
                 diagnostics.push(embedded_runtime_diagnostic(&err));
                 let state = failure.as_ref().map_or(
                     InternalWorkerOperationState::Rejected,
@@ -6954,6 +6947,11 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti
                 message,
             ),
         ),
+        EmbeddedRuntimeError::WorkerCreateFailed { failure, .. } => diagnostic(
+            "embedded_worker_create_failed",
+            HostDiagnosticSeverity::Warning,
+            worker::startup_failure::internal_startup_failure_summary(&failure.startup().primary),
+        ),
         EmbeddedRuntimeError::WorkerExecutionRejected {
             operation,
             outcome,
@@ -7036,6 +7034,9 @@ fn embedded_runtime_diagnostic(error: &EmbeddedRuntimeError) -> RuntimeDiagnosti
             })
         });
     }
+    if let EmbeddedRuntimeError::WorkerCreateFailed { failure, .. } = error {
+        projection.startup = internal_startup_transfer(failure.startup().clone());
+    }
     projection
 }
 
@@ -8904,10 +8905,9 @@ mod tests {
         assert!(spawned.acceptance_evidence.is_empty());
         assert_eq!(spawned.diagnostics.len(), 1);
         assert!(spawned.diagnostics.iter().any(|diagnostic| {
-            diagnostic.code == "embedded_worker_execution_rejected"
+            diagnostic.code == "embedded_worker_create_failed"
-                && diagnostic
+                && diagnostic.startup.is_some()
-                    .message
+                && !diagnostic.message.contains("provider setup failed")
-                    .contains("provider setup failed at [redacted]")
                 && !diagnostic.message.contains("/tmp/secret-provider-config")
                 && !diagnostic.message.contains("secret-value")
                 && !diagnostic.message.contains("session-42")
@@ -10064,7 +10064,9 @@ fn startup_failure_operation_state(
     failure: &runtime_api::startup::WorkerCreateFailure,
 ) -> InternalWorkerOperationState {
     use runtime_api::startup::{CreationAcceptance, ExecutionOutcome};
-    if failure.operation().acceptance == CreationAcceptance::Unknown {
+    if failure.operation().acceptance == CreationAcceptance::Unknown
+        || failure.operation().execution == ExecutionOutcome::Unknown
+    {
         InternalWorkerOperationState::Unknown
     } else if failure.operation().execution == ExecutionOutcome::Unsupported {
         InternalWorkerOperationState::Unsupported

crates/workspace-server/src/server.rs

@@ -35096,12 +35096,15 @@ fn browser_worker_response_for_committed_summary(
     let workspace_id = api.workspace_id().to_string();
     let resource_key = api
         .store
-        .worker_resource_key(&workspace_id, &worker.worker)?
+        .worker_resource_key(&workspace_id, &worker.worker)
-        .ok_or_else(|| {
+        .and_then(|key| {
-            Error::Store(format!(
+            key.ok_or_else(|| {
-                "Workspace Worker `{worker_id}` has no resource key after registration"
+                Error::Store(format!(
-            ))
+                    "Workspace Worker `{worker_id}` has no resource key after registration"
-        })?;
+                ))
+            })
+        })
+        .map_err(|error| worker_startup::publication_error(api, &worker.worker, error.into()))?;
     let worker = worker_launch_worker_summary(worker);
     let console_href = browser_worker_console_href(&workspace_id, &resource_key);
     Ok(BrowserCreateWorkerResponse {

crates/workspace-server/src/server/tests/startup_result_tests.rs

@@ -1,8 +1,9 @@
-//! Backend result fixture: real SQLite reservation -> coordinator helper -> typed
+//! Backend result fixtures use real SQLite reservations, coordinator helpers, typed
-//! read -> public ApiError response -> HTTP middleware observation. Only the route's
+//! reads, public ApiError responses and HTTP middleware observation. Publication
-//! startup producer is supplied by the fixture; this does not test authorization,
+//! faults enter the production committed-summary response helper with a fixture
-//! Session callbacks, Runtime startup or the Browser creation route. Those belong
+//! Runtime Worker; only the earlier Runtime startup is substituted. These cases do
-//! to session_registration_connected_tests. Runs in the ordinary server lib suite;
+//! not claim creation-route authorization or Session callback coverage (see
+//! session_registration_connected_tests). Runs in the ordinary server lib suite;
 //! each case owns its tempdir/DB and subscriber, with no external services.
 
 use super::*;
@@ -82,6 +83,10 @@ fn assert_semantics(actual: &WorkerCreateFailure, expected: &WorkerCreateFailure
 }
 
 fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef {
+    reserve_for_runtime(api, EMBEDDED_WORKER_RUNTIME_ID)
+}
+
+fn reserve_for_runtime(api: &WorkspaceApi, runtime_id: &str) -> RuntimeWorkerRef {
     let settings = api
         .config_store
         .get_workspace_memory_settings(TEST_WORKSPACE_ID)
@@ -90,7 +95,7 @@ fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef {
         .config_store
         .reserve_worker_create(
             TEST_WORKSPACE_ID,
-            EMBEDDED_WORKER_RUNTIME_ID,
+            runtime_id,
             "manual:startup-result-fixture",
             "sha256:startup-result-fixture",
             None,
@@ -98,7 +103,7 @@ fn reserve(api: &WorkspaceApi) -> RuntimeWorkerRef {
         )
         .unwrap();
     RuntimeWorkerRef {
-        runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
+        runtime_id: runtime_id.into(),
         worker_id: reservation.worker_id.to_string(),
     }
 }
@@ -123,6 +128,9 @@ fn lifecycle_snapshot(api: &WorkspaceApi) -> Vec<Vec<rusqlite::types::Value>> {
             format!("SELECT {columns} FROM worker_create_reservations ORDER BY workspace_id,allocation_key"),
             "SELECT * FROM workspace_resource_keys ORDER BY workspace_id,resource_kind,resource_id".into(),
             "SELECT * FROM worker_singleton_owners ORDER BY workspace_id,singleton_key".into(),
+            "SELECT * FROM worker_registry ORDER BY workspace_id,runtime_id,worker_id".into(),
+            "SELECT * FROM ticket_worker_assignments ORDER BY workspace_id,ticket_id,assignment_id".into(),
+            "SELECT * FROM ticket_assignment_operations ORDER BY workspace_id,operation_id".into(),
         ];
         let mut snapshot = Vec::new();
         for query in queries {
@@ -455,6 +463,290 @@ async fn context_and_public_projection_are_silent_until_explicit_http_observatio
     assert_log(&events[0], &expected, StatusCode::CONFLICT);
 }
 
+async fn assert_publication_fault_retains_committed_worker(query_failure: bool) {
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    // Freeze unrelated observation writes before deliberately damaging the DB.
+    api.worker_projection.shutdown_async().await;
+    api.store.db_executor().drain().await.unwrap();
+    register_test_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID).await;
+    let stops = Arc::new(AtomicUsize::new(0));
+    let stop_count = stops.clone();
+    let runtime = WorkdirlessFixtureRuntime {
+        stop_callback: Some(Arc::new(move || {
+            stop_count.fetch_add(1, Ordering::SeqCst);
+        })),
+        ..Default::default()
+    };
+    api.runtime.register_or_replace(runtime.clone());
+    let worker = reserve_for_runtime(&api, WorkdirlessFixtureRuntime::RUNTIME_ID);
+    let spawned = crate::hosts::WorkspaceWorkerRuntime::spawn_worker(
+        &runtime,
+        WorkerCreateBinding {
+            plugin_execution: None,
+            worker_id: WorkerId::parse(&worker.worker_id).unwrap(),
+            create_fingerprint: "sha256:startup-result-fixture".into(),
+        },
+        WorkerSpawnRequest {
+            intent: WorkerSpawnIntent::WorkspaceCompanion,
+            requested_worker_name: Some("Publication fixture".into()),
+            singleton_key: None,
+            acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                expected_segments: 0,
+            },
+            profile: ProfileSelector::Builtin("builtin:companion".into()),
+            ticket_assignment: None,
+            initial_submit: Vec::new(),
+            workdir_attachment_requests: Vec::new(),
+            resolved_workdir_attachment_requests: Vec::new(),
+            resolved_workdir_attachments: Vec::new(),
+            resolved_config_bundle: None,
+            resolved_worker_observation_enabled: false,
+            resolved_worker_observation_grants: Vec::new(),
+            resolved_workspace_api: None,
+            resolved_memory_settings: None,
+            resolved_subjektiv_attached: false,
+            resolved_control_operation: None,
+            resolved_execution_grant: None,
+        },
+    );
+    assert_eq!(spawned.state, InternalWorkerOperationState::Accepted);
+    let summary = spawned.worker.unwrap();
+    assert_eq!(summary.worker, worker);
+    let record = record_worker_summary(
+        &api,
+        &summary,
+        &summary.display_name,
+        summary.profile.clone(),
+        WorkerRegistryDisplayNamePolicy::UseProvided,
+    )
+    .unwrap();
+    api.config_store
+        .complete_worker_create_reservation(
+            TEST_WORKSPACE_ID,
+            WorkerId::parse(&worker.worker_id).unwrap(),
+        )
+        .unwrap();
+    let ticket = browser_ticket_backend(&api)
+        .unwrap()
+        .create(ticket::NewTicket::new("Publication retains assignment"))
+        .unwrap();
+    let assignment = TicketWorkerAssignmentRecord {
+        workspace_id: TEST_WORKSPACE_ID.into(),
+        ticket_id: ticket.id.clone(),
+        assignment_id: "publication-assignment".into(),
+        worker: worker.clone(),
+        assigned_by: "test-user".into(),
+        assigned_at: TEST_CREATED_AT.into(),
+    };
+    api.store
+        .set_current_ticket_worker_assignment(
+            &assignment,
+            None,
+            "publication-assignment-event",
+            "publication-assignment-operation",
+            false,
+        )
+        .unwrap();
+    // Establish a working publication read first; the fault changes only that
+    // final read, not reservation persistence or an earlier caller-identity check.
+    let key = api
+        .store
+        .worker_resource_key(TEST_WORKSPACE_ID, &worker)
+        .unwrap()
+        .unwrap();
+    api.config_store
+        .with_conn(|conn| {
+            if query_failure {
+                conn.execute_batch(
+                    "ALTER TABLE workspace_resource_keys RENAME COLUMN resource_key TO \"sqlite-secret\";",
+                )?;
+            } else {
+                let deleted = conn.execute(
+                    "DELETE FROM workspace_resource_keys WHERE workspace_id=?1 AND resource_kind='worker' AND resource_key=?2",
+                    rusqlite::params![TEST_WORKSPACE_ID, key],
+                )?;
+                assert_eq!(deleted, 1);
+            }
+            Ok(())
+        })
+        .unwrap();
+    let before = lifecycle_snapshot(&api);
+    let capture = Capture::default();
+    let route_api = api.clone();
+    let app = Router::new()
+        .route(
+            "/api/startup-result-fixture",
+            axum::routing::get(move || {
+                let api = route_api.clone();
+                let summary = summary.clone();
+                async move {
+                    let error = browser_worker_response_for_committed_summary(
+                        &api,
+                        summary,
+                        "Publication fixture".into(),
+                        Vec::new(),
+                    )
+                    .unwrap_err();
+                    let Error::WorkerCreateFailed { source, .. } = &error.error else {
+                        panic!("final publication read must return a typed result: {error:?}");
+                    };
+                    let source = source
+                        .as_deref()
+                        .expect("keep the actual read error internal");
+                    if query_failure {
+                        assert!(matches!(source, Error::Sqlite(_)), "{source:?}");
+                    } else {
+                        assert!(matches!(source, Error::Store(_)), "{source:?}");
+                        assert!(source.to_string().contains("no resource key"));
+                    }
+                    error.into_response()
+                }
+            }),
+        )
+        .layer(middleware::from_fn(log_failed_api_response));
+    let (status, problem, public) = public_result(app)
+        .with_subscriber(subscriber(capture.clone()))
+        .await;
+    assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR);
+    assert_eq!(
+        primary_cause(&public),
+        StartupCause::Boundary(runtime_api::startup::StartupBoundaryCause {
+            reason: runtime_api::startup::StartupBoundaryReason::FinalizeFailed,
+        })
+    );
+    assert_eq!(public.operation().acceptance, CreationAcceptance::Accepted);
+    assert_eq!(public.operation().execution, ExecutionOutcome::Accepted);
+    assert_eq!(
+        public.operation().finalize,
+        Some(FinalizeFacts {
+            stage: FinalizeStage::Publication,
+            compensation: CompensationOutcome::Residual,
+        })
+    );
+    assert_eq!(public.operation().backend_compensation, None);
+    assert_eq!(
+        public.startup().resources,
+        Some(protocol::startup_failure::StartupResourceFacts {
+            resources: protocol::startup_failure::StartupResources::WorkerStartup,
+            cleanup: protocol::startup_failure::StartupCleanup::NotNeeded,
+            durable_effects_possible: true,
+        })
+    );
+    assert_eq!(
+        public.backend_result_persistence(),
+        Some(StartupPersistenceOutcome::Persisted)
+    );
+    assert_eq!(
+        public.operation().correlation.as_ref().unwrap().as_str(),
+        worker_startup::correlation(&worker.runtime_id, &worker.worker_id)
+    );
+    assert!(
+        problem["instance"]
+            .as_str()
+            .unwrap()
+            .ends_with(public.operation().correlation.as_ref().unwrap().as_str())
+    );
+    assert_semantics(&retained(&api, &worker), &public);
+    let events = capture.events();
+    assert_eq!(events.len(), 1, "one primary HTTP event: {events:?}");
+    let event = &events[0];
+    assert_eq!(event["target"], "yoi::worker_startup");
+    assert_eq!(event["event"], "worker_create_result");
+    assert_eq!(event["domain"], "startup_boundary");
+    assert_eq!(event["reason"], "finalize_failed");
+    assert_eq!(event["acceptance"], "Accepted");
+    assert_eq!(event["execution"], "Accepted");
+    assert_eq!(
+        event["finalize"],
+        format!("{:?}", public.operation().finalize)
+    );
+    assert_eq!(event["backend_compensation"], "None");
+    assert_eq!(event["backend_result_persistence"], "Some(Persisted)");
+    assert_eq!(
+        event["resources"],
+        format!("{:?}", public.startup().resources)
+    );
+    assert_eq!(event["status"], status.as_u16());
+    assert_eq!(
+        event["correlation"],
+        problem["yoi_startup"]["operation"]["correlation"]
+    );
+    let exposed = format!("{problem} {events:?}");
+    for internal in [
+        worker.worker_id.as_str(),
+        "no such column",
+        "no resource key",
+        "workspace_resource_keys",
+        dir.path().to_str().unwrap(),
+    ] {
+        assert!(
+            !exposed.contains(internal),
+            "internal source leaked: {exposed}"
+        );
+    }
+    assert_eq!(lifecycle_snapshot(&api), before);
+    assert_eq!(
+        api.store
+            .get_worker_registry(TEST_WORKSPACE_ID, &worker)
+            .unwrap(),
+        Some(record)
+    );
+    assert_eq!(
+        api.store
+            .get_current_ticket_worker_assignment(TEST_WORKSPACE_ID, &ticket.id)
+            .unwrap(),
+        Some(assignment)
+    );
+    assert_eq!(stops.load(Ordering::SeqCst), 0);
+    assert_eq!(runtime.spawn_requests().len(), 1, "no reexecution");
+    assert_eq!(
+        runtime.workers.lock().unwrap().len(),
+        1,
+        "no Runtime deletion"
+    );
+}
+
+#[tokio::test]
+async fn publication_sqlite_read_failure_retains_committed_worker_and_typed_result() {
+    assert_publication_fault_retains_committed_worker(true).await;
+}
+
+#[tokio::test]
+async fn publication_missing_resource_key_retains_committed_worker_and_typed_result() {
+    assert_publication_fault_retains_committed_worker(false).await;
+}
+
+#[tokio::test]
+async fn publication_error_preserves_existing_primary_without_claiming_compensation() {
+    let dir = tempfile::tempdir().unwrap();
+    let api = test_api(dir.path()).await;
+    let worker = reserve(&api);
+    let original = failure_fixture(CompensationOutcome::Completed);
+    let error = worker_startup::publication_error(&api, &worker, api_error(original.clone()));
+    let Error::WorkerCreateFailed { failure, source } = &error.error else {
+        panic!("publication must retain a typed startup result");
+    };
+    assert!(matches!(
+        source.as_deref(),
+        Some(Error::WorkerCreateFailed { .. })
+    ));
+    assert_eq!(failure.startup().primary, original.startup().primary);
+    assert_eq!(failure.startup().secondary, original.startup().secondary);
+    assert_eq!(
+        failure.startup().persistence,
+        original.startup().persistence
+    );
+    assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+    assert_eq!(failure.operation().execution, ExecutionOutcome::Accepted);
+    assert_eq!(failure.operation().backend_compensation, None);
+    assert_eq!(
+        failure.operation().finalize.as_ref().unwrap().stage,
+        FinalizeStage::Publication
+    );
+    assert_semantics(&retained(&api, &worker), failure);
+}
+
 #[derive(Clone)]
 struct FailingWriter(Arc<AtomicUsize>);
 

crates/workspace-server/src/server/tests/ticket_worker_launch_tests.rs

@@ -12,6 +12,7 @@ use worker_runtime::execution::{
 // published Profile projection; leave Worker creation/control to the real Runtime.
 #[derive(Default)]
 struct LaunchExecutionBackend {
+    lose_initial_ack: std::sync::atomic::AtomicBool,
     fail_stop: std::sync::atomic::AtomicBool,
     execution: DeterministicExecutionBackend,
     provider_close_gate: Mutex<Option<Arc<super::workdir_session_close_tests::ProviderCloseGate>>>,
@@ -197,7 +198,16 @@ impl WorkerExecutionBackend for LaunchExecutionBackend {
             .lock()
             .unwrap()
             .push((worker.clone(), input.clone()));
-        self.execution.dispatch_input(worker, input)
+        let mut result = self.execution.dispatch_input(worker, input);
+        if self
+            .lose_initial_ack
+            .swap(false, std::sync::atomic::Ordering::SeqCst)
+        {
+            // Lose only the response acknowledgement, after the real fixture
+            // execution committed its durable receipt. Never redispatch it.
+            result.submission = None;
+        }
+        result
     }
 
     fn termination_receipt(
@@ -1160,6 +1170,90 @@ async fn read_only_ticket_targets_allow_extra_external_grants_without_widening_c
     );
 }
 
+/// Production create owner and Backend adapters share the same executor; only
+/// its response acknowledgement and cleanup boundary are fault-injected.
+#[tokio::test(flavor = "multi_thread")]
+async fn initial_input_ack_loss_preserves_creation_facts_across_embedded_and_remote() {
+    use runtime_api::startup::*;
+    for failed_cleanup in [false, true] {
+        let mut observed = Vec::new();
+        for remote in [false, true] {
+            let fixture = LaunchFixture::new(remote).await;
+            let ticket = fixture.ticket();
+            let payload = fixture.payload(&ticket, "lost-initial-ack", "builtin:companion", false);
+            fixture
+                .backend
+                .lose_initial_ack
+                .store(true, std::sync::atomic::Ordering::SeqCst);
+            fixture
+                .backend
+                .fail_stop
+                .store(failed_cleanup, std::sync::atomic::Ordering::SeqCst);
+            let (status, body) = fixture.post(TEST_WORKSPACE_ID, payload).await;
+            assert_eq!(status, StatusCode::BAD_GATEWAY, "{body}");
+            let DecodeStartupProblem::Known(failure) = decode_startup_problem(
+                &serde_json::to_vec(&body).unwrap(),
+                status.as_u16(),
+                PROBLEM_CONTENT_TYPE,
+                StartupBoundary::Browser,
+            ) else {
+                panic!("must retain typed creation facts: {body}");
+            };
+            assert_eq!(failure.operation().acceptance, CreationAcceptance::Accepted);
+            assert_eq!(failure.operation().execution, ExecutionOutcome::Unknown);
+            let finalize = failure.operation().finalize.unwrap();
+            assert_eq!(finalize.stage, FinalizeStage::InitialInput);
+            assert_eq!(
+                finalize.compensation,
+                if failed_cleanup {
+                    CompensationOutcome::Unknown
+                } else {
+                    CompensationOutcome::Completed
+                }
+            );
+            assert_eq!(body["yoi_failure"]["reason"], "invalid_response");
+            assert!(
+                !body
+                    .to_string()
+                    .contains("TEST_ONLY_PRIVATE_STOP_CREDENTIAL")
+            );
+            let creates = fixture.backend.creates.lock().unwrap().clone();
+            assert_eq!(creates.len(), 1);
+            let worker_id = creates[0].worker_id.to_string();
+            let worker = RuntimeWorkerRef::new(&fixture.runtime_id, &worker_id);
+            let Some(DecodeStoredWorkerCreateFailure::Known(saved)) =
+                fixture.api.worker_create_failure(&worker).unwrap()
+            else {
+                panic!("retain queryable create result");
+            };
+            assert_eq!(saved.public_facts(), failure.public_facts());
+            assert_eq!(
+                fixture.backend.submits_for(&worker_id).len(),
+                1,
+                "observation failure never redispatches"
+            );
+            assert_eq!(
+                fixture
+                    .backend
+                    .execution
+                    .contexts
+                    .lock()
+                    .unwrap()
+                    .keys()
+                    .any(|worker| worker.worker_id.to_string() == worker_id),
+                failed_cleanup
+            );
+            let mut facts = failure.public_facts();
+            facts.operation.correlation = None;
+            observed.push((status, body["yoi_failure"]["reason"].clone(), facts));
+        }
+        assert_eq!(
+            observed[0], observed[1],
+            "same owner facts through Embedded and Remote"
+        );
+    }
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn worker_assignment_finalize_failure_compensates_real_embedded_runtime_and_preserves_open_case()
  {

crates/workspace-server/src/server/worker_startup.rs

@@ -109,6 +109,60 @@ pub(super) fn finalize_failure(
     .expect("locally constructed finalization facts fit the shared boundary budget")
 }
 
+/// Publication failure cannot undo committed creation or authorize reexecution.
+/// Keep the Worker and its assignment: `Residual` describes deliberate retention,
+/// not attempted rollback; Backend compensation remains unattempted (`None`).
+/// The original store error stays internal while the independent reservation sink
+/// retains only the typed result and a path-free correlation.
+pub(super) fn publication_error(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+    source: ApiError,
+) -> ApiError {
+    use protocol::startup_failure::{StartupCleanup, StartupResourceFacts, StartupResources};
+    use runtime_api::startup::{
+        CompensationOutcome, FinalizeFacts, FinalizeStage, StartupBoundaryCause,
+        StartupBoundaryReason,
+    };
+
+    let operation = StartupOperationFacts {
+        acceptance: CreationAcceptance::Accepted,
+        execution: ExecutionOutcome::Accepted,
+        finalize: Some(FinalizeFacts {
+            stage: FinalizeStage::Publication,
+            compensation: CompensationOutcome::Residual,
+        }),
+        backend_compensation: None,
+        correlation: None,
+    };
+    let failure = match &source.error {
+        Error::WorkerCreateFailed { failure, .. } => failure.as_ref().clone(),
+        _ => WorkerCreateFailure::from_cause(
+            StartupCause::Boundary(StartupBoundaryCause {
+                reason: StartupBoundaryReason::FinalizeFailed,
+            }),
+            operation.clone(),
+        )
+        .expect("enum-only publication cause fits the boundary budget"),
+    };
+    let (mut startup, _) = failure.into_parts();
+    startup.resources = Some(StartupResourceFacts {
+        resources: StartupResources::WorkerStartup,
+        cleanup: StartupCleanup::NotNeeded,
+        durable_effects_possible: true,
+    });
+    let failure = WorkerCreateFailure::new(startup, operation)
+        .expect("locally constructed publication facts fit the shared boundary budget");
+    let failure = persist(api, worker, failure);
+    ApiError::with_diagnostics(
+        Error::WorkerCreateFailed {
+            failure: Box::new(failure),
+            source: Some(Box::new(source.error)),
+        },
+        source.diagnostics,
+    )
+}
+
 pub(super) fn unknown_result() -> WorkerCreateFailure {
     WorkerCreateFailure::from_cause(
         StartupCause::Boundary(runtime_api::startup::StartupBoundaryCause {
@@ -196,17 +250,9 @@ pub(super) fn persist(
 }
 
 pub(super) fn correlation(runtime_id: &str, worker_id: &str) -> String {
-    use sha2::{Digest, Sha256};
+    runtime_api::startup::StartupCorrelation::for_worker(runtime_id, worker_id)
-    let mut hash = Sha256::new();
+        .as_str()
-    hash.update(runtime_id.as_bytes());
+        .to_owned()
-    hash.update([0]);
-    hash.update(worker_id.as_bytes());
-    let digest = hash
-        .finalize()
-        .iter()
-        .map(|byte| format!("{byte:02x}"))
-        .collect::<String>();
-    format!("worker-{digest}")
 }
 
 /// Only the HTTP result boundary records the primary failure. Context wrappers,

crates/workspace-server/src/subject_assignment_migration_tests.rs

@@ -852,8 +852,6 @@ mod subject_assignment_migration_tests {
         };
 
         const PLAN_NAME: &str = "Immutable Worker Plugin plans and retained pins";
-        const SUBJECT_NAME: &str =
-            "Subject Ticket responsibility and current-body notification routing";
         type Rows = Vec<Vec<rusqlite::types::Value>>;
 
         fn raw(path: &Path) -> Connection {
@@ -1038,8 +1036,23 @@ mod subject_assignment_migration_tests {
                     } else {
                         String::new()
                     };
+                    // Keep append identities for rowid tables; current content
+                    // history tables are WITHOUT ROWID, so compare every column
+                    // in deterministic value order instead of inventing rowids.
+                    let without_rowid: bool = conn
+                        .query_row(
+                            "SELECT wr FROM pragma_table_list WHERE schema='main' AND name=?1",
+                            [&table],
+                            |row| row.get(0),
+                        )
+                        .unwrap();
+                    let (projection, order) = if without_rowid {
+                        (columns.clone(), columns)
+                    } else {
+                        (format!("rowid,{columns}"), "rowid".into())
+                    };
                     let query = format!(
-                        "SELECT rowid,{columns} FROM \"{table}\"{sequence_filter} ORDER BY rowid"
+                        "SELECT {projection} FROM \"{table}\"{sequence_filter} ORDER BY {order}"
                     );
                     let rows = fixture_query_rows(conn, &query);
                     (query, rows)
@@ -1120,49 +1133,15 @@ mod subject_assignment_migration_tests {
         }
 
         fn expected_tail(version: i64) -> Vec<(i64, String)> {
-            let mut tail = match version {
+            assert!(
-                96 => vec![
+                matches!(version, 96 | 106 | 107),
-                    (
+                "unsupported fixture origin"
-                        97,
+            );
-                        "MR notifications and authenticated implementation results".into(),
+            MIGRATIONS
-                    ),
+                .iter()
-                    (98, "Assignment-independent implementation reports".into()),
+                .filter(|migration| migration.version > version)
-                    (
+                .map(|migration| (migration.version, migration.name.into()))
-                        99,
+                .collect()
-                        "Explicit hosting administrators and self-registration policy".into(),
-                    ),
-                    (
-                        100,
-                        "Provisional users and scoped passkey activation".into(),
-                    ),
-                    (101, "Durable explicit Subject body release".into()),
-                    (102, "Durable failed Subject creation recovery".into()),
-                    (103, "Passkey metadata and atomic addition results".into()),
-                    (
-                        104,
-                        "Converge Passkey results and Job input retention".into(),
-                    ),
-                    (
-                        105,
-                        "Durable Subject deletion snapshots and Job admission fences".into(),
-                    ),
-                    (106, crate::plugin_catalog::MIGRATION_NAME.into()),
-                    (107, PLAN_NAME.into()),
-                    (108, SUBJECT_NAME.into()),
-                ],
-                106 => vec![(107, PLAN_NAME.into()), (108, SUBJECT_NAME.into())],
-                107 => vec![(108, SUBJECT_NAME.into())],
-                _ => panic!("unsupported fixture origin"),
-            };
-            tail.push((109, "Durable Plugin Host invocations and HTTPS effect receipts".into()));
-            tail.push((110, "Worker Plugin launch selection associations".into()));
-            tail.push((
-                111,
-                "Typed Plugin Ticket extension data and atomic mutation receipts".into(),
-            ));
-            tail.push((112, "Ticket Service outbox and fenced consumers".into()));
-            tail.push((113, "Ticket work-decomposition parent links".into()));
-            tail
         }
 
         fn plan_steps(version: i64) -> Vec<WorkspaceSchemaMigrationStep> {

crates/workspace-server/src/ticket_post_upgrade_tests.rs

@@ -56,12 +56,17 @@ fn ticket_posts_schema93_preserves_initial_body_creator_comments_and_archived_au
     drop(conn);
     let plan = SqliteWorkspaceStore::migration_plan(&path).unwrap();
     assert_eq!(plan.current_schema_version, 93);
+    assert_eq!(plan.target_schema_version, LATEST_SCHEMA_VERSION);
     assert_eq!(
-        plan.migrations
+        plan.migrations,
+        MIGRATIONS
             .iter()
-            .map(|step| step.version)
+            .filter(|migration| migration.version > 93)
-            .collect::<Vec<_>>(),
+            .map(|migration| WorkspaceSchemaMigrationStep {
-        (94..=LATEST_SCHEMA_VERSION).collect::<Vec<_>>()
+                version: migration.version,
+                name: migration.name.into(),
+            })
+            .collect::<Vec<_>>()
     );
     let conn = Connection::open(&path).unwrap();
     assert!(