fix: reconcile Memory Journal links with handoff fencing

crates/subjektiv/README.md

@@ -3,6 +3,7 @@
 Shared Host domain implementation for per-Subject Memory, change history, candidate dispositions, derived surfaces, and Session activity Journals.
 `feature-storage` manages physical SQLite storage; the calling Host handles authentication and execution authority.
 Journal storage does not generate prose, launch Jobs, or fetch Session history.
+Hosts retain body ownership across inactivity, reserve actual execution quiescence, and reconcile final committed history before releasing a Journal writer lifetime. [`journal::binding`] fences new mutations inside storage transactions; historical attribution and exact receipt recovery cannot renew a released lifetime. [`journal::startup`] selects bounded reference context independently of source retention and long-term Memory.
 
 ## Main entry points
 

crates/subjektiv/src/journal/binding.rs

@@ -0,0 +1,205 @@
+//! Durable Journal writer lifetime, issued only from the Host's body binding. Kernel leases
+//! and Runtime liveness do not end ownership. The immutable binding ID fences retained contexts
+//! even when the same Worker later reconnects. Hosts serialize release with body lifecycle and
+//! reconcile final committed history before calling release; this store checks saved effects.
+
+use super::*;
+use rusqlite::{Connection, OptionalExtension, params};
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalWriterBinding {
+    pub subject_id: String,
+    pub runtime_id: String,
+    pub worker_id: String,
+    pub binding_id: String,
+}
+
+pub(crate) fn migrate(tx: &Transaction<'_>) -> Result<(), FeatureStorageError> {
+    tx.execute_batch("CREATE TABLE journal_writer_bindings (
+        binding_id TEXT PRIMARY KEY,
+        subject_id TEXT NOT NULL REFERENCES subjects(subject_id) ON DELETE RESTRICT,
+        runtime_id TEXT NOT NULL,
+        worker_id TEXT NOT NULL,
+        active INTEGER NOT NULL CHECK(active IN (0,1))
+    );
+    CREATE UNIQUE INDEX journal_one_body_binding ON journal_writer_bindings(subject_id) WHERE active=1;
+    CREATE TABLE journal_session_bindings (
+        session_id TEXT PRIMARY KEY REFERENCES subject_session_attributions(session_id) ON DELETE RESTRICT,
+        binding_id TEXT NOT NULL REFERENCES journal_writer_bindings(binding_id) ON DELETE RESTRICT
+    );
+    CREATE INDEX journal_body_history ON journal_writer_bindings(subject_id,active);
+    CREATE INDEX journal_body_sessions ON journal_session_bindings(binding_id);
+    CREATE INDEX journal_recent_subject ON journals(subject_id,updated_at DESC,id DESC);")?;
+    Ok(())
+}
+
+fn denied() -> JournalError {
+    JournalError::PermissionDenied("Journal writer binding is absent, released or replaced".into())
+}
+
+pub(super) fn check_writer(tx: &Connection, host: &JournalHostContext) -> JournalResult<()> {
+    let attr = host.attribution.as_ref().ok_or_else(denied)?;
+    let binding = host.writer_binding_id.as_ref().ok_or_else(denied)?;
+    if !tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_writer_bindings b JOIN journal_session_bindings s ON s.binding_id=b.binding_id WHERE b.binding_id=?1 AND b.subject_id=?2 AND b.runtime_id=?3 AND b.worker_id=?4 AND b.active=1 AND s.session_id=?5)", params![binding,host.subject_id,attr.runtime_id,attr.worker_id,attr.session_id], |r| r.get::<_,bool>(0))? {
+        return Err(denied());
+    }
+    Ok(())
+}
+
+pub(super) fn require_saved_position(tx: &Connection, generation: &str) -> JournalResult<()> {
+    let saved: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_operations p WHERE p.generation_id=?1 AND json_extract(p.request_json,'$.action.action')='update_position' AND p.rowid > COALESCE((SELECT MAX(r.rowid) FROM journal_operations r WHERE r.generation_id=?1 AND json_extract(r.request_json,'$.action.action') IN ('create','rewrite','fix','confirm_unchanged')),0))",[generation],|r|r.get(0))?;
+    if !saved {
+        return Err(JournalError::Conflict("Journal current position must be saved after the generation's activity records before completion".into()));
+    }
+    Ok(())
+}
+
+impl SubjektivStore {
+    /// The last released body's most recently updated Journal Session, independent of source
+    /// retention. Session attachment order may differ from the order of recorded activity.
+    pub fn journal_predecessor_session(
+        &self,
+        binding: &JournalWriterBinding,
+    ) -> JournalResult<Option<String>> {
+        self.database.try_with_connection(|tx| {
+            Ok(tx.query_row("SELECT j.source_session_id FROM journals j JOIN journal_session_bindings s ON s.session_id=j.source_session_id WHERE j.subject_id=?1 AND s.binding_id=(SELECT binding_id FROM journal_writer_bindings WHERE subject_id=?1 AND active=0 ORDER BY rowid DESC LIMIT 1) ORDER BY j.updated_at DESC,j.id DESC LIMIT 1", [&binding.subject_id], |r| r.get(0)).optional()?)
+        })
+    }
+
+    pub fn journal_worker_was_released(
+        &self,
+        subject: &str,
+        runtime: &str,
+        worker: &str,
+    ) -> JournalResult<bool> {
+        self.database.try_with_connection(|tx| {
+            Ok(tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_writer_bindings WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3 AND active=0)", params![subject,runtime,worker], |r| r.get(0))?)
+        })
+    }
+
+    pub fn journal_writer_active(&self, binding: &JournalWriterBinding) -> JournalResult<bool> {
+        self.database.try_with_connection(|tx| {
+            Ok(tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_writer_bindings WHERE binding_id=?1 AND subject_id=?2 AND runtime_id=?3 AND worker_id=?4 AND active=1)", params![binding.binding_id,binding.subject_id,binding.runtime_id,binding.worker_id], |r| r.get(0))?)
+        })
+    }
+
+    pub fn journal_session_writer_matches(
+        &self,
+        binding: &JournalWriterBinding,
+        session: &str,
+    ) -> JournalResult<bool> {
+        self.database.try_with_connection(|tx| {
+            Ok(tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_session_bindings WHERE binding_id=?1 AND session_id=?2)", params![binding.binding_id,session], |r| r.get(0))?)
+        })
+    }
+
+    /// Called under the existing Host Subject lifecycle authority. A stopped or unavailable
+    /// owner remains the owner; only the exact same body may resume an unreleased binding.
+    pub fn claim_journal_writer(
+        &self,
+        subject: &str,
+        runtime: &str,
+        worker: &str,
+    ) -> JournalResult<JournalWriterBinding> {
+        for identifier in [subject, runtime, worker] {
+            if identifier.is_empty()
+                || identifier.len() > 128
+                || identifier.trim() != identifier
+                || identifier.chars().any(char::is_control)
+            {
+                return Err(JournalError::InvalidInput(
+                    "invalid Journal binding identity".into(),
+                ));
+            }
+        }
+        self.database.try_transaction(|tx| {
+            let state: Option<String> = tx.query_row("SELECT state FROM subjects WHERE subject_id=?1", [subject], |r| r.get(0)).optional()?;
+            if state.as_deref() != Some("active") { return Err(JournalError::PermissionDenied("Subject is absent or retired".into())); }
+            let current: Option<JournalWriterBinding> = tx.query_row("SELECT subject_id,runtime_id,worker_id,binding_id FROM journal_writer_bindings WHERE subject_id=?1 AND active=1", [subject], |r| Ok(JournalWriterBinding { subject_id:r.get(0)?,runtime_id:r.get(1)?,worker_id:r.get(2)?,binding_id:r.get(3)? })).optional()?;
+            if let Some(current) = current {
+                if current.runtime_id != runtime || current.worker_id != worker {
+                    return Err(JournalError::Conflict("Subject is still occupied; explicit Idle and Journal-complete release required".into()));
+                }
+                return Ok(current);
+            }
+            // Upgraded T-734 data can contain a removed body's durable recording handoff.
+            // Retention is not permission for a different body to occupy that Subject.
+            let legacy_busy: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM journals j JOIN subject_session_attributions a ON a.session_id=j.source_session_id WHERE j.subject_id=?1 AND (a.runtime_id<>?2 OR a.worker_id<>?3) AND NOT EXISTS(SELECT 1 FROM journal_session_bindings b WHERE b.session_id=j.source_session_id) AND (EXISTS(SELECT 1 FROM journal_generations g WHERE g.journal_id=j.id AND g.active=1) OR EXISTS(SELECT 1 FROM journal_sources s,json_each(s.accepted_json) r WHERE s.journal_id=j.id AND json_extract(r.value,'$.end')>s.reflected_bytes)))", params![subject,runtime,worker], |r| r.get(0))?;
+            if legacy_busy { return Err(JournalError::Conflict("previous body's retained Journal handoff is not complete".into())); }
+            let binding = JournalWriterBinding { subject_id:subject.into(),runtime_id:runtime.into(),worker_id:worker.into(),binding_id:crate::issued_id("journal-binding") };
+            tx.execute("INSERT INTO journal_writer_bindings(binding_id,subject_id,runtime_id,worker_id,active) VALUES(?1,?2,?3,?4,1)", params![binding.binding_id,subject,runtime,worker])?;
+            Ok(binding)
+        })
+    }
+
+    /// Binds a new Session once. Reconnecting a Worker under a later binding cannot acquire
+    /// mutation authority for its old Session. Attribution and binding are Host-only input.
+    pub fn bind_journal_session(
+        &self,
+        binding: &JournalWriterBinding,
+        attribution: SubjectSessionAttribution,
+    ) -> JournalResult<JournalHostContext> {
+        let mut host = JournalHostContext::validated_session(self, attribution)?;
+        self.database.try_transaction(|tx| {
+            let attr = host.attribution.as_ref().expect("validated Session");
+            if binding.subject_id != attr.subject_id || binding.runtime_id != attr.runtime_id || binding.worker_id != attr.worker_id {
+                return Err(denied());
+            }
+            tx.execute("INSERT OR IGNORE INTO journal_session_bindings(session_id,binding_id) SELECT ?1,binding_id FROM journal_writer_bindings WHERE binding_id=?2 AND subject_id=?3 AND runtime_id=?4 AND worker_id=?5 AND active=1",params![attr.session_id,binding.binding_id,attr.subject_id,attr.runtime_id,attr.worker_id])?;
+            host.writer_binding_id = Some(binding.binding_id.clone());
+            check_writer(tx,&host)?;
+            Ok(host.clone())
+        })
+    }
+
+    /// Destructive release is distinct from a continuity handoff: every attributed Session
+    /// of this body must already have a permanent erasure tombstone. This also records a
+    /// released lifetime for pre-binding bodies, without granting them new mutation authority.
+    pub fn release_erased_journal_body(&self, host: &JournalHostContext) -> JournalResult<bool> {
+        self.database.try_transaction(|tx| {
+            super::store::check_session_host(self, tx, host)?;
+            let attr = host.attribution.as_ref().expect("checked Session authority");
+            let remaining: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM subject_session_attributions a WHERE a.subject_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 AND NOT EXISTS(SELECT 1 FROM journal_erased_sessions e WHERE e.session_id=a.session_id AND e.subject_id=a.subject_id))", params![host.subject_id,attr.runtime_id,attr.worker_id], |r| r.get(0))?;
+            if remaining {
+                return Ok(false);
+            }
+            let existing: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_writer_bindings WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3)", params![host.subject_id,attr.runtime_id,attr.worker_id], |r| r.get(0))?;
+            tx.execute("UPDATE journal_writer_bindings SET active=0 WHERE subject_id=?1 AND runtime_id=?2 AND worker_id=?3",params![host.subject_id,attr.runtime_id,attr.worker_id])?;
+            if !existing {
+                tx.execute("INSERT INTO journal_writer_bindings(binding_id,subject_id,runtime_id,worker_id,active) VALUES(?1,?2,?3,?4,0)", params![crate::issued_id("journal-erased-body"),host.subject_id,attr.runtime_id,attr.worker_id])?;
+            }
+            Ok(true)
+        })
+    }
+
+    /// The Host must hold its lifecycle fence, confirm Idle, reconcile all committed original
+    /// Sessions (including the save/notification gap), and exclude queued/running/Unknown Jobs.
+    /// This transaction rejects retained writers, pending source bytes and active generations.
+    /// It never fixes an open entry or declares the represented work complete.
+    pub fn release_journal_writer(&self, binding: &JournalWriterBinding) -> JournalResult<()> {
+        self.database.try_transaction(|tx| {
+            let active: Option<bool> = tx.query_row("SELECT active FROM journal_writer_bindings WHERE binding_id=?1 AND subject_id=?2 AND runtime_id=?3 AND worker_id=?4",params![binding.binding_id,binding.subject_id,binding.runtime_id,binding.worker_id],|r|r.get(0)).optional()?;
+            let Some(active) = active else { return Err(denied()); };
+            if !active { return Ok(()); }
+            let unbound: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM subject_session_attributions a WHERE a.subject_id=?1 AND a.runtime_id=?2 AND a.worker_id=?3 AND NOT EXISTS(SELECT 1 FROM journal_session_bindings s WHERE s.session_id=a.session_id) AND NOT EXISTS(SELECT 1 FROM journal_erased_sessions e WHERE e.session_id=a.session_id AND e.subject_id=a.subject_id))",params![binding.subject_id,binding.runtime_id,binding.worker_id],|r|r.get(0))?;
+            if unbound { return Err(JournalError::Conflict("body has legacy Sessions whose committed history has not been reconciled under its writer binding".into())); }
+            let busy: bool = tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_generations g JOIN journals j ON j.id=g.journal_id JOIN journal_session_bindings s ON s.session_id=j.source_session_id WHERE s.binding_id=?1 AND g.active=1)",[&binding.binding_id],|r|r.get(0))?;
+            if busy { return Err(JournalError::Conflict("Journal generation remains active or its result is unconfirmed".into())); }
+            let mut stmt = tx.prepare("SELECT s.accepted_json,s.reflected_bytes,s.unavailable_json FROM journal_sources s JOIN journals j ON j.id=s.journal_id JOIN journal_session_bindings b ON b.session_id=j.source_session_id WHERE b.binding_id=?1")?;
+            let mut rows = stmt.query([&binding.binding_id])?;
+            while let Some(row) = rows.next()? {
+                let accepted: Vec<JournalByteRange> = serde_json::from_str(&row.get::<_,String>(0)?)?;
+                let reflected: u64 = row.get(1)?;
+                if accepted.iter().any(|r| r.end > reflected) {
+                    return Err(JournalError::Conflict("Journal activity/position/checkpoint has unreflected source bytes".into()));
+                }
+            }
+            let mut checkpoints = tx.prepare("SELECT p.generation_id FROM journal_operations p JOIN journals j ON j.id=p.journal_id JOIN journal_session_bindings b ON b.session_id=j.source_session_id WHERE b.binding_id=?1 AND p.rowid=(SELECT MAX(c.rowid) FROM journal_operations c WHERE c.journal_id=j.id AND json_extract(c.request_json,'$.action.action')='checkpoint')")?;
+            let generations = checkpoints.query_map([&binding.binding_id], |r| r.get::<_,String>(0))?;
+            for generation in generations { require_saved_position(tx, &generation?)?; }
+            tx.execute("UPDATE journal_writer_bindings SET active=0 WHERE binding_id=?1 AND active=1",[&binding.binding_id])?;
+            Ok(())
+        })
+    }
+}

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

@@ -41,7 +41,10 @@ impl Fixture {
         let attr =
             SubjectSessionAttribution::new(&subject.id, "runtime", "worker", "session").unwrap();
         store.record_session_attribution(attr.clone()).unwrap();
-        let host = JournalHostContext::validated_session(&store, attr).unwrap();
+        let binding = store
+            .claim_journal_writer(&subject.id, "runtime", "worker")
+            .unwrap();
+        let host = store.bind_journal_session(&binding, attr).unwrap();
         // Canonical order crosses two Segments; legacy numeric ranges below intentionally do not
         // agree with this order. Stable E identities, not display positions, are the join contract.
         let segments = [Uuid::now_v7(), Uuid::now_v7()];
@@ -179,8 +182,14 @@ impl Fixture {
                         .read_journal(&self.host, &self.journal_id)
                         .unwrap()
                         .state_token,
+                    covered_ranges: match action {
+                        JournalAction::UpdatePosition { .. } => vec![],
+                        // Leave the final source pending so a subsequent Fix is a valid
+                        // activity write, while the staged anchor is already checkpointed.
+                        JournalAction::Checkpoint { .. } => self.ranges[..3].to_vec(),
+                        _ => self.ranges.clone(),
+                    },
                     action,
-                    covered_ranges: self.ranges.clone(),
                 },
             )
             .unwrap();
@@ -567,6 +576,12 @@ fn delayed_fixed_endpoint_connects_only_on_explicit_reconcile_without_using_chec
         )
         .entry_id
         .unwrap();
+    f.action(
+        "position-before-checkpoint",
+        JournalAction::UpdatePosition {
+            current_position: "Still working; no final endpoint is available".into(),
+        },
+    );
     f.action(
         "checkpoint",
         JournalAction::Checkpoint {
@@ -742,15 +757,17 @@ fn missing_target_never_retargets_saved_edge_to_a_later_event() {
 fn ambiguous_saved_anchor_does_not_choose_between_attributed_sessions() {
     let f = Fixture::new();
     f.fixed("first", 1, vec![]);
-    let attr = SubjectSessionAttribution::new(
-        &f.host.subject_id,
-        "runtime",
-        "other-worker",
-        "other-session",
-    )
-    .unwrap();
+    // Two Sessions of the same current body may contain an identical saved anchor.
+    // Bind both through the live writer; cross-body ownership is not part of this ambiguity.
+    let attr =
+        SubjectSessionAttribution::new(&f.host.subject_id, "runtime", "worker", "other-session")
+            .unwrap();
     f.store().record_session_attribution(attr.clone()).unwrap();
-    let other_host = JournalHostContext::validated_session(f.store(), attr).unwrap();
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    let other_host = f.store().bind_journal_session(&binding, attr).unwrap();
     let mut range = f.ranges[0].clone();
     range.source.session_id = "other-session".into();
     f.store()
@@ -950,7 +967,7 @@ fn historical_candidates_page_all_immutable_sources_without_reextracting_or_losi
 fn before_links() -> &'static [feature_storage::FeatureMigration] {
     &crate::MIGRATIONS[..crate::MIGRATIONS
         .iter()
-        .position(|m| m.version() == 10)
+        .position(|m| m.version() == 11)
         .unwrap()]
 }
 
@@ -1086,9 +1103,9 @@ fn failed_link_migration_rolls_back_ddl_and_preserves_existing_data_for_retry()
             crate::MIGRATIONS
                 .iter()
                 .map(|migration| {
-                    if migration.version() == 10 {
+                    if migration.version() == 11 {
                         feature_storage::FeatureMigration::new(
-                            10,
+                            11,
                             migration.name(),
                             fail_after_link_schema,
                         )
@@ -1118,7 +1135,7 @@ fn failed_link_migration_rolls_back_ddl_and_preserves_existing_data_for_retry()
         assert!(matches!(
             SubjektivStore::open(&workspace, &registration),
             Err(crate::SubjektivError::Storage(
-                feature_storage::FeatureStorageError::MigrationFailed { version: 10, .. }
+                feature_storage::FeatureStorageError::MigrationFailed { version: 11, .. }
             ))
         ));
         drop(workspace);

crates/subjektiv/src/journal/mod.rs

@@ -37,12 +37,16 @@
 //! Hosts resuming at the processing cursor can use a bounded source point lookup instead of
 //! traversing already reflected pages. Receipt/entry keysets remain append-stable.
 
+pub mod binding;
+pub use binding::JournalWriterBinding;
+pub(crate) use binding::migrate as migrate_binding;
 mod lifecycle;
 pub mod memory_links;
 pub(crate) use lifecycle::migrate as migrate_lifecycle;
 pub mod recording;
 pub mod service;
 mod source_status;
+pub mod startup;
 mod store;
 #[cfg(test)]
 mod tests;
@@ -112,8 +116,14 @@ pub struct JournalHostContext {
     scope_id: String,
     subject_id: String,
     attribution: Option<SubjectSessionAttribution>,
+    writer_binding_id: Option<String>,
 }
 impl JournalHostContext {
+    /// Evidence for a retained Job grant, not permission to mint or renew a binding.
+    pub fn writer_binding_id(&self) -> Option<&str> {
+        self.writer_binding_id.as_deref()
+    }
+
     pub fn validated_reader(store: &SubjektivStore, subject_id: &str) -> JournalResult<Self> {
         store
             .subject(subject_id)?
@@ -122,6 +132,7 @@ impl JournalHostContext {
             scope_id: store.scope_id().into(),
             subject_id: subject_id.into(),
             attribution: None,
+            writer_binding_id: None,
         })
     }
     /// External authentication, public-history validation and execution authority are Host duties.
@@ -135,6 +146,17 @@ impl JournalHostContext {
             ));
         }
         let mut context = Self::validated_reader(store, &attribution.subject_id)?;
+        use rusqlite::OptionalExtension;
+        context.writer_binding_id = store.database.try_with_connection(|tx| {
+            Ok::<_, JournalError>(
+                tx.query_row(
+                    "SELECT binding_id FROM journal_session_bindings WHERE session_id=?1",
+                    [&attribution.session_id],
+                    |r| r.get(0),
+                )
+                .optional()?,
+            )
+        })?;
         context.attribution = Some(attribution);
         Ok(context)
     }

crates/subjektiv/src/journal/service.rs

@@ -386,6 +386,7 @@ pub fn finish(
             return Ok(serde_json::from_str(&saved)?);
         }
         super::store::job(store, tx, context, true)?;
+        super::binding::require_saved_position(tx, &generation.id)?;
         let last = generation
             .selected_ranges
             .last()

crates/subjektiv/src/journal/startup.rs

@@ -0,0 +1,126 @@
+//! Read-only, bounded Journal context for a new body Session. Selection and text projection
+//! share one storage transaction; no source history, Job, admission or checkpoint is consulted.
+//! This is derived reference material, never a new observation or an execution capability.
+
+use super::*;
+use rusqlite::{Connection, OptionalExtension, params};
+
+pub const STARTUP_MAX_BYTES: usize = 32 * 1024;
+const JOURNALS: usize = 3;
+const ENTRIES: usize = 3;
+const TEXT_BYTES: usize = 2048;
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalStartupSnapshot {
+    pub subject_id: String,
+    pub journals: Vec<StartupJournal>,
+    pub omitted: bool,
+}
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct StartupJournal {
+    pub journal_id: String,
+    pub source_session_id: String,
+    pub state_token: String,
+    pub process_order: u64,
+    pub process_byte: u64,
+    pub current_position: String,
+    pub entries: Vec<StartupEntry>,
+    pub omitted: bool,
+}
+#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct StartupEntry {
+    pub entry_id: String,
+    pub position: u64,
+    pub state: JournalEntryState,
+    pub title: String,
+    pub body: String,
+    pub ongoing: String,
+    pub source_end: Option<JournalSourcePointer>,
+    pub correction_refs: Vec<String>,
+    pub omitted: bool,
+}
+
+// Read a bounded UTF-8 prefix at SQLite, not a full megabyte field followed by truncation.
+fn text(tx: &Connection, table: &str, id: &str, field: &str) -> JournalResult<(String, bool)> {
+    let (mut bytes, length): (Vec<u8>, usize) = tx.query_row(
+        &format!("SELECT COALESCE(substr(CAST({field} AS BLOB),1,?2),X''),length(CAST({field} AS BLOB)) FROM {table} WHERE id=?1"),
+        params![id, TEXT_BYTES],
+        |r| Ok((r.get(0)?, r.get(1)?)),
+    )?;
+    let length_read = bytes.len();
+    match std::str::from_utf8(&bytes) {
+        Ok(_) => {}
+        Err(error) if error.error_len().is_none() => bytes.truncate(error.valid_up_to()),
+        Err(_) => {
+            return Err(JournalError::Conflict(
+                "stored Journal text is not UTF-8".into(),
+            ));
+        }
+    }
+    let omitted = bytes.len() < length || length_read < length;
+    Ok((String::from_utf8(bytes).expect("validated UTF-8"), omitted))
+}
+
+impl SubjektivStore {
+    /// `previous_session` comes from the Host's retained predecessor binding, not model input.
+    /// Recent Subject-local Journals supplement that predecessor under independent fixed budgets.
+    pub fn journal_startup_snapshot(
+        &self,
+        host: &JournalHostContext,
+        previous_session: Option<&str>,
+    ) -> JournalResult<JournalStartupSnapshot> {
+        self.database.try_transaction(|tx| {
+            super::store::check_host(self, tx, host, false)?;
+            let mut ids = Vec::new();
+            if let Some(session) = previous_session {
+                let previous: Option<(String, String)> = tx.query_row(
+                    "SELECT id,subject_id FROM journals WHERE source_session_id=?1",
+                    [session], |r| Ok((r.get(0)?, r.get(1)?)),
+                ).optional()?;
+                if let Some((id, subject)) = previous {
+                    if subject != host.subject_id { return Err(JournalError::PermissionDenied("predecessor Journal belongs to another Subject".into())); }
+                    ids.push(id);
+                }
+            }
+            let mut query = tx.prepare("SELECT id FROM journals WHERE subject_id=?1 ORDER BY updated_at DESC,id DESC LIMIT ?2")?;
+            let recent = query.query_map(params![host.subject_id, JOURNALS + 1], |r| r.get::<_, String>(0))?;
+            for id in recent { let id = id?; if !ids.contains(&id) { ids.push(id); } }
+            let omitted = ids.len() > JOURNALS;
+            ids.truncate(JOURNALS);
+            let mut snapshot = JournalStartupSnapshot { subject_id: host.subject_id.clone(), journals: Vec::new(), omitted };
+            for id in ids {
+                let (session, token, order, byte) = tx.query_row(
+                    "SELECT source_session_id,state_token,process_order,process_byte FROM journals WHERE id=?1 AND subject_id=?2",
+                    params![id, host.subject_id], |r| Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?)),
+                )?;
+                let (current_position, mut omitted) = text(tx, "journals", &id, "current_position")?;
+                let mut stmt = tx.prepare("SELECT id,position,state,source_end_json FROM journal_entries WHERE journal_id=?1 ORDER BY position DESC LIMIT ?2")?;
+                let rows = stmt.query_map(params![id, ENTRIES + 1], |r| Ok((r.get::<_,String>(0)?,r.get::<_,u64>(1)?,r.get::<_,String>(2)?,r.get::<_,Option<String>>(3)?)))?.collect::<Result<Vec<_>,_>>()?;
+                omitted |= rows.len() > ENTRIES;
+                let mut entries = Vec::new();
+                for (entry_id, position, state, source_end) in rows.into_iter().take(ENTRIES) {
+                    let (title, title_cut) = text(tx, "journal_entries", &entry_id, "title")?;
+                    let (body, body_cut) = text(tx, "journal_entries", &entry_id, "body")?;
+                    let (ongoing, ongoing_cut) = text(tx, "journal_entries", &entry_id, "ongoing")?;
+                    let mut refs = tx.prepare("SELECT target_id FROM journal_corrections WHERE entry_id=?1 ORDER BY target_id LIMIT 17")?;
+                    let mut correction_refs = refs.query_map([&entry_id], |r| r.get::<_,String>(0))?.collect::<Result<Vec<_>,_>>()?;
+                    let omitted = title_cut || body_cut || ongoing_cut || correction_refs.len() > 16;
+                    correction_refs.truncate(16);
+                    entries.push(StartupEntry { entry_id, position, state: match state.as_str() { "open" => JournalEntryState::Open, "fixed" => JournalEntryState::Fixed, _ => return Err(JournalError::Conflict("invalid stored Journal entry state".into())) }, title, body, ongoing, source_end: source_end.map(|s| serde_json::from_str(&s)).transpose()?, correction_refs, omitted });
+                }
+                snapshot.journals.push(StartupJournal { journal_id: id, source_session_id: session, state_token: token, process_order: order, process_byte: byte, current_position, entries, omitted });
+            }
+            // JSON escaping also consumes the independent output budget. Drop least-priority
+            // entries/Journals rather than returning an over-budget or silently partial document.
+            while serde_json::to_vec(&snapshot)?.len() > STARTUP_MAX_BYTES {
+                snapshot.omitted = true;
+                let journal = snapshot.journals.last_mut().ok_or_else(|| JournalError::InvalidInput("startup metadata exceeds budget".into()))?;
+                if journal.entries.pop().is_some() { journal.omitted = true; } else { snapshot.journals.pop(); }
+            }
+            Ok(snapshot)
+        })
+    }
+}

crates/subjektiv/src/journal/store.rs

@@ -141,6 +141,7 @@ pub(super) fn check_host(
             return Err(JournalError::PermissionDenied("Subject is retired".into()));
         }
         check_session_host(store, tx, host)?;
+        super::binding::check_writer(tx, host)?;
     }
     Ok(())
 }
@@ -373,7 +374,10 @@ pub(super) fn job(
     ctx: &JournalJobContext,
     active: bool,
 ) -> JournalResult<JournalGeneration> {
-    check_host(store, tx, &ctx.host, true)?;
+    // Recovery can inspect the original generation and exact saved receipts after release;
+    // only a new mutation requires the still-current writer lifetime.
+    check_host(store, tx, &ctx.host, active)?;
+    check_session_host(store, tx, &ctx.host)?;
     let (g, is_active) = generation(tx, &ctx.host, &ctx.generation_id)?;
     if g.job != ctx.binding {
         return Err(JournalError::PermissionDenied(
@@ -629,6 +633,13 @@ impl SubjektivStore {
             if operation.expected_state!=j.state_token {return Err(conflict("stored Journal content changed; reread before applying"));}
             in_snapshot(&g,&operation.covered_ranges)?;
             if !operation.covered_ranges.is_empty() {check_ranges(tx,&j,&operation.covered_ranges,true)?;}
+            // A fully checkpointed selection has no pending source to recover another
+            // activity write if the Job subsequently fails. Receipts and position repair
+            // remain available, but new activity belongs to the next selected prefix.
+            if matches!(operation.action, JournalAction::Create {..}|JournalAction::Rewrite {..}|JournalAction::Fix {..}|JournalAction::ConfirmUnchanged)
+                && (j.process_order,j.process_byte)==prefix(&g.initial_journal,&g.selected_ranges)? {
+                return Err(conflict("generation activity is already fully checkpointed; finish this Job before recording new input"));
+            }
             let mut entry_id=None;
             match &operation.action {
                 JournalAction::Create {title,body,ongoing,state,source_end,correction_refs}=> {
@@ -680,6 +691,9 @@ impl SubjektivStore {
                         if !contains(&merged,r.bytes) {return Err(invalid("checkpoint has input with no durable record correspondence"));}
                         tx.execute("UPDATE journal_sources SET reflected_bytes=?3 WHERE journal_id=?1 AND source_order=?2",params![j.id,r.source.order,r.bytes.end])?;
                     }
+                    // Advancing reflection without position would strand a terminal failed
+                    // Job: source selection would see no pending bytes to repair that memo.
+                    super::binding::require_saved_position(tx,&g.id)?;
                     j.process_order=order;j.process_byte=byte;
                 }
             }

crates/subjektiv/src/journal/tests.rs

@@ -10,6 +10,450 @@ use sha2::{Digest, Sha256};
 use std::sync::{Arc, Barrier};
 use std::thread;
 
+#[test]
+fn explicit_erasure_releases_only_after_all_body_sessions_including_unadmitted_history_are_erased()
+{
+    let f = Fixture::new();
+    let subject = f.host.subject_id.clone();
+    let binding = f
+        .store()
+        .claim_journal_writer(&subject, "runtime", "worker")
+        .unwrap();
+    let attr = f
+        .store()
+        .record_session_attribution(
+            SubjectSessionAttribution::new(&subject, "runtime", "worker", "saved-before-admission")
+                .unwrap(),
+        )
+        .unwrap();
+    let unadmitted = f.store().bind_journal_session(&binding, attr).unwrap();
+    assert!(!f.store().release_erased_journal_body(&f.host).unwrap());
+    f.store()
+        .erase_journal_session(&f.host, "erase-first")
+        .unwrap();
+    assert!(!f.store().release_erased_journal_body(&f.host).unwrap());
+    assert!(f.store().journal_writer_active(&binding).unwrap());
+    f.store()
+        .erase_journal_session(&unadmitted, "erase-last")
+        .unwrap();
+    assert!(f.store().release_erased_journal_body(&unadmitted).unwrap());
+    assert!(f.store().release_erased_journal_body(&unadmitted).unwrap());
+    assert!(!f.store().journal_writer_active(&binding).unwrap());
+    assert!(
+        f.store()
+            .journal_worker_was_released(&subject, "runtime", "worker")
+            .unwrap()
+    );
+    let next = f
+        .store()
+        .claim_journal_writer(&subject, "runtime", "replacement")
+        .unwrap();
+    assert!(f.store().release_erased_journal_body(&f.host).unwrap());
+    assert!(
+        f.store().journal_writer_active(&next).unwrap(),
+        "late erase retry cannot revoke another body"
+    );
+}
+
+#[test]
+fn released_binding_preserves_open_and_receipts_but_never_revives_old_session_writes() {
+    let mut f = Fixture::new();
+    let subject = f.host.subject_id.clone();
+    let binding = f
+        .store()
+        .claim_journal_writer(&subject, "runtime", "worker")
+        .unwrap();
+    assert!(
+        f.store()
+            .claim_journal_writer(&subject, "runtime", "other-worker")
+            .is_err()
+    );
+    assert!(f.store().release_journal_writer(&binding).is_err());
+    let (generation, context) = f.generation("first-body", f.ranges.clone());
+    let create = f.operation(
+        "create",
+        open_action("A was proposed; user corrected a constraint; B selected and partly executed"),
+        f.ranges.clone(),
+    );
+    let receipt = f.store().execute_journal(&context, create.clone()).unwrap();
+    f.execute(
+        &context,
+        "position",
+        JournalAction::UpdatePosition {
+            current_position:
+                "B: configuration changed; tests not executed; external result Unknown".into(),
+        },
+        vec![],
+    );
+    assert!(f.store().release_journal_writer(&binding).is_err());
+    f.execute(
+        &context,
+        "checkpoint",
+        JournalAction::Checkpoint {
+            record_operation_ids: vec!["create".into()],
+        },
+        f.ranges.clone(),
+    );
+    // A saved checkpoint is not proof that the generation has stopped executing.
+    assert!(f.store().release_journal_writer(&binding).is_err());
+    f.close(&generation);
+    f.store().release_journal_writer(&binding).unwrap();
+    f.store().release_journal_writer(&binding).unwrap();
+    f.reopen();
+    let old = f
+        .store()
+        .query_journal_operation(&f.host, &f.journal_id, "create")
+        .unwrap();
+    assert_eq!(old, Some(receipt.clone()));
+    assert_eq!(
+        f.store().execute_journal(&context, create).unwrap(),
+        receipt
+    );
+    let reader = JournalHostContext::validated_reader(f.store(), &subject).unwrap();
+    let snapshot = f
+        .store()
+        .journal_startup_snapshot(&reader, Some("source-session"))
+        .unwrap();
+    assert_eq!(
+        snapshot.journals[0].entries[0].state,
+        JournalEntryState::Open
+    );
+    assert!(snapshot.journals[0].current_position.contains("Unknown"));
+    let new = f
+        .store()
+        .claim_journal_writer(&subject, "runtime", "worker")
+        .unwrap();
+    assert_ne!(new.binding_id, binding.binding_id);
+    assert!(
+        f.store()
+            .bind_journal_session(&new, f.host.attribution.clone().unwrap())
+            .is_err()
+    );
+    let late = source_range("source-session", 2, 64);
+    assert!(matches!(
+        f.store().accept_journal_sources(&f.host, "late", &[late]),
+        Err(JournalError::PermissionDenied(_))
+    ));
+    assert!(
+        f.store()
+            .prepare_journal_generation(
+                &f.host,
+                &f.journal_id,
+                JobAttemptBinding {
+                    job_id: "late".into(),
+                    attempt_id: "late".into()
+                },
+                f.ranges.clone()
+            )
+            .is_err()
+    );
+    let operation = JournalOperation {
+        request_id: "late-position".into(),
+        expected_state: snapshot.journals[0].state_token.clone(),
+        action: JournalAction::UpdatePosition {
+            current_position: "forged".into(),
+        },
+        covered_ranges: vec![],
+    };
+    assert!(f.store().execute_journal(&context, operation).is_err());
+}
+
+#[test]
+fn release_rejects_unbound_legacy_body_sessions_even_without_admitted_sources() {
+    let f = Fixture::new();
+    let (generation, context) = f.generation("recorded-body", f.ranges.clone());
+    f.execute(
+        &context,
+        "record",
+        JournalAction::ConfirmUnchanged,
+        f.ranges.clone(),
+    );
+    save_position(f.store(), &context, "position");
+    f.execute(
+        &context,
+        "checkpoint",
+        checkpoint(&["record"]),
+        f.ranges.clone(),
+    );
+    f.close(&generation);
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    let legacy = f
+        .store()
+        .record_session_attribution(
+            SubjectSessionAttribution::new(
+                &f.host.subject_id,
+                "runtime",
+                "worker",
+                "legacy-not-admitted",
+            )
+            .unwrap(),
+        )
+        .unwrap();
+    assert!(matches!(
+        f.store().release_journal_writer(&binding),
+        Err(JournalError::Conflict(_))
+    ));
+    assert!(f.store().journal_writer_active(&binding).unwrap());
+    // The Host must bind and reconcile this original Session, even if its canonical
+    // history proves empty and therefore creates no Journal.
+    f.store().bind_journal_session(&binding, legacy).unwrap();
+    f.store().release_journal_writer(&binding).unwrap();
+}
+
+#[test]
+fn predecessor_prefers_latest_journal_activity_not_session_attachment_order() {
+    let f = Fixture::new();
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    let second = session_host(f.store(), &f.host.subject_id, "later-attached-session");
+    let second_ranges = vec![source_range("later-attached-session", 0, 8)];
+    let second_id = f
+        .store()
+        .accept_journal_sources(&second, "Earlier activity", &second_ranges)
+        .unwrap();
+    for (host, journal_id, ranges) in [
+        (&f.host, &f.journal_id, &f.ranges),
+        (&second, &second_id, &second_ranges),
+    ] {
+        let job = JobAttemptBinding {
+            job_id: journal_id.clone(),
+            attempt_id: "recording".into(),
+        };
+        let generation = f
+            .store()
+            .prepare_journal_generation(host, journal_id, job.clone(), ranges.clone())
+            .unwrap();
+        let context = JournalJobContext::validated(f.store(), host, &generation.id, job).unwrap();
+        let state = f
+            .store()
+            .read_journal(host, journal_id)
+            .unwrap()
+            .state_token;
+        f.store()
+            .execute_journal(
+                &context,
+                JournalOperation {
+                    request_id: "record".into(),
+                    expected_state: state,
+                    action: open_action("Unfinished activity"),
+                    covered_ranges: ranges.clone(),
+                },
+            )
+            .unwrap();
+        let position = save_position(f.store(), &context, "position");
+        f.store()
+            .checkpoint_journal(
+                &context,
+                JournalOperation {
+                    request_id: "checkpoint".into(),
+                    expected_state: position.state_token,
+                    action: checkpoint(&["record"]),
+                    covered_ranges: ranges.clone(),
+                },
+            )
+            .unwrap();
+        f.store()
+            .close_journal_generation(host, &generation.id, &generation.job)
+            .unwrap();
+    }
+    // Stored timestamps model a later update to the earlier-attached Session without
+    // using elapsed wall time as a synchronization or ordering assertion.
+    f.store()
+        .database
+        .try_transaction::<_, JournalError>(|tx| {
+            tx.execute(
+                "UPDATE journals SET updated_at='2001-01-01T00:00:00Z' WHERE id=?1",
+                [&f.journal_id],
+            )?;
+            tx.execute(
+                "UPDATE journals SET updated_at='2000-01-01T00:00:00Z' WHERE id=?1",
+                [&second_id],
+            )?;
+            Ok(())
+        })
+        .unwrap();
+    let _empty = session_host(
+        f.store(),
+        &f.host.subject_id,
+        "latest-attachment-no-journal",
+    );
+    f.store().release_journal_writer(&binding).unwrap();
+    let replacement = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "replacement")
+        .unwrap();
+    assert_eq!(
+        f.store()
+            .journal_predecessor_session(&replacement)
+            .unwrap()
+            .as_deref(),
+        Some("source-session")
+    );
+}
+
+#[test]
+fn concurrent_body_claims_from_independent_connections_bind_only_one_worker() {
+    let f = Fixture::new();
+    let subject = f
+        .store()
+        .create_subject(SubjectRole::new("claim-race").unwrap())
+        .unwrap();
+    let manager = FeatureStorage::new(f.temp.path().join("storage"));
+    let workspace = manager.workspace("journal-tests").unwrap();
+    let registration = SubjektivStore::register(&workspace).unwrap();
+    let second = SubjektivStore::open(&workspace, &registration).unwrap();
+    let barrier = Arc::new(Barrier::new(3));
+    let spawn = |store: SubjektivStore, worker: &'static str| {
+        let barrier = barrier.clone();
+        let subject = subject.id.clone();
+        thread::spawn(move || {
+            barrier.wait();
+            store.claim_journal_writer(&subject, "runtime", worker)
+        })
+    };
+    let first = spawn(f.store().clone(), "first");
+    let second = spawn(second, "second");
+    barrier.wait();
+    let outcomes = [first.join().unwrap(), second.join().unwrap()];
+    assert_eq!(outcomes.iter().filter(|r| r.is_ok()).count(), 1);
+    let winner = outcomes.into_iter().find_map(Result::ok).unwrap();
+    assert!(f.store().journal_writer_active(&winner).unwrap());
+    let loser = if winner.worker_id == "first" {
+        "second"
+    } else {
+        "first"
+    };
+    assert!(
+        f.store()
+            .claim_journal_writer(&subject.id, "runtime", loser)
+            .is_err()
+    );
+    drop(workspace);
+    manager.shutdown().unwrap();
+}
+
+#[test]
+fn release_and_late_admission_are_serialized_by_the_persisted_binding_check() {
+    let f = Fixture::new();
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    let (generation, context) = f.generation("release-race", f.ranges.clone());
+    f.execute(
+        &context,
+        "record",
+        open_action("unfinished"),
+        f.ranges.clone(),
+    );
+    f.execute(
+        &context,
+        "position",
+        JournalAction::UpdatePosition {
+            current_position: "remaining work".into(),
+        },
+        vec![],
+    );
+    f.execute(
+        &context,
+        "checkpoint",
+        JournalAction::Checkpoint {
+            record_operation_ids: vec!["record".into()],
+        },
+        f.ranges.clone(),
+    );
+    f.close(&generation);
+    let manager = FeatureStorage::new(f.temp.path().join("storage"));
+    let workspace = manager.workspace("journal-tests").unwrap();
+    let registration = SubjektivStore::register(&workspace).unwrap();
+    let second = SubjektivStore::open(&workspace, &registration).unwrap();
+    let barrier = Arc::new(Barrier::new(3));
+    let b = barrier.clone();
+    let store = f.store().clone();
+    let released = binding.clone();
+    let release = thread::spawn(move || {
+        b.wait();
+        store.release_journal_writer(&released)
+    });
+    let b = barrier.clone();
+    let host = f.host.clone();
+    let late = source_range("source-session", 2, 16);
+    let admission = thread::spawn(move || {
+        b.wait();
+        second.accept_journal_sources(&host, "late", &[late])
+    });
+    barrier.wait();
+    let release = release.join().unwrap();
+    let admission = admission.join().unwrap();
+    assert_ne!(
+        release.is_ok(),
+        admission.is_ok(),
+        "exactly one ordered effect may commit"
+    );
+    assert_eq!(
+        f.store().journal_writer_active(&binding).unwrap(),
+        admission.is_ok()
+    );
+    assert_eq!(
+        f.store()
+            .read_journal_source(&f.host, &f.journal_id, 2)
+            .is_ok(),
+        admission.is_ok()
+    );
+    drop(workspace);
+    manager.shutdown().unwrap();
+}
+
+#[test]
+fn startup_snapshot_distinguishes_none_is_subject_local_and_bounds_utf8_at_storage() {
+    let f = Fixture::new();
+    let (_, context) = f.generation("snapshot", f.ranges.clone());
+    f.execute(
+        &context,
+        "large-open",
+        open_action(&"日本語\\\"\\n".repeat(6000)),
+        f.ranges.clone(),
+    );
+    let reader = JournalHostContext::validated_reader(f.store(), &f.host.subject_id).unwrap();
+    let snapshot = f
+        .store()
+        .journal_startup_snapshot(&reader, Some("source-session"))
+        .unwrap();
+    let encoded = serde_json::to_vec(&snapshot).unwrap();
+    assert!(encoded.len() <= startup::STARTUP_MAX_BYTES);
+    assert!(snapshot.journals[0].entries[0].omitted);
+    assert!(snapshot.journals[0].entries[0].body.len() <= 2048);
+    assert_eq!(snapshot.journals[0].source_session_id, "source-session");
+    let other = f
+        .store()
+        .create_subject_with_behavior(
+            SubjectRole::new("other").unwrap(),
+            "behavior unchanged".into(),
+        )
+        .unwrap();
+    let other_reader = JournalHostContext::validated_reader(f.store(), &other.id).unwrap();
+    assert!(matches!(
+        f.store()
+            .journal_startup_snapshot(&other_reader, Some("source-session")),
+        Err(JournalError::PermissionDenied(_))
+    ));
+    let none = f
+        .store()
+        .journal_startup_snapshot(&other_reader, None)
+        .unwrap();
+    assert!(none.journals.is_empty());
+    assert!(!none.omitted);
+    assert_eq!(
+        f.store().subject(&other.id).unwrap().unwrap().behavior_md,
+        "behavior unchanged"
+    );
+}
+
 #[test]
 fn stable_legacy_run_references_are_accepted_without_relaxing_character_or_size_limits() {
     let f = Fixture::new();
@@ -430,6 +874,111 @@ fn erasure_failure_rolls_back_tombstone_and_content_then_exact_retry_succeeds()
     );
 }
 
+#[test]
+fn checkpointed_generation_rejects_late_activity_but_retains_saved_receipts() {
+    let f = Fixture::new();
+    let (generation, context) = f.generation("checkpointed-job", f.ranges.clone());
+    let operation = f.operation("record", open_action("B partly executed"), f.ranges.clone());
+    let receipt = f
+        .store()
+        .execute_journal(&context, operation.clone())
+        .unwrap();
+    let entry_id = receipt.entry_id.clone().unwrap();
+    save_position(f.store(), &context, "position");
+    f.execute(
+        &context,
+        "checkpoint",
+        checkpoint(&["record"]),
+        f.ranges.clone(),
+    );
+    let before = f.state();
+    for (index, action) in [
+        open_action("late create"),
+        JournalAction::Rewrite {
+            entry_id: entry_id.clone(),
+            title: "late rewrite".into(),
+            body: "late".into(),
+            ongoing: "late".into(),
+        },
+        JournalAction::Fix {
+            entry_id,
+            source_end: f.ranges.last().unwrap().source.pointer.clone(),
+        },
+        JournalAction::ConfirmUnchanged,
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let late = f.operation(&format!("late-{index}"), action, f.ranges.clone());
+        assert!(matches!(
+            f.store().execute_journal(&context, late),
+            Err(JournalError::Conflict(_))
+        ));
+        assert_eq!(f.state(), before);
+    }
+    assert_eq!(
+        f.store().execute_journal(&context, operation).unwrap(),
+        receipt
+    );
+    f.close(&generation);
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    f.store().release_journal_writer(&binding).unwrap();
+}
+
+#[test]
+fn checkpoint_without_current_position_leaves_failed_job_input_recoverable_after_reopen() {
+    let mut f = Fixture::new();
+    let (failed, context) = f.generation("incomplete-position", f.ranges.clone());
+    save_position(f.store(), &context, "outdated-position");
+    f.execute(
+        &context,
+        "record",
+        open_action("B selected; tests not run"),
+        f.ranges.clone(),
+    );
+    let checkpoint_operation = f.operation("checkpoint", checkpoint(&["record"]), f.ranges.clone());
+    let before = f.state();
+    assert!(matches!(
+        f.store().execute_journal(&context, checkpoint_operation),
+        Err(JournalError::Conflict(_))
+    ));
+    assert_eq!(f.state(), before);
+    assert!(f.sources().iter().all(|source| source.reflected_bytes == 0));
+    assert!(f.receipt("checkpoint").is_none());
+    f.close(&failed);
+    f.reopen();
+    let (replacement, context) = f.generation("recover-position", f.ranges.clone());
+    f.execute(
+        &context,
+        "recovered-position",
+        JournalAction::UpdatePosition {
+            current_position: "B partly performed; tests unperformed; external result Unknown"
+                .into(),
+        },
+        vec![],
+    );
+    // Use the saved record receipt rather than duplicating the old activity entry.
+    f.execute(
+        &context,
+        "checkpoint",
+        checkpoint(&["record"]),
+        f.ranges.clone(),
+    );
+    f.close(&replacement);
+    let binding = f
+        .store()
+        .claim_journal_writer(&f.host.subject_id, "runtime", "worker")
+        .unwrap();
+    assert_eq!(f.entries().len(), 1);
+    assert_eq!(f.entries()[0].state, JournalEntryState::Open);
+    assert!(f.state().0.current_position.contains("Unknown"));
+    f.store().release_journal_writer(&binding).unwrap();
+    assert!(!f.store().journal_writer_active(&binding).unwrap());
+}
+
 #[test]
 fn job_service_completion_requires_checkpoint_and_replays_saved_completion_after_new_generation() {
     let mut f = Fixture::new();
@@ -476,22 +1025,43 @@ fn job_service_completion_requires_checkpoint_and_replays_saved_completion_after
         service::finish(f.store(), &context),
         Err(JournalError::Conflict(_))
     ));
-    service::execute(
-        f.store(),
-        &context,
-        transport::JournalJobRequest::Execute {
-            operation: JournalOperation {
-                request_id: "service-checkpoint".into(),
-                expected_state: receipt.state_token,
-                action: JournalAction::Checkpoint {
-                    record_operation_ids: vec!["service-create".into()],
-                },
-                covered_ranges: f.ranges.clone(),
-            },
+    let checkpoint_operation = JournalOperation {
+        request_id: "service-checkpoint".into(),
+        expected_state: receipt.state_token,
+        action: JournalAction::Checkpoint {
+            record_operation_ids: vec!["service-create".into()],
         },
-        |_| unreachable!(),
-    )
-    .unwrap();
+        covered_ranges: f.ranges.clone(),
+    };
+    // Reject the checkpoint atomically: failed/terminal Jobs must leave pending input
+    // recoverable rather than stranding a fully reflected prefix without a position memo.
+    assert!(matches!(
+        service::execute(
+            f.store(),
+            &context,
+            transport::JournalJobRequest::Execute {
+                operation: checkpoint_operation
+            },
+            |_| unreachable!(),
+        ),
+        Err(JournalError::Conflict(_))
+    ));
+    assert!(f.sources().iter().all(|source| source.reflected_bytes == 0));
+    assert!(f.receipt("service-checkpoint").is_none());
+    f.execute(
+        &context,
+        "service-position",
+        JournalAction::UpdatePosition {
+            current_position: "ongoing; work remains".into(),
+        },
+        vec![],
+    );
+    f.execute(
+        &context,
+        "service-checkpoint",
+        checkpoint(&["service-create"]),
+        f.ranges.clone(),
+    );
     let completion = service::finish(f.store(), &context).unwrap();
     assert_eq!(completion.generation_id, generation.id);
     assert_eq!((completion.process_order, completion.process_byte), (2, 0));
@@ -685,7 +1255,7 @@ fn retention_release_requires_checkpoint_not_saved_entry_or_unavailability() {
     let f = Fixture::new();
     let (_, job) = f.generation("retained-job", f.ranges.clone());
     let operation = f.operation("record", open_action("ongoing activity"), f.ranges.clone());
-    let receipt = f.store().execute_journal(&job, operation).unwrap();
+    f.store().execute_journal(&job, operation).unwrap();
     assert!(f.store().journal_retention_required(&f.host).unwrap());
     f.store()
         .mark_journal_sources_unavailable(&f.host, &f.journal_id, &f.ranges)
@@ -694,6 +1264,7 @@ fn retention_release_requires_checkpoint_not_saved_entry_or_unavailability() {
     f.store()
         .mark_journal_sources_available(&f.host, &f.journal_id, &f.ranges)
         .unwrap();
+    let receipt = save_position(f.store(), &job, "retained-position");
     f.store()
         .checkpoint_journal(
             &job,
@@ -805,8 +1376,7 @@ fn recording_materials_resume_exact_saved_byte_prefix_after_reopen() {
         .unwrap();
     let job = JournalJobContext::validated(f.store(), &host, &generation.id, binding).unwrap();
     let state = f.store().read_journal(&host, &id).unwrap().state_token;
-    let receipt = f
-        .store()
+    f.store()
         .execute_journal(
             &job,
             JournalOperation {
@@ -817,6 +1387,7 @@ fn recording_materials_resume_exact_saved_byte_prefix_after_reopen() {
             },
         )
         .unwrap();
+    let receipt = save_position(f.store(), &job, "recording-position");
     f.store()
         .checkpoint_journal(
             &job,
@@ -891,7 +1462,12 @@ impl Fixture {
                 },
             )
             .unwrap();
-        let host = session_host(&store, &subject.id, "source-session");
+        let attribution =
+            SubjectSessionAttribution::new(&subject.id, "runtime", "worker", "source-session")
+                .unwrap();
+        store
+            .record_session_attribution(attribution.clone())
+            .unwrap();
         let (manager, workspace, store) = if upgrade {
             let has_journals: bool = store.database.try_with_connection::<_, JournalError>(|connection| {
                 Ok(connection.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='journals')", [], |row| row.get(0))?)
@@ -905,15 +1481,16 @@ impl Fixture {
             let workspace = manager.workspace("journal-tests").unwrap();
             let registration = SubjektivStore::register(&workspace).unwrap();
             let store = SubjektivStore::open(&workspace, &registration).unwrap();
-            assert_eq!(store.subject(&subject.id).unwrap(), Some(subject));
+            assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
             assert_eq!(
                 store.session_attribution("source-session").unwrap(),
-                host.attribution
+                Some(attribution)
             );
             (manager, workspace, store)
         } else {
             (manager, workspace, store)
         };
+        let host = session_host(&store, &subject.id, "source-session");
         let ranges = sizes
             .iter()
             .enumerate()
@@ -1070,10 +1647,11 @@ impl Drop for Fixture {
 fn session_host(store: &SubjektivStore, subject: &str, session: &str) -> JournalHostContext {
     let attribution =
         SubjectSessionAttribution::new(subject, "runtime", "worker", session).unwrap();
-    store
-        .record_session_attribution(attribution.clone())
+    let attribution = store.record_session_attribution(attribution).unwrap();
+    let binding = store
+        .claim_journal_writer(subject, "runtime", "worker")
         .unwrap();
-    JournalHostContext::validated_session(store, attribution).unwrap()
+    store.bind_journal_session(&binding, attribution).unwrap()
 }
 
 fn source_range(session: &str, order: u64, total_bytes: u64) -> JournalSourceRange {
@@ -1127,6 +1705,32 @@ fn fixed_action(body: &str, range: &JournalSourceRange, corrections: Vec<String>
     }
 }
 
+fn save_position(
+    store: &SubjektivStore,
+    context: &JournalJobContext,
+    request_id: &str,
+) -> JournalReceipt {
+    let generation = store
+        .read_journal_generation(&context.host, &context.generation_id)
+        .unwrap();
+    let (journal, _) = store
+        .journal_resume_state(&context.host, &generation.journal_id)
+        .unwrap();
+    store
+        .execute_journal(
+            context,
+            JournalOperation {
+                request_id: request_id.into(),
+                expected_state: journal.state_token,
+                action: JournalAction::UpdatePosition {
+                    current_position: journal.current_position,
+                },
+                covered_ranges: vec![],
+            },
+        )
+        .unwrap()
+}
+
 fn checkpoint(ids: &[&str]) -> JournalAction {
     JournalAction::Checkpoint {
         record_operation_ids: ids.iter().map(|id| (*id).into()).collect(),
@@ -1249,6 +1853,7 @@ fn sequential_open_rewrite_fix_and_correction_preserve_fixed_history() {
     assert_eq!(entries[1].correction_refs, vec![first_id]);
     assert_eq!(entries[1].previous_fixed_end, fixed.source_end);
     assert_eq!(fixture.state().0.process_order, 0);
+    save_position(fixture.store(), &ctx, "final-position");
     let cp = fixture.execute(
         &ctx,
         "checkpoint",
@@ -1966,6 +2571,7 @@ fn checkpoint_requires_saved_record_correspondence_and_never_position_only_evide
         JournalAction::ConfirmUnchanged,
         vec![fixture.ranges[1].clone()],
     );
+    save_position(fixture.store(), &ctx, "checked-position");
     let receipt = fixture.execute(
         &ctx,
         "valid-checkpoint",
@@ -2017,6 +2623,7 @@ fn huge_source_partial_generations_checkpoint_only_read_prefix_and_fix_only_at_e
         fixture.store().execute_journal(&first_ctx, unread_end),
         Err(JournalError::InvalidInput(_))
     ));
+    save_position(fixture.store(), &first_ctx, "partial-position");
     let cp = fixture.execute(
         &first_ctx,
         "partial-checkpoint",
@@ -2082,6 +2689,7 @@ fn huge_source_partial_generations_checkpoint_only_read_prefix_and_fix_only_at_e
         vec![],
     );
     assert_eq!(fixture.state().0.process_byte, head.bytes.end);
+    save_position(fixture.store(), &next_ctx, "tail-position");
     let cp = fixture.execute(
         &next_ctx,
         "tail-checkpoint",
@@ -2913,6 +3521,7 @@ fn failure_before_receipt_rolls_back_record_and_checkpoint_then_exact_retry_succ
         fixture.store().execute_journal(&ctx, record).unwrap(),
         saved
     );
+    save_position(fixture.store(), &ctx, "fault-position");
     let cp = fixture.operation(
         "fault-checkpoint",
         checkpoint(&["fault-record"]),
@@ -2994,6 +3603,7 @@ fn per_operation_reopen_recovers_a_realistic_investigation_and_later_correction_
     assert!(fixture.state().1.is_none());
     assert_eq!(fixture.state().0.process_order, 0);
     let fixed_before_correction = fixture.entries()[0].clone();
+    save_position(fixture.store(), &first_ctx, "investigation-fixed-position");
     fixture.execute_and_reopen(
         &first_ctx,
         "investigation-checkpoint",
@@ -3117,6 +3727,7 @@ fn split_input_and_replacement_job_reconcile_old_record_coverage_without_duplica
         .unwrap();
     assert_eq!(coverage.items, vec![whole.clone()]);
     let (_, replacement) = fixture.generation("replacement-job", vec![whole.clone()]);
+    save_position(fixture.store(), &replacement, "replacement-position");
     let cp = fixture.operation("recover-reflection", checkpoint(&["record"]), vec![whole]);
     let result = fixture
         .store()
@@ -3200,86 +3811,95 @@ fn serialized_metadata_and_text_budgets_preserve_continuation_and_availability_r
 
 #[test]
 fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existing_subject() {
-    let temp = tempfile::tempdir().unwrap();
-    let root = temp.path().join("storage");
-    let manager = FeatureStorage::new(&root);
-    let scope = manager.scope("migration-rollback").unwrap();
-    let registration = scope
-        .register(feature_storage::FeatureRegistration::new(
-            crate::SUBJEKTIV_FEATURE_ID,
-            &crate::MIGRATIONS[..7],
-        ))
-        .unwrap();
-    let store = SubjektivStore::open(&scope, &registration).unwrap();
-    let subject = store
-        .create_subject_with_behavior(
-            SubjectRole::new("preserved").unwrap(),
-            "Original behavior".into(),
-        )
-        .unwrap();
-    store
-        .database
-        .try_transaction::<_, JournalError>(|tx| {
-            tx.execute_batch("CREATE TABLE journal_entries(injected INTEGER)")?;
-            Ok(())
-        })
-        .unwrap();
-    drop(store);
-    drop(scope);
-    manager.shutdown().unwrap();
-    drop(manager);
-    let manager = FeatureStorage::new(&root);
-    let scope = manager.scope("migration-rollback").unwrap();
-    let registration = SubjektivStore::register(&scope).unwrap();
-    assert!(matches!(
-        SubjektivStore::open(&scope, &registration),
-        Err(crate::SubjektivError::Storage(
-            feature_storage::FeatureStorageError::MigrationFailed { .. }
-        ))
-    ));
-    drop(scope);
-    manager.shutdown().unwrap();
-    drop(manager);
-    let manager = FeatureStorage::new(&root);
-    let scope = manager.scope("migration-rollback").unwrap();
-    let registration = scope
-        .register(feature_storage::FeatureRegistration::new(
-            crate::SUBJEKTIV_FEATURE_ID,
-            &crate::MIGRATIONS[..7],
-        ))
-        .unwrap();
-    let store = SubjektivStore::open(&scope, &registration).unwrap();
-    assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
-    store
-        .database
-        .try_transaction::<_, JournalError>(|tx| {
-            assert!(!tx.query_row(
-                "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='journals')",
-                [],
-                |r| r.get::<_, bool>(0)
-            )?);
-            tx.execute_batch("DROP TABLE journal_entries")?;
-            Ok(())
-        })
-        .unwrap();
-    drop(store);
-    drop(scope);
-    manager.shutdown().unwrap();
-    drop(manager);
-    let manager = FeatureStorage::new(&root);
-    let scope = manager.scope("migration-rollback").unwrap();
-    let registration = SubjektivStore::register(&scope).unwrap();
-    let store = SubjektivStore::open(&scope, &registration).unwrap();
-    assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
-    let host = session_host(&store, &subject.id, "post-migration-session");
-    let range = source_range("post-migration-session", 0, 8);
-    let id = store
-        .accept_journal_sources(&host, "Recovered migration", &[range])
-        .unwrap();
-    assert_eq!(
-        store.read_journal(&host, &id).unwrap().source_session_id,
-        "post-migration-session"
-    );
+    for (baseline, collision, rolled_back) in [
+        (&crate::MIGRATIONS[..7], "journal_entries", "journals"),
+        (
+            &crate::MIGRATIONS[..9],
+            "journal_session_bindings",
+            "journal_writer_bindings",
+        ),
+    ] {
+        let temp = tempfile::tempdir().unwrap();
+        let root = temp.path().join("storage");
+        let manager = FeatureStorage::new(&root);
+        let scope = manager.scope("migration-rollback").unwrap();
+        let registration = scope
+            .register(feature_storage::FeatureRegistration::new(
+                crate::SUBJEKTIV_FEATURE_ID,
+                baseline,
+            ))
+            .unwrap();
+        let store = SubjektivStore::open(&scope, &registration).unwrap();
+        let subject = store
+            .create_subject_with_behavior(
+                SubjectRole::new("preserved").unwrap(),
+                "Original behavior".into(),
+            )
+            .unwrap();
+        store
+            .database
+            .try_transaction::<_, JournalError>(|tx| {
+                tx.execute_batch(&format!("CREATE TABLE {collision}(injected INTEGER)"))?;
+                Ok(())
+            })
+            .unwrap();
+        drop(store);
+        drop(scope);
+        manager.shutdown().unwrap();
+        drop(manager);
+        let manager = FeatureStorage::new(&root);
+        let scope = manager.scope("migration-rollback").unwrap();
+        let registration = SubjektivStore::register(&scope).unwrap();
+        assert!(matches!(
+            SubjektivStore::open(&scope, &registration),
+            Err(crate::SubjektivError::Storage(
+                feature_storage::FeatureStorageError::MigrationFailed { .. }
+            ))
+        ));
+        drop(scope);
+        manager.shutdown().unwrap();
+        drop(manager);
+        let manager = FeatureStorage::new(&root);
+        let scope = manager.scope("migration-rollback").unwrap();
+        let registration = scope
+            .register(feature_storage::FeatureRegistration::new(
+                crate::SUBJEKTIV_FEATURE_ID,
+                baseline,
+            ))
+            .unwrap();
+        let store = SubjektivStore::open(&scope, &registration).unwrap();
+        assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
+        store
+            .database
+            .try_transaction::<_, JournalError>(|tx| {
+                assert!(!tx.query_row(
+                    "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name=?1)",
+                    [rolled_back],
+                    |r| r.get::<_, bool>(0)
+                )?);
+                tx.execute_batch(&format!("DROP TABLE {collision}"))?;
+                Ok(())
+            })
+            .unwrap();
+        drop(store);
+        drop(scope);
+        manager.shutdown().unwrap();
+        drop(manager);
+        let manager = FeatureStorage::new(&root);
+        let scope = manager.scope("migration-rollback").unwrap();
+        let registration = SubjektivStore::register(&scope).unwrap();
+        let store = SubjektivStore::open(&scope, &registration).unwrap();
+        assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone()));
+        let host = session_host(&store, &subject.id, "post-migration-session");
+        let range = source_range("post-migration-session", 0, 8);
+        let id = store
+            .accept_journal_sources(&host, "Recovered migration", &[range])
+            .unwrap();
+        assert_eq!(
+            store.read_journal(&host, &id).unwrap().source_session_id,
+            "post-migration-session"
+        );
+    }
 }
 
 #[test]
@@ -3365,6 +3985,7 @@ fn late_gap_admission_and_partial_updates_reject_stale_source_page_cursors() {
             .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2)
             .is_ok()
     );
+    save_position(fixture.store(), &job, "listing-position");
     fixture.execute(
         &job,
         "checkpoint",

crates/subjektiv/src/legacy_migrations.rs

@@ -953,11 +953,12 @@ mod tests {
             store.session_attribution("session-1").unwrap().unwrap(),
             stored_attribution
         );
-        let host = crate::journal::JournalHostContext::validated_session(
-            &store,
-            stored_attribution.clone(),
-        )
-        .unwrap();
+        let binding = store
+            .claim_journal_writer(&subject_id, "runtime-1", "worker-1")
+            .unwrap();
+        let host = store
+            .bind_journal_session(&binding, stored_attribution.clone())
+            .unwrap();
         let range = crate::journal::JournalSourceRange {
             source: crate::journal::JournalSource {
                 session_id: stored_attribution.session_id.clone(),

crates/subjektiv/src/lib.rs

@@ -615,6 +615,11 @@ static MIGRATIONS: &[FeatureMigration] = &[
     ),
     FeatureMigration::new(
         10,
+        "fence journal writers by explicit body binding lifetime",
+        journal::migrate_binding,
+    ),
+    FeatureMigration::new(
+        11,
         "add immutable candidate Journal context links",
         journal::memory_links::migrate,
     ),