feat: fence Subject handoff on recorded Journal and seed new bodies once

crates/worker/src/controller.rs

@@ -78,7 +78,45 @@ impl WorkerHandle {
         self.method_tx.same_channel(&other.method_tx)
     }
 
+    /// Ask the actual Controller to acknowledge quiescence. No catalog status,
+    /// profile, or model protocol message grants this Host lifecycle capability.
+    pub async fn reserve_subject_release(
+        &self,
+    ) -> Result<crate::SubjectReleaseReservation, String> {
+        if self.protocol_is_closed() {
+            return Err("Controller is unavailable".into());
+        }
+        if self.shared_state.snapshot().state != WorkerState::Idle {
+            return Err("Controller is not Idle".into());
+        }
+        let control = self
+            .shared_state
+            .subject_release
+            .get()
+            .ok_or("Controller release authority unavailable")?;
+        tokio::select! {
+            result = control.request() => result,
+            _ = self.method_tx.closed() => Err("Controller closed before release acknowledgement".into()),
+        }
+    }
+
     pub async fn send(&self, method: Method) -> Result<(), mpsc::error::SendError<Method>> {
+        // The lease covers enqueue itself. The Controller's exclusive barrier
+        // cannot overlook an accepted send between its queue check and ack.
+        let _admission = if release_fenced_method(&method) {
+            match self
+                .shared_state
+                .subject_release
+                .get()
+                .map(|control| control.authority.enter())
+                .transpose()
+            {
+                Ok(lease) => lease,
+                Err(_) => return Err(mpsc::error::SendError(method)),
+            }
+        } else {
+            None
+        };
         // Reject known-busy sends before channel dispatch, including maintenance
         // that cannot receive methods until it finishes. The controller checks
         // again at admission to cover an Idle -> busy race after this snapshot.
@@ -290,6 +328,49 @@ impl WorkerHandle {
     }
 }
 
+fn answer_idle_release<St: Store + Clone>(
+    control: &Arc<crate::subject_release::ReleaseControl>,
+    shared_state: &WorkerSharedState,
+    methods: &mpsc::Receiver<Method>,
+    staged: bool,
+    pending: &PendingSubmissionHandle<St>,
+    notifications: &NotifyBuffer,
+    attention: bool,
+) {
+    control.respond(|| {
+        // Read ingress and durable pending state only after acquiring exclusivity.
+        // Actor-local staged/deferred/attention flags cannot change during this call.
+        if shared_state.snapshot().state != WorkerState::Idle
+            || staged
+            || attention
+            || !methods.is_empty()
+            || methods.is_closed()
+            || !notifications.is_empty()
+            || pending.has_unfinished_admission()
+        {
+            return Err("Controller has pending, staged, busy, or unavailable work".into());
+        }
+        Ok(())
+    });
+}
+
+fn release_fenced_method(method: &Method) -> bool {
+    !matches!(
+        method,
+        Method::Shutdown { .. }
+            | Method::ListWorkers
+            | Method::ListPendingSubmissions
+            | Method::ListCompletions { .. }
+    )
+}
+
+async fn release_requested(shared_state: &WorkerSharedState) {
+    match shared_state.subject_release.get() {
+        Some(control) => control.notified().await,
+        None => std::future::pending::<()>().await,
+    }
+}
+
 fn command_admission_disposition(
     admission: WorkerCommandAdmission,
 ) -> Result<(), WorkerCommandDisposition> {
@@ -802,6 +883,10 @@ impl WorkerController {
         C: LlmClient + Clone + 'static,
         St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
     {
+        worker
+            .validate_subjektiv_body_publication()
+            .map_err(std::io::Error::other)?;
+
         // === 1. Initialization (channels / RuntimeDir / worker-immutable
         //         snapshots / SpawnedWorkerRegistry / alerter attach /
         //         bash-output scope) ===
@@ -911,13 +996,13 @@ impl WorkerController {
         // the actual registered set instead of a hand-maintained mirror.
         worker.engine().tool_server_handle().flush_pending();
 
-        // Runtime-owned Workers persist their current model-visible head before
+        // Fresh Workers persist their current model-visible head before
         // controller exposure. Feature installation above must happen first so a
         // fresh subject Worker captures the current resident surface in its durable
         // prompt. Restored Workers defer generic Feature restore notification until
         // the next model-visible context materialization; only an installed
         // Feature may then request a typed durable history append.
-        if runtime_managed && worker.needs_initial_session_head_materialization() {
+        if worker.needs_initial_session_head_materialization() {
             worker
                 .materialize_durable_session_head()
                 .await
@@ -958,6 +1043,13 @@ impl WorkerController {
         let artifact_store: Arc<dyn Store> = Arc::new(worker.store().clone());
         let session_id = worker.session_id();
         let pending_activations = worker.pending_activation_state();
+        let release_control =
+            crate::subject_release::ReleaseControl::new(worker.mutation_authority());
+        shared_state
+            .subject_release
+            .set(release_control)
+            .ok()
+            .expect("one Controller release authority");
         let handle = WorkerHandle {
             method_tx,
             working_event_tx: working_event_tx.clone(),
@@ -1561,6 +1653,18 @@ where
     {
         feature_registry.add_module(subjektiv_sessions);
     }
+    if ordinary_subjektiv_features_enabled
+        && lifecycle_enabled
+        && let Some(host) = subjektiv_host.as_ref()
+        && host.host.journal_startup_granted()
+    {
+        worker.install_initial_context_source(Arc::new(
+            crate::feature::builtin::journal::JournalStartupContextSource::new(
+                host.clone(),
+                worker.prompts(),
+            ),
+        ));
+    }
     if ordinary_subjektiv_features_enabled
         && lifecycle_enabled
         && let Some(host) = subjektiv_host.as_ref()
@@ -2232,6 +2336,12 @@ async fn controller_loop<C, St>(
     C: LlmClient + Clone + 'static,
     St: Store + WorkerMetadataStore + Clone + 'static,
 {
+    let release_control = shared_state
+        .subject_release
+        .get()
+        .expect("Controller release authority installed")
+        .clone();
+    let _release_lifetime = crate::subject_release::ControllerLifetime(release_control.clone());
     // Hold an optional external attach server alive for the controller lifetime.
     // In-process runtimes retain and drive the WorkerHandle directly.
     let _socket_server = socket_server;
@@ -2452,6 +2562,13 @@ async fn controller_loop<C, St>(
                         break 'controller;
                     }
                 },
+                _ = release_control.notified() => {
+                    answer_idle_release(&release_control, &shared_state, &method_rx,
+                        pending.is_some() || !deferred_methods.is_empty(),
+                        &pending_submissions, &notify_buffer,
+                        worker.has_pending_compaction_cleanup() || worker.has_failed_segment_activation());
+                    continue;
+                }
                 _ = notify_buffer.notified() => continue,
                 _ = tokio::time::sleep(remaining) => {
                     match prepare_pending_run(
@@ -2486,10 +2603,32 @@ async fn controller_loop<C, St>(
                         break 'controller;
                     }
                 },
+                _ = release_control.notified() => {
+                    answer_idle_release(&release_control, &shared_state, &method_rx,
+                        pending.is_some() || !deferred_methods.is_empty(),
+                        &pending_submissions, &notify_buffer,
+                        worker.has_pending_compaction_cleanup() || worker.has_failed_segment_activation());
+                    continue;
+                }
                 _ = notify_buffer.notified() => continue,
             }
         };
 
+        let _method_admission = if release_fenced_method(&method) {
+            match release_control.authority.enter() {
+                Ok(lease) => Some(lease),
+                Err(error) => {
+                    let _ = working_event_tx.send(Event::Error {
+                        code: ErrorCode::InvalidRequest,
+                        message: error.to_string(),
+                    });
+                    continue;
+                }
+            }
+        } else {
+            None
+        };
+
         if shared_state.snapshot().state != WorkerState::Idle
             && reject_idle_only_submission(&method, &working_event_tx)
         {
@@ -2885,6 +3024,7 @@ async fn controller_loop<C, St>(
                     let mut compact = Box::pin(worker.manual_compact_with_cancel(cancel_rx));
                     loop {
                         tokio::select! {
+                            _ = release_control.notified() => { release_control.reject_busy(); }
                             result = &mut compact => break result,
                             method = method_rx.recv() => {
                                 match method {
@@ -3142,6 +3282,7 @@ async fn controller_loop<C, St>(
         }
     }
 
+    release_control.close();
     // Close method admission before any fallible child/session cleanup. Existing
     // senders then fail instead of accepting work that this execution can no
     // longer process.
@@ -3286,6 +3427,9 @@ where
             // Running only after processing the commit fence. This makes the
             // Running snapshot contract deterministic even for immediate clients.
             biased;
+            _ = release_requested(shared_state) => {
+                if let Some(control) = shared_state.subject_release.get() { control.reject_busy(); }
+            }
             committed = async {
                 input_commit
                     .as_mut()
@@ -3906,6 +4050,10 @@ fn worker_error_code(e: &WorkerError) -> ErrorCode {
     }
 }
 
+#[cfg(test)]
+#[path = "controller_subject_release_tests.rs"]
+mod subject_release_tests;
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -4251,7 +4399,7 @@ mod tests {
     }
 
     #[test]
-    fn runtime_session_head_is_materialized_after_feature_installation_before_exposure() {
+    fn fresh_session_head_is_materialized_after_feature_installation_before_exposure() {
         let production = include_str!("controller.rs")
             .split_once("#[cfg(test)]\nmod tests")
             .map(|(production, _)| production)
@@ -4270,8 +4418,8 @@ mod tests {
             .find("let handle = WorkerHandle")
             .expect("controller handle exposure");
         let materialize_gate = startup[..materialize]
-            .rfind("if runtime_managed && worker.needs_initial_session_head_materialization() {")
-            .expect("fresh runtime-managed materialization gate");
+            .rfind("if worker.needs_initial_session_head_materialization() {")
+            .expect("fresh materialization gate");
 
         assert!(feature_install < materialize_gate);
         assert!(materialize_gate < materialize);

crates/worker/src/controller_subject_release_tests.rs

@@ -0,0 +1,474 @@
+//! Production Direct bootstrap, Controller admission, pending persistence and FsStore.
+//! Only the provider is replaced by an explicitly released async boundary; no live Host,
+//! Journal policy, socket or model continuation semantics are claimed.
+use super::*;
+use crate::{
+    PreparedWorker, WorkerBootstrap, WorkerBootstrapLayout, WorkerFilesystemAuthority,
+    WorkerManifest, WorkerWorkspaceContext,
+};
+use session_store::{CombinedStore, FsStore, FsWorkerStore};
+use tempfile::TempDir;
+
+#[derive(Clone)]
+struct HeldProvider {
+    started: mpsc::UnboundedSender<()>,
+    finish: Arc<tokio::sync::Notify>,
+}
+#[async_trait::async_trait]
+impl LlmClient for HeldProvider {
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+    async fn stream(
+        &self,
+        _: agen::llm_client::Request,
+    ) -> Result<
+        std::pin::Pin<
+            Box<
+                dyn futures_util::Stream<
+                        Item = Result<
+                            agen::llm_client::event::Event,
+                            agen::llm_client::ClientError,
+                        >,
+                    > + Send,
+            >,
+        >,
+        agen::llm_client::ClientError,
+    > {
+        self.started.send(()).unwrap();
+        self.finish.notified().await;
+        Ok(Box::pin(futures_util::stream::iter([Ok(
+            agen::llm_client::event::Event::Status(agen::llm_client::event::StatusEvent {
+                status: agen::llm_client::event::ResponseStatus::Completed,
+            }),
+        )])))
+    }
+}
+async fn prepared(
+    temp: &TempDir,
+    client: HeldProvider,
+) -> PreparedWorker<Box<dyn LlmClient>, CombinedStore<FsStore, FsWorkerStore>> {
+    let mut manifest = WorkerManifest::from_toml(&format!("[worker]\nname='release-{}'\n[model]\nscheme='anthropic'\nmodel_id='scripted'\n[engine]\n[scope]\nallow=[]", uuid::Uuid::now_v7())).unwrap();
+    manifest.compaction = Some(manifest::CompactionConfig {
+        retained_tokens: 0,
+        ..Default::default()
+    });
+    WorkerBootstrap::new(
+        manifest,
+        CombinedStore::new(
+            FsStore::new(temp.path().join("sessions")).unwrap(),
+            FsWorkerStore::new(temp.path().join("workers")).unwrap(),
+        ),
+        crate::PromptCatalogSource::builtins_only(),
+        WorkerWorkspaceContext::no_workspace(),
+        WorkerFilesystemAuthority::None,
+        WorkerBootstrapLayout::Direct {
+            runtime_base: temp.path().join("runtime"),
+            bash_output_dir: temp.path().join("bash"),
+        },
+        WorkerControllerTransport::InProcess,
+    )
+    .with_model_client(client)
+    .prepare()
+    .await
+    .unwrap()
+}
+async fn reserve(handle: &WorkerHandle) -> Result<crate::SubjectReleaseReservation, String> {
+    tokio::time::timeout(Duration::from_secs(5), handle.reserve_subject_release())
+        .await
+        .unwrap()
+}
+async fn shutdown(controller: crate::BootstrappedWorker) {
+    controller
+        .handle
+        .send(Method::Shutdown {
+            command: WorkerCommandEnvelope::new(1),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), controller.shutdown)
+        .await
+        .unwrap()
+        .unwrap();
+    controller.controller_task.await.unwrap();
+}
+fn extension() -> LogEntry {
+    LogEntry::Extension {
+        ts: 1,
+        domain: "release-test".into(),
+        payload: serde_json::json!({}),
+    }
+}
+
+#[tokio::test]
+async fn reservation_fences_ingress_fallback_and_history_drop_recovers_commit_stays_fenced() {
+    let temp = TempDir::new().unwrap();
+    let _sandbox =
+        crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+    let (started, mut starts) = mpsc::unbounded_channel();
+    let prepared = prepared(
+        &temp,
+        HeldProvider {
+            started,
+            finish: Arc::default(),
+        },
+    )
+    .await;
+    let writer = prepared.worker().log_writer_handle();
+    let pending = prepared.worker().pending_submission_handle();
+    let buffer = prepared.worker().notify_buffer_handle();
+    let capture = prepared.worker().committed_session_capture_handle();
+    let extensions = prepared.worker().session_extension_handle();
+    let fallback = durable_parent_notification_target(pending.clone(), buffer.clone());
+    let controller = prepared.start().await.unwrap();
+    let before = controller.handle.committed_entries().len();
+    let expected = capture.capture().unwrap().location();
+    let guard = reserve(&controller.handle).await.unwrap();
+    assert!(
+        controller
+            .handle
+            .send(Method::submit_text(
+                protocol::new_submission_request_id(),
+                "blocked"
+            ))
+            .await
+            .is_err()
+    );
+    assert!(
+        controller
+            .handle
+            .send(Method::Notify {
+                notification_request_id: protocol::new_submission_request_id(),
+                message: "blocked".into()
+            })
+            .await
+            .is_err()
+    );
+    assert!(
+        pending
+            .accept_notification("blocked".into(), "blocked".into())
+            .is_err()
+    );
+    fallback.notify("child-session".into(), "must not be admitted".into());
+    assert!(buffer.is_empty());
+    assert!(!pending.has_unfinished_admission());
+    assert!(writer.append_entry(extension()).is_err());
+    assert!(
+        extensions
+            .append_if_current(&expected, "release-test", serde_json::json!({}))
+            .is_err()
+    );
+    assert_eq!(controller.handle.committed_entries().len(), before);
+    assert!(starts.try_recv().is_err());
+    drop(guard);
+    writer.append_entry(extension()).unwrap();
+    let mut guard = reserve(&controller.handle).await.unwrap();
+    guard.commit_disconnect().unwrap();
+    guard.commit_disconnect().unwrap();
+    drop(guard);
+    assert!(writer.append_entry(extension()).is_err());
+    assert!(
+        pending
+            .accept_notification("after-commit".into(), "blocked".into())
+            .is_err()
+    );
+    assert!(
+        controller
+            .handle
+            .send(Method::Notify {
+                notification_request_id: protocol::new_submission_request_id(),
+                message: "blocked".into()
+            })
+            .await
+            .is_err()
+    );
+    assert!(reserve(&controller.handle).await.is_err());
+    let _ = controller.handle.snapshot_event();
+    controller
+        .handle
+        .send(Method::ListPendingSubmissions)
+        .await
+        .unwrap();
+    let handle = controller.handle.clone();
+    shutdown(controller).await;
+    assert!(handle.protocol_is_closed());
+    assert!(reserve(&handle).await.is_err());
+}
+
+#[tokio::test]
+async fn accepted_input_cannot_be_overtaken_by_release_and_running_request_is_rejected() {
+    let temp = TempDir::new().unwrap();
+    let _sandbox =
+        crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+    let (started, mut starts) = mpsc::unbounded_channel();
+    let finish = Arc::new(tokio::sync::Notify::new());
+    let controller = prepared(
+        &temp,
+        HeldProvider {
+            started,
+            finish: finish.clone(),
+        },
+    )
+    .await
+    .start()
+    .await
+    .unwrap();
+    let mut events = controller.handle.subscribe();
+    // Enqueue succeeds before submitting the separate internal release request.
+    // Either the queue or the held provider is observed; neither is quiescence.
+    controller
+        .handle
+        .send(Method::submit_text(
+            protocol::new_submission_request_id(),
+            "accepted first",
+        ))
+        .await
+        .unwrap();
+    assert!(reserve(&controller.handle).await.is_err());
+    tokio::time::timeout(Duration::from_secs(5), starts.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(reserve(&controller.handle).await.is_err());
+    finish.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop { if matches!(events.recv().await.unwrap(), Event::WorkerState { snapshot } if snapshot.state == WorkerState::Idle) { break; } }
+    }).await.unwrap();
+    let guard = reserve(&controller.handle).await.unwrap();
+    drop(guard);
+    shutdown(controller).await;
+}
+
+#[tokio::test]
+async fn staged_notification_fallback_blocks_release_until_the_actual_turn_finishes() {
+    let temp = TempDir::new().unwrap();
+    let _sandbox =
+        crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+    let (started, mut starts) = mpsc::unbounded_channel();
+    let finish = Arc::new(tokio::sync::Notify::new());
+    let prepared = prepared(
+        &temp,
+        HeldProvider {
+            started,
+            finish: finish.clone(),
+        },
+    )
+    .await;
+    let pending = prepared.worker().pending_submission_handle();
+    let buffer = prepared.worker().notify_buffer_handle();
+    let fallback = durable_parent_notification_target(pending.clone(), buffer);
+    let controller = prepared.start().await.unwrap();
+    let mut events = controller.handle.subscribe();
+    fallback.notify("child".into(), "durable accepted notification".into());
+    assert!(reserve(&controller.handle).await.is_err());
+    tokio::time::timeout(Duration::from_secs(5), starts.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    finish.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop { if matches!(events.recv().await.unwrap(), Event::WorkerState { snapshot } if snapshot.state == WorkerState::Idle) { break; } }
+    }).await.unwrap();
+    let guard = reserve(&controller.handle).await.unwrap();
+    drop(guard);
+    shutdown(controller).await;
+}
+
+#[derive(Clone)]
+struct HeldRewrite {
+    started: mpsc::UnboundedSender<()>,
+    finish: Arc<tokio::sync::Notify>,
+}
+impl crate::feature::FeatureModule for HeldRewrite {
+    fn descriptor(&self) -> crate::feature::FeatureDescriptor {
+        crate::feature::FeatureDescriptor::builtin("held-rewrite", "Held rewrite").with_hook(
+            crate::feature::HookDeclaration::new(
+                "hold",
+                crate::feature::FeatureHookPoint::BeforeSessionRewrite,
+            ),
+        )
+    }
+    fn install(
+        &self,
+        context: &mut crate::feature::FeatureInstallContext<'_>,
+    ) -> Result<(), crate::feature::FeatureInstallError> {
+        context.hooks().add_before_session_rewrite(
+            "hold",
+            crate::hook::HookExecutionPolicy::fail_closed(),
+            self.clone(),
+        )?;
+        Ok(())
+    }
+}
+#[async_trait::async_trait]
+impl crate::hook::Hook<crate::hook::BeforeSessionRewrite> for HeldRewrite {
+    async fn call(
+        &self,
+        context: &crate::hook::BeforeSessionRewriteContext,
+    ) -> Result<crate::hook::BeforeSessionRewriteAction, crate::hook::HookError> {
+        if context.kind == crate::hook::SessionRewriteKind::Compact {
+            self.started.send(()).unwrap();
+            self.finish.notified().await;
+            return Ok(crate::hook::BeforeSessionRewriteAction::Deny(
+                "stop before compactor boundary".into(),
+            ));
+        }
+        Ok(crate::hook::BeforeSessionRewriteAction::Continue)
+    }
+}
+
+#[tokio::test]
+async fn compact_maintenance_rejects_release_until_the_real_rewrite_boundary_exits() {
+    let temp = TempDir::new().unwrap();
+    let _sandbox =
+        crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+    let (started, mut starts) = mpsc::unbounded_channel();
+    let finish = Arc::new(tokio::sync::Notify::new());
+    let mut prepared = prepared(
+        &temp,
+        HeldProvider {
+            started,
+            finish: finish.clone(),
+        },
+    )
+    .await;
+    let (rewrite_started, mut rewrites) = mpsc::unbounded_channel();
+    let rewrite_finish = Arc::new(tokio::sync::Notify::new());
+    let mut registry = FeatureRegistryBuilder::new();
+    registry.add_module(HeldRewrite {
+        started: rewrite_started,
+        finish: rewrite_finish.clone(),
+    });
+    assert!(
+        !prepared
+            .worker_mut()
+            .install_features(registry)
+            .has_errors()
+    );
+    let controller = prepared.start().await.unwrap();
+    let mut events = controller.handle.subscribe();
+    controller
+        .handle
+        .send(Method::submit_text(
+            protocol::new_submission_request_id(),
+            "history to compact ".repeat(128),
+        ))
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), starts.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    finish.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop { if matches!(events.recv().await.unwrap(), Event::WorkerState { snapshot } if snapshot.state == WorkerState::Idle) { break; } }
+    }).await.unwrap();
+    controller
+        .handle
+        .send(Method::Compact {
+            command: WorkerCommandEnvelope::new(1),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), rewrites.recv())
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(reserve(&controller.handle).await.is_err());
+    assert!(matches!(
+        controller.handle.shared_state.snapshot().state,
+        WorkerState::Busy(WorkerBusyState::Maintenance(_))
+    ));
+    rewrite_finish.notify_one();
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop { if matches!(events.recv().await.unwrap(), Event::WorkerState { snapshot } if snapshot.state == WorkerState::Idle) { break; } }
+    }).await.unwrap();
+    let guard = reserve(&controller.handle).await.unwrap();
+    drop(guard);
+    controller
+        .handle
+        .send(Method::Shutdown {
+            command: WorkerCommandEnvelope::new(2),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), controller.shutdown)
+        .await
+        .unwrap()
+        .unwrap();
+    controller.controller_task.await.unwrap();
+}
+
+#[tokio::test]
+async fn queued_or_actor_staged_work_cannot_pass_the_exclusive_ack_boundary() {
+    let temp = TempDir::new().unwrap();
+    let _sandbox =
+        crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+    let (started, _) = mpsc::unbounded_channel();
+    let mut prepared = prepared(
+        &temp,
+        HeldProvider {
+            started,
+            finish: Arc::default(),
+        },
+    )
+    .await;
+    prepared
+        .worker_mut()
+        .materialize_durable_session_head()
+        .await
+        .unwrap();
+    let pending = prepared.worker().pending_submission_handle();
+    let buffer = prepared.worker().notify_buffer_handle();
+    let shared = WorkerSharedState::new(
+        "queue-boundary".into(),
+        prepared.worker().segment_id(),
+        String::new(),
+        build_greeting(prepared.worker()),
+    );
+    let control =
+        crate::subject_release::ReleaseControl::new(prepared.worker().mutation_authority());
+    for condition in [
+        "queued",
+        "staged",
+        "attention",
+        "closed",
+        "durable-notification",
+    ] {
+        let (sender, receiver) = mpsc::channel(1);
+        if condition == "queued" {
+            sender
+                .send(Method::submit_text(
+                    protocol::new_submission_request_id(),
+                    "already enqueued",
+                ))
+                .await
+                .unwrap();
+        }
+        if condition == "durable-notification" {
+            pending
+                .accept_notification("already-admitted".into(), "pending".into())
+                .unwrap();
+        }
+        if condition == "closed" {
+            drop(sender);
+        }
+        let request = tokio::spawn({
+            let control = control.clone();
+            async move { control.request().await }
+        });
+        control.notified().await;
+        answer_idle_release(
+            &control,
+            &shared,
+            &receiver,
+            condition == "staged",
+            &pending,
+            &buffer,
+            condition == "attention",
+        );
+        assert!(request.await.unwrap().is_err(), "{condition}");
+        // A failed request must not leave admission fenced for recovery.
+        assert!(prepared.worker().mutation_authority().enter().is_ok());
+    }
+}

crates/worker/src/feature/builtin/journal.rs

@@ -26,6 +26,59 @@ use subjektiv::journal::{
     MAX_JOURNAL_RESPONSE_BYTES,
 };
 
+/// Subjektiv owns startup projection; Core only commits its typed initial items.
+pub(crate) struct JournalStartupContextSource {
+    host: SubjektivHostConnection,
+    prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
+}
+impl JournalStartupContextSource {
+    pub(crate) fn new(
+        host: SubjektivHostConnection,
+        prompts: Arc<arc_swap::ArcSwap<crate::prompt::catalog::PromptCatalog>>,
+    ) -> Self {
+        Self { host, prompts }
+    }
+}
+#[async_trait]
+impl crate::worker::InitialContextSource for JournalStartupContextSource {
+    async fn load(&self) -> Result<Vec<session_store::SystemItem>, String> {
+        use sha2::{Digest, Sha256};
+        let snapshot = self
+            .host
+            .journal_startup_context()
+            .map_err(|e| e.to_string())?;
+        // Hash the same canonical JSON value retained by SystemItem, so replay
+        // can verify provenance without depending on domain struct field order.
+        let snapshot = serde_json::to_value(snapshot).map_err(|e| e.to_string())?;
+        let snapshot_json = serde_json::to_string(&snapshot).map_err(|e| e.to_string())?;
+        if snapshot_json.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES {
+            return Err("Host Journal startup snapshot exceeds independent byte budget".into());
+        }
+        let prompts = self.prompts.load();
+        let body = prompts
+            .subjektiv_journal_startup(&snapshot_json)
+            .map_err(|e| e.to_string())?;
+        if body.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES + 4096 {
+            return Err("Journal startup prompt exceeds rendered byte budget".into());
+        }
+        let projection = prompts.projection();
+        Ok(vec![session_store::SystemItem::JournalStartup {
+            body,
+            snapshot,
+            snapshot_sha256: Sha256::digest(snapshot_json.as_bytes())
+                .iter()
+                .map(|byte| format!("{byte:02x}"))
+                .collect(),
+            prompt_provenance: session_store::PromptRenderProvenance {
+                workspace_id: self.host.host.workspace_id(),
+                source_digest: projection.source_digest.clone(),
+                projection_digest: projection.catalog_digest.clone(),
+                logical_name: "internal.subjektiv_journal_startup".into(),
+            },
+        }])
+    }
+}
+
 #[derive(Clone)]
 pub struct JournalRecordingFeature {
     host: SubjektivHostConnection,

crates/worker/src/ipc/notify_buffer.rs

@@ -76,9 +76,16 @@ impl PendingNotify {
 pub struct NotifyBuffer {
     inner: Arc<Mutex<VecDeque<PendingNotify>>>,
     wake: Arc<tokio::sync::Notify>,
+    admission: Arc<std::sync::OnceLock<Arc<crate::subject_release::MutationAuthority>>>,
 }
 
 impl NotifyBuffer {
+    pub(crate) fn bind_mutation_authority(
+        &self,
+        authority: Arc<crate::subject_release::MutationAuthority>,
+    ) {
+        let _ = self.admission.set(authority);
+    }
     pub fn new() -> Self {
         Self::default()
     }
@@ -113,6 +120,18 @@ impl NotifyBuffer {
     }
 
     fn push_entry(&self, entry: PendingNotify) {
+        let _admission = match self
+            .admission
+            .get()
+            .map(|authority| authority.enter())
+            .transpose()
+        {
+            Ok(lease) => lease,
+            Err(error) => {
+                tracing::warn!(%error, "notification rejected by Subject release fence");
+                return;
+            }
+        };
         self.inner
             .lock()
             .expect("notify buffer poisoned")

crates/worker/src/lib.rs

@@ -25,6 +25,8 @@ mod session_capture;
 mod session_history;
 pub mod shared_state;
 mod shutdown_after_idle;
+mod subject_release;
+pub use subject_release::SubjectReleaseReservation;
 pub mod skill;
 pub mod spawn;
 pub mod subjektiv;

crates/worker/src/prompt/catalog.rs

@@ -102,6 +102,7 @@ pub enum WorkerPrompt {
     ResidentMemorySummarySection,
     ResidentMemoryRestoreSection,
     SubjektivResidentContext,
+    SubjektivJournalStartup,
     WorkerOrchestrationGuidanceSection,
     SubWorkerSpawnToolDescription,
 }
@@ -126,6 +127,7 @@ impl WorkerPrompt {
             Self::AgentsMdSection => "internal.agents_md_section",
             Self::ResidentMemorySummarySection => "internal.resident_memory_summary_section",
             Self::ResidentMemoryRestoreSection => "internal.resident_memory_restore_section",
+            Self::SubjektivJournalStartup => "internal.subjektiv_journal_startup",
             Self::SubjektivResidentContext => "internal.subjektiv_resident_context",
             Self::WorkerOrchestrationGuidanceSection => {
                 "internal.worker_orchestration_guidance_section"
@@ -151,6 +153,7 @@ impl WorkerPrompt {
         WorkerPrompt::ResidentMemorySummarySection,
         WorkerPrompt::ResidentMemoryRestoreSection,
         WorkerPrompt::SubjektivResidentContext,
+        WorkerPrompt::SubjektivJournalStartup,
         WorkerPrompt::WorkerOrchestrationGuidanceSection,
         WorkerPrompt::SubWorkerSpawnToolDescription,
     ];
@@ -540,6 +543,14 @@ impl PromptCatalog {
             },
         )
     }
+    pub fn subjektiv_journal_startup(&self, snapshot_json: &str) -> Result<String, CatalogError> {
+        self.render(
+            WorkerPrompt::SubjektivJournalStartup,
+            minijinja::context! {
+                snapshot_json => snapshot_json,
+            },
+        )
+    }
     pub fn subjektiv_resident_context(
         &self,
         behavior_md: &str,

crates/worker/src/shared_state.rs

@@ -50,6 +50,7 @@ pub struct WorkerSharedState {
     fs_view: OnceLock<WorkerFsView>,
     feature_invocations: RwLock<crate::feature::FeatureInvocationRegistry>,
     flow_transition_enabled: AtomicBool,
+    pub(crate) subject_release: OnceLock<std::sync::Arc<crate::subject_release::ReleaseControl>>,
 }
 
 impl WorkerSharedState {
@@ -69,6 +70,7 @@ impl WorkerSharedState {
             fs_view: OnceLock::new(),
             feature_invocations: RwLock::new(crate::feature::FeatureInvocationRegistry::default()),
             flow_transition_enabled: AtomicBool::new(false),
+            subject_release: OnceLock::new(),
         }
     }
 

crates/worker/src/subject_release.rs

@@ -0,0 +1,222 @@
+//! Controller-owned quiescence with a structural persistence/admission fence.
+//!
+//! A reservation is acknowledged only by the actual idle Controller. Shared read
+//! leases cover input enqueue, durable pending admission, and history writes; the
+//! exclusive lease spans the Host's reconciliation and ownership release. No
+//! Journal policy or catalog-status inference belongs here.
+
+use session_store::StoreError;
+use std::collections::VecDeque;
+use std::sync::{
+    Arc, Mutex,
+    atomic::{AtomicBool, Ordering},
+};
+use tokio::sync::{Notify, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock, oneshot};
+
+#[derive(Default)]
+pub(crate) struct MutationAuthority {
+    barrier: Arc<RwLock<()>>,
+    disconnected: AtomicBool,
+}
+impl MutationAuthority {
+    pub(crate) fn enter(&self) -> Result<OwnedRwLockReadGuard<()>, StoreError> {
+        let lease = self
+            .barrier
+            .clone()
+            .try_read_owned()
+            .map_err(|_| denied())?;
+        if self.disconnected.load(Ordering::Acquire) {
+            return Err(denied());
+        }
+        Ok(lease)
+    }
+}
+fn denied() -> StoreError {
+    StoreError::Io(std::io::Error::new(
+        std::io::ErrorKind::PermissionDenied,
+        "Subject release has fenced Worker admission and history mutation",
+    ))
+}
+
+type Reply = oneshot::Sender<Result<SubjectReleaseReservation, String>>;
+struct Requests {
+    alive: bool,
+    queue: VecDeque<Reply>,
+}
+pub(crate) struct ReleaseControl {
+    pub(crate) authority: Arc<MutationAuthority>,
+    requests: Mutex<Requests>,
+    changed: Notify,
+}
+impl ReleaseControl {
+    pub(crate) fn new(authority: Arc<MutationAuthority>) -> Arc<Self> {
+        Arc::new(Self {
+            authority,
+            requests: Mutex::new(Requests {
+                alive: true,
+                queue: VecDeque::new(),
+            }),
+            changed: Notify::new(),
+        })
+    }
+    pub(crate) async fn request(self: &Arc<Self>) -> Result<SubjectReleaseReservation, String> {
+        let (reply, receive) = oneshot::channel();
+        {
+            let mut requests = self
+                .requests
+                .lock()
+                .map_err(|_| "Controller release requests poisoned")?;
+            if !requests.alive {
+                return Err("Controller is unavailable".into());
+            }
+            requests.queue.push_back(reply);
+        }
+        self.changed.notify_one();
+        receive
+            .await
+            .map_err(|_| "Controller closed before release acknowledgement".to_string())?
+    }
+    pub(crate) async fn notified(&self) {
+        self.changed.notified().await;
+    }
+    pub(crate) fn respond(self: &Arc<Self>, check_quiescent: impl Fn() -> Result<(), String>) {
+        let requests = {
+            let mut state = self
+                .requests
+                .lock()
+                .expect("Controller release requests poisoned");
+            state.queue.drain(..).collect::<Vec<_>>()
+        };
+        for reply in requests {
+            if reply.is_closed() {
+                continue;
+            }
+            let result = self
+                .authority
+                .barrier
+                .clone()
+                .try_write_owned()
+                .map_err(|_| "Controller admission or history mutation is active".to_string())
+                .and_then(|exclusive| {
+                    if self.authority.disconnected.load(Ordering::Acquire) {
+                        return Err("Controller body is disconnected".into());
+                    }
+                    check_quiescent()?;
+                    Ok(SubjectReleaseReservation {
+                        control: self.clone(),
+                        _exclusive: exclusive,
+                        committed: false,
+                    })
+                });
+            // If the waiter disappeared, dropping the unsent guard cancels the fence.
+            let _ = reply.send(result);
+        }
+    }
+    pub(crate) fn reject_busy(self: &Arc<Self>) {
+        self.respond(|| Err("Controller is not quiescent Idle".into()));
+    }
+    pub(crate) fn close(&self) {
+        let mut state = self
+            .requests
+            .lock()
+            .expect("Controller release requests poisoned");
+        state.alive = false;
+        for reply in state.queue.drain(..) {
+            let _ = reply.send(Err("Controller is unavailable".into()));
+        }
+    }
+}
+
+pub(crate) struct ControllerLifetime(pub(crate) Arc<ReleaseControl>);
+impl Drop for ControllerLifetime {
+    fn drop(&mut self) {
+        self.0.close();
+    }
+}
+
+/// Exclusive permission to perform Host-side Subject handoff while the old body
+/// cannot accept input or change source history. Dropping an uncommitted guard
+/// cancels the reservation. Committing permanently revokes body admission; it
+/// does not stop the Controller or disable read-only observation and shutdown.
+pub struct SubjectReleaseReservation {
+    control: Arc<ReleaseControl>,
+    _exclusive: OwnedRwLockWriteGuard<()>,
+    committed: bool,
+}
+impl SubjectReleaseReservation {
+    pub fn commit_disconnect(&mut self) -> Result<(), String> {
+        if self.committed {
+            return Ok(());
+        }
+        let state = self
+            .control
+            .requests
+            .lock()
+            .map_err(|_| "Controller release requests poisoned")?;
+        if !state.alive {
+            return Err("Controller stopped before disconnect commit".into());
+        }
+        self.control
+            .authority
+            .disconnected
+            .store(true, Ordering::Release);
+        self.committed = true;
+        Ok(())
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    // Structural lease test: the Controller's grant path is production code;
+    // the held reader represents an already-entered synchronous persistence call.
+    #[tokio::test]
+    async fn in_flight_mutation_rejects_release_without_waiting_and_retry_recovers() {
+        let authority = Arc::new(MutationAuthority::default());
+        let control = ReleaseControl::new(authority.clone());
+        let (entered, entry) = oneshot::channel();
+        let (finish, finished) = oneshot::channel();
+        let writer = tokio::spawn({
+            let authority = authority.clone();
+            async move {
+                let _lease = authority.enter().unwrap();
+                entered.send(()).unwrap();
+                finished.await.unwrap();
+            }
+        });
+        entry.await.unwrap();
+        let request = tokio::spawn({
+            let control = control.clone();
+            async move { control.request().await }
+        });
+        control.notified().await;
+        control.respond(|| Ok(()));
+        assert!(request.await.unwrap().is_err());
+        finish.send(()).unwrap();
+        writer.await.unwrap();
+        let request = tokio::spawn({
+            let control = control.clone();
+            async move { control.request().await }
+        });
+        control.notified().await;
+        control.respond(|| Ok(()));
+        let guard = request.await.unwrap().unwrap();
+        assert!(authority.enter().is_err());
+        drop(guard);
+        assert!(authority.enter().is_ok());
+    }
+    #[tokio::test]
+    async fn closed_controller_rejects_unanswered_request_and_guard_disconnect() {
+        let control = ReleaseControl::new(Arc::default());
+        let request = tokio::spawn({
+            let control = control.clone();
+            async move { control.request().await }
+        });
+        control.notified().await;
+        control.respond(|| Ok(()));
+        let mut guard = request.await.unwrap().unwrap();
+        control.close();
+        assert!(guard.commit_disconnect().is_err());
+        assert!(control.request().await.is_err());
+    }
+}

crates/worker/src/subjektiv.rs

@@ -10,6 +10,7 @@ pub use crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifec
 pub use crate::feature::builtin::journal::{
     JournalJobFinalizer, JournalJobHost, JournalJobToolsFeature, JournalRecordingFeature,
 };
+use subjektiv::journal::startup::JournalStartupSnapshot;
 pub use subjektiv::journal::transport;
 use transport::{JournalRecordingReason, JournalRecordingRequest, JournalRecordingResponse};
 
@@ -66,6 +67,18 @@ impl From<SubjektivHostError> for ToolError {
 /// identifier is not a grant. No method grants ambient filesystem access.
 pub trait SubjektivHost: Send + Sync + 'static {
     fn settings(&self) -> SubjektivHostSettings;
+    /// Startup reads are a body-only capability, independent of recording Jobs.
+    fn journal_startup_granted(&self) -> bool {
+        false
+    }
+    fn journal_startup_context(
+        &self,
+        _context: &SubjektivHostContext,
+    ) -> Result<JournalStartupSnapshot, SubjektivHostError> {
+        Err(SubjektivHostError::Unavailable(
+            "Journal startup not granted".into(),
+        ))
+    }
     /// Recording admission is separate from both Memory and Journal Job operation grants.
     fn journal_recording_granted(&self) -> bool {
         false
@@ -207,6 +220,25 @@ fn decode<T: serde::de::DeserializeOwned>(
 }
 
 impl SubjektivHost for BackendSubjektivHost {
+    fn journal_startup_granted(&self) -> bool {
+        true
+    }
+    fn journal_startup_context(
+        &self,
+        _context: &SubjektivHostContext,
+    ) -> Result<JournalStartupSnapshot, SubjektivHostError> {
+        decode(
+            self.client
+                .execute_with_timeout(
+                    WorkspaceRequest::get(format!(
+                        "/api/w/{}/subjektiv/journal/startup",
+                        self.workspace_id
+                    )),
+                    Duration::from_secs(5),
+                )
+                .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
+        )
+    }
     fn journal_recording_granted(&self) -> bool {
         // This adapter exists only for explicitly enabled, authenticated Host connections.
         // The Backend still verifies the main Worker/Subject binding on every request.
@@ -420,6 +452,14 @@ pub struct SubjektivHostConnection {
 }
 
 impl SubjektivHostConnection {
+    pub fn journal_startup_context(&self) -> Result<JournalStartupSnapshot, SubjektivHostError> {
+        if !self.host.journal_startup_granted() {
+            return Err(SubjektivHostError::Unavailable(
+                "Journal startup not granted".into(),
+            ));
+        }
+        self.host.journal_startup_context(&self.context)
+    }
     pub fn new(
         host: Arc<dyn SubjektivHost>,
         context: SubjektivHostContext,
@@ -546,12 +586,33 @@ mod tests {
 
     #[derive(Default)]
     struct LocalHost {
+        startup: Option<JournalStartupSnapshot>,
+        startup_error: bool,
+        startup_calls: std::sync::atomic::AtomicUsize,
         contexts: Mutex<Vec<SubjektivHostContext>>,
         attributed: Mutex<Vec<(SubjektivHostContext, bool)>>,
         journal_observer:
             Option<Arc<dyn Fn(&SubjektivHostContext, JournalRecordingReason) + Send + Sync>>,
     }
     impl SubjektivHost for LocalHost {
+        fn journal_startup_granted(&self) -> bool {
+            self.startup.is_some() || self.startup_error
+        }
+        fn journal_startup_context(
+            &self,
+            _context: &SubjektivHostContext,
+        ) -> Result<JournalStartupSnapshot, SubjektivHostError> {
+            self.startup_calls
+                .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if self.startup_error {
+                return Err(SubjektivHostError::Unavailable(
+                    "startup read failed".into(),
+                ));
+            }
+            self.startup
+                .clone()
+                .ok_or_else(|| SubjektivHostError::Unavailable("startup not granted".into()))
+        }
         fn journal_recording_granted(&self) -> bool {
             self.journal_observer.is_some()
         }
@@ -919,6 +980,223 @@ enabled = true
         );
     }
 
+    // Production Direct bootstrap, Feature selection/rendering, Core, FsStore,
+    // and model Request. The authenticated Host boundary and provider are fake;
+    // real Backend/standalone authority and live continuation are not claimed.
+    #[tokio::test]
+    async fn direct_startup_journal_is_durable_before_first_user_and_provider_exactly_once() {
+        use session_store::{LogEntry, LoggedSessionHistoryOrigin, Store, SystemItem};
+        use std::sync::atomic::Ordering;
+        for empty in [false, true] {
+            let temp = tempfile::tempdir().unwrap();
+            let _sandbox =
+                crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+            let name = format!("startup-{}", uuid::Uuid::now_v7());
+            let mut manifest = WorkerManifest::from_toml(&format!("[worker]\nname='{name}'\n[model]\nscheme='anthropic'\nmodel_id='scripted'\n[engine]\n[scope]\nallow=[]")).unwrap();
+            manifest.feature.subjektiv.profile.enabled = true;
+            manifest.feature.subjektiv.profile.extraction.enabled = false;
+            let snapshot = JournalStartupSnapshot {
+                subject_id: "subject".into(),
+                omitted: false,
+                journals: if empty {
+                    vec![]
+                } else {
+                    vec![subjektiv::journal::startup::StartupJournal {
+                        journal_id: "journal".into(),
+                        source_session_id: "previous-session".into(),
+                        state_token: "consistent-token".into(),
+                        process_order: 4,
+                        process_byte: 0,
+                        current_position: "B adopted after correcting A; implementation unfinished"
+                            .into(),
+                        entries: vec![],
+                        omitted: true,
+                    }]
+                },
+            };
+            let host = Arc::new(LocalHost {
+                startup: Some(snapshot.clone()),
+                ..LocalHost::default()
+            });
+            let store = CombinedStore::new(
+                FsStore::new(temp.path().join("sessions")).unwrap(),
+                FsWorkerStore::new(temp.path().join("workers")).unwrap(),
+            );
+            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+            let prepared = WorkerBootstrap::new(
+                manifest,
+                store.clone(),
+                crate::PromptCatalogSource::builtins_only(),
+                WorkerWorkspaceContext::local_filesystem(None),
+                WorkerFilesystemAuthority::None,
+                WorkerBootstrapLayout::Direct {
+                    runtime_base: temp.path().join("runtime"),
+                    bash_output_dir: temp.path().join("bash"),
+                },
+                WorkerControllerTransport::InProcess,
+            )
+            .with_model_client(ScriptedClient(tx))
+            .with_subjektiv_host(host.clone())
+            .prepare()
+            .await
+            .unwrap();
+            let (session, segment) = (
+                prepared.worker().session_id(),
+                prepared.worker().segment_id(),
+            );
+            let controller = prepared.start().await.unwrap();
+            let initial = store.read_all(session, segment).unwrap();
+            let journal = initial
+                .iter()
+                .find_map(|entry| match entry {
+                    LogEntry::AnnotatedSystemItem { entry, .. }
+                        if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
+                    {
+                        Some(entry)
+                    }
+                    _ => None,
+                })
+                .unwrap();
+            assert_eq!(
+                journal.metadata.origin,
+                LoggedSessionHistoryOrigin::DerivedSummary
+            );
+            let SystemItem::JournalStartup {
+                body,
+                snapshot: saved,
+                snapshot_sha256,
+                prompt_provenance,
+            } = &journal.item
+            else {
+                unreachable!()
+            };
+            assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap());
+            use sha2::{Digest, Sha256};
+            assert_eq!(
+                *snapshot_sha256,
+                Sha256::digest(serde_json::to_vec(saved).unwrap())
+                    .iter()
+                    .map(|byte| format!("{byte:02x}"))
+                    .collect::<String>()
+            );
+            assert_eq!(
+                prompt_provenance.logical_name,
+                "internal.subjektiv_journal_startup"
+            );
+            assert!(
+                !initial
+                    .iter()
+                    .any(|e| matches!(e, LogEntry::AnnotatedUserInput { .. }))
+            );
+            let mut events = controller.handle.subscribe();
+            for user in ["FIRST_USER_REQUEST", "SECOND_USER_REQUEST"] {
+                controller
+                    .handle
+                    .send(Method::submit_text(
+                        protocol::new_submission_request_id(),
+                        user,
+                    ))
+                    .await
+                    .unwrap();
+                let request = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+                    .await
+                    .unwrap()
+                    .unwrap();
+                let index = request
+                    .items
+                    .iter()
+                    .position(|item| item.as_text() == Some(body.as_str()))
+                    .unwrap();
+                let user_index = request
+                    .items
+                    .iter()
+                    .position(|item| item.as_text() == Some("FIRST_USER_REQUEST"))
+                    .unwrap();
+                assert!(index < user_index);
+                assert_eq!(
+                    request
+                        .items
+                        .iter()
+                        .filter(|item| item.as_text() == Some(body.as_str()))
+                        .count(),
+                    1
+                );
+                tokio::time::timeout(Duration::from_secs(5), async {
+                    loop {
+                        if matches!(events.recv().await.unwrap(), protocol::Event::RunEnd { .. }) {
+                            break;
+                        }
+                    }
+                })
+                .await
+                .unwrap();
+            }
+            assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
+            let logs = store.read_all(session, segment).unwrap();
+            assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).count(), 1);
+            controller
+                .handle
+                .send(Method::Shutdown {
+                    command: protocol::WorkerCommandEnvelope::new(1),
+                })
+                .await
+                .unwrap();
+            tokio::time::timeout(Duration::from_secs(5), controller.shutdown)
+                .await
+                .unwrap()
+                .unwrap();
+            controller.controller_task.await.unwrap();
+        }
+    }
+
+    #[tokio::test]
+    async fn direct_startup_read_error_rejects_exposure_without_initialized_head() {
+        use session_store::Store;
+        let temp = tempfile::tempdir().unwrap();
+        let _sandbox =
+            crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+        let mut manifest = WorkerManifest::from_toml(&format!("[worker]\nname='startup-error-{}'\n[model]\nscheme='anthropic'\nmodel_id='scripted'\n[engine]\n[scope]\nallow=[]", uuid::Uuid::now_v7())).unwrap();
+        manifest.feature.subjektiv.profile.enabled = true;
+        manifest.feature.subjektiv.profile.extraction.enabled = false;
+        let store = CombinedStore::new(
+            FsStore::new(temp.path().join("sessions")).unwrap(),
+            FsWorkerStore::new(temp.path().join("workers")).unwrap(),
+        );
+        let host = Arc::new(LocalHost {
+            startup_error: true,
+            ..LocalHost::default()
+        });
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let prepared = WorkerBootstrap::new(
+            manifest,
+            store.clone(),
+            crate::PromptCatalogSource::builtins_only(),
+            WorkerWorkspaceContext::local_filesystem(None),
+            WorkerFilesystemAuthority::None,
+            WorkerBootstrapLayout::Direct {
+                runtime_base: temp.path().join("runtime"),
+                bash_output_dir: temp.path().join("bash"),
+            },
+            WorkerControllerTransport::InProcess,
+        )
+        .with_model_client(ScriptedClient(tx))
+        .with_subjektiv_host(host.clone())
+        .prepare()
+        .await
+        .unwrap();
+        let (session, segment) = (
+            prepared.worker().session_id(),
+            prepared.worker().segment_id(),
+        );
+        assert!(prepared.start().await.is_err());
+        assert!(!store.exists(session, segment).unwrap());
+        assert!(rx.try_recv().is_err());
+        assert_eq!(
+            host.startup_calls.load(std::sync::atomic::Ordering::SeqCst),
+            1
+        );
+    }
+
     async fn launch_case(enabled: bool, attach: bool) {
         let temp = tempfile::tempdir().unwrap();
         let _runtime_sandbox =
@@ -1038,7 +1316,7 @@ enabled = true
         if enabled && attach {
             assert_eq!(
                 *host.attributed.lock().unwrap(),
-                vec![(expected.clone(), true)]
+                vec![(expected.clone(), true), (expected.clone(), false)]
             );
             assert!(
                 host.contexts