fix: bind startup cause persistence to confirmed Session transitions

2026-11-04 12:18:37 +09:00
parent 9e404b0021
commit 21fe1a7a31
10 changed files with 516 additions and 74 deletions
@@ -196,6 +196,10 @@ pub enum StartupFailurePhase {
    RuntimeRecordDelete,
    RuntimeSnapshotSave,
    RuntimeRemovalPublish,
+    /// Retaining create reconciliation observations in the canonical Worker record.
+    RuntimeReconciliationRecordSave,
+    /// Publishing a retained record to Runtime subscribers; not a durable write.
+    RuntimeReconciliationPublish,
    MetadataSave,
 }

@@ -323,7 +323,8 @@ impl StartupFailureExpectation {
        })
    }

-    fn matches(&self, metadata: &WorkerMetadata) -> bool {
+    /// Compare the complete observation without adopting newer ownership or causes.
+    pub fn matches(&self, metadata: &WorkerMetadata) -> bool {
        metadata.worker_name == self.worker_name
            && metadata.workspace_id == self.workspace_id
            && metadata.active.as_ref().map(|active| active.session_id) == Some(self.session_id)
@@ -1569,7 +1569,7 @@ impl Runtime {
                });
            }
            WorkerExecutionSpawnResult::ReconciliationRequired {
-                result,
+                mut result,
                worker_state,
                workdir_attachments,
            } => {
@@ -1582,12 +1582,33 @@ impl Runtime {
                } else {
                    self.retain_create_reconciliation_pending(&worker_ref, workdir_attachments)
                };
-                if let Err(error) = retain_result {
+                if let Err((phase, error)) = retain_result {
+                    let outcome = if matches!(error, RuntimeError::StoreCommitOutcomeUnknown { .. })
+                    {
+                        StartupPersistenceOutcome::OutcomeUnknown
+                    } else {
+                        StartupPersistenceOutcome::NotSaved
+                    };
+                    if phase == StartupFailurePhase::RuntimeReconciliationPublish {
+                        result.push_secondary_failure(StartupSecondaryFailure::Persistence {
+                            phase: StartupFailurePhase::RuntimeReconciliationRecordSave,
+                            outcome: StartupPersistenceOutcome::Persisted,
+                        });
+                    }
+                    result.push_secondary_failure(StartupSecondaryFailure::Persistence {
+                        phase,
+                        outcome,
+                    });
                    tracing::error!(
                        worker_id = %worker_ref.worker_id,
                        %error,
                        "failed to persist Worker create reconciliation evidence"
                    );
+                } else {
+                    result.push_secondary_failure(StartupSecondaryFailure::Persistence {
+                        phase: StartupFailurePhase::RuntimeReconciliationRecordSave,
+                        outcome: StartupPersistenceOutcome::Persisted,
+                    });
                }
                return Err(RuntimeError::WorkerExecutionRejected {
                    worker_id: worker_ref.worker_id.clone(),
@@ -5193,9 +5214,10 @@ impl Runtime {

        worker_state: protocol::WorkerStateSnapshot,
        workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
-    ) -> Result<(), RuntimeError> {
-        let mut state = self.lock()?;
-        let mut candidate = state.worker(worker_ref)?.clone();
+    ) -> Result<(), (StartupFailurePhase, RuntimeError)> {
+        let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error);
+        let mut state = self.lock().map_err(record_error)?;
+        let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone();

        candidate.restore_guard.rotate();
        candidate.execution_metadata_available = true;
@@ -5204,9 +5226,13 @@ impl Runtime {
        candidate.apply_worker_state(&worker_state);
        candidate.restore_intent = WorkerRestoreIntent::Automatic;
        candidate.workdir_attachments = workdir_attachments;
-        state.persist_worker_record(&candidate)?;
+        state
+            .persist_worker_record(&candidate)
+            .map_err(record_error)?;
        state.workers.insert(worker_ref.worker_id, candidate);
-        state.publish_worker_upsert(worker_ref.worker_id)?;
+        state
+            .publish_worker_upsert(worker_ref.worker_id)
+            .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?;
        Ok(())
    }

@@ -5228,9 +5254,10 @@ impl Runtime {
        &self,
        worker_ref: &WorkerRef,
        workdir_attachments: Vec<crate::catalog::WorkingDirectoryAttachmentStatus>,
-    ) -> Result<(), RuntimeError> {
-        let mut state = self.lock()?;
-        let mut candidate = state.worker(worker_ref)?.clone();
+    ) -> Result<(), (StartupFailurePhase, RuntimeError)> {
+        let record_error = |error| (StartupFailurePhase::RuntimeReconciliationRecordSave, error);
+        let mut state = self.lock().map_err(record_error)?;
+        let mut candidate = state.worker(worker_ref).map_err(record_error)?.clone();

        candidate.restore_guard.rotate();
        candidate.execution_metadata_available = false;
@@ -5246,8 +5273,12 @@ impl Runtime {
            logical_workdir_attachments_from_statuses(&workdir_attachments, claims);
        candidate.workdir_attachments = workdir_attachments;
        state.workers.insert(worker_ref.worker_id, candidate);
-        state.persist_worker(&worker_ref.worker_id)?;
-        state.publish_worker_upsert(worker_ref.worker_id)?;
+        state
+            .persist_worker(&worker_ref.worker_id)
+            .map_err(record_error)?;
+        state
+            .publish_worker_upsert(worker_ref.worker_id)
+            .map_err(|error| (StartupFailurePhase::RuntimeReconciliationPublish, error))?;
        Ok(())
    }

@@ -8379,6 +8410,8 @@ mod tests {
            result
        }

+        include!("runtime/create_reconciliation_tests.rs");
+
        fn check_rollback(fault: Fault) {
            let primary = primary_failure();
            for spawn_result in [
@@ -0,0 +1,113 @@
+mod create_reconciliation_tests {
+    //! The real create retention and FsRuntimeStore write/reopen path; only spawn
+    //! disposition and the existing deterministic atomic-write hook are substituted.
+    use super::*;
+    use crate::fs_store::AtomicWriteFault;
+
+    fn check_reconciliation(fault: AtomicWriteFault) {
+        for has_state in [false, true] {
+            let root = tempfile::tempdir().unwrap();
+            let options = FsRuntimeStoreOptions::new(root.path().join("runtime"))
+                .with_runtime_id("create-reconciliation");
+            let backend = Arc::new(TestExecutionBackend::default());
+            let runtime =
+                Runtime::with_fs_store_and_execution_backend(options.clone(), backend.clone())
+                    .unwrap();
+            runtime.store_config_bundle(test_bundle()).unwrap();
+            let request = task_request("reconciliation failure");
+            let worker_ref = WorkerRef::new(request.worker_id);
+            let primary = primary_failure();
+            *backend.spawn_result.lock().unwrap() =
+                Some(WorkerExecutionSpawnResult::ReconciliationRequired {
+                    result: primary.clone(),
+                    worker_state: has_state.then(|| protocol::WorkerStatus::Stopped.into()),
+                    workdir_attachments: Vec::new(),
+                });
+            let initial = Arc::new(Mutex::new(None));
+            *backend.before_spawn_result.lock().unwrap() = Some(Box::new({
+                let store = runtime_store(&runtime);
+                let initial = initial.clone();
+                let worker_ref = worker_ref.clone();
+                move || {
+                    *initial.lock().unwrap() = Some(store.read_worker_record(&worker_ref).unwrap());
+                    store.fail_next_worker_write(fault);
+                }
+            }));
+            let RuntimeError::WorkerExecutionRejected { result, .. } =
+                runtime.create_worker(request).unwrap_err()
+            else {
+                panic!("retention failure must not replace primary result")
+            };
+            assert_eq!(result.startup_failure, primary.startup_failure);
+            assert_eq!(result.startup_resources, primary.startup_resources);
+            assert_eq!(result.outcome, primary.outcome);
+            assert!(
+                result
+                    .secondary_failures
+                    .contains(&primary.secondary_failures[0])
+            );
+            let outcome = match fault {
+                AtomicWriteFault::BeforeRename => StartupPersistenceOutcome::NotSaved,
+                AtomicWriteFault::AfterRename => StartupPersistenceOutcome::OutcomeUnknown,
+                AtomicWriteFault::None => StartupPersistenceOutcome::Persisted,
+            };
+            assert!(result.secondary_failures.iter().any(|failure| matches!(failure, StartupSecondaryFailure::Persistence { phase: StartupFailurePhase::RuntimeReconciliationRecordSave, outcome: observed } if *observed == outcome)), "retention persistence must be explicit: {result:?}");
+            let disk = runtime_store(&runtime)
+                .read_worker_record(&worker_ref)
+                .unwrap();
+            let initial = initial.lock().unwrap().take().unwrap();
+            if fault == AtomicWriteFault::BeforeRename {
+                assert_eq!(disk.execution_state, initial.execution_state);
+                assert_eq!(
+                    disk.restore_guard.observation_token,
+                    initial.restore_guard.observation_token
+                );
+            } else {
+                assert_ne!(
+                    disk.restore_guard.observation_token,
+                    initial.restore_guard.observation_token
+                );
+                if has_state {
+                    assert!(
+                        matches!(disk.execution_state, crate::fs_store::PersistedWorkerExecutionState::Available(ref execution) if execution.binding.is_some())
+                    );
+                } else {
+                    assert_eq!(
+                        disk.execution_state,
+                        crate::fs_store::PersistedWorkerExecutionState::Unavailable
+                    );
+                }
+            }
+            // A pre-rename failure left the initial record; an uncertain response
+            // here followed a real rename and is reconciled from disk, not guessed.
+            let on_disk = std::fs::read(
+                root.path()
+                    .join("runtime/workers")
+                    .join(worker_ref.worker_id.to_string())
+                    .join("worker.json"),
+            )
+            .unwrap();
+            drop(runtime);
+            drop(backend);
+            let reopened = Runtime::with_fs_store(options).unwrap();
+            let reread = runtime_store(&reopened)
+                .read_worker_record(&worker_ref)
+                .unwrap();
+            assert_eq!(reread.execution_state, disk.execution_state);
+            assert!(reopened.worker_detail(&worker_ref).is_ok());
+            assert!(!on_disk.is_empty());
+        }
+    }
+    #[test]
+    fn create_reconciliation_write_failure_retains_primary_and_unsaved_evidence() {
+        check_reconciliation(AtomicWriteFault::BeforeRename);
+    }
+    #[test]
+    fn create_reconciliation_unknown_write_retains_primary_and_reopens_actual_record() {
+        check_reconciliation(AtomicWriteFault::AfterRename);
+    }
+    #[test]
+    fn create_reconciliation_success_retains_primary_and_durable_evidence() {
+        check_reconciliation(AtomicWriteFault::None);
+    }
+}
@@ -2251,34 +2251,7 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory {
                .map_err(str::to_string)?;
        }
        validate_worker_memory_settings(worker.manifest(), &request.request)?;
-        let expected = worker
-            .store()
-            .read_by_name(&worker_name)?
-            .and_then(|metadata| {
-                session_store::worker_metadata::StartupFailureExpectation::from_metadata(&metadata)
-            });
-        worker
-            .finalize_subjektiv_session_attribution(
-                request.request.subjektiv_attached,
-                SubjektivSessionAttributionLifecycle::NewSession,
-            )
-            .map_err(|error| {
-                let error = if let Some(expected) = &expected {
-                    worker::startup_failure::persist_startup_failure::<
-                        _,
-                        protocol::startup_failure::RuntimeStartupCause,
-                    >(
-                        worker.store(),
-                        expected,
-                        error,
-                        protocol::startup_failure::StartupStage::Binding,
-                        &(),
-                    )
-                } else {
-                    error
-                };
-                RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error)
-            })?;
+        startup_failure::finalize_startup_binding(worker, request.request.subjektiv_attached)?;
        let workdir_capabilities = request
            .workdir_attachments
            .keys()
@@ -7187,6 +7160,7 @@ mod tests {
    use worker::PreparedWorker;

    include!("subject_delete_backend_tests.rs");
+    include!("worker_backend/startup_binding_tests.rs");
    #[cfg(feature = "fs-store")]
    include!("subject_delete_create_tests.rs");

@@ -0,0 +1,253 @@
+mod startup_binding_tests {
+    //! Exercise the factory's production binding step with a constructed Worker,
+    //! real aggregate/Session stores and reopen. Only the Workspace response is
+    //! synthetic; this suite does not test domain authorization or network transport.
+    use super::*;
+    use protocol::startup_failure::{RuntimeStartupCause, StartupPersistenceOutcome};
+    use session_store::worker_metadata::StartupFailureRead;
+    use worker::{WorkspaceClientError, WorkspaceRequest, WorkspaceResponse};
+
+    #[derive(Debug, Clone, Copy)]
+    enum Response {
+        Rejected,
+        Lost,
+        BeforeSession,
+        AfterSession,
+        BeforeOwner,
+        AfterOwner,
+        AfterAttribution,
+    }
+    struct AttributionClient {
+        response: Response,
+        store: WorkerAggregateStore,
+        name: String,
+    }
+    impl std::fmt::Debug for AttributionClient {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            f.write_str("AttributionClient")
+        }
+    }
+    impl WorkspaceClient for AttributionClient {
+        fn workspace_id(&self) -> Option<&str> {
+            Some("startup-binding-workspace")
+        }
+        fn kind(&self) -> &str {
+            "startup-binding-fixture"
+        }
+        fn is_available(&self) -> bool {
+            true
+        }
+        fn execute(&self, _: WorkspaceRequest) -> Result<WorkspaceResponse, WorkspaceClientError> {
+            match self.response {
+                Response::AfterSession => self
+                    .store
+                    .update_by_name(&self.name, |metadata| {
+                        metadata.active = Some(WorkerActiveSegmentRef::pending_segment(
+                            session_store::new_session_id(),
+                        ));
+                        metadata.subjektiv_session_attribution = None;
+                    })
+                    .unwrap(),
+                Response::AfterOwner => self
+                    .store
+                    .update_by_name(&self.name, |metadata| {
+                        metadata.workspace_id = Some("replacement-owner".into());
+                    })
+                    .unwrap(),
+                Response::AfterAttribution => self
+                    .store
+                    .update_by_name(&self.name, |metadata| {
+                        metadata.subjektiv_session_attribution =
+                            Some(session_store::SubjektivSessionAttributionState::Confirmed {
+                                session_id: metadata.active.as_ref().unwrap().session_id,
+                                subject_id: "replacement-subject".into(),
+                            });
+                    })
+                    .unwrap(),
+                _ => self.store.read_by_name(&self.name).unwrap().unwrap(),
+            };
+            if matches!(self.response, Response::Lost) {
+                return Err(WorkspaceClientError::Unavailable(
+                    "response deliberately lost".into(),
+                ));
+            }
+            Ok(WorkspaceResponse {
+                status: 403,
+                body: "synthetic rejection with private token /internal/path".into(),
+            })
+        }
+    }
+    async fn check_binding(response: Response) {
+        require_restore_test_environment();
+        let root = tempfile::tempdir().unwrap();
+        let name = "startup-binding";
+        let store = WorkerAggregateStore::new(root.path(), name).unwrap();
+        let session_store = WorkerSessionStore::new(root.path().join("session")).unwrap();
+        let client = Arc::new(AttributionClient {
+            response,
+            store: store.clone(),
+            name: name.into(),
+        });
+        let manifest = WorkerManifest::from_toml(
+            r#"
+            [worker]
+            name = "startup-binding"
+            [model]
+            scheme = "anthropic"
+            model_id = "unused"
+            auth = { kind = "none" }
+            [engine]
+            max_tokens = 100
+            [scope]
+            allow = []
+        "#,
+        )
+        .unwrap();
+        let llm: Box<dyn LlmClient> = Box::new(MockClient::new(Vec::new()));
+        let mut worker = Worker::new(
+            manifest,
+            Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(llm),
+            CombinedStore::new(session_store, store.clone()),
+            WorkerWorkspaceContext::with_client(
+                Some(WorkspaceId::new("startup-binding-workspace").unwrap()),
+                client,
+            ),
+            WorkerFilesystemAuthority::None,
+            Scope::empty(),
+        )
+        .await
+        .unwrap();
+        worker.enable_worker_metadata_write_through().unwrap();
+        let origin = worker.session_id();
+        match response {
+            Response::BeforeSession => store
+                .update_by_name(name, |metadata| {
+                    metadata.active = Some(WorkerActiveSegmentRef::pending_segment(
+                        session_store::new_session_id(),
+                    ));
+                })
+                .unwrap(),
+            Response::BeforeOwner => store
+                .update_by_name(name, |metadata| {
+                    metadata.workspace_id = Some("replacement-owner".into());
+                })
+                .unwrap(),
+            _ => store.read_by_name(name).unwrap().unwrap(),
+        };
+        let failure = startup_failure::finalize_startup_binding(&worker, true)
+            .err()
+            .expect("failure must remain a failure");
+        let result = failure.execution_result(WorkerExecutionOperation::Spawn);
+        let primary = result.startup_failure::<RuntimeStartupCause>().unwrap();
+        assert!(matches!(primary, error_contract::DecodeFailure::Known(_)));
+        drop(worker);
+        drop(store);
+        let reopened = WorkerAggregateStore::new(root.path(), name).unwrap();
+        let metadata = reopened.read_by_name(name).unwrap().unwrap();
+        match response {
+            Response::Rejected | Response::Lost => {
+                assert_eq!(
+                    result.startup_failure_persistence,
+                    Some(StartupPersistenceOutcome::Persisted)
+                );
+                let StartupFailureRead::Present(error_contract::DecodeFailure::Known(cause)) =
+                    metadata.decode_startup_failure::<RuntimeStartupCause>()
+                else {
+                    panic!("same-Session failure must actually save a typed cause")
+                };
+                assert_eq!(
+                    cause.facts().cause().reason,
+                    protocol::startup_failure::RuntimeStartupReason::AttributionFailed
+                );
+                assert_eq!(metadata.active.unwrap().session_id, origin);
+                let saved = serde_json::to_string(&metadata.startup_failure).unwrap();
+                assert!(!saved.contains("private token"));
+                assert!(!saved.contains("/internal/path"));
+                assert_eq!(
+                    metadata
+                        .subjektiv_session_attribution
+                        .as_ref()
+                        .unwrap()
+                        .session_id(),
+                    origin
+                );
+                assert_eq!(
+                    matches!(
+                        metadata.subjektiv_session_attribution,
+                        Some(
+                            session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. }
+                        )
+                    ),
+                    matches!(response, Response::Lost)
+                );
+            }
+            _ => {
+                assert!(
+                    matches!(
+                        metadata.decode_startup_failure::<RuntimeStartupCause>(),
+                        StartupFailureRead::Missing
+                    ),
+                    "replacement Session/owner must not acquire stale cause"
+                );
+                assert_ne!(
+                    result.startup_failure_persistence,
+                    Some(StartupPersistenceOutcome::Persisted)
+                );
+                if matches!(response, Response::BeforeSession | Response::AfterSession) {
+                    assert_ne!(metadata.active.unwrap().session_id, origin);
+                }
+                if matches!(
+                    response,
+                    Response::BeforeOwner | Response::BeforeSession | Response::AfterSession
+                ) {
+                    assert!(metadata.subjektiv_session_attribution.is_none());
+                }
+                if matches!(response, Response::AfterOwner) {
+                    assert!(matches!(
+                        metadata.subjektiv_session_attribution,
+                        Some(
+                            session_store::SubjektivSessionAttributionState::OutcomeUnknown { .. }
+                        )
+                    ));
+                }
+                if matches!(response, Response::BeforeOwner | Response::AfterOwner) {
+                    assert_eq!(metadata.workspace_id.as_deref(), Some("replacement-owner"));
+                }
+                if matches!(response, Response::AfterAttribution) {
+                    assert!(matches!(
+                        metadata.subjektiv_session_attribution,
+                        Some(session_store::SubjektivSessionAttributionState::Confirmed { .. })
+                    ));
+                }
+            }
+        }
+    }
+    #[tokio::test]
+    async fn factory_rejection_saves_same_session_cause_after_own_attribution_transition() {
+        check_binding(Response::Rejected).await;
+    }
+    #[tokio::test]
+    async fn factory_unknown_response_saves_cause_without_claiming_attribution_success() {
+        check_binding(Response::Lost).await;
+    }
+    #[tokio::test]
+    async fn factory_session_replaced_before_capture_rejects_stale_cause() {
+        check_binding(Response::BeforeSession).await;
+    }
+    #[tokio::test]
+    async fn factory_session_replaced_after_capture_rejects_stale_cause() {
+        check_binding(Response::AfterSession).await;
+    }
+    #[tokio::test]
+    async fn factory_owner_replaced_before_capture_rejects_stale_cause() {
+        check_binding(Response::BeforeOwner).await;
+    }
+    #[tokio::test]
+    async fn factory_owner_replaced_after_capture_rejects_stale_cause() {
+        check_binding(Response::AfterOwner).await;
+    }
+    #[tokio::test]
+    async fn factory_attribution_replaced_after_capture_preserves_new_attribution() {
+        check_binding(Response::AfterAttribution).await;
+    }
+}
@@ -102,6 +102,21 @@ impl PendingRuntimeRestore {
    }
 }

+pub(super) fn finalize_startup_binding<St>(
+    worker: &Worker<Box<dyn agen::llm_client::LlmClient>, St>,
+    attached: bool,
+) -> Result<(), RuntimeRestoreFailure>
+where
+    St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
+{
+    worker
+        .finalize_subjektiv_session_attribution_retaining_startup_failure(
+            attached,
+            SubjektivSessionAttributionLifecycle::NewSession,
+        )
+        .map_err(|error| RuntimeRestoreFailure::from_worker(RuntimeRestoreStage::Binding, &error))
+}
+
 #[cfg(test)]
 mod tests {
    //! Connect the typed factory seam to the real Tokio adapter and protocol codec.
@@ -3,8 +3,7 @@

 use crate::metadata_lifecycle::SessionMetadataLifecycle;
 use session_store::{
-    SessionId, SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata,
-    WorkerMetadataStore,
+    SubjektivSessionAttributionState, WorkerActiveSegmentRef, WorkerMetadata, WorkerMetadataStore,
 };

 pub(crate) struct SessionAttributionMetadata;
@@ -25,18 +24,13 @@ impl SessionMetadataLifecycle for SessionAttributionMetadata {

 pub(crate) fn compare_and_swap_attribution<St: WorkerMetadataStore>(
    store: &St,
-    worker_name: &str,
-    session_id: SessionId,
-    expected: Option<&SubjektivSessionAttributionState>,
+    expected: &mut session_store::worker_metadata::StartupFailureExpectation,
    replacement: SubjektivSessionAttributionState,
 ) -> Result<(), crate::WorkerError> {
    let mut matched = false;
-    store.update_by_name(worker_name, |metadata| {
-        if metadata.active.as_ref().map(|active| active.session_id) == Some(session_id)
-            && metadata.subjektiv_session_attribution.as_ref() == expected
-            && replacement.session_id() == session_id
-        {
-            metadata.subjektiv_session_attribution = Some(replacement);
+    store.update_by_name(&expected.worker_name, |metadata| {
+        if expected.matches(metadata) && replacement.session_id() == expected.session_id {
+            metadata.subjektiv_session_attribution = Some(replacement.clone());
            matched = true;
        }
    })?;
@@ -47,6 +41,8 @@ pub(crate) fn compare_and_swap_attribution<St: WorkerMetadataStore>(
                .into(),
        });
    }
+    // Advance only after the exact owner/Session CAS returned confirmed success.
+    expected.attribution = Some(replacement);
    Ok(())
 }

@@ -9698,6 +9698,48 @@ where
        &self,
        subjektiv_attached: bool,
        lifecycle: SubjektivSessionAttributionLifecycle,
+    ) -> Result<(), WorkerError> {
+        self.finalize_subjektiv_session_attribution_observed(
+            subjektiv_attached,
+            lifecycle,
+            &mut None,
+        )
+    }
+
+    /// Retain a failed startup against its constructed Session and Workspace. Only
+    /// this operation's successful CAS transitions advance the attribution expectation;
+    /// a reread never adopts a replacement owner or another operation's attribution.
+    pub fn finalize_subjektiv_session_attribution_retaining_startup_failure(
+        &self,
+        subjektiv_attached: bool,
+        lifecycle: SubjektivSessionAttributionLifecycle,
+    ) -> Result<(), WorkerError> {
+        let mut expected = None;
+        self.finalize_subjektiv_session_attribution_observed(
+            subjektiv_attached,
+            lifecycle,
+            &mut expected,
+        )
+        .map_err(|error| match expected {
+            Some(expected) => crate::startup_failure::persist_startup_failure::<
+                _,
+                protocol::startup_failure::RuntimeStartupCause,
+            >(
+                &self.store,
+                &expected,
+                error,
+                protocol::startup_failure::StartupStage::Binding,
+                &(),
+            ),
+            None => error,
+        })
+    }
+
+    fn finalize_subjektiv_session_attribution_observed(
+        &self,
+        subjektiv_attached: bool,
+        lifecycle: SubjektivSessionAttributionLifecycle,
+        observation: &mut Option<session_store::worker_metadata::StartupFailureExpectation>,
    ) -> Result<(), WorkerError> {
        let _admission = self.segment_state.mutation_authority.enter()?;
        let worker_name = self.manifest.worker.name.as_str();
@@ -9726,6 +9768,31 @@ where
            });
        }

+        if metadata.worker_name != worker_name
+            || metadata.workspace_id.as_deref() != self.workspace_id().map(|id| id.as_str())
+        {
+            return Err(WorkerError::SubjektivSessionAttribution {
+                state: "owner_mismatch",
+                message: "Worker metadata does not match constructed Workspace ownership".into(),
+            });
+        }
+        *observation =
+            session_store::worker_metadata::StartupFailureExpectation::from_metadata(&metadata);
+        let mut persist_attribution =
+            |previous: Option<&SubjektivSessionAttributionState>,
+             next: SubjektivSessionAttributionState| {
+                let expected = observation
+                    .as_mut()
+                    .expect("validated active Session has an expectation");
+                if expected.attribution.as_ref() != previous {
+                    return Err(WorkerError::SubjektivSessionAttribution {
+                        state: "attribution_conflict",
+                        message: "startup attribution expectation changed".into(),
+                    });
+                }
+                crate::subjektiv_metadata::compare_and_swap_attribution(&self.store, expected, next)
+            };
+
        if self
            .backend_job()
            .is_some_and(crate::BackendJobExecutionBinding::has_subjektiv_job_grant)
@@ -9783,7 +9850,7 @@ where
            create_if_missing,
            initial_diagnostic.to_string(),
        );
-        self.persist_subjektiv_session_attribution(
+        persist_attribution(
            metadata.subjektiv_session_attribution.as_ref(),
            pending_attribution.clone(),
        )?;
@@ -9819,7 +9886,7 @@ where
                            message.clone(),
                        )
                    };
-                    self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?;
+                    persist_attribution(Some(&pending_attribution), state)?;
                    return Err(WorkerError::SubjektivSessionAttribution {
                        state: if create_if_missing && rejected {
                            "failed"
@@ -9844,7 +9911,7 @@ where
                    let message = bounded_attribution_diagnostic(format!(
                        "Session attribution transport outcome is unknown: {error}"
                    ));
-                    self.persist_subjektiv_session_attribution(
+                    persist_attribution(
                        Some(&pending_attribution),
                        incomplete_subjektiv_attribution_state(
                            session_id,
@@ -9874,7 +9941,7 @@ where
                        diagnostic: message.clone(),
                    }
                };
-                self.persist_subjektiv_session_attribution(Some(&pending_attribution), state)?;
+                persist_attribution(Some(&pending_attribution), state)?;
                return Err(WorkerError::SubjektivSessionAttribution {
                    state: if create_if_missing {
                        "failed"
@@ -9891,7 +9958,7 @@ where
                        let message = bounded_attribution_diagnostic(format!(
                            "Session attribution response outcome is unknown: {error}"
                        ));
-                        self.persist_subjektiv_session_attribution(
+                        persist_attribution(
                            Some(&pending_attribution),
                            incomplete_subjektiv_attribution_state(
                                session_id,
@@ -9914,7 +9981,7 @@ where
            let message =
                "Session attribution response did not confirm the requested immutable identity"
                    .to_string();
-            self.persist_subjektiv_session_attribution(
+            persist_attribution(
                Some(&pending_attribution),
                incomplete_subjektiv_attribution_state(
                    session_id,
@@ -9927,7 +9994,7 @@ where
                message,
            });
        }
-        self.persist_subjektiv_session_attribution(
+        persist_attribution(
            Some(&pending_attribution),
            SubjektivSessionAttributionState::Confirmed {
                session_id,
@@ -9972,20 +10039,6 @@ where
        Ok(())
    }

-    fn persist_subjektiv_session_attribution(
-        &self,
-        expected: Option<&SubjektivSessionAttributionState>,
-        state: SubjektivSessionAttributionState,
-    ) -> Result<(), WorkerError> {
-        crate::subjektiv_metadata::compare_and_swap_attribution(
-            &self.store,
-            &self.manifest.worker.name,
-            self.session_id(),
-            expected,
-            state,
-        )
-    }
-
    /// Build an in-process Internal Worker without machine-wide allocation or durable Worker metadata.
    pub(crate) async fn from_internal_manifest_with_context(
        manifest: WorkerManifest,
@@ -12,7 +12,7 @@ export type RuntimeStartupReason = "startup_failed" | "segment_empty" | "system_

 export type RuntimeStartupCause = { stage: StartupStage, reason: RuntimeStartupReason, };

-export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "metadata_save";
+export type StartupFailurePhase = "cleanup" | "runtime_record_delete" | "runtime_snapshot_save" | "runtime_removal_publish" | "runtime_reconciliation_record_save" | "runtime_reconciliation_publish" | "metadata_save";

 export type StartupCleanupFailure = "ownership" | "workdir" | "artifacts" | "allocation" | "feature" | "unsupported";