fix: fence Journal originals before publication and reconcile removal intent

Protect attributed originals before the first public commit; distinguish pending Session material from missing committed sources. Gate removal admission by the Journal Feature and resolve Subject ownership from durable singleton authority. Resume explicit erasure through signed delete transport and proof-bound purge plans without creating a Journal Job or archive.

Validation: changed-crate suites, root check, doc/fmt/diff/naming checks, Server/Runtime builds and worker_environment isolation suite passed. Public standalone scripted dialog connects Host, automatic Job, sequential tools and drain while dialog continues. Backend public embedded WorkerRemove tests connect removal forwarding, durable Jobs, retention, Runtime and SQLite; Session/attribution/profile fixture setup bypasses creation/auth routes. Runtime signed generated-client tests separately exercise request authentication.

Limits: real-model semantic acceptance condition 6 remains unevaluated without a new explicit model/credential/budget grant. Conditions 1/7/8/11 do not have complete product E2E coverage: worker_environment only verifies isolated startup/connection/cleanup. Compaction tests connect local hooks/reader/store rather than the full public dialog-to-Journal chain. Erasure racing a previously prepared Archive removal plan fails closed with conflict; automatic stale-plan replacement is not implemented.

crates/runtime-api/src/lib.rs

@@ -803,6 +803,20 @@ pub struct SessionMaterialDescriptor {
     pub total_bytes: u64,
 }
 
+/// Explicit Host erasure, distinct from ordinary retention policy or pin release.
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct SessionMaterialEraseRequest {
+    pub workspace_id: String,
+    pub source: SessionPublicSource,
+    pub expected_session_id: String,
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+pub struct SessionMaterialEraseResponse {
+    pub session_id: String,
+}
+
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct SessionMaterialPinRequest {
@@ -860,6 +874,11 @@ pub struct SessionMaterialReadPage {
 #[derive(Clone, Debug, Serialize, Deserialize)]
 #[serde(tag = "availability", rename_all = "snake_case")]
 pub enum SessionMaterialCatalogAvailability {
+    /// The saved active pointer is pending and storage proves no Session material
+    /// has ever been published. Missing/corrupt materialized sources are not empty.
+    NotMaterialized {
+        session_id: String,
+    },
     Page {
         page: SessionMaterialCatalogPage,
     },
@@ -1123,6 +1142,16 @@ pub trait RuntimeApi {
         #[query] request: WorkerSessionHistoryRequest,
     ) -> Result<WorkerSessionHistoryAvailability, RuntimeApiError>;
 
+    #[post(
+        "/v1/session-public/material/retention/erase",
+        status = 200,
+        error_status = 400
+    )]
+    async fn session_material_erase(
+        &self,
+        #[body] request: SessionMaterialEraseRequest,
+    ) -> Result<SessionMaterialEraseResponse, RuntimeApiError>;
+
     #[post(
         "/v1/session-public/material/retention/pin",
         status = 200,
@@ -1588,6 +1617,13 @@ mod tests {
             })
         }
 
+        async fn session_material_erase(
+            &self,
+            _request: SessionMaterialEraseRequest,
+        ) -> Result<SessionMaterialEraseResponse, RuntimeApiError> {
+            Err(test_error(501))
+        }
+
         async fn session_material_pin(
             &self,
             _request: SessionMaterialPinRequest,

crates/session-store/src/fs_store.rs

@@ -329,6 +329,14 @@ impl FsStore {
 }
 
 impl Store for FsStore {
+    fn retain_session_before_publication(
+        &self,
+        session_id: SessionId,
+        pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        self.retention().retain_before_publication(session_id, pin)
+    }
+
     fn append(
         &self,
         session_id: SessionId,

crates/session-store/src/retention.rs

@@ -109,6 +109,23 @@ impl SessionRetentionStore {
         Ok(guard)
     }
 
+    /// Protect a Host-owned Session before its first source publication. This is
+    /// retention intent, not proof that any original material exists. The owner must
+    /// establish it before accepting input and use the same control store for writes
+    /// and deletion. Source-serving APIs must still verify actual material separately.
+    pub fn retain_before_publication(
+        &self,
+        session: SessionId,
+        pin: SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        let _guard = self.lock_live(session)?;
+        let mut state = self.read(session)?;
+        if state.pins.insert(pin) {
+            self.save(session, &state)?;
+        }
+        Ok(())
+    }
+
     /// Call `source_exists` under the same lock used by deletion. Replaying the
     /// same pin is idempotent; missing source never becomes a successful lease.
     pub fn pin(
@@ -184,6 +201,43 @@ pub(crate) fn fs_retention(root: &Path) -> SessionRetentionStore {
 mod tests {
     use super::*;
 
+    #[test]
+    fn publication_intent_survives_reopen_without_attesting_missing_originals() {
+        let temp = tempfile::tempdir().unwrap();
+        let source = temp.path().join("source");
+        let controls = temp.path().join("controls");
+        let session = crate::new_session_id();
+        let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+        let retention = SessionRetentionStore::new(&controls);
+        retention.retain_before_publication(session, pin).unwrap();
+        let reopened = SessionRetentionStore::new(&controls);
+        assert_eq!(reopened.pins(session).unwrap(), [pin]);
+        // An early owner is not a successful source-availability receipt.
+        assert!(
+            matches!(reopened.pin(session, pin, || Ok(source.is_file())), Err(StoreError::SessionSourceMissing(id)) if id == session)
+        );
+        {
+            let _publication = reopened.lock_live(session).unwrap();
+            fs::write(&source, "committed before notification").unwrap();
+        }
+        assert!(
+            matches!(reopened.delete(session, SessionDeletionIntent::Ordinary, || { fs::remove_file(&source)?; Ok(()) }), Err(StoreError::SessionPinned(id)) if id == session)
+        );
+        assert_eq!(
+            fs::read_to_string(&source).unwrap(),
+            "committed before notification"
+        );
+        reopened
+            .delete(session, SessionDeletionIntent::ErasePermanently, || {
+                fs::remove_file(&source)?;
+                Ok(())
+            })
+            .unwrap();
+        assert!(
+            matches!(retention.retain_before_publication(session, pin), Err(StoreError::SessionErased(id)) if id == session)
+        );
+    }
+
     #[test]
     fn pin_survives_reopen_blocks_delete_and_releases_idempotently() {
         let temp = tempfile::tempdir().unwrap();

crates/session-store/src/store.rs

@@ -42,6 +42,9 @@ pub enum StoreError {
     #[error("session was permanently erased: {0}")]
     SessionErased(SessionId),
 
+    #[error("session source retention is unsupported by this store")]
+    SessionRetentionUnsupported,
+
     #[error("retained observation read limit exceeded")]
     ReadLimitExceeded,
 
@@ -89,6 +92,17 @@ pub enum StoreError {
 /// take `(SessionId, SegmentId)` so segments can be physically grouped
 /// per Session on disk (or per session_id in a DB).
 pub trait Store: Send + Sync {
+    /// Establish durable retention intent before the Host accepts public input.
+    /// This does not attest source availability; it must use the same control store
+    /// as publishers and physical deletion. Unsupported stores fail closed.
+    fn retain_session_before_publication(
+        &self,
+        _session_id: SessionId,
+        _pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        Err(StoreError::SessionRetentionUnsupported)
+    }
+
     /// Append a single log entry to the segment log.
     ///
     /// One committed line per successful call. Implementations must not expose

crates/session-store/src/worker_metadata.rs

@@ -806,6 +806,15 @@ where
     S: crate::Store,
     W: Send + Sync,
 {
+    fn retain_session_before_publication(
+        &self,
+        session_id: SessionId,
+        pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), crate::StoreError> {
+        self.session_store
+            .retain_session_before_publication(session_id, pin)
+    }
+
     fn append(
         &self,
         session_id: SessionId,

crates/session-store/src/worker_session_store.rs

@@ -716,6 +716,17 @@ impl WorkerSessionStore {
 }
 
 impl Store for WorkerSessionStore {
+    fn retain_session_before_publication(
+        &self,
+        session_id: SessionId,
+        pin: crate::retention::SessionRetentionPin,
+    ) -> Result<(), StoreError> {
+        self.retention
+            .as_ref()
+            .ok_or(StoreError::SessionRetentionUnsupported)?
+            .retain_before_publication(session_id, pin)
+    }
+
     fn append(
         &self,
         session_id: SessionId,

crates/standalone/src/subjektiv/journal/tests.rs

@@ -659,6 +659,253 @@ impl agen::llm_client::LlmClient for JournalScript {
     }
 }
 
+#[derive(Clone, Default)]
+struct DialogAndJournal {
+    journal: JournalScript,
+    dialog_calls: Arc<std::sync::atomic::AtomicUsize>,
+}
+#[async_trait::async_trait]
+impl agen::llm_client::LlmClient for DialogAndJournal {
+    async fn stream(
+        &self,
+        request: agen::llm_client::Request,
+    ) -> Result<
+        std::pin::Pin<
+            Box<
+                dyn futures::Stream<
+                        Item = Result<
+                            agen::llm_client::event::Event,
+                            agen::llm_client::error::ClientError,
+                        >,
+                    > + Send,
+            >,
+        >,
+        agen::llm_client::error::ClientError,
+    > {
+        if request
+            .tools
+            .iter()
+            .any(|tool| tool.name == "ReadJournalState")
+        {
+            return agen::llm_client::LlmClient::stream(&self.journal, request).await;
+        }
+        use agen::llm_client::{
+            event::{Event, StopReason},
+            types::Item,
+        };
+        let surface = request
+            .tools
+            .iter()
+            .any(|tool| tool.name == "SubmitMemorySurface");
+        if surface
+            || request
+                .tools
+                .iter()
+                .any(|tool| tool.name == "SubmitJobResult")
+        {
+            let events = if request.items.iter().any(|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "empty-memory")) {
+                vec![Event::text_block_start(0), Event::text_delta(0, "empty Memory acknowledged"), Event::text_block_stop(0, Some(StopReason::EndTurn))]
+            } else {
+                let (tool, value) = if surface {
+                    ("SubmitMemorySurface", serde_json::json!({"points":[]}))
+                } else {
+                    let input: serde_json::Value = request.items.iter().filter_map(Item::as_text).find_map(|text| text.lines().find_map(|line| line.strip_prefix("input_json: "))).map(|json| serde_json::from_str(json).unwrap()).expect("Memory Job input");
+                    assert_eq!(input["candidate_ids"], serde_json::json!([]), "fixture has no staged Memory; do not synthesize decisions");
+                    ("SubmitJobResult", serde_json::json!({"result":input}))
+                };
+                vec![Event::tool_use_start(0, "empty-memory", tool), Event::tool_input_delta(0, value.to_string()), Event::tool_use_stop(0)]
+            };
+            return Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))));
+        }
+        self.dialog_calls.fetch_add(1, Ordering::SeqCst);
+        Ok(Box::pin(futures::stream::iter(
+            [
+                Event::text_block_start(0),
+                Event::text_delta(0, "public scripted response"),
+                Event::text_block_stop(0, Some(StopReason::EndTurn)),
+            ]
+            .into_iter()
+            .map(Ok),
+        )))
+    }
+    fn clone_boxed(&self) -> Box<dyn agen::llm_client::LlmClient> {
+        Box::new(self.clone())
+    }
+}
+
+async fn submit_dialog(host: &crate::StandaloneHost, input: &str) {
+    let mut client = host.connect();
+    client
+        .send(&protocol::Method::submit_text(
+            protocol::new_submission_request_id(),
+            input,
+        ))
+        .await
+        .unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(20), async {
+        let mut answered = false;
+        loop {
+            match client.next_event().await.unwrap().unwrap() {
+                protocol::Event::TextDelta { text }
+                    if text.contains("public scripted response") =>
+                {
+                    answered = true
+                }
+                protocol::Event::RunEnd { .. } => {
+                    assert!(answered);
+                    return;
+                }
+                protocol::Event::Error { code, message } => panic!("{code:?}: {message}"),
+                _ => {}
+            }
+        }
+    })
+    .await
+    .expect("bounded main dialog completion");
+}
+
+/// Public standalone launch/protocol -> real controller/Worker Feature -> LocalSubjectHost ->
+/// original FsStore -> generic Job runner/tools -> Journal read API. Only the two model clients
+/// are scripted; no manual admission, generation reservation, source append or domain mutation.
+#[tokio::test]
+async fn public_host_dialog_auto_records_while_generation_waits_and_drains_without_another_input() {
+    let temp = tempfile::tempdir().unwrap();
+    let state = temp.path().join("state");
+    let catalog = StandaloneSubjects::open(&state).unwrap();
+    let subject = catalog.create("public-dialog-owner", None).unwrap();
+    let model = DialogAndJournal {
+        journal: JournalScript {
+            gate_first: true,
+            ..Default::default()
+        },
+        ..Default::default()
+    };
+    let launch = crate::StandaloneLaunchConfig::new(
+        temp.path(),
+        &state,
+        manifest::ProfileSelector::source_named(
+            manifest::ProfileRegistrySource::Builtin,
+            "standalone-subjektiv",
+        ),
+        "public-journal-dialog",
+    )
+    .with_subject(subject.id.clone())
+    .resolve()
+    .unwrap();
+    let host = crate::StandaloneHost::start_with_model_client(launch, model.clone())
+        .await
+        .unwrap();
+    let session = host.record().active_session_id;
+    assert_eq!(
+        crate::store::session_retention(&state)
+            .pins(session)
+            .unwrap(),
+        [session_store::retention::SessionRetentionPin(session)]
+    );
+    submit_dialog(&host, "first small public turn").await;
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        model.journal.entered.notified(),
+    )
+    .await
+    .unwrap();
+    let reader = JournalHostContext::validated_reader(&catalog.store, &subject.id).unwrap();
+    let journal = catalog
+        .store
+        .list_journals(&reader, None, 1)
+        .unwrap()
+        .items
+        .pop()
+        .unwrap();
+    let attr = catalog
+        .store
+        .session_attribution(&session.to_string())
+        .unwrap()
+        .unwrap();
+    let context = JournalHostContext::validated_session(&catalog.store, attr).unwrap();
+    let first = catalog
+        .store
+        .active_journal_generation(&context, &journal.id)
+        .unwrap()
+        .unwrap();
+    submit_dialog(&host, "second small public turn").await;
+    submit_dialog(&host, "last input; no further turn follows").await;
+    assert_eq!(model.dialog_calls.load(Ordering::SeqCst), 3);
+    assert_eq!(
+        catalog
+            .store
+            .active_journal_generation(&context, &journal.id)
+            .unwrap()
+            .unwrap(),
+        first,
+        "running generation/input remain frozen while dialog proceeds"
+    );
+    model.journal.release.notify_one();
+    let jobs = host.jobs();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        jobs.wait(&first.job.job_id),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        model.journal.entered.notified(),
+    )
+    .await
+    .unwrap();
+    let second = catalog
+        .store
+        .active_journal_generation(&context, &journal.id)
+        .unwrap()
+        .unwrap();
+    assert_ne!(first.id, second.id);
+    model.journal.release.notify_one();
+    tokio::time::timeout(
+        std::time::Duration::from_secs(20),
+        jobs.wait(&second.job.job_id),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    assert!(!catalog.store.journal_retention_required(&context).unwrap());
+    let saved = catalog.store.read_journal(&reader, &journal.id).unwrap();
+    assert!(saved.current_position_bytes > 0);
+    let position = catalog
+        .store
+        .read_journal_text(
+            &reader,
+            &journal.id,
+            None,
+            JournalTextField::CurrentPosition,
+            None,
+            0,
+            MAX_JOURNAL_READ_BYTES,
+        )
+        .unwrap();
+    assert!(!position.text.is_empty());
+    let (_, open) = catalog
+        .store
+        .journal_resume_state(&context, &journal.id)
+        .unwrap();
+    let open = open.unwrap();
+    let body = catalog
+        .store
+        .read_journal_text(
+            &reader,
+            &journal.id,
+            Some(&open.id),
+            JournalTextField::Body,
+            None,
+            0,
+            MAX_JOURNAL_READ_BYTES,
+        )
+        .unwrap();
+    assert!(body.text.contains("last input"));
+    host.shutdown().await.unwrap();
+}
+
 #[tokio::test]
 async fn real_job_tools_save_and_drain_last_input_without_another_turn() {
     let fixture = Fixture::new("Proposed A, corrected to B; partly performed; interrupted");

crates/subjektiv/src/journal/mod.rs

@@ -62,6 +62,25 @@ pub const MAX_JOURNAL_READ_BYTES: usize = 16 * 1024;
 /// Whole pretty-serialized public page budget, including escaped text and metadata.
 pub const MAX_JOURNAL_RESPONSE_BYTES: usize = 64 * 1024;
 
+/// Stable owner of Backend source retention, shared by attribution-time protection
+/// and later source admission/release. This identifier confers no read authority.
+/// Local standalone storage uses its scope-exclusive Session UUID as the owner.
+pub fn source_retention_pin(workspace: &str, subject: &str, session: &str) -> Uuid {
+    use sha2::{Digest, Sha256};
+    let digest = Sha256::digest(
+        serde_json::to_vec(&(
+            "subjektiv-journal-retention-v1",
+            workspace,
+            subject,
+            session,
+        ))
+        .expect("string tuple serialization"),
+    );
+    let mut bytes = [0; 16];
+    bytes.copy_from_slice(&digest[..16]);
+    Uuid::from_bytes(bytes)
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum JournalError {
     #[error("journal permission denied: {0}")]

crates/worker-runtime/src/http_server/runtime_management_api.rs

@@ -369,6 +369,43 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
             .map_err(api_error)
     }
 
+    async fn session_material_erase(
+        &self,
+        request: runtime_api::SessionMaterialEraseRequest,
+    ) -> Result<runtime_api::SessionMaterialEraseResponse, runtime_api::RuntimeApiError> {
+        let scope = material_scope(&self.state, &request.workspace_id)?;
+        #[cfg(feature = "fs-store")]
+        {
+            self.state
+                .runtime
+                .delete_session_public_source_scoped(
+                    &scope,
+                    &request.source,
+                    &request.expected_session_id,
+                    session_store::retention::SessionDeletionIntent::ErasePermanently,
+                )
+                .map_err(|_| {
+                    runtime_api::RuntimeApiError::new(
+                        409,
+                        "session_erasure_incomplete",
+                        "Session erasure could not complete; retain erasure intent and retry",
+                    )
+                })?;
+            Ok(runtime_api::SessionMaterialEraseResponse {
+                session_id: request.expected_session_id,
+            })
+        }
+        #[cfg(not(feature = "fs-store"))]
+        {
+            let _ = scope;
+            Err(runtime_api::RuntimeApiError::new(
+                503,
+                "session_retention_unavailable",
+                "Session erasure requires filesystem storage",
+            ))
+        }
+    }
+
     async fn session_material_pin(
         &self,
         request: runtime_api::SessionMaterialPinRequest,

crates/worker-runtime/src/http_server/tests/session_material_tests.rs

@@ -194,6 +194,101 @@ impl Fixture {
     }
 }
 
+#[test]
+fn pending_session_proof_is_not_missing_material_and_pin_release_allows_empty_archive() {
+    let f = Fixture::new();
+    std::fs::remove_dir_all(f.root()).unwrap();
+    let aggregate = f.root().parent().unwrap().to_path_buf();
+    let metadata =
+        WorkerAggregateStore::new(&aggregate, format!("worker-runtime-{}", f.worker.worker_id))
+            .unwrap();
+    metadata
+        .write(
+            &session_store::WorkerMetadata::new(
+                format!("worker-runtime-{}", f.worker.worker_id),
+                Some(session_store::WorkerActiveSegmentRef::pending_segment(
+                    f.session,
+                )),
+            )
+            .with_workspace_id(&f.scope.workspace_id),
+        )
+        .unwrap();
+    let pin = SessionRetentionPin(uuid::Uuid::now_v7());
+    let retention = runtime_session_retention(f.temp.path());
+    retention.retain_before_publication(f.session, pin).unwrap();
+    assert!(
+        matches!(f.runtime.session_material_catalog_scoped(&f.scope, &f.request()), Catalog::NotMaterialized { session_id } if session_id == f.session.to_string())
+    );
+    assert!(
+        f.runtime
+            .pin_session_public_source_scoped(&f.scope, &f.source(), &f.session.to_string(), pin)
+            .is_err(),
+        "retention intent is not source availability"
+    );
+    // A missing committed source is never the same fact as a pending pointer.
+    metadata
+        .write(
+            &session_store::WorkerMetadata::new(
+                format!("worker-runtime-{}", f.worker.worker_id),
+                Some(session_store::WorkerActiveSegmentRef::active_segment(
+                    f.session, f.segment,
+                )),
+            )
+            .with_workspace_id(&f.scope.workspace_id),
+        )
+        .unwrap();
+    assert!(matches!(
+        f.runtime
+            .session_material_catalog_scoped(&f.scope, &f.request()),
+        Catalog::Unavailable { .. }
+    ));
+    metadata
+        .write(
+            &session_store::WorkerMetadata::new(
+                format!("worker-runtime-{}", f.worker.worker_id),
+                Some(session_store::WorkerActiveSegmentRef::pending_segment(
+                    f.session,
+                )),
+            )
+            .with_workspace_id(&f.scope.workspace_id),
+        )
+        .unwrap();
+    // Any unexpected source file invalidates the empty proof, too.
+    std::fs::create_dir_all(f.root()).unwrap();
+    std::fs::write(f.root().join("unattested-original"), "do not discard").unwrap();
+    assert!(matches!(
+        f.runtime
+            .session_material_catalog_scoped(&f.scope, &f.request()),
+        Catalog::Unavailable { .. }
+    ));
+    std::fs::remove_file(f.root().join("unattested-original")).unwrap();
+    f.runtime
+        .release_session_public_source_pin_scoped(
+            &f.scope,
+            &f.source(),
+            &f.session.to_string(),
+            pin,
+        )
+        .unwrap();
+    assert!(retention.pins(f.session).unwrap().is_empty());
+    f.stop();
+    let inventory = f
+        .runtime
+        .worker_retention_inventory(&f.scope.workspace_id, &f.worker)
+        .unwrap();
+    assert_eq!(inventory.session_id, None);
+    assert_eq!(inventory.session_bytes, 0);
+    let removed = f
+        .runtime
+        .execute_worker_retention(&f.retention_request(SessionDisposition::Archive))
+        .unwrap();
+    let archive = removed.archive.unwrap();
+    assert!(removed.source_removed);
+    assert_eq!(archive.source_session_id, f.session.to_string());
+    assert_eq!(archive.content_file_count, 0);
+    assert!(archive.source_active_segment_id.is_none());
+}
+
 #[test]
 fn canonical_pages_read_frozen_originals_after_tail_append_and_reject_forged_descriptors() {
     let f = Fixture::new();
@@ -524,14 +619,10 @@ fn explicit_erase_invalidates_pins_and_fences_old_readers_publishers_and_retry()
             SessionDeletionIntent::ErasePermanently,
         )
         .unwrap();
-    // Removing the remaining Worker registration/metadata still requires the
-    // caller's explicit erase intent; ordinary deletion may not infer it.
+    // Ordinary registration removal can resume a pre-existing permanent erase
+    // decision. It cannot establish one or recreate original Session material.
     f.runtime
-        .delete_worker_with_intent_scoped(
-            &f.scope,
-            &f.worker,
-            SessionDeletionIntent::ErasePermanently,
-        )
+        .delete_worker_with_intent_scoped(&f.scope, &f.worker, SessionDeletionIntent::Ordinary)
         .unwrap();
     assert!(!f.root().parent().unwrap().exists());
     assert!(matches!(
@@ -808,13 +899,47 @@ async fn signed_generated_client_reads_canonical_pages_and_pins_with_real_router
                 .len(),
             1
         );
-        client.session_material_release(pin).await.unwrap();
+        client.session_material_release(pin.clone()).await.unwrap();
         assert!(
             runtime_session_retention(f.temp.path())
                 .pins(f.session)
                 .unwrap()
                 .is_empty()
         );
+        client.session_material_pin(pin).await.unwrap();
+        f.stop();
+        let erase = runtime_api::SessionMaterialEraseRequest {
+            workspace_id: f.scope.workspace_id.clone(),
+            source: f.source(),
+            expected_session_id: f.session.to_string(),
+        };
+        for _ in 0..2 {
+            assert_eq!(
+                client
+                    .session_material_erase(erase.clone())
+                    .await
+                    .unwrap()
+                    .session_id,
+                f.session.to_string()
+            );
+        }
+        assert!(!f.root().exists());
+        assert!(matches!(
+            client.session_material_catalog(f.request()).await.unwrap(),
+            Catalog::Unavailable { .. }
+        ));
+        assert!(matches!(
+            runtime_session_retention(f.temp.path()).ensure_live(f.session),
+            Err(session_store::StoreError::SessionErased(_))
+        ));
+        // Register/lifecycle cleanup after erasure must not need a fresh archive or Job.
+        let result = f
+            .runtime
+            .execute_worker_retention(&f.retention_request(SessionDisposition::Purge))
+            .unwrap();
+        assert!(result.source_removed);
+        assert!(result.archive.is_none());
+        assert!(!f.root().parent().unwrap().exists());
     })
     .await
     .expect("bounded signed client integration");
@@ -866,3 +991,39 @@ async fn canonical_http_rejects_replay_tampered_body_and_workspace_mismatch() {
         .unwrap();
     assert_eq!(unsigned.status(), StatusCode::UNAUTHORIZED);
 }
+
+#[tokio::test]
+async fn explicit_material_erasure_requires_delete_capability_and_matching_scope() {
+    let f = Fixture::new();
+    f.stop();
+    let identity = RuntimeIdentityMaterial::generate("erase-key").unwrap();
+    let trust = workspace_trust(&identity);
+    let app = runtime_http_router_with_workspace_auth(
+        f.runtime.clone(),
+        None,
+        workspace_test_auth(vec![trust.clone()]),
+    );
+    let path = "/v1/session-public/material/retention/erase";
+    let mut request = runtime_api::SessionMaterialEraseRequest {
+        workspace_id: f.scope.workspace_id.clone(),
+        source: f.source(),
+        expected_session_id: f.session.to_string(),
+    };
+    let body = serde_json::to_vec(&request).unwrap();
+    let mut claims = signed_claims(&trust, "POST", path, &body);
+    claims.operation = "workers:read".into();
+    let response = signed_request(app.clone(), &identity, &claims, body, "workspace-a").await;
+    assert_eq!(response.status(), StatusCode::UNAUTHORIZED);
+    assert!(f.root().exists());
+    request.workspace_id = "other-workspace".into();
+    let body = serde_json::to_vec(&request).unwrap();
+    let claims = signed_claims(&trust, "POST", path, &body);
+    let response = signed_request(app, &identity, &claims, body, "workspace-a").await;
+    assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    assert!(
+        runtime_session_retention(f.temp.path())
+            .ensure_live(f.session)
+            .is_ok()
+    );
+    assert!(f.root().exists());
+}

crates/worker-runtime/src/retention.rs

@@ -178,6 +178,40 @@ pub(crate) trait WorkerRetentionProvider: Send + Sync {
 
 pub use crate::session_retention::runtime_session_retention;
 
+/// Prove the legitimate pre-publication state, rather than interpreting a missing
+/// source as empty. A committed active pointer or any source file defeats this proof.
+pub(crate) fn unmaterialized_session(
+    worker_dir: &Path,
+    workspace_id: &str,
+) -> Result<Option<session_store::SessionId>, RuntimeError> {
+    let metadata_path = worker_dir.join("metadata.json");
+    let bytes = match fs::read(&metadata_path) {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
+        Err(source) => {
+            return Err(RuntimeError::StoreIo {
+                operation: "inspect pending Session",
+                path: metadata_path,
+                source,
+            });
+        }
+    };
+    let metadata: session_store::WorkerMetadata = serde_json::from_slice(&bytes)
+        .map_err(|_| RuntimeError::InvalidRequest("Pending Session metadata is invalid".into()))?;
+    if metadata.workspace_id.as_deref() != Some(workspace_id) {
+        return Ok(None);
+    }
+    let Some(active) = metadata.active.filter(|active| active.segment_id.is_none()) else {
+        return Ok(None);
+    };
+    let root = worker_dir.join("session");
+    if collect_files(&root, "inspect pending Session")?.is_empty() {
+        Ok(Some(active.session_id))
+    } else {
+        Ok(None)
+    }
+}
+
 pub(crate) fn aggregate_session_id(
     worker_dir: &Path,
 ) -> Result<Option<session_store::SessionId>, RuntimeError> {
@@ -303,6 +337,19 @@ pub(crate) fn delete_worker_aggregate(
     intent: session_store::retention::SessionDeletionIntent,
 ) -> Result<(), RuntimeError> {
     let session = aggregate_session_id(worker_dir)?;
+    let retention = runtime_session_retention(runtime_root);
+    // Ordinary policy never creates erase authority. A pre-existing permanent
+    // tombstone is an irrevocable decision whose physical cleanup may be resumed.
+    let intent = match session
+        .map(|session| retention.ensure_live(session))
+        .transpose()
+    {
+        Err(session_store::StoreError::SessionErased(_)) => {
+            session_store::retention::SessionDeletionIntent::ErasePermanently
+        }
+        Err(error) => return Err(retention_error(error)),
+        Ok(_) => intent,
+    };
     let remove = || -> Result<(), session_store::StoreError> {
         if intent == session_store::retention::SessionDeletionIntent::ErasePermanently {
             if let Some(session) = session {
@@ -561,33 +608,36 @@ impl WorkerRetentionProvider for FsWorkerRetentionProvider {
             return Err(RuntimeError::WorkerNotFound { worker_id });
         }
         let session_dir = worker_dir.join("session");
-        let (session_id, segment_ids, session_bytes) = if session_dir.is_dir() {
-            let manifest: CanonicalSessionManifest = read_json(
-                &session_dir.join("session.json"),
-                "inventory Worker retention",
-            )?;
-            let files = collect_files(&session_dir, "inventory Worker retention")?;
-            let mut segment_ids = BTreeSet::new();
-            let mut bytes = 0_u64;
-            for (relative, path) in files {
-                bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?);
-                if let Some(name) = relative.file_name().and_then(|value| value.to_str()) {
-                    let segment = name
-                        .strip_suffix(".trace.jsonl")
-                        .or_else(|| name.strip_suffix(".jsonl"));
-                    if let Some(segment) = segment {
-                        segment_ids.insert(segment.to_string());
+        let (session_id, segment_ids, session_bytes) =
+            if unmaterialized_session(&worker_dir, workspace_id)?.is_some() {
+                (None, Vec::new(), 0)
+            } else if session_dir.is_dir() {
+                let manifest: CanonicalSessionManifest = read_json(
+                    &session_dir.join("session.json"),
+                    "inventory Worker retention",
+                )?;
+                let files = collect_files(&session_dir, "inventory Worker retention")?;
+                let mut segment_ids = BTreeSet::new();
+                let mut bytes = 0_u64;
+                for (relative, path) in files {
+                    bytes = bytes.saturating_add(file_len(&path, "inventory Worker retention")?);
+                    if let Some(name) = relative.file_name().and_then(|value| value.to_str()) {
+                        let segment = name
+                            .strip_suffix(".trace.jsonl")
+                            .or_else(|| name.strip_suffix(".jsonl"));
+                        if let Some(segment) = segment {
+                            segment_ids.insert(segment.to_string());
+                        }
                     }
                 }
-            }
-            (
-                Some(manifest.session_id),
-                segment_ids.into_iter().collect(),
-                bytes,
-            )
-        } else {
-            (None, Vec::new(), 0)
-        };
+                (
+                    Some(manifest.session_id),
+                    segment_ids.into_iter().collect(),
+                    bytes,
+                )
+            } else {
+                (None, Vec::new(), 0)
+            };
         let diagnostics_bytes = diagnostics_files(&worker_dir, "inventory Worker retention")?
             .into_iter()
             .try_fold(0_u64, |total, path| {
@@ -909,14 +959,14 @@ fn commit_session_archive(
         RuntimeError::InvalidRequest("archive disposition requires archive_id".to_string())
     })?;
     let session_dir = worker_dir.join("session");
-    if !session_dir.is_dir() {
-        return Err(RuntimeError::StoreMissing {
-            operation: "archive Worker Session",
-            path: session_dir,
-        });
-    }
-    let session: CanonicalSessionManifest =
-        read_json(&session_dir.join("session.json"), "archive Worker Session")?;
+    let pending = unmaterialized_session(worker_dir, &request.workspace_id)?;
+    let session: CanonicalSessionManifest = if let Some(session) = pending {
+        CanonicalSessionManifest {
+            session_id: session.to_string(),
+        }
+    } else {
+        read_json(&session_dir.join("session.json"), "archive Worker Session")?
+    };
     let source_files = collect_files(&session_dir, "archive Worker Session")?;
     let (checksum, bytes, count) = checksum_files(&source_files, "archive Worker Session")?;
     let mut segment_ids = BTreeSet::new();

crates/worker-runtime/src/runtime.rs

@@ -4842,15 +4842,20 @@ impl Runtime {
         expected_session_id: &str,
         pin: session_store::retention::SessionRetentionPin,
     ) -> Result<(), RuntimeError> {
-        let (root, _) = self.resolve_session_public_source(scope, source)?;
+        let (root, archive) =
+            self.resolve_session_public_source_with_validation(scope, source, false)?;
         let session = expected_session_id
             .parse::<session_store::SessionId>()
             .map_err(|_| RuntimeError::InvalidRequest("Invalid Session identity".into()))?;
-        if session_store::WorkerSessionStore::open_read_only(&root)
-            .and_then(|store| store.session_id())
-            .map_err(crate::retention::retention_error)?
-            != Some(session)
-        {
+        let owned_session = match archive {
+            Some(manifest) => manifest.source_session_id.parse().ok(),
+            None => {
+                crate::retention::aggregate_session_id(root.parent().ok_or_else(|| {
+                    RuntimeError::InvalidRequest("Session root unavailable".into())
+                })?)?
+            }
+        };
+        if owned_session != Some(session) {
             return Err(RuntimeError::InvalidRequest(
                 "Session source identity mismatch".into(),
             ));
@@ -4995,6 +5000,33 @@ impl Runtime {
         ))
     }
 
+    /// Prove an attributed but not yet materialized Session without treating a
+    /// missing committed log as empty. The source/Workspace binding is resolved first.
+    #[cfg(feature = "fs-store")]
+    pub(crate) fn session_not_materialized_scoped(
+        &self,
+        scope: &RuntimeWorkspaceScope,
+        source: &runtime_api::SessionPublicSource,
+        expected_session_id: &str,
+    ) -> Result<bool, RuntimeError> {
+        let (root, archive) = self.resolve_session_public_source(scope, source)?;
+        if archive.is_some() {
+            return Ok(false);
+        }
+        let session = expected_session_id
+            .parse::<session_store::SessionId>()
+            .map_err(|_| RuntimeError::InvalidRequest("Invalid Session identity".into()))?;
+        let retention = self.session_retention_store()?;
+        let _guard = retention
+            .lock_live(session)
+            .map_err(crate::retention::retention_error)?;
+        Ok(crate::retention::unmaterialized_session(
+            root.parent()
+                .ok_or_else(|| RuntimeError::InvalidRequest("Session root unavailable".into()))?,
+            &scope.workspace_id,
+        )? == Some(session))
+    }
+
     /// Resolve the Backend-selected source without exposing local storage paths.
     /// The saved active pointer, never a caller-provided Segment, selects lineage.
     #[cfg(feature = "fs-store")]

crates/worker-runtime/src/session_material.rs

@@ -57,6 +57,18 @@ pub(crate) fn catalog(
     scope: &RuntimeWorkspaceScope,
     request: &SessionMaterialCatalogRequest,
 ) -> SessionMaterialCatalogAvailability {
+    if scope.workspace_id == request.workspace_id
+        && (1..=100).contains(&request.limit)
+        && request.offset == 0
+        && request.expected_catalog_sha256.is_none()
+        && runtime
+            .session_not_materialized_scoped(scope, &request.source, &request.expected_session_id)
+            .is_ok_and(|empty| empty)
+    {
+        return SessionMaterialCatalogAvailability::NotMaterialized {
+            session_id: request.expected_session_id.clone(),
+        };
+    }
     let result = (|| -> Result<_, SessionMaterialSourceError> {
         if scope.workspace_id != request.workspace_id {
             return Err("Session Workspace scope mismatch".into());

crates/worker/src/worker.rs

@@ -7812,7 +7812,10 @@ where
                     });
                 }
                 match state {
-                    SubjektivSessionAttributionState::Confirmed { .. } => return Ok(()),
+                    SubjektivSessionAttributionState::Confirmed { subject_id, .. } => {
+                        return self
+                            .retain_journal_source_before_publication(session_id, subject_id);
+                    }
                     SubjektivSessionAttributionState::OutcomeUnknown { .. }
                     | SubjektivSessionAttributionState::Failed { .. } => true,
                     SubjektivSessionAttributionState::LegacyUnknown { .. } => false,
@@ -7967,8 +7970,44 @@ where
         }
         self.persist_subjektiv_session_attribution(SubjektivSessionAttributionState::Confirmed {
             session_id,
-            subject_id: output.subject_id,
-        })
+            subject_id: output.subject_id.clone(),
+        })?;
+        self.retain_journal_source_before_publication(session_id, &output.subject_id)
+    }
+
+    fn retain_journal_source_before_publication(
+        &self,
+        session_id: SessionId,
+        subject_id: &str,
+    ) -> Result<(), WorkerError> {
+        let Some(connection) = self.subjektiv_host_connection().map_err(|error| {
+            WorkerError::SubjektivSessionAttribution {
+                state: "retention_unconfirmed",
+                message: error.to_string(),
+            }
+        })?
+        else {
+            return Ok(());
+        };
+        if !connection.host.journal_recording_granted() {
+            return Ok(());
+        }
+        let pin = session_store::retention::SessionRetentionPin(
+            match self.workspace_client().workspace_id() {
+                Some(workspace) => subjektiv::journal::source_retention_pin(
+                    workspace,
+                    subject_id,
+                    &session_id.to_string(),
+                ),
+                None => session_id,
+            },
+        );
+        // Confirmation can be replayed after response loss. Persist this intent before
+        // exposing the Worker to input, not after the first RunCommitted notification.
+        // It does not claim that an original Session has materialized yet.
+        self.store
+            .retain_session_before_publication(session_id, pin)?;
+        Ok(())
     }
 
     fn persist_subjektiv_session_attribution(
@@ -11116,6 +11155,82 @@ mod build_summary_prompt_tests {
         assert!(worker.backend_job().is_none());
     }
 
+    #[tokio::test]
+    async fn journal_attribution_protects_first_publication_before_any_admission_notification() {
+        for enabled in [false, true] {
+            let client = Arc::new(AttributionLifecycleClient::new([]));
+            let (mut worker, store, _temp) = attribution_lifecycle_worker(client.clone()).await;
+            if enabled {
+                worker.manifest.feature.subjektiv.profile.enabled = true;
+                worker
+                    .manifest
+                    .feature
+                    .subjektiv
+                    .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot {
+                        workspace_id: "workspace-test".into(),
+                        language: "English".into(),
+                    })
+                    .unwrap();
+            }
+            let session = worker.session_id();
+            assert!(store.session_store.list_sessions().unwrap().is_empty());
+            worker
+                .finalize_subjektiv_session_attribution(
+                    true,
+                    SubjektivSessionAttributionLifecycle::NewSession,
+                )
+                .unwrap();
+            let pin = session_store::retention::SessionRetentionPin(
+                subjektiv::journal::source_retention_pin(
+                    "workspace-test",
+                    "subject-test",
+                    &session.to_string(),
+                ),
+            );
+            let expected = if enabled { vec![pin] } else { vec![] };
+            assert_eq!(
+                store.session_store.retention().pins(session).unwrap(),
+                expected
+            );
+            // Reopen/replayed confirmation does not depend on a notification or add an owner.
+            worker
+                .finalize_subjektiv_session_attribution(
+                    true,
+                    SubjektivSessionAttributionLifecycle::RestoredSession,
+                )
+                .unwrap();
+            assert_eq!(client.requests.lock().unwrap().len(), 1);
+            assert_eq!(
+                store.session_store.retention().pins(session).unwrap(),
+                expected
+            );
+            worker.materialize_durable_session_head().await.unwrap();
+            let segment = worker.segment_id();
+            store
+                .append(
+                    session,
+                    segment,
+                    &LogEntry::TurnEnd {
+                        ts: 1,
+                        turn_count: 1,
+                    },
+                )
+                .unwrap();
+            // Simulate stop after the public commit but before RunCommitted/Host admission.
+            if enabled {
+                assert!(
+                    matches!(store.session_store.delete_session(session), Err(StoreError::SessionPinned(id)) if id == session)
+                );
+                assert!(store.session_store.exists(session, segment).unwrap());
+                store
+                    .session_store
+                    .release_session_pin(session, pin)
+                    .unwrap();
+            }
+            assert!(store.session_store.delete_session(session).unwrap());
+        }
+    }
+
     #[tokio::test]
     async fn subject_maintenance_job_binding_is_immutable_and_has_no_subject_body_attribution() {
         for (subjektiv_consolidation, subjektiv_journal) in [(true, false), (false, true)] {

crates/workspace-server/src/hosts.rs

@@ -1301,6 +1301,13 @@ pub trait WorkspaceWorkerRuntime: Send + Sync {
         Err("Session source retention release is unavailable on this Runtime".into())
     }
 
+    fn session_material_erase(
+        &self,
+        _request: runtime_api::SessionMaterialEraseRequest,
+    ) -> Result<runtime_api::SessionMaterialEraseResponse, String> {
+        Err("Session source permanent erasure is unavailable on this Runtime".into())
+    }
+
     fn session_material_catalog(
         &self,
         _request: runtime_api::SessionMaterialCatalogRequest,
@@ -2266,6 +2273,21 @@ impl RuntimeRegistry {
             })
     }
 
+    pub fn session_material_erase(
+        &self,
+        runtime_id: &str,
+        request: runtime_api::SessionMaterialEraseRequest,
+    ) -> Result<runtime_api::SessionMaterialEraseResponse, RuntimeRegistryError> {
+        validate_backend_identifier("runtime_id", runtime_id)?;
+        self.runtime(runtime_id)?
+            .session_material_erase(request)
+            .map_err(|message| RuntimeRegistryError::RuntimeOperationFailed {
+                runtime_id: runtime_id.into(),
+                code: "session_material_erase_failed".into(),
+                message,
+            })
+    }
+
     pub fn session_material_catalog(
         &self,
         runtime_id: &str,
@@ -3324,6 +3346,27 @@ impl WorkspaceWorkerRuntime for EmbeddedWorkerRuntime {
         })
     }
 
+    fn session_material_erase(
+        &self,
+        request: runtime_api::SessionMaterialEraseRequest,
+    ) -> Result<runtime_api::SessionMaterialEraseResponse, String> {
+        if request.workspace_id != self.workspace_id {
+            return Err("Session erase Workspace mismatch".into());
+        }
+        let scope = RuntimeWorkspaceScope::new(&request.workspace_id, "embedded-backend");
+        self.runtime
+            .delete_session_public_source_scoped(
+                &scope,
+                &request.source,
+                &request.expected_session_id,
+                session_store::retention::SessionDeletionIntent::ErasePermanently,
+            )
+            .map_err(|e| e.to_string())?;
+        Ok(runtime_api::SessionMaterialEraseResponse {
+            session_id: request.expected_session_id,
+        })
+    }
+
     fn session_material_catalog(
         &self,
         request: runtime_api::SessionMaterialCatalogRequest,
@@ -5503,6 +5546,18 @@ impl WorkspaceWorkerRuntime for RemoteWorkerRuntime {
         .map_err(|diagnostic| diagnostic.message)
     }
 
+    fn session_material_erase(
+        &self,
+        request: runtime_api::SessionMaterialEraseRequest,
+    ) -> Result<runtime_api::SessionMaterialEraseResponse, String> {
+        self.run_runtime_api(
+            self.request_timeout,
+            MAX_REMOTE_RUNTIME_RESPONSE_BYTES,
+            move |client| async move { client.session_material_erase(request).await },
+        )
+        .map_err(|diagnostic| diagnostic.message)
+    }
+
     fn session_material_catalog(
         &self,
         request: runtime_api::SessionMaterialCatalogRequest,
@@ -8262,12 +8317,58 @@ mod tests {
             ),
             "workers:read"
         );
+        assert_eq!(
+            workspace_runtime_operation("POST", "/v1/session-public/material/retention/erase"),
+            "workers:delete"
+        );
         assert_eq!(
             workspace_runtime_operation("POST", SSH_HOST_KEY_PROBE_PATH),
             SSH_HOST_KEY_PROBE_OPERATION
         );
     }
 
+    #[test]
+    fn remote_session_erasure_uses_the_generated_management_transport() {
+        // The loopback fixture substitutes only the remote Runtime. This covers
+        // Registry/remote-adapter DTO and route wiring, not Runtime authorization or deletion.
+        let session_id = uuid::Uuid::new_v4().to_string();
+        let (base_url, server) = serve_mock_http(vec![mock_response(
+            "POST",
+            "/v1/session-public/material/retention/erase",
+            true,
+            200,
+            json!({"session_id": session_id}).to_string(),
+        )]);
+        let registry = RuntimeRegistry::new(Vec::new());
+        registry.register(
+            RemoteWorkerRuntime::new(
+                RemoteRuntimeConfig::new(
+                    "remote:erase",
+                    "Erase transport fixture",
+                    base_url,
+                    Some("secret-token".into()),
+                ),
+                "workspace-test".into(),
+                "http://127.0.0.1:8787".into(),
+            )
+            .unwrap(),
+        );
+        let response = registry
+            .session_material_erase(
+                "remote:erase",
+                runtime_api::SessionMaterialEraseRequest {
+                    workspace_id: "workspace-test".into(),
+                    source: runtime_api::SessionPublicSource::Retained {
+                        worker_id: EmbeddedWorkerId::from_legacy_u64(1),
+                    },
+                    expected_session_id: session_id.clone(),
+                },
+            )
+            .unwrap();
+        assert_eq!(response.session_id, session_id);
+        server.join().unwrap();
+    }
+
     #[test]
     fn remote_runtime_registry_routes_commands_without_browser_secret_leaks() {
         let worker_id = EmbeddedWorkerId::from_legacy_u64(1).to_string();

crates/workspace-server/src/retention.rs

@@ -63,6 +63,51 @@ pub struct WorkerRemovalPlanRequest {
     pub workspace_id: String,
     pub worker: RuntimeWorkerRef,
     pub reason: String,
+    /// Internal evidence cannot be supplied through serialized deletion requests.
+    #[serde(skip)]
+    pub(crate) journal_erasure: Option<ErasedJournalSession>,
+}
+
+const ERASED_JOURNAL_POLICY_ID: &str = "backend-journal-erased-session-v1";
+
+/// Monotonic domain erasure proof, bound to the exact source Worker and Session.
+/// The only constructor rereads the Journal tombstone; ordinary Purge grants no erase authority.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(crate) struct ErasedJournalSession {
+    workspace_id: String,
+    worker: RuntimeWorkerRef,
+    session_id: String,
+}
+impl ErasedJournalSession {
+    pub(crate) fn validated(
+        workspace_id: &str,
+        store: &crate::subjektiv::SubjektivStore,
+        attr: &crate::subjektiv::SubjectSessionAttribution,
+    ) -> crate::Result<Option<Self>> {
+        let host = subjektiv::journal::JournalHostContext::validated_session(store, attr.clone())
+            .map_err(|e| StoreError::Store(e.to_string()))?;
+        Ok(store
+            .journal_session_erased(&host)
+            .map_err(|e| StoreError::Store(e.to_string()))?
+            .then(|| Self {
+                workspace_id: workspace_id.into(),
+                worker: RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id),
+                session_id: attr.session_id.clone(),
+            }))
+    }
+
+    fn matches(
+        &self,
+        req: &WorkerRemovalPlanRequest,
+        inventory: &WorkerRetentionInventory,
+    ) -> bool {
+        self.workspace_id == req.workspace_id
+            && self.worker == req.worker
+            && inventory
+                .session_id
+                .as_ref()
+                .is_none_or(|id| *id == self.session_id)
+    }
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
@@ -223,10 +268,20 @@ impl SqliteWorkspaceStore {
         inv: &WorkerRetentionInventory,
     ) -> Result<WorkerRemovalPlan, WorkerRetentionError> {
         validate_plan(req, inv)?;
+        if req
+            .journal_erasure
+            .as_ref()
+            .is_some_and(|proof| !proof.matches(req, inv))
+        {
+            return Err(WorkerRetentionError::Invalid(
+                "Journal erasure proof does not match removal source".into(),
+            ));
+        }
+        let erased_journal = req.journal_erasure.is_some();
         let now = Utc::now().to_rfc3339();
         self.with_conn_mut(|conn| {
             let tx=conn.transaction_with_behavior(TransactionBehavior::Immediate)?;
-            let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?;
+            let policy=load_removal_policy(&tx,&req.workspace_id,&req.worker,erased_journal)?.ok_or_else(|| StoreError::InvalidInput(format!("policy-missing:{}",req.workspace_id)))?;
             let worker=match load_worker(&tx,&req.workspace_id,&req.worker)? {
                 Some(v)=>v,
                 None=>{
@@ -272,7 +327,7 @@ impl SqliteWorkspaceStore {
                 return Err(StoreError::InvalidInput(format!("stale:{plan_id}:plan state {} is not executable", state_s(plan.state))));
             }
             if !plan.blockers.is_empty(){return Err(StoreError::InvalidInput(format!("blocked:{}",serde_json::to_string(&plan.blockers).unwrap())));}
-            let policy=load_removal_policy(&tx,workspace_id,&plan.worker)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?;
+            let policy=load_removal_policy(&tx,workspace_id,&plan.worker,plan.policy_id==ERASED_JOURNAL_POLICY_ID)?.ok_or_else(||StoreError::InvalidInput(format!("policy-missing:{workspace_id}")))?;
             if policy.policy_id!=plan.policy_id||policy.policy_digest!=plan.policy_digest{
                 mark_stale(&tx,&plan,"policy content changed")?;tx.commit()?;
                 return Err(stale_error(&plan,"policy content changed"));
@@ -785,20 +840,28 @@ struct WorkerRow {
 fn load_worker(c: &Connection, w: &str, r: &RuntimeWorkerRef) -> crate::Result<Option<WorkerRow>> {
     c.query_row("SELECT display_name,profile,retention_state,created_at,updated_at FROM worker_registry WHERE workspace_id=?1 AND runtime_id=?2 AND worker_id=?3",params![w,r.runtime_id,r.worker_id],|x|Ok(WorkerRow{display_name:x.get(0)?,profile:x.get(1)?,retention_state:x.get(2)?,created_at:x.get(3)?,updated_at:x.get(4)?})).optional().map_err(StoreError::from)
 }
-/// Recording Sessions may contain original text in tool results. Only a terminal,
-/// Host-bound Journal attempt grants this exception; profile names grant nothing.
+/// Recording Sessions may contain original text in tool results. A terminal,
+/// Host-bound Journal attempt or a validated source-erasure proof grants this
+/// exception; profile names and serialized deletion input grant nothing.
+/// A saved erasure plan retains the monotonic decision across restart, but creates
+/// no Runtime erase authority: the domain/physical tombstones must already exist.
 /// This plan-local policy never changes the Workspace conversation policy.
 fn load_removal_policy(
     c: &Connection,
     w: &str,
     worker: &RuntimeWorkerRef,
+    erased_journal: bool,
 ) -> crate::Result<Option<WorkerRetentionPolicy>> {
     let Some(mut policy) = load_policy(c, w)? else {
         return Ok(None);
     };
-    if crate::store::terminal_journal_job_worker(c, w, worker)? {
+    if erased_journal || crate::store::terminal_journal_job_worker(c, w, worker)? {
         let update = WorkerRetentionPolicyUpdate {
-            policy_id: "backend-journal-ephemeral-session-v1".into(),
+            policy_id: if erased_journal {
+                ERASED_JOURNAL_POLICY_ID.into()
+            } else {
+                "backend-journal-ephemeral-session-v1".into()
+            },
             session_disposition: SessionDisposition::Purge,
             metadata_disposition: MetadataDisposition::Tombstone,
             archive_retention: ArchiveRetention::Forever,
@@ -1194,6 +1257,7 @@ mod tests {
                 worker_id: worker_id().to_string(),
             },
             reason: "cleanup".into(),
+            journal_erasure: None,
         }
     }
     #[test]

crates/workspace-server/src/server.rs

@@ -2190,13 +2190,17 @@ impl WorkerRemovalService {
             ));
         }
 
-        if let Err(error) = subjektiv_journal::prepare_worker_removal(self, runtime, target) {
-            return Ok(worker_remove_error_response(
-                StatusCode::CONFLICT,
-                "journal_handoff_incomplete",
-                &error.to_string(),
-            ));
-        }
+        let journal_erasure = match subjektiv_journal::prepare_worker_removal(self, runtime, target)
+        {
+            Ok(proof) => proof,
+            Err(error) => {
+                return Ok(worker_remove_error_response(
+                    StatusCode::CONFLICT,
+                    "journal_handoff_incomplete",
+                    &error.to_string(),
+                ));
+            }
+        };
         let inventory = match runtime.worker_retention_inventory(&target) {
             Ok(inventory) => inventory,
             Err(_) => {
@@ -2211,6 +2215,7 @@ impl WorkerRemovalService {
             workspace_id: self.workspace_id.clone(),
             worker: target.clone(),
             reason: reason.to_string(),
+            journal_erasure,
         };
         let plan = match self.store.plan_worker_removal(&request, &inventory) {
             Ok(plan) => plan,

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

@@ -320,6 +320,22 @@ fn admit(
         ));
     }
     let mut id = store.journal_for_session(&host).map_err(domain)?;
+    let reader = material_reader(api, &attribution)?;
+    if reader.not_materialized {
+        if store.journal_retention_required(&host).map_err(domain)? {
+            return Err(Error::RepositoryConflict(
+                "Admitted Journal original Session is not materialized".into(),
+            ));
+        }
+        retain_source(
+            api.runtime.as_ref(),
+            api.workspace_id(),
+            &attribution,
+            material_source(api, &attribution)?,
+            true,
+        )?;
+        return Ok(JournalRecordingResponse::NoNewSources);
+    }
     retain_source(
         api.runtime.as_ref(),
         api.workspace_id(),
@@ -327,7 +343,10 @@ fn admit(
         material_source(api, &attribution)?,
         false,
     )?;
-    let sources = canonical_catalog(api, &attribution)?;
+    let sources: Vec<_> = reader
+        .descriptors()
+        .map(|d| source_from_descriptor(&attribution.session_id, d))
+        .collect();
     for batch in sources.chunks(MAX_JOURNAL_RANGES) {
         id = Some(
             store
@@ -374,7 +393,7 @@ pub(super) async fn recording_route(
                 )
                 .into());
             }
-            require_enabled(api.clone(), &worker)?;
+            require_enabled(&api, &worker)?;
             let store = store(&api)?;
             let attributions = store
                 .subject_session_attributions(subject)
@@ -424,7 +443,16 @@ pub(super) async fn job_route(
         .await
 }
 
-fn require_enabled(api: WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> {
+fn require_enabled(api: &WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> {
+    if !enabled(api, worker)? {
+        return Err(Error::WorkspacePermissionDenied(
+            "Journal Feature is disabled".into(),
+        ));
+    }
+    Ok(())
+}
+
+fn enabled(api: &WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<bool> {
     let projection = api
         .store
         .worker_registry_projection(api.workspace_id(), worker)?
@@ -459,12 +487,7 @@ fn require_enabled(api: WorkspaceApi, worker: &RuntimeWorkerRef) -> Result<()> {
         .ok_or_else(|| Error::Config("Journal Worker Profile bundle missing".into()))?;
     let manifest =
         crate::profile_settings::resolve_profile_manifest_from_config_bundle(&bundle, key)?;
-    if !manifest.feature.subjektiv.enabled() {
-        return Err(Error::WorkspacePermissionDenied(
-            "Journal Feature is disabled".into(),
-        ));
-    }
-    Ok(())
+    Ok(manifest.feature.subjektiv.enabled())
 }
 
 pub(super) fn drain_after_completion(api: &WorkspaceApi, request: &BackendJobRequest) {
@@ -539,16 +562,6 @@ fn source_from_descriptor(
         },
     }
 }
-fn canonical_catalog(
-    api: &WorkspaceApi,
-    attribution: &crate::subjektiv::SubjectSessionAttribution,
-) -> Result<Vec<JournalSourceRange>> {
-    let reader = material_reader(api, attribution)?;
-    Ok(reader
-        .descriptors()
-        .map(|d| source_from_descriptor(&attribution.session_id, d))
-        .collect())
-}
 fn read_canonical(
     reader: &MaterialReader<'_>,
     source: &JournalSource,
@@ -626,20 +639,14 @@ pub(super) fn recover(api: &WorkspaceApi) -> Result<()> {
                     }
                     if erased {
                         if let Err(error) = material_source(api, attr).and_then(|source| {
-                            retain_source(
-                                api.runtime.as_ref(),
-                                api.workspace_id(),
-                                attr,
-                                source,
-                                true,
-                            )
+                            erase_source(api.runtime.as_ref(), api.workspace_id(), attr, source)
                         }) {
-                            tracing::warn!(%error,session_id=%attr.session_id,"Erased Session source pin release remains recoverable");
+                            tracing::warn!(%error,session_id=%attr.session_id,"Erased Session physical cleanup remains recoverable");
                         }
                         continue;
                     }
                     let source = RuntimeWorkerRef::new(&attr.runtime_id, &attr.worker_id);
-                    if require_enabled(api.clone(), &source).is_err() {
+                    if require_enabled(api, &source).is_err() {
                         // Disablement or removal does not erase previously admitted work, but
                         // also does not authorize collecting a new disabled Worker's history.
                         let host = JournalHostContext::validated_session(&store, attr.clone())
@@ -710,16 +717,13 @@ fn removal_attribution(
     service: &WorkerRemovalService,
     target: &RuntimeWorkerRef,
 ) -> Result<Option<crate::subjektiv::SubjectSessionAttribution>> {
-    let Some(p) = service
+    let Some(lease) = service
         .store
-        .worker_registry_projection(&service.workspace_id, target)?
+        .require_current_worker_singleton_owner(&service.workspace_id, target)?
     else {
         return Ok(None);
     };
-    let Some(key) = p.resource_key.as_deref() else {
-        return Ok(None);
-    };
-    let Some(subject) = key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else {
+    let Some(subject) = lease.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else {
         return Ok(None);
     };
     let store = crate::subjektiv::SubjektivStore::open(
@@ -767,9 +771,9 @@ pub(super) fn prepare_worker_removal(
     service: &WorkerRemovalService,
     runtime: &RuntimeRegistry,
     target: &RuntimeWorkerRef,
-) -> Result<()> {
+) -> Result<Option<crate::retention::ErasedJournalSession>> {
     let Some(attr) = removal_attribution(service, target)? else {
-        return Ok(());
+        return Ok(None);
     };
     let store = crate::subjektiv::SubjektivStore::open(
         &service.journal_storage,
@@ -781,31 +785,59 @@ pub(super) fn prepare_worker_removal(
         worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id)
             .ok_or_else(|| Error::Store("Worker identity invalid".into()))?,
     };
-    retain_source(runtime, &service.workspace_id, &attr, source.clone(), false)?;
+    let api = service.journal_dispatch_api.upgrade().ok_or_else(|| {
+        Error::Store("Workspace API unavailable for Journal removal Feature gate".into())
+    })?;
+    if let Some(proof) =
+        crate::retention::ErasedJournalSession::validated(&service.workspace_id, &store, &attr)?
+    {
+        // Erasure is monotonic. Retry the same physical intent, never admission or generation.
+        reconcile_session_jobs(&api, &attr, true)?;
+        erase_source(runtime, &service.workspace_id, &attr, source)?;
+        return Ok(Some(proof));
+    }
+    let collect_new_sources = enabled(&api, target)?;
+    let mut id = store.journal_for_session(&host).map_err(domain)?;
+    // Disablement forbids new admission, not recovery of an already admitted prefix.
+    // With no admitted remainder there is no source reader or Job to create at removal.
+    if !collect_new_sources && !store.journal_retention_required(&host).map_err(domain)? {
+        return Ok(None);
+    }
     let reader = MaterialReader::open(
         runtime,
         &service.workspace_id,
         &target.runtime_id,
-        source,
+        source.clone(),
         &attr.session_id,
     )?;
-    let ranges: Vec<_> = reader
-        .descriptors()
-        .map(|d| source_from_descriptor(&attr.session_id, d))
-        .collect();
-    let mut id = store.journal_for_session(&host).map_err(domain)?;
-    for batch in ranges.chunks(MAX_JOURNAL_RANGES) {
-        id = Some(
-            store
-                .accept_journal_sources(&host, "Session activity", batch)
-                .map_err(domain)?,
-        );
+    if reader.not_materialized {
+        if store.journal_retention_required(&host).map_err(domain)? {
+            return Err(Error::RepositoryConflict(
+                "Admitted Journal original Session is not materialized".into(),
+            ));
+        }
+        retain_source(runtime, &service.workspace_id, &attr, source, true)?;
+        return Ok(None);
+    }
+    retain_source(runtime, &service.workspace_id, &attr, source, false)?;
+    if collect_new_sources {
+        let ranges: Vec<_> = reader
+            .descriptors()
+            .map(|d| source_from_descriptor(&attr.session_id, d))
+            .collect();
+        for batch in ranges.chunks(MAX_JOURNAL_RANGES) {
+            id = Some(
+                store
+                    .accept_journal_sources(&host, "Session activity", batch)
+                    .map_err(domain)?,
+            );
+        }
     }
     let Some(id) = id else {
-        return Ok(());
+        return Ok(None);
     };
     if !store.journal_retention_required(&host).map_err(domain)? {
-        return Ok(());
+        return Ok(None);
     }
     if let Some(g) = store
         .active_journal_generation(&host, &id)
@@ -817,7 +849,7 @@ pub(super) fn prepare_worker_removal(
             .get_backend_job(&service.workspace_id, &g.job.job_id)?
             .is_some()
         {
-            return Ok(());
+            return Ok(None);
         }
         return Err(Error::Store(
             "Journal generation's existing Job intent is missing".into(),
@@ -843,7 +875,7 @@ pub(super) fn prepare_worker_removal(
         // only starts the existing bounded Job; it never waits for Journal generation.
         api.drain_reserved_backend_jobs();
     }
-    Ok(())
+    Ok(None)
 }
 
 pub(super) fn validate_removal_retention(
@@ -860,6 +892,22 @@ pub(super) fn validate_removal_retention(
     )
     .map_err(|e| Error::Store(e.to_string()))?;
     let host = JournalHostContext::validated_session(&store, attr.clone()).map_err(domain)?;
+    if store.journal_session_erased(&host).map_err(domain)? {
+        if prepared.runtime_request.session_disposition
+            != worker_runtime::retention::SessionDisposition::Purge
+            || prepared.runtime_request.diagnostics_disposition
+                != worker_runtime::retention::DiagnosticsDisposition::Purge
+            || prepared.runtime_request.archive_id.is_some()
+        {
+            return Err(Error::RepositoryConflict(
+                "Erased Journal Session requires source and diagnostics Purge, not an archive"
+                    .into(),
+            ));
+        }
+        // Physical erasure precedes the plan. Runtime resumes aggregate cleanup with
+        // its retained permanent fence, including recovery after the aggregate is gone.
+        return Ok(());
+    }
     let source = runtime_api::SessionPublicSource::Retained {
         worker_id: worker_runtime::identity::WorkerId::parse(&target.worker_id)
             .ok_or_else(|| Error::Store("Worker identity invalid".into()))?,
@@ -926,6 +974,7 @@ struct MaterialReader<'a> {
     runtime_id: String,
     source: runtime_api::SessionPublicSource,
     session_id: String,
+    not_materialized: bool,
     descriptors: Vec<session_store::public_material::PublicMaterialDescriptor>,
 }
 fn material_descriptor(
@@ -968,6 +1017,25 @@ impl<'a> MaterialReader<'a> {
                 )
                 .map_err(|e| e.into_error())?
             {
+                runtime_api::SessionMaterialCatalogAvailability::NotMaterialized {
+                    session_id: observed,
+                } => {
+                    if observed != session_id || offset != 0 || condition.is_some() {
+                        return Err(Error::RepositoryConflict(
+                            "Journal Session materialization proof changed during catalog read"
+                                .into(),
+                        ));
+                    }
+                    return Ok(Self {
+                        runtime,
+                        workspace_id,
+                        runtime_id: runtime_id.into(),
+                        source,
+                        session_id: session_id.into(),
+                        not_materialized: true,
+                        descriptors,
+                    });
+                }
                 runtime_api::SessionMaterialCatalogAvailability::Page { page } => page,
                 runtime_api::SessionMaterialCatalogAvailability::Unavailable {
                     reason,
@@ -1015,6 +1083,7 @@ impl<'a> MaterialReader<'a> {
             runtime_id: runtime_id.into(),
             source,
             session_id: session_id.into(),
+            not_materialized: false,
             descriptors,
         })
     }
@@ -1075,24 +1144,30 @@ impl<'a> MaterialReader<'a> {
     }
 }
 
-pub(super) fn source_pin_id(
+fn erase_source(
+    runtime: &RuntimeRegistry,
     workspace: &str,
     attr: &crate::subjektiv::SubjectSessionAttribution,
-) -> String {
-    use sha2::{Digest, Sha256};
-    let digest = Sha256::digest(
-        serde_json::to_vec(&(
-            "subjektiv-journal-retention-v1",
-            workspace,
-            &attr.subject_id,
-            &attr.session_id,
-        ))
-        .expect("string tuple serialization"),
-    );
-    let mut bytes = [0; 16];
-    bytes.copy_from_slice(&digest[..16]);
-    Uuid::from_bytes(bytes).to_string()
+    source: runtime_api::SessionPublicSource,
+) -> Result<()> {
+    let response = runtime
+        .session_material_erase(
+            &attr.runtime_id,
+            runtime_api::SessionMaterialEraseRequest {
+                workspace_id: workspace.into(),
+                source,
+                expected_session_id: attr.session_id.clone(),
+            },
+        )
+        .map_err(|e| e.into_error())?;
+    if response.session_id != attr.session_id {
+        return Err(Error::RepositoryConflict(
+            "Session erase receipt differs from the requested identity".into(),
+        ));
+    }
+    Ok(())
 }
+
 fn retain_source(
     runtime: &RuntimeRegistry,
     workspace: &str,
@@ -1100,7 +1175,9 @@ fn retain_source(
     source: runtime_api::SessionPublicSource,
     release: bool,
 ) -> Result<()> {
-    let pin_id = source_pin_id(workspace, attr);
+    let pin_id =
+        subjektiv::journal::source_retention_pin(workspace, &attr.subject_id, &attr.session_id)
+            .to_string();
     let request = runtime_api::SessionMaterialPinRequest {
         workspace_id: workspace.into(),
         source,
@@ -1340,8 +1417,10 @@ fn reconcile_session_jobs(
 }
 
 impl WorkspaceApi {
-    /// Host-only explicit erasure hook. The existing domain tombstone is the durable
-    /// cancellation intent; interrupted Worker Purge is retried by ordinary Job cleanup.
+    /// Host-only erasure saves the domain fence, cancels Jobs, then erases originals.
+    /// Runtime requires a stopped source Worker. Physical failure leaves the domain
+    /// tombstone and cancellations committed for exact retry or removal preflight;
+    /// terminal Job execution Sessions remain owned by ordinary Job cleanup.
     pub fn erase_subjektiv_journal_session(
         &self,
         session_id: &str,
@@ -1359,12 +1438,11 @@ impl WorkspaceApi {
             .erase_journal_session(&host, request_id)
             .map_err(domain)?;
         reconcile_session_jobs(self, &attr, true)?;
-        retain_source(
+        erase_source(
             self.runtime.as_ref(),
             self.workspace_id(),
             &attr,
             material_source(self, &attr)?,
-            true,
         )
     }
 }

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

@@ -394,6 +394,24 @@ fn seed_session(
     session: session_store::SessionId,
     item: session_store::LoggedItem,
     origin: session_store::LoggedSessionHistoryOrigin,
+) {
+    seed_session_history(
+        root,
+        session,
+        vec![session_store::LoggedHistoryEntry {
+            item,
+            metadata: session_store::LoggedSessionHistoryMetadata {
+                entry_id: session_store::LoggedSessionHistoryEntryId("read-source-original".into()),
+                origin,
+                derivation: None,
+            },
+        }],
+    );
+}
+fn seed_session_history(
+    root: &std::path::Path,
+    session: session_store::SessionId,
+    history: Vec<session_store::LoggedHistoryEntry>,
 ) {
     use session_store::{LogEntry, Store as _, WorkerSessionStore};
     let segment = Uuid::new_v4();
@@ -407,16 +425,7 @@ fn seed_session(
                 session_id: session,
                 system_prompt: None,
                 config: agen::llm_client::RequestConfig::default(),
-                history: vec![session_store::LoggedHistoryEntry {
-                    item,
-                    metadata: session_store::LoggedSessionHistoryMetadata {
-                        entry_id: session_store::LoggedSessionHistoryEntryId(
-                            "read-source-original".into(),
-                        ),
-                        origin,
-                        derivation: None,
-                    },
-                }],
+                history,
                 forked_from: None,
                 compacted_from: None,
             }],
@@ -486,7 +495,12 @@ impl RuntimeFixture {
                     workspace_id: TEST_WORKSPACE_ID.into(),
                     source: src.clone(),
                     expected_session_id: session.to_string(),
-                    pin_id: subjektiv_journal::source_pin_id(TEST_WORKSPACE_ID, &attr),
+                    pin_id: subjektiv::journal::source_retention_pin(
+                        TEST_WORKSPACE_ID,
+                        &attr.subject_id,
+                        &attr.session_id,
+                    )
+                    .to_string(),
                 },
             )
             .unwrap();
@@ -559,6 +573,18 @@ impl RuntimeFixture {
             id,
         }
     }
+    fn stop_source(&self) {
+        self.api
+            .runtime
+            .stop_worker(
+                &self.source,
+                WorkerLifecycleRequest {
+                    reason: Some("settle original source before permanent erase".into()),
+                    ticket_assignment: None,
+                },
+            )
+            .unwrap();
+    }
     fn dispatch_and_read(&self) -> BackendJobReservation {
         let dispatched = self
             .api
@@ -664,9 +690,10 @@ impl RuntimeFixture {
                 .unwrap()
                 .is_none()
         );
-        assert!(
+        assert_eq!(
             session_root(self.dir.path(), &self.source).exists(),
-            "the original conversation was not purged"
+            !self.store.journal_session_erased(&self.host).unwrap(),
+            "only explicit source erasure, never Job cleanup, removes the original conversation"
         );
     }
 }
@@ -713,6 +740,7 @@ async fn terminal_journal_attempts_purge_real_tool_result_sessions_without_chang
                     .unwrap();
             }
             "erased" => {
+                f.stop_source();
                 f.api
                     .erase_subjektiv_journal_session(
                         &f.session.to_string(),
@@ -1126,6 +1154,7 @@ async fn journal_profile_name_without_host_job_binding_keeps_normal_archive_poli
                 workspace_id: TEST_WORKSPACE_ID.into(),
                 worker: f.source,
                 reason: "profile is not purge authority".into(),
+                journal_erasure: None,
             },
             &inventory,
         )
@@ -1364,6 +1393,7 @@ async fn failed_cancelled_and_erased_job_sessions_retry_purge_after_cleanup_fail
             )
             .unwrap();
         if outcome == "erased" {
+            f.stop_source();
             f.api
                 .erase_subjektiv_journal_session(
                     &f.session.to_string(),
@@ -1472,6 +1502,7 @@ async fn explicit_erasure_interrupts_completed_journal_cleanup_waiting_for_run_t
         session_root(f.dir.path(), worker).exists(),
         "normal cleanup still waits for tracked response teardown"
     );
+    f.stop_source();
     f.api
         .erase_subjektiv_journal_session(&f.session.to_string(), "erase-while-completed-run-busy")
         .unwrap();
@@ -1497,3 +1528,616 @@ async fn explicit_erasure_interrupts_completed_journal_cleanup_waiting_for_run_t
     assert!(f.store.journal_session_erased(&f.host).unwrap());
     assert!(subjektiv_journal::execute(&f.api, worker, JournalJobRequest::ReadState).is_err());
 }
+
+/// Removal uses the production embedded WorkerRemove forwarder, target-bound proof,
+/// common reception, catalog/retention planner, Runtime and SQLite. Source creation
+/// calls the create handler directly (no HTTP auth); canonical Session contents,
+/// attribution and profile disablement are fixtures, not model execution or proofs
+/// of the normal publication/attribution connection.
+struct SubjectRemovalFixture {
+    dir: tempfile::TempDir,
+    api: WorkspaceApi,
+    dispatcher: Arc<crate::worker_source::EmbeddedServerWorkerMutationDispatcher>,
+    controller: RuntimeWorkerRef,
+    target: RuntimeWorkerRef,
+    store: crate::subjektiv::SubjektivStore,
+    attr: crate::subjektiv::SubjectSessionAttribution,
+    host: JournalHostContext,
+}
+impl SubjectRemovalFixture {
+    async fn new() -> Self {
+        Self::with_materialized_source(true).await
+    }
+    async fn with_materialized_source(materialized: bool) -> Self {
+        let dir = tempfile::tempdir().unwrap();
+        let config = test_server_config(dir.path());
+        let control = Arc::new(test_control_store(&config));
+        seed_test_registered_workspace(control.as_ref(), &config)
+            .await
+            .unwrap();
+        let dispatcher = Arc::new(
+            crate::worker_source::EmbeddedServerWorkerMutationDispatcher::new(
+                config.clone(),
+                control.clone(),
+            ),
+        );
+        let api = WorkspaceApi::new_with_execution_backend_and_broker(
+            config,
+            control,
+            Arc::new(DeterministicExecutionBackend::default()),
+            BackendResourceBroker::default(),
+            HostExecutors::default(),
+            Some(dispatcher.clone()),
+        )
+        .await
+        .unwrap();
+        let Json(orchestrator) = scoped_start_workspace_orchestrator(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+            }),
+        )
+        .await
+        .unwrap();
+        let controller = orchestrator.worker.unwrap();
+        let controller = RuntimeWorkerRef::new(controller.runtime_id, controller.worker_id);
+        let store = open_subjektiv_store(&api).unwrap();
+        let subject = store
+            .create_subject(crate::subjektiv::SubjectRole::new("removal-source").unwrap())
+            .unwrap();
+        let Json(created) = scoped_create_workspace_worker(
+            State(api.clone()),
+            AxumPath(ScopedWorkspacePath {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+            }),
+            HeaderMap::new(),
+            Json(CreateWorkspaceWorkerRequest {
+                runtime_id: EMBEDDED_WORKER_RUNTIME_ID.into(),
+                display_name: "removal-source".into(),
+                singleton_key: None,
+                profile: Some("builtin:companion".into()),
+                ticket_assignment: None,
+                initial_submit: vec![],
+                workdir_attachments: vec![],
+                feature_connections: server_api::WorkspaceWorkerFeatureConnectionsRequest {
+                    subjektiv: Some(server_api::SubjektivWorkerConnectionRequest {
+                        subject_id: subject.id.clone(),
+                    }),
+                },
+                control_operation_id: None,
+            }),
+        )
+        .await
+        .unwrap();
+        let target = RuntimeWorkerRef::new(created.runtime_id, created.worker_id);
+        let session = store
+            .subject_session_attributions(&subject.id)
+            .unwrap()
+            .first()
+            .map(|attr| attr.session_id.parse().unwrap())
+            .unwrap_or_else(session_store::new_session_id);
+        if materialized {
+            seed_session_history(
+                &session_root(dir.path(), &target),
+                session,
+                ["admitted-prefix", "disabled-new-history"]
+                    .into_iter()
+                    .map(|text| session_store::LoggedHistoryEntry {
+                        item: message(text),
+                        metadata: session_store::LoggedSessionHistoryMetadata {
+                            entry_id: session_store::LoggedSessionHistoryEntryId(text.into()),
+                            origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
+                            derivation: None,
+                        },
+                    })
+                    .collect(),
+            );
+        } else {
+            use session_store::WorkerMetadataStore as _;
+            let root = session_root(dir.path(), &target);
+            let aggregate = root.parent().unwrap();
+            let name = format!("worker-runtime-{}", target.worker_id);
+            session_store::WorkerAggregateStore::new(aggregate, &name)
+                .unwrap()
+                .write(
+                    &session_store::WorkerMetadata::new(
+                        &name,
+                        Some(session_store::WorkerActiveSegmentRef::pending_segment(
+                            session,
+                        )),
+                    )
+                    .with_workspace_id(TEST_WORKSPACE_ID),
+                )
+                .unwrap();
+        }
+        let attr = store
+            .record_session_attribution(
+                crate::subjektiv::SubjectSessionAttribution::new(
+                    &subject.id,
+                    &target.runtime_id,
+                    &target.worker_id,
+                    session.to_string(),
+                )
+                .unwrap(),
+            )
+            .unwrap();
+        let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap();
+        seed_worker_control_grant(&api, &controller, &target, "subject-removal-fixture");
+        api.runtime
+            .stop_worker(
+                &target,
+                WorkerLifecycleRequest {
+                    reason: Some("test stopped Subject removal".into()),
+                    ticket_assignment: None,
+                },
+            )
+            .unwrap();
+        let summary = api.runtime.worker(&target).unwrap();
+        assert_eq!(summary.state, "stopped");
+        sync_worker_observation(&api, &summary).unwrap();
+        Self {
+            dir,
+            api,
+            dispatcher,
+            controller,
+            target,
+            store,
+            attr,
+            host,
+        }
+    }
+    fn disable(&self) {
+        let mut registry = self
+            .api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &self.target)
+            .unwrap()
+            .unwrap();
+        registry.profile = Some("builtin:default".into());
+        self.api.store.upsert_worker_registry(&registry).unwrap();
+    }
+    fn remove(&self) -> worker::WorkspaceResponse {
+        let key = self
+            .api
+            .store
+            .worker_resource_key(TEST_WORKSPACE_ID, &self.target)
+            .unwrap()
+            .unwrap();
+        worker_runtime::worker_source::RuntimeWorkerMutationForwarder::embedded(
+            &self.controller.runtime_id,
+            worker_runtime::RuntimeWorkspaceScope::new(TEST_WORKSPACE_ID, "embedded-backend"),
+            &self.controller.worker_id,
+            self.dispatcher.clone(),
+        )
+        .execute_worker_remove(&self.target.runtime_id, &key, "remove stopped Subject")
+        .unwrap()
+    }
+}
+
+#[tokio::test]
+async fn disabled_subject_removal_does_not_admit_new_public_history_or_reserve_a_job() {
+    let f = SubjectRemovalFixture::new().await;
+    f.disable();
+    let response = f.remove();
+    assert_eq!(
+        response.status,
+        StatusCode::OK.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_none()
+    );
+    assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
+    assert!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+async fn disabled_subject_removal_retains_only_the_already_admitted_prefix() {
+    let f = SubjectRemovalFixture::new().await;
+    let runtime_api::SessionMaterialCatalogAvailability::Page { page } = f
+        .api
+        .runtime
+        .session_material_catalog(
+            &f.target.runtime_id,
+            runtime_api::SessionMaterialCatalogRequest {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                source: runtime_api::SessionPublicSource::Retained {
+                    worker_id: worker_runtime::identity::WorkerId::parse(&f.target.worker_id)
+                        .unwrap(),
+                },
+                expected_session_id: f.attr.session_id.clone(),
+                expected_catalog_sha256: None,
+                offset: 0,
+                limit: 100,
+            },
+        )
+        .unwrap()
+    else {
+        panic!("canonical source catalog required")
+    };
+    let ranges: Vec<_> = page
+        .descriptors
+        .into_iter()
+        .map(|d| JournalSourceRange {
+            source: JournalSource {
+                session_id: f.attr.session_id.clone(),
+                pointer: JournalSourcePointer {
+                    segment_id: d.segment_id.parse().unwrap(),
+                    entry_ref: d.entry_ref,
+                },
+                order: d.canonical_order,
+                content_sha256: d.sha256,
+                total_bytes: d.total_bytes,
+            },
+            bytes: JournalByteRange {
+                start: 0,
+                end: d.total_bytes,
+            },
+        })
+        .collect();
+    assert_eq!(ranges.len(), 2);
+    let id = f
+        .store
+        .accept_journal_sources(&f.host, "accepted before disablement", &ranges[..1])
+        .unwrap();
+    f.disable();
+    let response = f.remove();
+    assert_eq!(
+        response.status,
+        StatusCode::OK.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_none()
+    );
+    let jobs = f
+        .api
+        .store
+        .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+        .unwrap();
+    assert_eq!(jobs.len(), 1);
+    assert_eq!(
+        jobs[0]
+            .request
+            .grants
+            .subjektiv_journal
+            .as_ref()
+            .unwrap()
+            .selected_ranges,
+        ranges[..1]
+    );
+    assert_eq!(
+        f.store.journal_for_session(&f.host).unwrap().as_deref(),
+        Some(id.as_str())
+    );
+    assert!(f.store.journal_retention_required(&f.host).unwrap());
+    let archive = f
+        .api
+        .config_store
+        .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id)
+        .unwrap()
+        .unwrap();
+    assert!(!session_root(f.dir.path(), &f.target).exists());
+    let source = runtime_api::SessionPublicSource::Archived {
+        archive_id: archive.archive_id,
+    };
+    let response = f
+        .api
+        .runtime
+        .session_material_catalog(
+            &f.target.runtime_id,
+            runtime_api::SessionMaterialCatalogRequest {
+                workspace_id: TEST_WORKSPACE_ID.into(),
+                source,
+                expected_session_id: f.attr.session_id.clone(),
+                expected_catalog_sha256: None,
+                offset: 0,
+                limit: 100,
+            },
+        )
+        .unwrap();
+    assert!(matches!(
+        response,
+        runtime_api::SessionMaterialCatalogAvailability::Page { .. }
+    ));
+}
+
+#[tokio::test]
+async fn attributed_unmaterialized_subject_removal_needs_no_journal_job() {
+    let f = SubjectRemovalFixture::with_materialized_source(false).await;
+    let retention = source_retention(f.dir.path());
+    let session = f.attr.session_id.parse().unwrap();
+    // Fixture the pre-publication pin without materializing the Session.
+    retention
+        .pin(
+            session,
+            session_store::retention::SessionRetentionPin(
+                subjektiv::journal::source_retention_pin(
+                    TEST_WORKSPACE_ID,
+                    &f.attr.subject_id,
+                    &f.attr.session_id,
+                )
+                .to_string()
+                .parse()
+                .unwrap(),
+            ),
+            || Ok(true),
+        )
+        .unwrap();
+    assert_eq!(retention.pins(session).unwrap().len(), 1);
+    let response = f.remove();
+    assert!(retention.pins(session).unwrap().is_empty());
+    assert_eq!(
+        response.status,
+        StatusCode::OK.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_none()
+    );
+    assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
+    assert!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+async fn missing_materialized_subject_source_blocks_removal_without_marking_it_empty() {
+    let f = SubjectRemovalFixture::new().await;
+    std::fs::remove_dir_all(session_root(f.dir.path(), &f.target)).unwrap();
+    let response = f.remove();
+    assert_eq!(
+        response.status,
+        StatusCode::CONFLICT.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(response.body.contains("journal_handoff_incomplete"));
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_some()
+    );
+    assert!(f.store.journal_for_session(&f.host).unwrap().is_none());
+    assert!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+            .unwrap()
+            .is_empty()
+    );
+}
+
+fn source_retention(root: &std::path::Path) -> session_store::retention::SessionRetentionStore {
+    session_store::retention::SessionRetentionStore::new(
+        root.join(".test-embedded-runtime-store/retention/session-sources"),
+    )
+}
+
+#[tokio::test]
+async fn explicitly_erased_subject_removal_purges_without_generating_or_archiving() {
+    let f = SubjectRemovalFixture::new().await;
+    let policy = f
+        .api
+        .config_store
+        .worker_retention_policy(TEST_WORKSPACE_ID)
+        .unwrap();
+    f.api
+        .erase_subjektiv_journal_session(&f.attr.session_id, "erase-before-normal-remove")
+        .unwrap();
+    assert!(f.store.journal_session_erased(&f.host).unwrap());
+    assert!(!session_root(f.dir.path(), &f.target).exists());
+    // Exact erase replay must resume the same permanent intent even after source removal.
+    f.api
+        .erase_subjektiv_journal_session(&f.attr.session_id, "erase-before-normal-remove")
+        .unwrap();
+    let response = f.remove();
+    assert_eq!(
+        response.status,
+        StatusCode::OK.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_none()
+    );
+    let saved = f
+        .api
+        .config_store
+        .recover_worker_removal_execution(TEST_WORKSPACE_ID, &f.target)
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        saved.plan.session_disposition,
+        worker_runtime::retention::SessionDisposition::Purge
+    );
+    assert_eq!(
+        saved.plan.diagnostics_disposition,
+        worker_runtime::retention::DiagnosticsDisposition::Purge
+    );
+    assert!(saved.plan.archive_id.is_none());
+    assert!(
+        f.api
+            .config_store
+            .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id)
+            .unwrap()
+            .is_none()
+    );
+    assert!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+            .unwrap()
+            .is_empty()
+    );
+    assert_eq!(
+        f.api
+            .config_store
+            .worker_retention_policy(TEST_WORKSPACE_ID)
+            .unwrap(),
+        policy
+    );
+    assert!(matches!(
+        source_retention(f.dir.path()).ensure_live(f.attr.session_id.parse().unwrap()),
+        Err(session_store::StoreError::SessionErased(_))
+    ));
+}
+
+#[tokio::test]
+async fn erased_subject_removal_retries_interrupted_physical_erasure_under_the_permanent_fence() {
+    let f = SubjectRemovalFixture::new().await;
+    // Fixture boundary: interrupt the real retention deletion after its durable fence,
+    // before physical removal. WorkerRemove below uses the real Backend/Runtime erase path.
+    f.store
+        .erase_journal_session(&f.host, "erase-interrupted")
+        .unwrap();
+    let retention = source_retention(f.dir.path());
+    let session = f.attr.session_id.parse().unwrap();
+    assert!(
+        retention
+            .delete(
+                session,
+                session_store::retention::SessionDeletionIntent::ErasePermanently,
+                || {
+                    Err::<(), _>(session_store::StoreError::Io(std::io::Error::other(
+                        "injected physical cleanup failure",
+                    )))
+                }
+            )
+            .is_err()
+    );
+    assert!(session_root(f.dir.path(), &f.target).exists());
+    let response = f.remove();
+    assert_eq!(
+        response.status,
+        StatusCode::OK.as_u16(),
+        "{}",
+        response.body
+    );
+    assert!(
+        !session_root(f.dir.path(), &f.target)
+            .parent()
+            .unwrap()
+            .exists()
+    );
+    assert!(
+        f.api
+            .store
+            .get_worker_registry(TEST_WORKSPACE_ID, &f.target)
+            .unwrap()
+            .is_none()
+    );
+    assert!(
+        f.api
+            .config_store
+            .worker_session_archive(TEST_WORKSPACE_ID, &f.target, &f.attr.session_id)
+            .unwrap()
+            .is_none()
+    );
+    assert!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.attr.session_id, None, 100)
+            .unwrap()
+            .is_empty()
+    );
+    assert!(matches!(
+        retention.ensure_live(session),
+        Err(session_store::StoreError::SessionErased(_))
+    ));
+}
+
+#[tokio::test]
+async fn physical_erase_failure_keeps_domain_tombstone_and_job_cancellation_for_exact_retry() {
+    let f = RuntimeFixture::new().await;
+    let dispatched = f.dispatch_and_read();
+    // Original source is not stopped: Runtime must reject physical erasure, after
+    // the Backend has durably fenced all Journal adoption and cancelled its Job.
+    assert!(
+        f.api
+            .erase_subjektiv_journal_session(&f.session.to_string(), "erase-unsettled-source")
+            .is_err()
+    );
+    assert!(f.store.journal_session_erased(&f.host).unwrap());
+    assert!(session_root(f.dir.path(), &f.source).exists());
+    let job = f
+        .api
+        .store
+        .get_backend_job(TEST_WORKSPACE_ID, &f.request.job_id)
+        .unwrap()
+        .unwrap();
+    assert_eq!(job.state, BackendJobState::Failed);
+    assert!(
+        subjektiv_journal::execute(
+            &f.api,
+            dispatched.attempt.worker.as_ref().unwrap(),
+            JournalJobRequest::ReadState
+        )
+        .is_err()
+    );
+    f.stop_source();
+    f.api
+        .erase_subjektiv_journal_session(&f.session.to_string(), "erase-unsettled-source")
+        .unwrap();
+    assert!(!session_root(f.dir.path(), &f.source).exists());
+    assert_eq!(
+        f.api
+            .store
+            .list_backend_journal_jobs(TEST_WORKSPACE_ID, &f.session.to_string(), None, 100)
+            .unwrap()
+            .len(),
+        1
+    );
+}
+
+#[tokio::test]
+async fn serialized_removal_input_cannot_supply_journal_erasure_authority() {
+    let f = SubjectRemovalFixture::new().await;
+    let request: crate::retention::WorkerRemovalPlanRequest = serde_json::from_value(json!({
+        "workspace_id": TEST_WORKSPACE_ID, "worker": f.target, "reason": "untrusted erase claim",
+        "journal_erasure": {"workspace_id": TEST_WORKSPACE_ID, "worker": f.target, "session_id": f.attr.session_id},
+    })).unwrap();
+    let inventory = f.api.runtime.worker_retention_inventory(&f.target).unwrap();
+    let plan = f
+        .api
+        .store
+        .plan_worker_removal(&request, &inventory)
+        .unwrap();
+    assert_eq!(
+        plan.session_disposition,
+        worker_runtime::retention::SessionDisposition::Archive
+    );
+    assert!(!f.store.journal_session_erased(&f.host).unwrap());
+    assert!(session_root(f.dir.path(), &f.target).exists());
+}