feat(journal): execute frozen body inputs with durable delivery and erasure fences

crates/subjektiv/Cargo.toml

@@ -15,6 +15,8 @@ serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
 sha2.workspace = true
 server-api.workspace = true
+session-store.workspace = true
+job.workspace = true
 thiserror.workspace = true
 uuid = { workspace = true, features = ["v7"] }
 

crates/subjektiv/src/journal/recording.rs

@@ -5,8 +5,221 @@
 //! original canonical identity while streaming and return only the requested bounded fragment.
 
 use super::*;
+use session_store::capture::{CaptureDto, CaptureLimits};
 use sha2::{Digest, Sha256};
 
+/// A Job owns only this bounded original-body prefix. The initial content condition binds
+/// the separate generation snapshot (whose potentially large Journal text remains paged).
+/// Neither possessing these bytes nor delivering them is a processing checkpoint.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalInput {
+    pub version: u32,
+    pub subject_id: String,
+    pub journal_id: String,
+    pub initial_state_token: String,
+    pub initial_open_id: Option<String>,
+    pub previous_fixed_end: Option<JournalSourcePointer>,
+    pub selected_ranges: Vec<JournalSourceRange>,
+    pub capture: CaptureDto,
+}
+
+impl JournalInput {
+    /// Capture callback runs once for this immutable Job, never once per page. Fitting uses
+    /// only that captured value and measures the complete serialized input, including the
+    /// initial Journal condition and ranges duplicated beside the capture's descriptors.
+    pub fn build(
+        store: &SubjektivStore,
+        host: &JournalHostContext,
+        journal_id: &str,
+        capture_once: impl FnOnce(u64, u64, usize) -> JournalResult<CaptureDto>,
+    ) -> JournalResult<Self> {
+        let (journal, open, previous_fixed_end) = store.database.try_with_connection(|tx| {
+            super::store::check_host(store, tx, host, true)?;
+            Ok::<_, JournalError>((
+                super::store::journal(tx, host, journal_id, true)?,
+                super::store::open(tx, journal_id)?,
+                super::store::previous_fixed(tx, journal_id)?.map(|p| p.0),
+            ))
+        })?;
+        let mut input = Self {
+            version: 1,
+            subject_id: journal.subject_id.clone(),
+            journal_id: journal_id.into(),
+            initial_state_token: journal.state_token,
+            initial_open_id: open.map(|e| e.id),
+            previous_fixed_end,
+            selected_ranges: Vec::new(),
+            capture: CaptureDto {
+                version: 1,
+                session_id: journal.source_session_id,
+                segment_id: "pending".into(),
+                view_entry_count: 0,
+                entries: Vec::new(),
+            },
+        };
+        let envelope = serde_json::to_vec(&input)?.len();
+        // Capture once at a fixed storage boundary. Any adjustment for the Journal envelope
+        // below operates on this value only, never by making another live capture request.
+        let budget = job::MAX_JOB_INPUT_BYTES.saturating_sub(envelope);
+        input.capture = capture_once(journal.process_order, journal.process_byte, budget)?;
+        input.selected_ranges = capture_ranges(&input.capture)?;
+        if input.selected_ranges.is_empty() {
+            return Ok(input);
+        }
+        input.fit()?;
+        let value = serde_json::to_value(&input)?;
+        Self::decode(&value)
+    }
+
+    fn fit(&mut self) -> JournalResult<()> {
+        let original = self.capture.clone();
+        let first = self
+            .selected_ranges
+            .first()
+            .ok_or_else(|| invalid("empty recording selection"))?;
+        let start = session_store::capture::RecordingPosition {
+            canonical_order: first.source.order,
+            byte_offset: first.bytes.start,
+        };
+        let mut budget = serde_json::to_vec(&original)?.len();
+        loop {
+            let bytes = serde_json::to_vec(self)?.len();
+            if bytes <= job::MAX_JOB_INPUT_BYTES {
+                return Ok(());
+            }
+            budget = budget
+                .checked_sub(bytes - job::MAX_JOB_INPUT_BYTES)
+                .ok_or_else(|| invalid("Journal envelope cannot fit source bytes"))?;
+            self.capture = original
+                .recording_slice(start, MAX_JOURNAL_RANGES, budget)
+                .map_err(|e| invalid(&e.to_string()))?
+                .capture;
+            self.selected_ranges = capture_ranges(&self.capture)?;
+            if self.selected_ranges.is_empty() {
+                return Err(invalid("Journal envelope cannot fit one source scalar"));
+            }
+        }
+    }
+
+    pub fn decode(value: &serde_json::Value) -> JournalResult<Self> {
+        if serde_json::to_vec(value)?.len() > job::MAX_JOB_INPUT_BYTES {
+            return Err(invalid(
+                "Journal input exceeds whole serialized Job input limit",
+            ));
+        }
+        let input: Self = serde_json::from_value(value.clone())?;
+        input
+            .capture
+            .validate(CaptureLimits {
+                max_bytes: job::MAX_JOB_INPUT_BYTES,
+                max_entries: MAX_JOURNAL_RANGES,
+                max_text_bytes: job::MAX_JOB_INPUT_BYTES,
+                ..CaptureLimits::default()
+            })
+            .map_err(|e| invalid(&e.to_string()))?;
+        if input.version != 1
+            || input.subject_id.is_empty()
+            || input.journal_id.is_empty()
+            || input.initial_state_token.is_empty()
+            || input.selected_ranges.is_empty()
+            || input.selected_ranges.len() > MAX_JOURNAL_RANGES
+            || input
+                .selected_ranges
+                .iter()
+                .any(|r| r.source.session_id != input.capture.session_id)
+        {
+            return Err(invalid("invalid Journal input identity or selection"));
+        }
+        if capture_ranges(&input.capture)? != input.selected_ranges {
+            return Err(conflict(
+                "Journal body snapshot does not match selected canonical ranges",
+            ));
+        }
+        Ok(input)
+    }
+
+    pub fn validate_generation(&self, generation: &JournalGeneration) -> JournalResult<()> {
+        if generation.journal_id != self.journal_id
+            || generation.initial_journal.subject_id != self.subject_id
+            || generation.initial_journal.source_session_id != self.capture.session_id
+            || generation.initial_journal.state_token != self.initial_state_token
+            || generation.initial_open.as_ref().map(|e| &e.id) != self.initial_open_id.as_ref()
+            || generation.previous_fixed_end != self.previous_fixed_end
+            || generation.selected_ranges != self.selected_ranges
+        {
+            return Err(conflict(
+                "saved Journal input differs from its initial generation snapshot",
+            ));
+        }
+        Ok(())
+    }
+}
+
+/// Converts the shared recording projection into Journal's coverage language without
+/// interpreting model text or constructing a second capture representation.
+pub fn capture_ranges(capture: &CaptureDto) -> JournalResult<Vec<JournalSourceRange>> {
+    use session_store::capture::CaptureBody;
+    let mut ranges = Vec::new();
+    for entry in &capture.entries {
+        let CaptureBody::CanonicalPublic {
+            descriptor,
+            byte_offset,
+            text,
+            ..
+        } = &entry.body
+        else {
+            return Err(invalid(
+                "Journal requires canonical original recording capture",
+            ));
+        };
+        if !entry.policy.recording_material
+            || text.is_empty()
+            || descriptor.identity.segment_id.to_string() != entry.source.segment_id
+            || descriptor.identity.entry_ref != format!("E{}", entry.entry_id)
+            || descriptor.canonical_order != entry.order
+            || byte_offset
+                .checked_add(text.len() as u64)
+                .is_none_or(|end| end > descriptor.total_bytes)
+        {
+            return Err(conflict(
+                "recording body identity, policy or range mismatch",
+            ));
+        }
+        let source = JournalSource {
+            session_id: capture.session_id.clone(),
+            pointer: JournalSourcePointer {
+                segment_id: descriptor.identity.segment_id,
+                entry_ref: descriptor.identity.entry_ref.clone(),
+            },
+            order: descriptor.canonical_order,
+            content_sha256: descriptor.sha256.clone(),
+            total_bytes: descriptor.total_bytes,
+        };
+        if *byte_offset == 0 && text.len() as u64 == descriptor.total_bytes {
+            verify_canonical_text(&source, text)?;
+        }
+        ranges.push(JournalSourceRange {
+            source,
+            bytes: JournalByteRange {
+                start: *byte_offset,
+                end: *byte_offset + text.len() as u64,
+            },
+        });
+    }
+    for pair in ranges.windows(2) {
+        if pair[0].source.order.checked_add(1) != Some(pair[1].source.order)
+            || pair[0].bytes.end != pair[0].source.total_bytes
+            || pair[1].bytes.start != 0
+        {
+            return Err(conflict(
+                "recording capture is not an oldest contiguous prefix",
+            ));
+        }
+    }
+    Ok(ranges)
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct JournalMaterial {

crates/subjektiv/src/journal/service.rs

@@ -66,8 +66,8 @@ fn text_page(
     })
 }
 
-/// Adapter for Host captures bounded by their storage contract. Production source reads should
-/// use [`execute_bounded`] to avoid allocating an entire source for one small read.
+/// Legacy adapter for Host captures bounded by their storage contract.
+/// Production Hosts use [`attest_input_delivery`] and [`execute_snapshot`] instead.
 pub fn execute(
     store: &SubjektivStore,
     context: &JournalJobContext,
@@ -101,23 +101,149 @@ pub fn execute(
     })
 }
 
-/// Reads are delivered through this service before mutation authority is enabled. The durable
-/// delivery ledger stores byte ranges only: Jobs may retain immutable descriptors without
-/// accumulating a permanent second copy of the original Session in their request ledgers.
-/// The Host verifies original provenance/digest/length and returns a bounded UTF-8 fragment.
+pub(crate) fn migrate_input_delivery(
+    tx: &rusqlite::Transaction<'_>,
+) -> feature_storage::Result<()> {
+    tx.execute_batch(
+        "CREATE TABLE journal_generation_input_deliveries (
+            generation_id TEXT PRIMARY KEY NOT NULL REFERENCES journal_generations(id) ON DELETE CASCADE,
+            input_digest TEXT NOT NULL CHECK(length(input_digest)=71 AND substr(input_digest,1,7)='sha256:' AND substr(input_digest,8) NOT GLOB '*[^0-9a-f]*')
+        );
+        CREATE TRIGGER journal_generation_input_delivery_no_update
+        BEFORE UPDATE ON journal_generation_input_deliveries
+        BEGIN SELECT RAISE(ABORT, 'Journal input delivery bindings are immutable'); END;",
+    )?;
+    Ok(())
+}
+
+fn validate_input_digest(input_digest: &str) -> JournalResult<()> {
+    // Preserve JobRequest::input_digest() verbatim, including its algorithm prefix.
+    if input_digest.strip_prefix("sha256:").is_none_or(|hex| {
+        hex.len() != 64
+            || !hex
+                .bytes()
+                .all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))
+    }) {
+        return Err(JournalError::InvalidInput(
+            "immutable Job input digest must be sha256: followed by 64 lowercase hex digits".into(),
+        ));
+    }
+    Ok(())
+}
+
+/// Trusted Host-only delivery evidence, never a model tool or a claim of understanding.
+/// Before calling, the Host must validate the saved, immutable body-bearing input against the
+/// bound Job's input digest, its Session/provenance/ranges, and the actual dispatch plan that
+/// supplies that exact input. The service cannot verify bytes it does not own. This transaction
+/// stores only the digest bound to the generation's immutable selection; it neither copies the
+/// body nor changes records, position, reflection or checkpoint. Exact replay is allowed only
+/// while the same generation and writer binding remain active; changed digests/ranges conflict.
+pub fn attest_input_delivery(
+    store: &SubjektivStore,
+    context: &JournalJobContext,
+    input_digest: &str,
+    selected_ranges: &[JournalSourceRange],
+) -> JournalResult<()> {
+    validate_input_digest(input_digest)?;
+    store.database.try_transaction(|tx| {
+        super::lifecycle::reject_erased(tx, &context.host)?;
+        let generation = super::store::job(store, tx, context, true)?;
+        if generation.selected_ranges != selected_ranges {
+            return Err(JournalError::Conflict(
+                "immutable Job input selection differs from generation".into(),
+            ));
+        }
+        let saved: Option<String> = tx.query_row(
+            "SELECT input_digest FROM journal_generation_input_deliveries WHERE generation_id=?1",
+            [&generation.id], |row| row.get(0),
+        ).optional()?;
+        if let Some(saved) = saved {
+            if saved != input_digest {
+                return Err(JournalError::Conflict(
+                    "generation is bound to a different immutable Job input digest".into(),
+                ));
+            }
+        } else {
+            tx.execute(
+                "INSERT INTO journal_generation_input_deliveries(generation_id,input_digest) VALUES(?1,?2)",
+                params![generation.id, input_digest],
+            )?;
+        }
+        Ok(())
+    })
+}
+
+fn require_input_delivery(
+    tx: &rusqlite::Connection,
+    context: &JournalJobContext,
+    input_digest: &str,
+) -> JournalResult<()> {
+    validate_input_digest(input_digest)?;
+    let saved: Option<String> = tx
+        .query_row(
+            "SELECT input_digest FROM journal_generation_input_deliveries WHERE generation_id=?1",
+            [&context.generation_id],
+            |row| row.get(0),
+        )
+        .optional()?;
+    if saved.as_deref() != Some(input_digest) {
+        return Err(JournalError::Conflict(
+            "trusted Host delivery of this immutable Job input is required".into(),
+        ));
+    }
+    Ok(())
+}
+
+/// Production sequential execution over Journal state and the Host-attested immutable input.
+/// The digest is taken from the bound saved Job by the Host, not from model arguments. Every
+/// request requires the matching delivery binding and an active generation/writer. Source text
+/// is already in immutable input: Source targets fail explicitly, with no Session reader or
+/// fallback. Receipt-backed record, position and checkpoint constraints remain unchanged.
+pub fn execute_snapshot(
+    store: &SubjektivStore,
+    context: &JournalJobContext,
+    input_digest: &str,
+    request: JournalJobRequest,
+) -> JournalResult<JournalJobResponse> {
+    execute_inner(store, context, request, Some(input_digest), |_, _, _| {
+        Err(JournalError::PermissionDenied(
+            "Source reads are unavailable for immutable Journal input".into(),
+        ))
+    })
+}
+
+/// Legacy/test source-read adapter. Production Hosts must use [`execute_snapshot`].
+/// This ledger attests bounded Source responses only and cannot satisfy the input-digest gate.
 pub fn execute_bounded(
     store: &SubjektivStore,
     context: &JournalJobContext,
     request: JournalJobRequest,
+    read_source: impl FnMut(&JournalSource, u64, usize) -> JournalResult<recording::JournalMaterial>,
+) -> JournalResult<JournalJobResponse> {
+    execute_inner(store, context, request, None, read_source)
+}
+
+fn execute_inner(
+    store: &SubjektivStore,
+    context: &JournalJobContext,
+    request: JournalJobRequest,
+    input_digest: Option<&str>,
     mut read_source: impl FnMut(&JournalSource, u64, usize) -> JournalResult<recording::JournalMaterial>,
 ) -> JournalResult<JournalJobResponse> {
-    let generation = store
-        .database
-        .try_with_connection(|tx| super::store::job(store, tx, context, true))?;
+    let generation = store.database.try_with_connection(|tx| {
+        super::lifecycle::reject_erased(tx, &context.host)?;
+        let generation = super::store::job(store, tx, context, true)?;
+        if let Some(input_digest) = input_digest {
+            require_input_delivery(tx, context, input_digest)?;
+        }
+        Ok::<_, JournalError>(generation)
+    })?;
     let mut delivered = None;
     let response = match request {
         JournalJobRequest::Execute { operation } => {
-            require_delivery(store, context, &generation.selected_ranges)?;
+            if input_digest.is_none() {
+                require_delivery(store, context, &generation.selected_ranges)?;
+            }
             let receipt = if matches!(operation.action, JournalAction::Checkpoint { .. }) {
                 crate::api::checkpoint_journal(store, context, operation)?
             } else {
@@ -197,6 +323,11 @@ pub fn execute_bounded(
         JournalJobRequest::ReadText { read } => {
             let text = match &read.target {
                 JournalTextTarget::Source { pointer } => {
+                    if input_digest.is_some() {
+                        return Err(JournalError::PermissionDenied(
+                            "Source target is unavailable: raw source bodies are in immutable Job input".into(),
+                        ));
+                    }
                     let range = generation
                         .selected_ranges
                         .iter()

crates/subjektiv/src/journal/store.rs

@@ -202,7 +202,7 @@ pub(super) fn open(tx: &Connection, journal_id: &str) -> JournalResult<Option<Jo
         .optional()?;
     id.map(|id| entry(tx, journal_id, &id)).transpose()
 }
-fn previous_fixed(
+pub(super) fn previous_fixed(
     tx: &Connection,
     journal_id: &str,
 ) -> JournalResult<Option<(JournalSourcePointer, u64)>> {
@@ -569,6 +569,60 @@ impl SubjektivStore {
             Ok(snapshot)
         })
     }
+    /// Rebind a saved body input to a new attempt without resampling its initial state.
+    /// The first binding freezes the full initial Journal snapshot; retries reuse it and
+    /// retain operation receipts. Different Jobs cannot borrow that snapshot as authority.
+    pub fn prepare_journal_input(
+        &self,
+        host: &JournalHostContext,
+        input: &super::recording::JournalInput,
+        binding: JobAttemptBinding,
+    ) -> JournalResult<JournalGeneration> {
+        label(&binding.job_id)?;
+        label(&binding.attempt_id)?;
+        self.database.try_transaction(|tx| {
+            check_host(self, tx, host, false)?;
+            check_session_host(self, tx, host)?;
+            let current = journal(tx, host, &input.journal_id, true)?;
+            let existing: Option<String> = tx.query_row(
+                "SELECT snapshot_json FROM journal_generations WHERE journal_id=?1 AND job_id=?2 AND attempt_id=?3",
+                params![input.journal_id, binding.job_id, binding.attempt_id], |r| r.get(0)).optional()?;
+            if let Some(raw) = existing {
+                let g = serde_json::from_str(&raw)?;
+                input.validate_generation(&g)?;
+                return Ok(g);
+            }
+            // Retained receipt lookup does not reacquire writer authority. Only a new
+            // generation requires the current active Subject/Session writer lifetime.
+            check_host(self, tx, host, true)?;
+            if tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_generations WHERE journal_id=?1 AND active=1)", [&input.journal_id], |r| r.get::<_, bool>(0))? {
+                return Err(conflict("another active generation owns this Journal"));
+            }
+            let prior: Option<String> = tx.query_row(
+                "SELECT snapshot_json FROM journal_generations WHERE journal_id=?1 AND job_id=?2 ORDER BY rowid LIMIT 1",
+                params![input.journal_id, binding.job_id], |r| r.get(0)).optional()?;
+            let mut g = if let Some(raw) = prior {
+                serde_json::from_str::<JournalGeneration>(&raw)?
+            } else {
+                JournalGeneration { id: String::new(), journal_id: input.journal_id.clone(), job: binding.clone(),
+                    selected_ranges: input.selected_ranges.clone(), initial_journal: current.clone(),
+                    initial_open: open(tx, &input.journal_id)?, previous_fixed_end: previous_fixed(tx, &input.journal_id)?.map(|p| p.0) }
+            };
+            input.validate_generation(&g)?;
+            check_ranges(tx, &current, &input.selected_ranges, true)?;
+            let end = prefix(&g.initial_journal, &input.selected_ranges)?;
+            let cursor = (current.process_order, current.process_byte);
+            if cursor < (g.initial_journal.process_order, g.initial_journal.process_byte) || cursor > end {
+                return Err(conflict("saved input no longer owns the current processing cursor"));
+            }
+            g.id = issued_id("journal-generation");
+            g.job = binding;
+            tx.execute("INSERT INTO journal_generations(id,journal_id,job_id,attempt_id,request_json,snapshot_json,active) VALUES(?1,?2,?3,?4,?5,?6,1)",
+                params![g.id,g.journal_id,g.job.job_id,g.job.attempt_id,serde_json::to_string(&g.selected_ranges)?,serde_json::to_string(&g)?])?;
+            Ok(g)
+        })
+    }
+
     pub(super) fn validate_journal_job(&self, ctx: &JournalJobContext) -> JournalResult<()> {
         self.database.try_with_connection(|tx| {
             job(self, tx, ctx, true)?;

crates/subjektiv/src/journal/transport.rs

@@ -1,8 +1,10 @@
 //! Bounded, Host-bound Journal recording and sequential Job transport.
 //!
 //! Subject, Journal and generation authority never come from these model inputs.
-//! Hosts validate live execution and public source bytes before selection; reading a
-//! preview must not select the enclosing entry. Operation retries retain the entire
+//! Hosts validate live execution and public source bytes before selection, save body-bearing
+//! immutable Job input, and attest its exact digest and dispatch plan through the service.
+//! Model requests cannot attest input delivery; Source targets are rejected by snapshot execution.
+//! Initial/current Journal reads remain separate from input delivery. Operation retries retain the entire
 //! JournalOperation, including its original expected_state and covered_ranges.
 
 use serde::{Deserialize, Serialize};
@@ -50,8 +52,8 @@ pub struct JournalJobState {
     pub current_journal: JournalMetadata,
     pub current_open: Option<JournalEntrySummary>,
     pub previous_fixed_end: Option<JournalSourcePointer>,
-    /// Frozen permission to fetch this prefix, not evidence of delivery or reflection.
-    /// The sequential service gates mutations on its durable Source-read delivery ledger.
+    /// Frozen source selection delivered in immutable Job input, not evidence of reflection.
+    /// Production execution requires the trusted Host's matching input-digest attestation.
     pub selected_ranges: Vec<JournalSourceRange>,
 }
 
@@ -83,7 +85,8 @@ pub enum JournalTextTarget {
     OpenOngoing {
         snapshot: JournalSnapshot,
     },
-    /// The Host restricts this pointer to this Job's frozen public source input.
+    /// Legacy source-read adapter only. Production snapshot execution rejects this target;
+    /// raw source bodies must already be present in the immutable Job input.
     Source {
         pointer: JournalSourcePointer,
     },

crates/subjektiv/src/lib.rs

@@ -639,6 +639,11 @@ static MIGRATIONS: &[FeatureMigration] = &[
         "add explicit Subject permission sources and immutable child execution grants",
         permissions::migrate,
     ),
+    FeatureMigration::new(
+        13,
+        "bind Journal generation delivery to immutable Job input digest",
+        journal::service::migrate_input_delivery,
+    ),
 ];
 
 pub const REGISTRATION: FeatureRegistration =