chore: merge develop into hare/develop

crates/standalone/src/host.rs

@@ -699,8 +699,11 @@ fn backing_store(
     store: &StandaloneWorkerStore,
     worker_id: WorkerId,
 ) -> Result<(StandaloneBackingStore, FsWorkerStore), StandaloneStartupError> {
-    let session_store = FsStore::new(store.sessions_dir(worker_id))
-        .map_err(|_| StandaloneStartupError::StateStore)?;
+    let session_store = FsStore::new_with_retention(
+        store.sessions_dir(worker_id),
+        crate::store::session_retention(store.root()),
+    )
+    .map_err(|_| StandaloneStartupError::StateStore)?;
     let worker_store = FsWorkerStore::new(store.worker_metadata_dir(worker_id))
         .map_err(|_| StandaloneStartupError::StateStore)?;
     Ok((
@@ -813,3 +816,51 @@ mod local_input_tests {
         }
     }
 }
+
+#[cfg(test)]
+mod source_retention_tests {
+    //! Host backing-store assembly uses the production publisher and an isolated external
+    //! erasure fence. No controller/model is substituted for the source authority under test.
+    use super::*;
+    use session_store::{LogEntry, Store, StoreError};
+
+    #[test]
+    fn backing_publisher_and_reopen_use_the_host_wide_erasure_fence() {
+        let temp = tempfile::tempdir().unwrap();
+        let workers = StandaloneWorkerStore::open(temp.path().join("state")).unwrap();
+        let allocation = workers
+            .allocate(temp.path(), StaleLeasePolicy::Reject)
+            .unwrap();
+        let worker = allocation.worker_id();
+        let (backing, _) = backing_store(&workers, worker).unwrap();
+        let session = session_store::SessionId::now_v7();
+        let segment = uuid::Uuid::now_v7();
+        let entry = LogEntry::TurnEnd {
+            ts: 1,
+            turn_count: 1,
+        };
+        backing
+            .create_segment(session, segment, &[entry.clone()])
+            .unwrap();
+        let retention = crate::store::session_retention(workers.root());
+        assert!(
+            retention
+                .delete(
+                    session,
+                    session_store::retention::SessionDeletionIntent::ErasePermanently,
+                    || Err::<(), StoreError>(
+                        std::io::Error::other("injected physical erase failure").into()
+                    )
+                )
+                .is_err()
+        );
+        assert!(
+            matches!(backing.append(session, segment, &entry), Err(StoreError::SessionErased(id)) if id == session)
+        );
+        let (reopened, _) = backing_store(&workers, worker).unwrap();
+        assert!(
+            matches!(reopened.append(session, segment, &entry), Err(StoreError::SessionErased(id)) if id == session)
+        );
+        assert!(!workers.sessions_dir(worker).join(".retention").exists());
+    }
+}

crates/standalone/src/job_store.rs

@@ -25,7 +25,7 @@ use thiserror::Error;
 #[path = "job_store/legacy_migrations.rs"]
 mod legacy_migrations;
 
-const SCHEMA_VERSION: i64 = 3;
+const SCHEMA_VERSION: i64 = 4;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "snake_case")]
@@ -189,6 +189,7 @@ impl JobStore {
             0 => {
                 tx.execute_batch(SCHEMA)?;
                 tx.execute_batch(GRANT_SCHEMA)?;
+                tx.execute_batch(RECONCILIATION_SCHEMA)?;
                 tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
             }
             1 | 2 => {
@@ -196,6 +197,11 @@ impl JobStore {
                     tx.execute_batch(GRANT_SCHEMA)?;
                 }
                 legacy_migrations::migrate_legacy_jobs(&tx)?;
+                tx.execute_batch(RECONCILIATION_SCHEMA)?;
+                tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
+            }
+            3 => {
+                tx.execute_batch(RECONCILIATION_SCHEMA)?;
                 tx.pragma_update(None, "user_version", SCHEMA_VERSION)?;
             }
             SCHEMA_VERSION => {}
@@ -278,6 +284,49 @@ impl JobStore {
         Ok(snapshot)
     }
 
+    /// Domain Host attests that this fenced attempt's only effects are reconciled in its
+    /// authoritative durable consumer. This releases serialization, not the original Unknown
+    /// history, and cannot authorize retry or change its result. The service checks cleanup.
+    pub(crate) fn reconcile_granted_attempt(
+        &mut self,
+        job_id: &str,
+        attempt_id: &str,
+        domain_receipt: &str,
+    ) -> Result<(), JobStoreError> {
+        if domain_receipt.is_empty()
+            || domain_receipt.len() > 256
+            || domain_receipt.chars().any(char::is_control)
+        {
+            return Err(JobStoreError::InvalidTransition(
+                "invalid domain reconciliation receipt",
+            ));
+        }
+        let tx = self
+            .conn
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let snapshot = require_snapshot(&tx, job_id)?;
+        check_attempt(&snapshot, attempt_id)?;
+        if snapshot.domain_grant.is_none()
+            || snapshot.attempt.state != JobAttemptState::Unknown
+            || !matches!(snapshot.state, JobState::Unknown | JobState::Cancelled)
+        {
+            return Err(JobStoreError::InvalidTransition(
+                "domain reconciliation requires a fenced unknown granted attempt",
+            ));
+        }
+        let old: Option<String> = tx.query_row("SELECT domain_receipt FROM job_domain_reconciliations WHERE job_id=?1 AND attempt_id=?2",
+            params![job_id,attempt_id], |row| row.get(0)).optional()?;
+        if old.as_deref().is_some_and(|old| old != domain_receipt) {
+            return Err(JobStoreError::InvalidTransition(
+                "domain reconciliation receipt changed",
+            ));
+        }
+        tx.execute("INSERT OR IGNORE INTO job_domain_reconciliations(job_id,attempt_id,domain_receipt) VALUES(?1,?2,?3)",
+            params![job_id,attempt_id,domain_receipt])?;
+        tx.commit()?;
+        Ok(())
+    }
+
     /// Only definitive failure can create a new immutable attempt. Unknown
     /// outcomes require caller reconciliation, never an automatic replay.
     pub(crate) fn retry(&mut self, job_id: &str) -> Result<JobSnapshot, JobStoreError> {
@@ -319,6 +368,16 @@ impl JobStore {
             .collect()
     }
 
+    pub(crate) fn unknown_granted(&self) -> Result<Vec<JobSnapshot>, JobStoreError> {
+        let mut statement = self.conn.prepare("SELECT i.job_id FROM job_intents i JOIN job_attempts a ON a.job_id=i.job_id AND a.number=i.current_attempt JOIN job_domain_grants g ON g.job_id=i.job_id WHERE a.state='unknown' AND NOT EXISTS (SELECT 1 FROM job_domain_reconciliations r WHERE r.job_id=a.job_id AND r.attempt_id=a.attempt_id) ORDER BY i.rowid")?;
+        let ids = statement
+            .query_map([], |row| row.get::<_, String>(0))?
+            .collect::<rusqlite::Result<Vec<_>>>()?;
+        ids.into_iter()
+            .map(|id| require_snapshot(&self.conn, &id))
+            .collect()
+    }
+
     /// Only unstarted current attempts are eligible for service startup/resume.
     pub(crate) fn pending(&self) -> Result<Vec<JobSnapshot>, JobStoreError> {
         let mut statement = self.conn.prepare(
@@ -527,8 +586,10 @@ fn check_serialization(
     if let Some(key) = key {
         let conflicting: bool = conn.query_row(
             "SELECT EXISTS(SELECT 1 FROM job_intents i WHERE serialization_key = ?1
-             AND job_id <> ?2 AND (state IN ('pending', 'unknown') OR EXISTS
-                 (SELECT 1 FROM job_attempts a WHERE a.job_id = i.job_id AND a.state = 'unknown')))",
+             AND job_id <> ?2 AND (state = 'pending' OR EXISTS
+                 (SELECT 1 FROM job_attempts a WHERE a.job_id = i.job_id AND a.state = 'unknown'
+                  AND NOT EXISTS (SELECT 1 FROM job_domain_reconciliations r
+                    WHERE r.job_id=a.job_id AND r.attempt_id=a.attempt_id))))",
             params![key, job_id],
             |row| row.get(0),
         )?;
@@ -658,6 +719,22 @@ fn read_snapshot(conn: &Connection, job_id: &str) -> Result<Option<JobSnapshot>,
     .transpose()
 }
 
+const RECONCILIATION_SCHEMA: &str = "
+DROP INDEX job_serialization_singleflight;
+CREATE UNIQUE INDEX job_serialization_singleflight ON job_intents(serialization_key)
+    WHERE serialization_key IS NOT NULL AND state = 'pending';
+CREATE TABLE job_domain_reconciliations (
+    job_id TEXT NOT NULL REFERENCES job_intents(job_id),
+    attempt_id TEXT NOT NULL,
+    domain_receipt TEXT NOT NULL,
+    PRIMARY KEY (job_id, attempt_id)
+);
+CREATE TRIGGER job_reconciliation_immutable BEFORE UPDATE ON job_domain_reconciliations
+BEGIN SELECT RAISE(ABORT, 'immutable domain reconciliation'); END;
+CREATE TRIGGER job_reconciliation_retained BEFORE DELETE ON job_domain_reconciliations
+BEGIN SELECT RAISE(ABORT, 'retained domain reconciliation'); END;
+";
+
 const GRANT_SCHEMA: &str = "
 CREATE TABLE job_domain_grants (
     job_id TEXT PRIMARY KEY NOT NULL REFERENCES job_intents(job_id),
@@ -732,6 +809,98 @@ mod tests {
         }
     }
 
+    #[test]
+    fn granted_unknown_reconciliation_releases_serialization_but_preserves_history() {
+        let (dir, mut store) = fixture();
+        let mut first = request("first");
+        first.serialization_key = Some("journal".into());
+        let reserved = store
+            .reserve_granted(first, Some(json!({"domain":"journal"})))
+            .unwrap();
+        store
+            .dispatch("first", &reserved.attempt.attempt_id, 8)
+            .unwrap();
+        let unknown = store
+            .finish(
+                "first",
+                &reserved.attempt.attempt_id,
+                JobAttemptState::Unknown,
+                Some("response lost"),
+            )
+            .unwrap();
+        let mut second = request("second");
+        second.serialization_key = Some("journal".into());
+        assert!(matches!(
+            store.reserve(second.clone()),
+            Err(JobStoreError::SerializationConflict(_))
+        ));
+        assert!(
+            store
+                .reconcile_granted_attempt("first", "foreign-attempt", "domain-fenced")
+                .is_err()
+        );
+        store
+            .reconcile_granted_attempt("first", &unknown.attempt.attempt_id, "domain-fenced")
+            .unwrap();
+        store
+            .reconcile_granted_attempt("first", &unknown.attempt.attempt_id, "domain-fenced")
+            .unwrap();
+        assert!(
+            store
+                .reconcile_granted_attempt("first", &unknown.attempt.attempt_id, "different-proof")
+                .is_err()
+        );
+        assert_eq!(store.get("first").unwrap(), unknown);
+        store.reserve(second).unwrap();
+        assert!(store.retry("first").is_err());
+        drop(store);
+        let mut reopened = JobStore::open(&dir.path().join("jobs.sqlite3")).unwrap();
+        assert_eq!(reopened.get("first").unwrap(), unknown);
+        assert!(reopened.unknown_granted().unwrap().is_empty());
+        assert!(
+            reopened
+                .conn
+                .execute("DELETE FROM job_domain_reconciliations", [])
+                .is_err()
+        );
+        assert!(reopened.retry("first").is_err());
+    }
+
+    #[test]
+    fn schema_three_migration_preserves_unknown_grant_and_its_serialization_fence() {
+        let (dir, mut store) = fixture();
+        let mut first = request("first");
+        first.serialization_key = Some("journal".into());
+        let reserved = store
+            .reserve_granted(first, Some(json!({"domain":"journal"})))
+            .unwrap();
+        store
+            .dispatch("first", &reserved.attempt.attempt_id, 8)
+            .unwrap();
+        let unknown = store
+            .finish(
+                "first",
+                &reserved.attempt.attempt_id,
+                JobAttemptState::Unknown,
+                Some("unconfirmed"),
+            )
+            .unwrap();
+        store.conn.execute_batch("DROP TRIGGER job_reconciliation_immutable; DROP TRIGGER job_reconciliation_retained; DROP TABLE job_domain_reconciliations; DROP INDEX job_serialization_singleflight; CREATE UNIQUE INDEX job_serialization_singleflight ON job_intents(serialization_key) WHERE serialization_key IS NOT NULL AND state IN ('pending','unknown'); PRAGMA user_version=3;").unwrap();
+        drop(store);
+        let mut reopened = JobStore::open(&dir.path().join("jobs.sqlite3")).unwrap();
+        assert_eq!(reopened.get("first").unwrap(), unknown);
+        let mut next = request("next");
+        next.serialization_key = Some("journal".into());
+        assert!(matches!(
+            reopened.reserve(next.clone()),
+            Err(JobStoreError::SerializationConflict(_))
+        ));
+        reopened
+            .reconcile_granted_attempt("first", &unknown.attempt.attempt_id, "domain-fenced")
+            .unwrap();
+        reopened.reserve(next).unwrap();
+    }
+
     fn fixture() -> (TempDir, JobStore) {
         let dir = tempfile::tempdir().unwrap();
         let store = JobStore::open(&dir.path().join("jobs.sqlite3")).unwrap();

crates/standalone/src/jobs.rs

@@ -157,6 +157,30 @@ impl StandaloneJobs {
         self.with_store(|store| store.unacknowledged())
     }
 
+    pub(crate) fn unknown_granted(&self) -> Result<Vec<JobSnapshot>, JobServiceError> {
+        self.with_store(|store| store.unknown_granted())
+    }
+
+    /// Release only an Unknown attempt's serialization reservation after the bound domain
+    /// has reconciled its receipts and durably fenced further effects. Never a generic retry.
+    pub(crate) fn reconcile_granted_attempt(
+        &self,
+        job_id: &str,
+        attempt_id: &str,
+        domain_receipt: &str,
+    ) -> Result<(), JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        if executions.jobs.contains_key(job_id) {
+            return Err(JobServiceError::Cleanup(
+                "domain reconciliation requires observed execution cleanup".into(),
+            ));
+        }
+        self.with_store(|store| store.reconcile_granted_attempt(job_id, attempt_id, domain_receipt))
+    }
+
     /// Explicit new attempt, bounded by immutable intent limits. Unknown
     /// outcomes cannot be retried without domain reconciliation/new intent.
     pub fn retry(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {

crates/standalone/src/store.rs

@@ -119,6 +119,16 @@ pub enum StaleLeasePolicy {
     Recover,
 }
 
+/// Source control outlives Worker relocation and physical Session pruning. Always derive this
+/// from the Host state root, never from an individual live or retained Worker directory.
+pub(crate) fn session_retention(
+    state_root: &Path,
+) -> session_store::retention::SessionRetentionStore {
+    session_store::retention::SessionRetentionStore::new(
+        state_root.join("retention/session-sources"),
+    )
+}
+
 #[derive(Debug, Clone)]
 pub struct StandaloneWorkerStore {
     root: PathBuf,
@@ -359,7 +369,19 @@ impl StandaloneWorkerStore {
     }
 
     pub fn delete(&self, id: WorkerId) -> Result<(), StandaloneStoreError> {
-        let record = self.load(id)?;
+        let record = match self.load(id) {
+            Err(StandaloneStoreError::WorkerNotFound(_)) => {
+                let retained_root = self.root.join("subjektiv").join("retained-workers");
+                let retained =
+                    Self::open_existing(retained_root).and_then(|store| store.load(id))?;
+                if retained.status == StandaloneWorkerStatus::Stopped && retained.subject.is_some()
+                {
+                    return Ok(());
+                }
+                return Err(StandaloneStoreError::InvalidRecord(id));
+            }
+            result => result?,
+        };
         if record.status != StandaloneWorkerStatus::Stopped {
             return Err(StandaloneStoreError::DeleteActive(id));
         }
@@ -374,7 +396,14 @@ impl StandaloneWorkerStore {
                 LeaseLiveness::Unknown => StandaloneStoreError::LeaseLivenessUnknown(id),
             });
         }
-        fs::remove_dir_all(self.worker_dir(id)).map_err(StandaloneStoreError::Io)?;
+        if let Some(retained) = crate::subjektiv::prepare_worker_deletion(&self.root, &record)? {
+            // Atomic relocation preserves the original Session, not a permanent duplicate.
+            // The Subject's durable consumer can read it after logical Worker removal.
+            fs::rename(self.worker_dir(id), &retained).map_err(StandaloneStoreError::Io)?;
+            sync_directory(retained.parent().expect("retained Worker parent"))?;
+        } else {
+            fs::remove_dir_all(self.worker_dir(id)).map_err(StandaloneStoreError::Io)?;
+        }
         sync_directory(&self.root)
     }
 
@@ -793,6 +822,8 @@ pub enum StandaloneStoreError {
     LeaseOwnershipLost,
     #[error("standalone Worker {0} must be stopped before deletion")]
     DeleteActive(WorkerId),
+    #[error("Journal deletion handoff/retention is unconfirmed: {0}")]
+    JournalRetention(String),
     #[error("standalone Worker {0} metadata changed since it was read")]
     MetadataChanged(WorkerId),
     #[error("system clock is before the Unix epoch or out of range")]

crates/standalone/src/subjektiv.rs

@@ -15,6 +15,9 @@
 //! Unknown attempts are not rerun; delivery failures for accepted results do not trigger model
 //! reexecution. Memory application and surface repair use shared-domain CAS/receipts.
 
+mod journal;
+pub(crate) use journal::prepare_worker_deletion;
+
 use std::fs::{self, File, OpenOptions};
 use std::io::Write;
 use std::path::{Path, PathBuf};
@@ -220,7 +223,9 @@ impl StandaloneSubjects {
             scope_id: self.scope_id.clone(),
             subject_id: subject_id.to_string(),
         };
-        let host = Arc::new(LocalSubjectHost {
+        let host = Arc::new_cyclic(|weak| LocalSubjectHost {
+            self_weak: weak.clone(),
+            journal_dispatch: Mutex::new(false),
             language: self.memory_language.clone(),
             catalog: self,
             binding: binding.clone(),
@@ -277,6 +282,8 @@ impl Drop for SubjectConnection {
 }
 
 pub(crate) struct LocalSubjectHost {
+    self_weak: std::sync::Weak<Self>,
+    journal_dispatch: Mutex<bool>,
     catalog: StandaloneSubjects,
     binding: StandaloneSubjectBinding,
     worker_id: protocol::WorkerId,
@@ -551,10 +558,11 @@ impl LocalSubjectHost {
                     .and_then(|v| serde_json::from_value::<LocalConsolidationGrant>(v.clone()).ok())
                     .is_some_and(|g| g.binding == self.binding)
                 {
-                    schedule_job(jobs.clone(), job.request.job_id);
+                    schedule_job(jobs.clone(), job.request.job_id, self.self_weak.clone());
                 }
             }
         }
+        self.recover_journals();
     }
     fn validate_evidence(
         &self,
@@ -690,6 +698,22 @@ impl LocalSubjectHost {
     }
 }
 impl SubjektivHost for LocalSubjectHost {
+    fn journal_recording_granted(&self) -> bool {
+        self.live.load(Ordering::Acquire)
+    }
+    fn request_journal_recording(
+        &self,
+        context: &SubjektivHostContext,
+        _reason: subjektiv::journal::transport::JournalRecordingReason,
+    ) -> Result<subjektiv::journal::transport::JournalRecordingResponse, SubjektivHostError> {
+        let admitted = self.admit_journal(context)?;
+        self.wake_journals();
+        Ok(if admitted {
+            subjektiv::journal::transport::JournalRecordingResponse::Accepted
+        } else {
+            subjektiv::journal::transport::JournalRecordingResponse::NoNewSources
+        })
+    }
     fn settings(&self) -> SubjektivHostSettings {
         SubjektivHostSettings {
             language: self.language.clone(),
@@ -788,6 +812,7 @@ impl SubjektivHost for LocalSubjectHost {
         create_if_missing: bool,
     ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
         let attribution = self.attribution(context, create_if_missing)?;
+        self.reserve_journal_session(context)?;
         Ok(SubjektivRecordSessionResponse {
             subject_id: attribution.subject_id,
             session_id: attribution.session_id,
@@ -811,7 +836,7 @@ impl SubjektivHost for LocalSubjectHost {
                     .is_some_and(|g| g.binding == self.binding)
             })
         {
-            schedule_job(jobs, pending.request.job_id);
+            schedule_job(jobs, pending.request.job_id, self.self_weak.clone());
             let (count, bytes) = self
                 .catalog
                 .store
@@ -874,7 +899,7 @@ impl SubjektivHost for LocalSubjectHost {
                     .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?,
             )
             .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?;
-        schedule_job(jobs, snapshot.request.job_id);
+        schedule_job(jobs, snapshot.request.job_id, self.self_weak.clone());
         Ok(memory::backend::MemoryConsolidationOutput { status:"started".into(),summary:"Requested a durable generic standalone consolidation Job; processing stops on Host shutdown".into(),candidate_count:count,total_bytes:bytes })
     }
 }
@@ -894,6 +919,9 @@ impl crate::jobs::JobDomainProvider for LocalDomainProvider {
     ) -> Result<worker::job::JobFeatureGrant, String> {
         let host = self.0.upgrade().ok_or("Subject Host unavailable")?;
         host.ensure_live().map_err(|e| e.to_string())?;
+        if snapshot.request.purpose == journal::PURPOSE {
+            return journal::grant(host, snapshot, fence);
+        }
         let grant: LocalConsolidationGrant = serde_json::from_value(
             snapshot
                 .domain_grant
@@ -1097,7 +1125,11 @@ impl worker::job::JobResultFinalizer for LocalConsolidationFinalizer {
     }
 }
 
-fn schedule_job(jobs: crate::jobs::StandaloneJobs, id: String) {
+fn schedule_job(
+    jobs: crate::jobs::StandaloneJobs,
+    id: String,
+    host: std::sync::Weak<LocalSubjectHost>,
+) {
     // Scheduling/consumer acknowledgement has no Internal Worker ownership.
     // The generic service fences and retains every execution until cleanup.
     tokio::spawn(async move {
@@ -1114,6 +1146,9 @@ fn schedule_job(jobs: crate::jobs::StandaloneJobs, id: String) {
                 "standalone Subject Job {id} cleanup/acknowledgement is unconfirmed: {error}"
             ),
         }
+        if let Some(host) = host.upgrade() {
+            host.wake_journals();
+        }
     });
 }
 

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

@@ -0,0 +1,429 @@
+//! Public standalone protocol through real mid-turn compaction and Journal Jobs.
+//! Only external model streams are scripted. No history, attribution, hook or Job
+//! result is injected; cancellation and rollback use the controller's public commands.
+//! Reported model usage and a low configured request threshold force two rewrites.
+//! The test attests durable public originals, not unfinished display deltas or meaning.
+
+use super::*;
+use agen::llm_client::{
+    ClientError, LlmClient, Request,
+    event::{Event, ResponseStatus, StatusEvent, StopReason, UsageEvent},
+    types::Item,
+};
+use futures::{Stream, StreamExt};
+use std::pin::Pin;
+use std::sync::atomic::AtomicUsize;
+
+type ResponseStream = Pin<Box<dyn Stream<Item = Result<Event, ClientError>> + Send>>;
+const SUMMARY: &str = "DERIVED_COMPACTION_SENTINEL_NOT_AN_ORIGINAL";
+const ROLLED_BACK: &str = "ROLLED_BACK_INPUT_MUST_NOT_REACH_JOURNAL";
+const PARTIAL: &str = "public partial answer before cancellation";
+
+#[derive(Clone, Default)]
+struct CompactingDialog {
+    shared: DialogAndJournal,
+    dialog_calls: Arc<AtomicUsize>,
+    compactions: Arc<AtomicUsize>,
+    rollback_entered: Arc<tokio::sync::Notify>,
+    partial_stream_paused: Arc<tokio::sync::Notify>,
+}
+
+fn events(events: Vec<Event>) -> ResponseStream {
+    Box::pin(futures::stream::iter(events.into_iter().map(Ok)))
+}
+fn done() -> ResponseStream {
+    events(vec![
+        Event::text_block_start(0),
+        Event::text_delta(0, "done"),
+        Event::text_block_stop(0, Some(StopReason::EndTurn)),
+    ])
+}
+
+#[async_trait::async_trait]
+impl LlmClient for CompactingDialog {
+    async fn stream(&self, request: Request) -> Result<ResponseStream, ClientError> {
+        if request
+            .tools
+            .iter()
+            .any(|tool| tool.name == "write_summary")
+        {
+            if request.items.iter().any(|item| matches!(item, Item::ToolResult { call_id, .. } if call_id == "compact-summary")) {
+                return Ok(done());
+            }
+            let count = self.compactions.fetch_add(1, Ordering::SeqCst);
+            assert!(count < 2, "unexpected additional compaction");
+            return Ok(events(vec![
+                Event::tool_use_start(0, "compact-summary", "write_summary"),
+                Event::tool_input_delta(0, serde_json::json!({"text": SUMMARY}).to_string()),
+                Event::tool_use_stop(0),
+                Event::Status(StatusEvent {
+                    status: ResponseStatus::Completed,
+                }),
+            ]));
+        }
+        if request.tools.iter().any(|tool| {
+            matches!(
+                tool.name.as_str(),
+                "ReadJournalState" | "SubmitMemorySurface" | "SubmitJobResult"
+            )
+        }) {
+            return self.shared.stream(request).await;
+        }
+        match self.dialog_calls.fetch_add(1, Ordering::SeqCst) {
+            step @ (0 | 1) => {
+                assert!(request.tools.iter().any(|tool| tool.name == "TaskCreate"));
+                Ok(events(vec![
+                    Event::tool_use_start(0, format!("public-task-{step}"), "TaskCreate"),
+                    Event::tool_input_delta(
+                        0,
+                        serde_json::json!({
+                            "subject": format!("original activity {step}"),
+                            "description": format!("original task result before compaction {step}")
+                        })
+                        .to_string(),
+                    ),
+                    Event::tool_use_stop(0),
+                    Event::Usage(UsageEvent {
+                        input_tokens: Some(100_000),
+                        output_tokens: Some(10),
+                        total_tokens: Some(100_010),
+                        ..Default::default()
+                    }),
+                    Event::Status(StatusEvent {
+                        status: ResponseStatus::Completed,
+                    }),
+                ]))
+            }
+            2 => {
+                let paused = self.partial_stream_paused.clone();
+                Ok(Box::pin(
+                    events(vec![
+                        Event::text_block_start(0),
+                        Event::text_delta(0, PARTIAL),
+                        Event::text_block_stop(0, None),
+                    ])
+                    .chain(futures::stream::once(async move {
+                        paused.notify_one();
+                        std::future::pending::<Result<Event, ClientError>>().await
+                    })),
+                ))
+            }
+            3 => {
+                // No AI event: cancelling this request must roll back the submitted input.
+                self.rollback_entered.notify_one();
+                std::future::pending::<()>().await;
+                unreachable!()
+            }
+            other => panic!("unexpected dialog request {other}"),
+        }
+    }
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+}
+
+async fn cancel_and_wait(
+    client: &mut client::Client<client::transport::in_process::Socket>,
+    sequence: u64,
+) {
+    client
+        .send(&protocol::Method::Cancel {
+            command: protocol::WorkerCommandEnvelope::new(sequence),
+        })
+        .await
+        .unwrap();
+    tokio::time::timeout(std::time::Duration::from_secs(20), async {
+        loop {
+            match client.next_event().await.unwrap().unwrap() {
+                protocol::Event::RunEnd { result } => {
+                    assert_ne!(result, protocol::RunResult::Finished);
+                    return;
+                }
+                protocol::Event::Error { code, message } => panic!("{code:?}: {message}"),
+                _ => {}
+            }
+        }
+    })
+    .await
+    .unwrap();
+}
+
+#[tokio::test]
+async fn public_mid_turn_compactions_preserve_originals_and_drain_after_cancel_and_rollback() {
+    tokio::time::timeout(std::time::Duration::from_secs(60), async {
+        let temp = tempfile::tempdir().unwrap();
+        let state = temp.path().join("state");
+        let catalog = StandaloneSubjects::open(&state).unwrap();
+        let subject = catalog.create("compaction-journal", None).unwrap();
+        let model = CompactingDialog {
+            shared: DialogAndJournal {
+                journal: JournalScript {
+                    gate_first: true,
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let mut launch = crate::StandaloneLaunchConfig::new(
+            temp.path(),
+            &state,
+            manifest::ProfileSelector::source_named(
+                manifest::ProfileRegistrySource::Builtin,
+                "standalone-subjektiv",
+            ),
+            "public-compaction-journal",
+        )
+        .with_subject(subject.id.clone())
+        .resolve()
+        .unwrap();
+        launch.profile.manifest.compaction = Some(manifest::CompactionConfig {
+            threshold: None,
+            request_threshold: Some(50_000),
+            retained_tokens: 0,
+            ..Default::default()
+        });
+        let host = crate::StandaloneHost::start_with_model_client(launch, model.clone())
+            .await
+            .unwrap();
+        let session = host.record().active_session_id;
+        let workers = StandaloneWorkerStore::open_existing(&state).unwrap();
+        let source_store = session_store::FsStore::new_with_retention(
+            workers.sessions_dir(host.record().worker_id),
+            crate::store::session_retention(&state),
+        )
+        .unwrap();
+        let mut client = host.connect();
+        client
+            .send(&protocol::Method::submit_text(
+                protocol::new_submission_request_id(),
+                "original long turn survives compaction",
+            ))
+            .await
+            .unwrap();
+        let mut completed_compactions = 0;
+        loop {
+            match client.next_event().await.unwrap().unwrap() {
+                protocol::Event::CompactionProgress {
+                    compaction: Some(progress),
+                } if progress.phase == protocol::CompactionPhase::Committing => {
+                    assert_eq!(
+                        progress.trigger,
+                        protocol::CompactionTrigger::RequestThreshold
+                    );
+                    completed_compactions += 1;
+                }
+                protocol::Event::TextDelta { text } if text.contains(PARTIAL) => break,
+                protocol::Event::RunEnd { .. } => panic!("dialog ended before cancellation"),
+                protocol::Event::Error { code, message } => panic!("{code:?}: {message}"),
+                _ => {}
+            }
+        }
+        assert_eq!(model.compactions.load(Ordering::SeqCst), 2);
+        assert_eq!(completed_compactions, 2);
+        model.shared.journal.entered.notified().await;
+        let context = JournalHostContext::validated_session(
+            &catalog.store,
+            catalog
+                .store
+                .session_attribution(&session.to_string())
+                .unwrap()
+                .unwrap(),
+        )
+        .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 first = catalog
+            .store
+            .active_journal_generation(&context, &journal.id)
+            .unwrap()
+            .unwrap();
+        assert!(
+            catalog
+                .store
+                .list_journal_entries(&reader, &journal.id, None, 10)
+                .unwrap()
+                .items
+                .is_empty(),
+            "compaction must finish before generation is released"
+        );
+        model.partial_stream_paused.notified().await;
+        cancel_and_wait(&mut client, 1).await;
+        client
+            .send(&protocol::Method::submit_text(
+                protocol::new_submission_request_id(),
+                ROLLED_BACK,
+            ))
+            .await
+            .unwrap();
+        model.rollback_entered.notified().await;
+        let before_rollback: Vec<_> = source_store
+            .list_segments(session)
+            .unwrap()
+            .iter()
+            .flat_map(|segment| source_store.read_all(session, *segment).unwrap())
+            .collect();
+        assert!(
+            serde_json::to_string(&before_rollback)
+                .unwrap()
+                .contains(ROLLED_BACK),
+            "the submitted input must reach the actual durable rollback boundary"
+        );
+        cancel_and_wait(&mut client, 2).await;
+        assert_eq!(
+            catalog
+                .store
+                .active_journal_generation(&context, &journal.id)
+                .unwrap()
+                .unwrap(),
+            first,
+            "compaction/cancellation must not alter the running generation's frozen input"
+        );
+
+        assert!(matches!(source_store.delete_session(session),
+            Err(session_store::StoreError::SessionPinned(id)) if id == session));
+        let segments = source_store.list_segments(session).unwrap();
+        assert_eq!(segments.len(), 3);
+        let raw: Vec<_> = segments
+            .iter()
+            .flat_map(|segment| source_store.read_all(session, *segment).unwrap())
+            .collect();
+        assert_eq!(
+            raw.iter()
+                .filter(|entry| matches!(
+                    entry,
+                    session_store::LogEntry::AnnotatedSegmentStart {
+                        compacted_from: Some(_),
+                        ..
+                    }
+                ))
+                .count(),
+            2
+        );
+        assert!(
+            !serde_json::to_string(&raw).unwrap().contains(ROLLED_BACK),
+            "pre-AI cancellation must roll back the speculative input"
+        );
+        assert!(
+            serde_json::to_string(&raw).unwrap().contains(SUMMARY),
+            "real rewrite did not persist its summary"
+        );
+        let metadata =
+            session_store::FsWorkerStore::new(workers.worker_metadata_dir(host.record().worker_id))
+                .unwrap()
+                .read_by_name(&host.record().storage_key)
+                .unwrap()
+                .unwrap();
+        let originals =
+            session_store::public_material::SessionPublicMaterialReader::open_fs_with_retention(
+                &workers.sessions_dir(host.record().worker_id),
+                session,
+                metadata.active.unwrap().segment_id.unwrap(),
+                session_store::public_material::PublicMaterialLimits::default(),
+                crate::store::session_retention(&state),
+            )
+            .unwrap();
+        let final_original = originals.descriptors().last().unwrap();
+        let expected_cursor = (final_original.canonical_order + 1, 0);
+        let canonical: Vec<_> = originals
+            .descriptors()
+            .map(|descriptor| {
+                let page = originals
+                    .read_content(descriptor, 0, MAX_JOURNAL_READ_BYTES)
+                    .unwrap();
+                assert_eq!(page.next_offset, descriptor.total_bytes);
+                page.content
+            })
+            .collect();
+        assert!(!canonical.join("\n").contains(ROLLED_BACK));
+        assert!(!canonical.join("\n").contains(SUMMARY));
+
+        // Resume the existing runner, then its one durable successor, with no more input.
+        let jobs = host.jobs();
+        model.shared.journal.release.notify_one();
+        jobs.wait(&first.job.job_id).await.unwrap();
+        model.shared.journal.entered.notified().await;
+        let second = catalog
+            .store
+            .active_journal_generation(&context, &journal.id)
+            .unwrap()
+            .unwrap();
+        assert_ne!(first.id, second.id);
+        model.shared.journal.release.notify_one();
+        jobs.wait(&second.job.job_id).await.unwrap();
+        assert!(!catalog.store.journal_retention_required(&context).unwrap());
+        let (saved, open) = catalog
+            .store
+            .journal_resume_state(&context, &journal.id)
+            .unwrap();
+        let open = open.unwrap();
+        assert!(!open.ongoing.is_empty());
+        let entries = catalog
+            .store
+            .list_journal_entries(&reader, &journal.id, None, 10)
+            .unwrap();
+        assert_eq!(
+            entries.items.len(),
+            1,
+            "compaction/turn boundaries must not fix or duplicate the open entry"
+        );
+        assert_eq!(entries.items[0].id, open.id);
+        let supplied = model
+            .shared
+            .journal
+            .source_reads
+            .lock()
+            .unwrap()
+            .values()
+            .cloned()
+            .collect::<Vec<_>>()
+            .join("\n");
+        for original in [
+            "original long turn survives compaction",
+            "original activity 0",
+            "original activity 1",
+        ] {
+            assert!(
+                supplied.contains(original),
+                "Journal did not receive {original}"
+            );
+        }
+        for original in canonical {
+            assert!(
+                supplied.contains(&original),
+                "a durable public original was skipped"
+            );
+        }
+        assert!(
+            !supplied.contains(SUMMARY),
+            "derived summary substituted for original input"
+        );
+        assert!(
+            !supplied.contains(ROLLED_BACK),
+            "cancelled input was admitted after rollback"
+        );
+        assert_eq!((saved.process_order, saved.process_byte), expected_cursor);
+        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());
+        host.shutdown().await.unwrap();
+    })
+    .await
+    .expect("bounded public compaction/Journal integration");
+}

crates/standalone/src/subjektiv_sessions.rs

@@ -155,6 +155,29 @@ pub(crate) fn authorized_index(
     )
 }
 
+pub(crate) fn authorized_sessions_path(
+    state_root: &Path,
+    scope_id: &str,
+    subject_id: &str,
+    store: &SubjektivStore,
+    session_id: &str,
+) -> Result<PathBuf, SubjektivHostError> {
+    let backend = Backend {
+        state_root,
+        scope_id,
+        subject_id,
+        store,
+        reader: session_store::public_index::read_fs_session_public_index,
+    };
+    (|| {
+        backend.check_scope()?;
+        required_selector(session_id)?;
+        let attribution = backend.attribution(session_id)?;
+        Ok(backend.authorize(attribution)?.sessions_path)
+    })()
+    .map_err(|error: Failure| SubjektivHostError::Unavailable(error.message))
+}
+
 fn authorized_index_with_reader(
     state_root: &Path,
     scope_id: &str,
@@ -221,13 +244,28 @@ impl Backend<'_> {
                 "Attributed standalone Worker state is missing",
             ));
         }
-        let workers = StandaloneWorkerStore::open_existing(self.state_root).map_err(|_| {
+        let mut workers = StandaloneWorkerStore::open_existing(self.state_root).map_err(|_| {
             failure(
                 SubjektivSessionDiagnosticCode::RuntimeUnavailable,
                 "Standalone Worker storage is inaccessible",
             )
         })?;
-        let record = workers.load(id).map_err(|error| match error {
+        let record = match workers.load(id) {
+            Err(StandaloneStoreError::WorkerNotFound(_)) => {
+                workers = StandaloneWorkerStore::open_existing(
+                    self.state_root.join("subjektiv").join("retained-workers"),
+                )
+                .map_err(|_| {
+                    failure(
+                        SubjektivSessionDiagnosticCode::RetentionMissing,
+                        "Original Session is not retained",
+                    )
+                })?;
+                workers.load(id)
+            }
+            result => result,
+        }
+        .map_err(|error| match error {
             StandaloneStoreError::WorkerNotFound(_) => failure(
                 SubjektivSessionDiagnosticCode::RetentionMissing,
                 "Attributed standalone Worker record is missing",
@@ -262,6 +300,13 @@ impl Backend<'_> {
                 "Attributed Session identity is invalid",
             )
         })?;
+        let retention = crate::store::session_retention(self.state_root);
+        let _guard = retention.lock_live(id).map_err(|_| {
+            failure(
+                SubjektivSessionDiagnosticCode::RetentionMissing,
+                "Original Session is erased or its retention fence is unavailable",
+            )
+        })?;
         let index = (self.reader)(&source.sessions_path, id, limits).map_err(index_failure)?;
         if index.session_id != source.attribution.session_id {
             return Err(failure(
@@ -1539,7 +1584,11 @@ allow = []
             )
         };
         let sessions = f.index_path.parent().unwrap();
-        let fs_store = session_store::FsStore::new(sessions).unwrap();
+        let fs_store = session_store::FsStore::new_with_retention(
+            sessions,
+            crate::store::session_retention(&f.temp.path().join("workers")),
+        )
+        .unwrap();
         let segment: SessionId = f.segment.parse().unwrap();
         fs_store
             .create_segment(

crates/standalone/tests/subjektiv.rs

@@ -354,6 +354,18 @@ impl LlmClient for ScriptedClient {
         request: Request,
     ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
     {
+        // This suite observes Memory and dialog requests. Journal's independent model
+        // boundary fails explicitly rather than consuming their script or using live transport;
+        // the Journal adapter suite exercises its successful sequential tool execution.
+        if request
+            .tools
+            .iter()
+            .any(|tool| tool.name == "ReadJournalState")
+        {
+            return Err(ClientError::Config(
+                "Journal model is outside this Memory suite".into(),
+            ));
+        }
         // Lifecycle-triggered consolidation uses a separate Internal Worker. It
         // must not consume the dialog script or silently become a live request.
         let job = request