feat: integrate durable Journal storage APIs

crates/subjektiv/README.md

@@ -1,7 +1,8 @@
 # subjektiv
 
-Shared Host domain implementation for per-Subject Memory, change history, candidate dispositions, and derived surfaces.
+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.
 
 ## Main entry points
 
@@ -9,3 +10,4 @@ Shared Host domain implementation for per-Subject Memory, change history, candid
 - [`MemoryRecord`], [`SurfaceSnapshot`]: change history and derived context
 - [`api`]: shared Host operations and bounded responses
 - [`job`]: consolidation selection rules and result validation
+- [`journal`]: individually committed activity operations, source coverage, retry receipts, and bounded reading

crates/subjektiv/src/api.rs

@@ -11,8 +11,10 @@
 
 mod context;
 mod operation;
-pub use context::{HostOperationContext, JobAttemptBinding};
-pub use operation::execute;
+/// Shared Journal domain operations; no model tools or Host-specific transport.
+pub use crate::journal;
+pub use context::{HostOperationContext, JobAttemptBinding, JournalHostContext, JournalJobContext};
+pub use operation::{checkpoint_journal, execute, execute_journal};
 
 use serde::{Deserialize, Serialize};
 use std::collections::HashSet;

crates/subjektiv/src/api/context.rs

@@ -1,8 +1,9 @@
 //! Trusted Host operation context, never deserialized from model input.
 use super::{OperationError, OperationResult, subjektiv_store_error};
+pub use crate::journal::{JournalHostContext, JournalJobContext};
 use crate::{SubjectSessionAttribution, SubjektivStore};
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 pub struct JobAttemptBinding {
     pub job_id: String,
     pub attempt_id: String,

crates/subjektiv/src/api/operation.rs

@@ -1,4 +1,22 @@
 use super::*;
+
+/// Commits one Journal effect with its exact retry response, using Journal-only authority.
+pub fn execute_journal(
+    store: &crate::SubjektivStore,
+    context: &JournalJobContext,
+    operation: journal::JournalOperation,
+) -> journal::JournalResult<journal::JournalReceipt> {
+    store.execute_journal(context, operation)
+}
+
+/// Advances only selected input backed by already committed Journal effects.
+pub fn checkpoint_journal(
+    store: &crate::SubjektivStore,
+    context: &JournalJobContext,
+    operation: journal::JournalOperation,
+) -> journal::JournalResult<journal::JournalReceipt> {
+    store.checkpoint_journal(context, operation)
+}
 use server_api::{
     SubjektivMemoryBackendOperation as Op, SubjektivMemoryBackendResponse as Response,
 };

crates/subjektiv/src/journal/mod.rs

@@ -0,0 +1,345 @@
+//! Durable activity records owned by Subjektiv, independent of Session retention.
+//!
+//! Hosts attest committed public source identity and canonical Session order, not models. Byte
+//! ranges are half-open in the Host's canonical UTF-8 public-entry serialization (including
+//! framing, so even an empty message has a nonzero length). Compaction copies must retain the
+//! original pointer/order/digest; derived summaries are not new sources. Rewind replacement is
+//! an identity conflict, never an implicit reset. This module does not fetch or retain history.
+//!
+//! Each record operation and its exact retry receipt commit together. A separate checkpoint
+//! advances only a contiguous, selected prefix backed by saved record/explicit no-change
+//! receipts. It cannot use a position update as evidence. Stopping between these boundaries
+//! leaves saved records and unreflected input; read receipts before retrying, even across reopen
+//! or replacement Jobs. No whole-Job result submission is required. The Host checks live Job
+//! execution leases before issuing contexts, and durably closes a generation on cancellation.
+//!
+//! Selected ranges describe concrete public bytes supplied to generation, not a preview's
+//! enclosing history range or a permission to claim unobserved bytes. The Host verifies canonical
+//! serialization, SHA-256 identity and UTF-8 slice boundaries before admission/selection. This
+//! store validates identities, byte bounds and coverage but cannot attest text it does not fetch.
+//! Availability is the latest Host attestation; readers must not equate it with a live fetch.
+//! Public text and metadata reads are paged. Full generation/resume snapshots are Host-only
+//! construction material bounded by per-field storage caps (three text fields plus current position),
+//! not model-facing reads. A restored snapshot never restores an external Job lease.
+//!
+//! A T-734 caller saves `Create`/`Rewrite`/`Fix` under stable request IDs, carries each returned
+//! `state_token` into the next operation, optionally saves `UpdatePosition`, then submits
+//! `Checkpoint { record_operation_ids }` for just the reflected prefix. After a lost response it
+//! queries the original receipt, status and coverage pages before submitting any new create.
+//! On restart it reads the active generation and current resume state, verifies the external Job
+//! binding, and continues missing operations; cancellation closes the generation before replacement.
+//! A replacement may checkpoint old saved record receipts only for identical selected source bytes.
+//! Tokens incorporate prior saved content conditions as well as current content, preventing A→B→A
+//! edits from resurrecting an old operation premise after intervening saved effects.
+//! Source-ledger pagination uses an opaque, scope/Subject/Journal-bound content cursor: late gap
+//! admission, accepted-fragment extension, reflection, or availability changes reject continuation.
+//! The fingerprint and bounded projection share one transaction; hashing streams bounded rows.
+//! 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.
+
+mod store;
+#[cfg(test)]
+mod tests;
+
+use crate::{SubjectSessionAttribution, SubjektivStore, api::JobAttemptBinding};
+use feature_storage::FeatureStorageError;
+use rusqlite::Transaction;
+use serde::{Deserialize, Serialize};
+use uuid::Uuid;
+
+pub type JournalId = String;
+pub type JournalEntryId = String;
+pub type JournalResult<T> = Result<T, JournalError>;
+pub const MAX_JOURNAL_TEXT_BYTES: usize = 1024 * 1024;
+pub const MAX_JOURNAL_RANGES: usize = 64;
+pub const MAX_JOURNAL_PAGE: usize = 32;
+pub const MAX_JOURNAL_READ_BYTES: usize = 16 * 1024;
+/// Whole pretty-serialized public page budget, including escaped text and metadata.
+pub const MAX_JOURNAL_RESPONSE_BYTES: usize = 64 * 1024;
+
+#[derive(Debug, thiserror::Error)]
+pub enum JournalError {
+    #[error("journal permission denied: {0}")]
+    PermissionDenied(String),
+    #[error("journal conflict: {0}")]
+    Conflict(String),
+    #[error("invalid journal input: {0}")]
+    InvalidInput(String),
+    #[error("journal record not found: {0}")]
+    NotFound(String),
+    #[error(transparent)]
+    Storage(#[from] FeatureStorageError),
+    #[error(transparent)]
+    Json(#[from] serde_json::Error),
+    #[error(transparent)]
+    Subject(#[from] crate::SubjektivError),
+}
+impl From<rusqlite::Error> for JournalError {
+    fn from(e: rusqlite::Error) -> Self {
+        Self::Storage(e.into())
+    }
+}
+
+/// Host-issued in-process authority, never a model request DTO. Reader contexts need no live Session.
+#[derive(Debug, Clone)]
+pub struct JournalHostContext {
+    scope_id: String,
+    subject_id: String,
+    attribution: Option<SubjectSessionAttribution>,
+}
+impl JournalHostContext {
+    pub fn validated_reader(store: &SubjektivStore, subject_id: &str) -> JournalResult<Self> {
+        store
+            .subject(subject_id)?
+            .ok_or_else(|| JournalError::NotFound(subject_id.into()))?;
+        Ok(Self {
+            scope_id: store.scope_id().into(),
+            subject_id: subject_id.into(),
+            attribution: None,
+        })
+    }
+    /// External authentication, public-history validation and execution authority are Host duties.
+    pub fn validated_session(
+        store: &SubjektivStore,
+        attribution: SubjectSessionAttribution,
+    ) -> JournalResult<Self> {
+        if store.session_attribution(&attribution.session_id)?.as_ref() != Some(&attribution) {
+            return Err(JournalError::PermissionDenied(
+                "immutable Session attribution mismatch".into(),
+            ));
+        }
+        let mut context = Self::validated_reader(store, &attribution.subject_id)?;
+        context.attribution = Some(attribution);
+        Ok(context)
+    }
+}
+
+/// A restricted Journal generation, not Memory consolidation authority.
+#[derive(Debug, Clone)]
+pub struct JournalJobContext {
+    host: JournalHostContext,
+    generation_id: String,
+    binding: JobAttemptBinding,
+}
+impl JournalJobContext {
+    pub fn validated(
+        store: &SubjektivStore,
+        host: &JournalHostContext,
+        generation_id: &str,
+        binding: JobAttemptBinding,
+    ) -> JournalResult<Self> {
+        let context = Self {
+            host: host.clone(),
+            generation_id: generation_id.into(),
+            binding,
+        };
+        store.validate_journal_job(&context)?;
+        Ok(context)
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalSourcePointer {
+    pub segment_id: Uuid,
+    /// Serialized public SessionEntryRef, including its `E` prefix.
+    pub entry_ref: String,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalSource {
+    pub session_id: String,
+    pub pointer: JournalSourcePointer,
+    /// Host-attested canonical public entry order across Segments, starting at zero.
+    pub order: u64,
+    pub content_sha256: String,
+    pub total_bytes: u64,
+}
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalByteRange {
+    pub start: u64,
+    pub end: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalSourceRange {
+    pub source: JournalSource,
+    pub bytes: JournalByteRange,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JournalEntryState {
+    Open,
+    Fixed,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct Journal {
+    pub id: JournalId,
+    pub subject_id: String,
+    pub source_session_id: String,
+    pub title: String,
+    pub current_position: String,
+    /// Processing cursor is independent of every fixed event endpoint.
+    pub process_order: u64,
+    pub process_byte: u64,
+    /// Content condition advanced by committed operations, never by acceptance notifications.
+    pub state_token: String,
+    pub created_at: String,
+    pub updated_at: String,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalMetadata {
+    pub id: JournalId,
+    pub subject_id: String,
+    pub source_session_id: String,
+    pub title: String,
+    pub current_position_bytes: usize,
+    pub process_order: u64,
+    pub process_byte: u64,
+    pub state_token: String,
+    pub created_at: String,
+    pub updated_at: String,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalEntry {
+    pub id: JournalEntryId,
+    pub journal_id: JournalId,
+    pub position: u64,
+    pub state: JournalEntryState,
+    pub title: String,
+    pub body: String,
+    pub ongoing: String,
+    pub source_end: Option<JournalSourcePointer>,
+    pub created_at: String,
+    pub updated_at: String,
+    pub fixed_at: Option<String>,
+    pub correction_refs: Vec<JournalEntryId>,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalGeneration {
+    pub id: String,
+    pub journal_id: JournalId,
+    pub job: JobAttemptBinding,
+    pub selected_ranges: Vec<JournalSourceRange>,
+    pub initial_journal: Journal,
+    pub initial_open: Option<JournalEntry>,
+    pub previous_fixed_end: Option<JournalSourcePointer>,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)]
+pub enum JournalAction {
+    Create {
+        title: String,
+        body: String,
+        ongoing: String,
+        state: JournalEntryState,
+        source_end: Option<JournalSourcePointer>,
+        correction_refs: Vec<JournalEntryId>,
+    },
+    Rewrite {
+        entry_id: JournalEntryId,
+        title: String,
+        body: String,
+        ongoing: String,
+    },
+    Fix {
+        entry_id: JournalEntryId,
+        source_end: JournalSourcePointer,
+    },
+    UpdatePosition {
+        current_position: String,
+    },
+    /// Explicitly records the caller's checked conclusion that no entry change is needed.
+    ConfirmUnchanged,
+    /// Links durable record/no-change receipts to actual reflected input; IDs are Journal scoped.
+    Checkpoint {
+        record_operation_ids: Vec<String>,
+    },
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JournalOperation {
+    pub request_id: String,
+    pub expected_state: String,
+    pub action: JournalAction,
+    pub covered_ranges: Vec<JournalSourceRange>,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalReceipt {
+    pub request_id: String,
+    pub generation_id: String,
+    pub entry_id: Option<JournalEntryId>,
+    pub state_token: String,
+    pub process_order: u64,
+    pub process_byte: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalSavedOperation {
+    pub operation: JournalOperation,
+    pub receipt: JournalReceipt,
+}
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JournalOperationKind {
+    Create,
+    Rewrite,
+    Fix,
+    UpdatePosition,
+    ConfirmUnchanged,
+    Checkpoint,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalOperationStatus {
+    pub receipt: JournalReceipt,
+    pub kind: JournalOperationKind,
+    pub covered_range_count: usize,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalSourceStatus {
+    pub source: JournalSource,
+    pub accepted: Vec<JournalByteRange>,
+    pub unavailable: Vec<JournalByteRange>,
+    pub reflected_bytes: u64,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalPage<T> {
+    pub items: Vec<T>,
+    pub next: Option<String>,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalEntrySummary {
+    pub id: JournalEntryId,
+    pub journal_id: JournalId,
+    pub position: u64,
+    pub state: JournalEntryState,
+    pub title: String,
+    pub source_end: Option<JournalSourcePointer>,
+    /// Exclusive start; None means Session start. The end is inclusive.
+    pub previous_fixed_end: Option<JournalSourcePointer>,
+    pub source_available: bool,
+    pub correction_refs: Vec<JournalEntryId>,
+    pub content_token: String,
+    pub created_at: String,
+    pub updated_at: String,
+    pub fixed_at: Option<String>,
+}
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JournalTextField {
+    Body,
+    Ongoing,
+    CurrentPosition,
+}
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JournalTextPage {
+    pub text: String,
+    pub next_byte: Option<usize>,
+    pub content_token: String,
+}
+
+pub(super) fn migrate(tx: &Transaction<'_>) -> feature_storage::Result<()> {
+    tx.execute_batch(include_str!("schema.sql"))?;
+    Ok(())
+}

crates/subjektiv/src/journal/schema.sql

@@ -0,0 +1,88 @@
+CREATE TABLE journals (
+    id TEXT PRIMARY KEY,
+    subject_id TEXT NOT NULL REFERENCES subjects(subject_id) ON DELETE RESTRICT,
+    source_session_id TEXT NOT NULL UNIQUE,
+    title TEXT NOT NULL,
+    current_position TEXT NOT NULL,
+    process_order INTEGER NOT NULL DEFAULT 0 CHECK (process_order >= 0),
+    process_byte INTEGER NOT NULL DEFAULT 0 CHECK (process_byte >= 0),
+    state_token TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    UNIQUE(subject_id, id)
+);
+CREATE TABLE journal_entries (
+    id TEXT PRIMARY KEY,
+    subject_id TEXT NOT NULL,
+    journal_id TEXT NOT NULL,
+    position INTEGER NOT NULL CHECK (position >= 0),
+    state TEXT NOT NULL CHECK (state IN ('open', 'fixed')),
+    title TEXT NOT NULL,
+    body TEXT NOT NULL,
+    ongoing TEXT NOT NULL,
+    source_end_json TEXT,
+    source_order INTEGER,
+    created_at TEXT NOT NULL,
+    updated_at TEXT NOT NULL,
+    fixed_at TEXT,
+    UNIQUE(journal_id, position),
+    UNIQUE(subject_id, id),
+    FOREIGN KEY(subject_id, journal_id) REFERENCES journals(subject_id, id) ON DELETE RESTRICT,
+    CHECK ((state = 'open' AND source_end_json IS NULL AND source_order IS NULL AND fixed_at IS NULL)
+        OR (state = 'fixed' AND source_end_json IS NOT NULL AND source_order IS NOT NULL AND source_order >= 0 AND fixed_at IS NOT NULL AND ongoing = ''))
+);
+CREATE UNIQUE INDEX journal_one_open ON journal_entries(journal_id) WHERE state = 'open';
+CREATE TRIGGER journal_append_after_open BEFORE INSERT ON journal_entries
+WHEN EXISTS (SELECT 1 FROM journal_entries WHERE journal_id = NEW.journal_id AND state = 'open')
+BEGIN SELECT RAISE(ABORT, 'journal open must be the tail'); END;
+CREATE TRIGGER journal_fixed_immutable BEFORE UPDATE ON journal_entries WHEN OLD.state = 'fixed'
+BEGIN SELECT RAISE(ABORT, 'journal fixed entries are immutable'); END;
+CREATE TRIGGER journal_entry_identity BEFORE UPDATE ON journal_entries
+WHEN OLD.id <> NEW.id OR OLD.subject_id <> NEW.subject_id OR OLD.journal_id <> NEW.journal_id
+    OR OLD.position <> NEW.position OR OLD.created_at <> NEW.created_at
+BEGIN SELECT RAISE(ABORT, 'journal entry identity is immutable'); END;
+CREATE TABLE journal_corrections (
+    subject_id TEXT NOT NULL,
+    entry_id TEXT NOT NULL,
+    target_id TEXT NOT NULL,
+    PRIMARY KEY(entry_id, target_id),
+    FOREIGN KEY(subject_id, entry_id) REFERENCES journal_entries(subject_id, id) ON DELETE RESTRICT,
+    FOREIGN KEY(subject_id, target_id) REFERENCES journal_entries(subject_id, id) ON DELETE RESTRICT,
+    CHECK(entry_id <> target_id)
+);
+CREATE TABLE journal_sources (
+    journal_id TEXT NOT NULL REFERENCES journals(id) ON DELETE RESTRICT,
+    source_order INTEGER NOT NULL CHECK(source_order >= 0),
+    pointer_json TEXT NOT NULL,
+    entry_ref TEXT NOT NULL,
+    source_json TEXT NOT NULL,
+    accepted_json TEXT NOT NULL,
+    unavailable_json TEXT NOT NULL,
+    reflected_bytes INTEGER NOT NULL DEFAULT 0 CHECK(reflected_bytes >= 0),
+    PRIMARY KEY(journal_id, source_order),
+    UNIQUE(journal_id, pointer_json),
+    UNIQUE(journal_id, entry_ref)
+);
+CREATE TABLE journal_generations (
+    id TEXT PRIMARY KEY,
+    journal_id TEXT NOT NULL REFERENCES journals(id) ON DELETE RESTRICT,
+    job_id TEXT NOT NULL,
+    attempt_id TEXT NOT NULL,
+    request_json TEXT NOT NULL,
+    snapshot_json TEXT NOT NULL,
+    active INTEGER NOT NULL CHECK(active IN (0, 1)),
+    UNIQUE(journal_id, job_id, attempt_id)
+);
+CREATE UNIQUE INDEX journal_one_generation ON journal_generations(journal_id) WHERE active = 1;
+CREATE TABLE journal_operations (
+    journal_id TEXT NOT NULL REFERENCES journals(id) ON DELETE RESTRICT,
+    request_id TEXT NOT NULL,
+    generation_id TEXT NOT NULL REFERENCES journal_generations(id) ON DELETE RESTRICT,
+    request_json TEXT NOT NULL,
+    receipt_json TEXT NOT NULL,
+    PRIMARY KEY(journal_id, request_id)
+);
+CREATE TRIGGER journal_operations_no_update BEFORE UPDATE ON journal_operations
+BEGIN SELECT RAISE(ABORT, 'journal operation receipts are immutable'); END;
+CREATE TRIGGER journal_operations_no_delete BEFORE DELETE ON journal_operations
+BEGIN SELECT RAISE(ABORT, 'journal operation receipts are retained'); END;

crates/subjektiv/src/journal/store.rs

@@ -0,0 +1,988 @@
+use super::*;
+use crate::{issued_id, now};
+use rusqlite::{Connection, OptionalExtension, params};
+use sha2::{Digest, Sha256};
+
+#[derive(Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SourcePageCursor {
+    scope_id: String,
+    subject_id: String,
+    journal_id: String,
+    source_fingerprint: String,
+    after_order: u64,
+}
+
+// Hash rows incrementally rather than materializing an unbounded coverage ledger. The caller
+// holds one storage transaction for both this content condition and the bounded page projection.
+fn source_fingerprint(tx: &Connection, journal_id: &str) -> JournalResult<String> {
+    let mut hash = Sha256::new();
+    let mut stmt=tx.prepare("SELECT source_order, source_json, accepted_json, unavailable_json, reflected_bytes FROM journal_sources WHERE journal_id=?1 ORDER BY source_order")?;
+    let mut rows = stmt.query([journal_id])?;
+    while let Some(row) = rows.next()? {
+        let value = (
+            row.get::<_, u64>(0)?,
+            row.get::<_, String>(1)?,
+            row.get::<_, String>(2)?,
+            row.get::<_, String>(3)?,
+            row.get::<_, u64>(4)?,
+        );
+        let bytes = serde_json::to_vec(&value)?;
+        hash.update((bytes.len() as u64).to_le_bytes());
+        hash.update(bytes);
+    }
+    Ok(hash.finalize().iter().map(|b| format!("{b:02x}")).collect())
+}
+
+fn invalid(message: &str) -> JournalError {
+    JournalError::InvalidInput(message.into())
+}
+fn conflict(message: &str) -> JournalError {
+    JournalError::Conflict(message.into())
+}
+fn label(value: &str) -> JournalResult<()> {
+    if value.is_empty()
+        || value.len() > 128
+        || value.trim() != value
+        || value.chars().any(char::is_control)
+    {
+        return Err(invalid(
+            "identifier must be 1..128 bytes without whitespace or controls",
+        ));
+    }
+    Ok(())
+}
+fn text(value: &str, maximum: usize) -> JournalResult<()> {
+    if value.len() > maximum {
+        return Err(invalid("text exceeds storage byte limit"));
+    }
+    Ok(())
+}
+fn hash<T: Serialize>(value: &T) -> JournalResult<String> {
+    Ok(Sha256::digest(serde_json::to_vec(value)?)
+        .iter()
+        .map(|b| format!("{b:02x}"))
+        .collect())
+}
+fn page_limit(limit: usize) -> JournalResult<usize> {
+    if !(1..=MAX_JOURNAL_PAGE).contains(&limit) {
+        return Err(invalid("page limit must be 1..32"));
+    }
+    Ok(limit)
+}
+fn fit_page<T: Serialize>(rows: Vec<(String, T)>, mut more: bool) -> JournalResult<JournalPage<T>> {
+    let mut page = JournalPage {
+        items: Vec::new(),
+        next: None,
+    };
+    let mut last = None;
+    for (key, item) in rows {
+        page.items.push(item);
+        page.next = Some(key.clone());
+        if serde_json::to_string_pretty(&page)?.len() > MAX_JOURNAL_RESPONSE_BYTES {
+            page.items.pop();
+            if page.items.is_empty() {
+                return Err(invalid("single metadata item exceeds page budget"));
+            }
+            more = true;
+            break;
+        }
+        last = Some(key);
+    }
+    page.next = if more { last } else { None };
+    Ok(page)
+}
+fn check_host(
+    store: &SubjektivStore,
+    tx: &Connection,
+    host: &JournalHostContext,
+    write: bool,
+) -> JournalResult<()> {
+    if host.scope_id != store.scope_id() {
+        return Err(JournalError::PermissionDenied(
+            "different storage scope".into(),
+        ));
+    }
+    if write {
+        let state: String = tx.query_row(
+            "SELECT state FROM subjects WHERE subject_id=?1",
+            [&host.subject_id],
+            |r| r.get(0),
+        )?;
+        if state != "active" {
+            return Err(JournalError::PermissionDenied("Subject is retired".into()));
+        }
+        let attr = host
+            .attribution
+            .as_ref()
+            .ok_or_else(|| JournalError::PermissionDenied("Session authority required".into()))?;
+        let raw: Option<String> = tx.query_row("SELECT record_json FROM subject_session_attributions WHERE session_id=?1 AND subject_id=?2", params![attr.session_id, host.subject_id], |r| r.get(0)).optional()?;
+        if raw
+            .map(|s| serde_json::from_str::<SubjectSessionAttribution>(&s))
+            .transpose()?
+            .as_ref()
+            != Some(attr)
+        {
+            return Err(JournalError::PermissionDenied(
+                "Session attribution mismatch".into(),
+            ));
+        }
+    }
+    Ok(())
+}
+fn journal(
+    tx: &Connection,
+    host: &JournalHostContext,
+    id: &str,
+    write: bool,
+) -> JournalResult<Journal> {
+    let j = tx.query_row("SELECT id, subject_id, source_session_id, title, current_position, process_order, process_byte, state_token, created_at, updated_at FROM journals WHERE id=?1 AND subject_id=?2", params![id, host.subject_id], |r| Ok(Journal { id:r.get(0)?, subject_id:r.get(1)?, source_session_id:r.get(2)?, title:r.get(3)?, current_position:r.get(4)?, process_order:r.get(5)?, process_byte:r.get(6)?, state_token:r.get(7)?, created_at:r.get(8)?, updated_at:r.get(9)? })).optional()?.ok_or_else(|| JournalError::NotFound(id.into()))?;
+    if write
+        && host.attribution.as_ref().map(|a| a.session_id.as_str())
+            != Some(j.source_session_id.as_str())
+    {
+        return Err(JournalError::PermissionDenied(
+            "different source Session".into(),
+        ));
+    }
+    Ok(j)
+}
+fn entry(tx: &Connection, journal_id: &str, id: &str) -> JournalResult<JournalEntry> {
+    let row = tx.query_row("SELECT id, journal_id, position, state, title, body, ongoing, source_end_json, created_at, updated_at, fixed_at FROM journal_entries WHERE journal_id=?1 AND id=?2", params![journal_id, id], |r| Ok((r.get::<_,String>(0)?,r.get::<_,String>(1)?,r.get::<_,u64>(2)?,r.get::<_,String>(3)?,r.get::<_,String>(4)?,r.get::<_,String>(5)?,r.get::<_,String>(6)?,r.get::<_,Option<String>>(7)?,r.get::<_,String>(8)?,r.get::<_,String>(9)?,r.get::<_,Option<String>>(10)?))).optional()?.ok_or_else(|| JournalError::NotFound(id.into()))?;
+    let mut stmt = tx.prepare(
+        "SELECT target_id FROM journal_corrections WHERE entry_id=?1 ORDER BY target_id",
+    )?;
+    let correction_refs = stmt
+        .query_map([id], |r| r.get(0))?
+        .collect::<Result<Vec<String>, _>>()?;
+    Ok(JournalEntry {
+        id: row.0,
+        journal_id: row.1,
+        position: row.2,
+        state: if row.3 == "open" {
+            JournalEntryState::Open
+        } else {
+            JournalEntryState::Fixed
+        },
+        title: row.4,
+        body: row.5,
+        ongoing: row.6,
+        source_end: row.7.map(|s| serde_json::from_str(&s)).transpose()?,
+        created_at: row.8,
+        updated_at: row.9,
+        fixed_at: row.10,
+        correction_refs,
+    })
+}
+fn open(tx: &Connection, journal_id: &str) -> JournalResult<Option<JournalEntry>> {
+    let id: Option<String> = tx
+        .query_row(
+            "SELECT id FROM journal_entries WHERE journal_id=?1 AND state='open'",
+            [journal_id],
+            |r| r.get(0),
+        )
+        .optional()?;
+    id.map(|id| entry(tx, journal_id, &id)).transpose()
+}
+fn previous_fixed(
+    tx: &Connection,
+    journal_id: &str,
+) -> JournalResult<Option<(JournalSourcePointer, u64)>> {
+    let raw: Option<(String,u64)> = tx.query_row("SELECT source_end_json, source_order FROM journal_entries WHERE journal_id=?1 AND state='fixed' ORDER BY position DESC LIMIT 1", [journal_id], |r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+    raw.map(|(s, o)| Ok((serde_json::from_str(&s)?, o)))
+        .transpose()
+}
+fn refresh_state(tx: &Connection, j: &mut Journal) -> JournalResult<()> {
+    let tail: Option<String> = tx
+        .query_row(
+            "SELECT id FROM journal_entries WHERE journal_id=?1 ORDER BY position DESC LIMIT 1",
+            [&j.id],
+            |r| r.get(0),
+        )
+        .optional()?;
+    let tail = tail.map(|id| entry(tx, &j.id, &id)).transpose()?;
+    j.state_token = hash(&(
+        &j.state_token,
+        &j.id,
+        &j.title,
+        &j.current_position,
+        j.process_order,
+        j.process_byte,
+        tail,
+    ))?;
+    j.updated_at = now();
+    tx.execute("UPDATE journals SET current_position=?2, process_order=?3, process_byte=?4, state_token=?5, updated_at=?6 WHERE id=?1", params![j.id,j.current_position,j.process_order,j.process_byte,j.state_token,j.updated_at])?;
+    Ok(())
+}
+fn check_source(range: &JournalSourceRange, session: &str) -> JournalResult<()> {
+    let s = &range.source;
+    label(&s.session_id)?;
+    let suffix = s
+        .pointer
+        .entry_ref
+        .strip_prefix('E')
+        .ok_or_else(|| invalid("source entry reference must be a public E reference"))?;
+    if suffix.is_empty()
+        || suffix.len() > 64
+        || !suffix
+            .bytes()
+            .all(|b| b.is_ascii_alphanumeric() || b == b'-')
+    {
+        return Err(invalid("invalid public entry reference"));
+    }
+    if s.session_id != session {
+        return Err(JournalError::PermissionDenied(
+            "source belongs to another Session".into(),
+        ));
+    }
+    if s.order >= i64::MAX as u64
+        || s.total_bytes == 0
+        || s.total_bytes > i64::MAX as u64
+        || range.bytes.start >= range.bytes.end
+        || range.bytes.end > s.total_bytes
+        || s.content_sha256.len() != 64
+        || !s.content_sha256.bytes().all(|b| b.is_ascii_hexdigit())
+    {
+        return Err(invalid("invalid source identity or half-open byte range"));
+    }
+    Ok(())
+}
+fn merge(mut ranges: Vec<JournalByteRange>) -> JournalResult<Vec<JournalByteRange>> {
+    ranges.sort_by_key(|r| r.start);
+    let mut out: Vec<JournalByteRange> = Vec::new();
+    for r in ranges {
+        if let Some(last) = out.last_mut().filter(|last| r.start <= last.end) {
+            last.end = last.end.max(r.end);
+        } else {
+            out.push(r);
+        }
+    }
+    if out.len() > 256 {
+        return Err(invalid(
+            "too many disjoint ranges; admit bounded contiguous fragments",
+        ));
+    }
+    Ok(out)
+}
+fn contains(ranges: &[JournalByteRange], r: JournalByteRange) -> bool {
+    ranges.iter().any(|a| a.start <= r.start && a.end >= r.end)
+}
+fn overlaps(ranges: &[JournalByteRange], r: JournalByteRange) -> bool {
+    ranges.iter().any(|a| a.start < r.end && r.start < a.end)
+}
+fn source(tx: &Connection, journal_id: &str, order: u64) -> JournalResult<JournalSourceStatus> {
+    let raw:Option<(String,String,String,u64)>=tx.query_row("SELECT source_json,accepted_json,unavailable_json,reflected_bytes FROM journal_sources WHERE journal_id=?1 AND source_order=?2",params![journal_id,order],|r|Ok((r.get(0)?,r.get(1)?,r.get(2)?,r.get(3)?))).optional()?;
+    let (s, a, u, r) =
+        raw.ok_or_else(|| JournalError::NotFound(format!("source order {order}")))?;
+    Ok(JournalSourceStatus {
+        source: serde_json::from_str(&s)?,
+        accepted: serde_json::from_str(&a)?,
+        unavailable: serde_json::from_str(&u)?,
+        reflected_bytes: r,
+    })
+}
+fn check_ranges(
+    tx: &Connection,
+    j: &Journal,
+    ranges: &[JournalSourceRange],
+    available: bool,
+) -> JournalResult<()> {
+    if ranges.is_empty() || ranges.len() > MAX_JOURNAL_RANGES {
+        return Err(invalid("source range count must be 1..64"));
+    }
+    for r in ranges {
+        check_source(r, &j.source_session_id)?;
+        let stored = source(tx, &j.id, r.source.order)?;
+        if stored.source != r.source {
+            return Err(conflict("source identity differs from accepted history"));
+        }
+        if !contains(&stored.accepted, r.bytes) {
+            return Err(invalid("source range was not accepted"));
+        }
+        if available && overlaps(&stored.unavailable, r.bytes) {
+            return Err(conflict("selected history is unavailable"));
+        }
+    }
+    Ok(())
+}
+fn prefix(j: &Journal, ranges: &[JournalSourceRange]) -> JournalResult<(u64, u64)> {
+    let (mut order, mut byte) = (j.process_order, j.process_byte);
+    for r in ranges {
+        if r.source.order != order || r.bytes.start != byte {
+            return Err(invalid(
+                "cannot skip a missing, unread or unreflected source prefix",
+            ));
+        }
+        byte = r.bytes.end;
+        if byte == r.source.total_bytes {
+            order += 1;
+            byte = 0;
+        }
+    }
+    Ok((order, byte))
+}
+fn in_snapshot(snapshot: &JournalGeneration, ranges: &[JournalSourceRange]) -> JournalResult<()> {
+    if ranges.len() > MAX_JOURNAL_RANGES {
+        return Err(invalid("too many source ranges"));
+    }
+    for r in ranges {
+        let selected = merge(
+            snapshot
+                .selected_ranges
+                .iter()
+                .filter(|a| a.source == r.source)
+                .map(|a| a.bytes)
+                .collect(),
+        )?;
+        if !contains(&selected, r.bytes) {
+            return Err(invalid("operation exceeds immutable generation input"));
+        }
+    }
+    Ok(())
+}
+fn generation(
+    tx: &Connection,
+    host: &JournalHostContext,
+    id: &str,
+) -> JournalResult<(JournalGeneration, bool)> {
+    let raw:Option<(String,bool)>=tx.query_row("SELECT g.snapshot_json,g.active FROM journal_generations g JOIN journals j ON j.id=g.journal_id WHERE g.id=?1 AND j.subject_id=?2",params![id,host.subject_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+    let (raw, active) = raw.ok_or_else(|| JournalError::NotFound(id.into()))?;
+    let snapshot: JournalGeneration = serde_json::from_str(&raw)?;
+    journal(tx, host, &snapshot.journal_id, true)?;
+    Ok((snapshot, active))
+}
+fn job(
+    store: &SubjektivStore,
+    tx: &Connection,
+    ctx: &JournalJobContext,
+    active: bool,
+) -> JournalResult<JournalGeneration> {
+    check_host(store, tx, &ctx.host, true)?;
+    let (g, is_active) = generation(tx, &ctx.host, &ctx.generation_id)?;
+    if g.job != ctx.binding {
+        return Err(JournalError::PermissionDenied(
+            "different Job attempt".into(),
+        ));
+    }
+    if active && !is_active {
+        return Err(conflict("generation is closed or cancelled"));
+    }
+    Ok(g)
+}
+fn saved(
+    tx: &Connection,
+    journal_id: &str,
+    id: &str,
+) -> JournalResult<Option<JournalSavedOperation>> {
+    let raw:Option<(String,String)>=tx.query_row("SELECT request_json,receipt_json FROM journal_operations WHERE journal_id=?1 AND request_id=?2",params![journal_id,id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+    raw.map(|(a, b)| {
+        Ok(JournalSavedOperation {
+            operation: serde_json::from_str(&a)?,
+            receipt: serde_json::from_str(&b)?,
+        })
+    })
+    .transpose()
+}
+fn endpoint(
+    tx: &Connection,
+    j: &Journal,
+    g: &JournalGeneration,
+    p: &JournalSourcePointer,
+) -> JournalResult<u64> {
+    let r = g
+        .selected_ranges
+        .iter()
+        .find(|r| &r.source.pointer == p && r.bytes.end == r.source.total_bytes)
+        .ok_or_else(|| invalid("endpoint must identify a fully selected entry end"))?;
+    let s = source(tx, &j.id, r.source.order)?;
+    if s.source != r.source
+        || !contains(
+            &s.accepted,
+            JournalByteRange {
+                start: 0,
+                end: s.source.total_bytes,
+            },
+        )
+        || !s.unavailable.is_empty()
+    {
+        return Err(conflict("endpoint identity/availability differs"));
+    }
+    // The earlier bytes may have been reflected by an earlier generation, but cannot be unread.
+    let selected = merge(
+        g.selected_ranges
+            .iter()
+            .filter(|a| a.source == r.source)
+            .map(|a| a.bytes)
+            .collect(),
+    )?;
+    if !contains(
+        &selected,
+        JournalByteRange {
+            start: s.reflected_bytes,
+            end: s.source.total_bytes,
+        },
+    ) && s.reflected_bytes < s.source.total_bytes
+    {
+        return Err(invalid("endpoint entry has an unread byte prefix"));
+    }
+    if previous_fixed(tx, &j.id)?.is_some_and(|(_, order)| order >= r.source.order) {
+        return Err(invalid(
+            "fixed endpoint must advance canonical Session order",
+        ));
+    }
+    Ok(r.source.order)
+}
+
+impl SubjektivStore {
+    /// Admits Host-verified canonical fragments without generating text or starting a Job.
+    /// Duplicate/overlapping fragments merge. Identity replacement aborts the entire admission.
+    pub fn accept_journal_sources(
+        &self,
+        host: &JournalHostContext,
+        title: &str,
+        ranges: &[JournalSourceRange],
+    ) -> JournalResult<JournalId> {
+        text(title, 1024)?;
+        self.database.try_transaction(|tx| {
+            check_host(self,tx,host,true)?;
+            let session=&host.attribution.as_ref().expect("checked").session_id;
+            if ranges.is_empty() || ranges.len()>MAX_JOURNAL_RANGES {return Err(invalid("admission count must be 1..64"));}
+            let existing:Option<String>=tx.query_row("SELECT id FROM journals WHERE source_session_id=?1",[session],|r|r.get(0)).optional()?;
+            let id=if let Some(id)=existing {journal(tx,host,&id,true)?;id} else {
+                let id=issued_id("journal");let time=now();
+                tx.execute("INSERT INTO journals(id,subject_id,source_session_id,title,current_position,state_token,created_at,updated_at) VALUES(?1,?2,?3,?4,'','',?5,?5)",params![id,host.subject_id,session,title,time])?;
+                let mut j=journal(tx,host,&id,true)?;refresh_state(tx,&mut j)?;id
+            };
+            for r in ranges {
+                check_source(r,session)?;
+                let raw:Option<String>=tx.query_row("SELECT source_json FROM journal_sources WHERE journal_id=?1 AND source_order=?2",params![id,r.source.order],|r|r.get(0)).optional()?;
+                if let Some(raw)=raw {
+                    if serde_json::from_str::<JournalSource>(&raw)?!=r.source {return Err(conflict("same source order has a different identity (rewind/replacement)"));}
+                    let s=source(tx,&id,r.source.order)?;
+                    let mut accepted=s.accepted;accepted.push(r.bytes);
+                    tx.execute("UPDATE journal_sources SET accepted_json=?3 WHERE journal_id=?1 AND source_order=?2",params![id,r.source.order,serde_json::to_string(&merge(accepted)?)?])?;
+                } else {
+                    let pointer=serde_json::to_string(&r.source.pointer)?;
+                    if tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_sources WHERE journal_id=?1 AND (pointer_json=?2 OR entry_ref=?3))",params![id,pointer,r.source.pointer.entry_ref],|r|r.get::<_,bool>(0))? {return Err(conflict("canonical entry already admitted at a different order"));}
+                    tx.execute("INSERT INTO journal_sources(journal_id,source_order,pointer_json,entry_ref,source_json,accepted_json,unavailable_json) VALUES(?1,?2,?3,?4,?5,?6,'[]')",params![id,r.source.order,pointer,r.source.pointer.entry_ref,serde_json::to_string(&r.source)?,serde_json::to_string(&vec![r.bytes])?])?;
+                }
+            }
+            Ok(id)
+        })
+    }
+    /// Availability is Host-attested and separate from reflection; disappearance never marks input processed.
+    pub fn mark_journal_sources_unavailable(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+        ranges: &[JournalSourceRange],
+    ) -> JournalResult<()> {
+        self.database.try_transaction(|tx| {
+            check_host(self,tx,host,true)?;let j=journal(tx,host,id,true)?;check_ranges(tx,&j,ranges,false)?;
+            for r in ranges {let s=source(tx,id,r.source.order)?;let mut u=s.unavailable;u.push(r.bytes);tx.execute("UPDATE journal_sources SET unavailable_json=?3 WHERE journal_id=?1 AND source_order=?2",params![id,r.source.order,serde_json::to_string(&merge(u)?)?])?;}
+            Ok(())
+        })
+    }
+    /// Recovery after a Host has reacquired the same canonical bytes. This changes availability,
+    /// never their identity, accepted coverage, or reflected cursor.
+    pub fn mark_journal_sources_available(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+        ranges: &[JournalSourceRange],
+    ) -> JournalResult<()> {
+        self.database.try_transaction(|tx| {
+            check_host(self,tx,host,true)?;let j=journal(tx,host,id,true)?;check_ranges(tx,&j,ranges,false)?;
+            for r in ranges {
+                let s=source(tx,id,r.source.order)?;let mut remaining=Vec::new();
+                for u in s.unavailable {
+                    if u.start<r.bytes.start {remaining.push(JournalByteRange {start:u.start,end:u.end.min(r.bytes.start)});}
+                    if u.end>r.bytes.end {remaining.push(JournalByteRange {start:u.start.max(r.bytes.end),end:u.end});}
+                }
+                tx.execute("UPDATE journal_sources SET unavailable_json=?3 WHERE journal_id=?1 AND source_order=?2",params![id,r.source.order,serde_json::to_string(&merge(remaining)?)?])?;
+            }
+            Ok(())
+        })
+    }
+    pub fn active_journal_generation(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+    ) -> JournalResult<Option<JournalGeneration>> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, true)?;
+            journal(tx, host, id, true)?;
+            let generation_id: Option<String> = tx
+                .query_row(
+                    "SELECT id FROM journal_generations WHERE journal_id=?1 AND active=1",
+                    [id],
+                    |r| r.get(0),
+                )
+                .optional()?;
+            generation_id
+                .map(|id| Ok(generation(tx, host, &id)?.0))
+                .transpose()
+        })
+    }
+    pub fn prepare_journal_generation(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+        binding: JobAttemptBinding,
+        selected_ranges: Vec<JournalSourceRange>,
+    ) -> JournalResult<JournalGeneration> {
+        label(&binding.job_id)?;
+        label(&binding.attempt_id)?;
+        self.database.try_transaction(|tx| {
+            check_host(self,tx,host,true)?;let j=journal(tx,host,id,true)?;
+            let request=serde_json::to_string(&selected_ranges)?;
+            let existing:Option<(String,String)>=tx.query_row("SELECT request_json,snapshot_json FROM journal_generations WHERE journal_id=?1 AND job_id=?2 AND attempt_id=?3",params![id,binding.job_id,binding.attempt_id],|r|Ok((r.get(0)?,r.get(1)?))).optional()?;
+            if let Some((old,snapshot))=existing {if old!=request {return Err(conflict("Job attempt already bound to different source selection"));}return Ok(serde_json::from_str(&snapshot)?);}
+            if tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_generations WHERE journal_id=?1 AND active=1)",[id],|r|r.get::<_,bool>(0))? {return Err(conflict("another active generation owns this Journal"));}
+            check_ranges(tx,&j,&selected_ranges,true)?;prefix(&j,&selected_ranges)?;
+            let snapshot=JournalGeneration {id:issued_id("journal-generation"),journal_id:id.into(),job:binding,selected_ranges,initial_journal:j,initial_open:open(tx,id)?,previous_fixed_end:previous_fixed(tx,id)?.map(|p|p.0)};
+            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![snapshot.id,id,snapshot.job.job_id,snapshot.job.attempt_id,request,serde_json::to_string(&snapshot)?])?;
+            Ok(snapshot)
+        })
+    }
+    pub(super) fn validate_journal_job(&self, ctx: &JournalJobContext) -> JournalResult<()> {
+        self.database.try_with_connection(|tx| {
+            job(self, tx, ctx, true)?;
+            Ok(())
+        })
+    }
+    /// Finish/cancel are Host lifecycle decisions. Both fence new operations, preserve successful receipts and do not fix open entries.
+    pub fn close_journal_generation(
+        &self,
+        host: &JournalHostContext,
+        generation_id: &str,
+        binding: &JobAttemptBinding,
+    ) -> JournalResult<()> {
+        self.database.try_transaction(|tx| {
+            check_host(self, tx, host, true)?;
+            let (g, _) = generation(tx, host, generation_id)?;
+            if &g.job != binding {
+                return Err(JournalError::PermissionDenied(
+                    "different Job attempt".into(),
+                ));
+            }
+            tx.execute(
+                "UPDATE journal_generations SET active=0 WHERE id=?1",
+                [generation_id],
+            )?;
+            Ok(())
+        })
+    }
+    pub fn read_journal_generation(
+        &self,
+        host: &JournalHostContext,
+        generation_id: &str,
+    ) -> JournalResult<JournalGeneration> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, true)?;
+            Ok(generation(tx, host, generation_id)?.0)
+        })
+    }
+    /// The immutable snapshot is for Host input construction, capped by the text/range storage limits.
+    /// User/model readers use the metadata and UTF-8 text pages instead of this full Host snapshot.
+    pub fn journal_resume_state(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+    ) -> JournalResult<(Journal, Option<JournalEntry>)> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, true)?;
+            Ok((journal(tx, host, id, true)?, open(tx, id)?))
+        })
+    }
+    pub fn execute_journal(
+        &self,
+        ctx: &JournalJobContext,
+        operation: JournalOperation,
+    ) -> JournalResult<JournalReceipt> {
+        label(&operation.request_id)?;
+        self.database.try_transaction(|tx| {
+            let g=job(self,tx,ctx,false)?;
+            if let Some(old)=saved(tx,&g.journal_id,&operation.request_id)? {
+                if old.receipt.generation_id!=g.id || old.operation!=operation {return Err(conflict("operation identifier already used with different input or generation"));}
+                return Ok(old.receipt);
+            }
+            job(self,tx,ctx,true)?;
+            let mut j=journal(tx,&ctx.host,&g.journal_id,true)?;
+            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)?;}
+            let mut entry_id=None;
+            match &operation.action {
+                JournalAction::Create {title,body,ongoing,state,source_end,correction_refs}=> {
+                    text(title,1024)?;text(body,MAX_JOURNAL_TEXT_BYTES)?;text(ongoing,MAX_JOURNAL_TEXT_BYTES)?;
+                    if correction_refs.len()>16 {return Err(invalid("at most 16 correction targets"));}
+                    if open(tx,&j.id)?.is_some() {return Err(conflict("fix existing open before appending"));}
+                    let source_order=match (state,source_end) {
+                        (JournalEntryState::Open,None)=>None,
+                        (JournalEntryState::Fixed,Some(p)) if ongoing.is_empty()=>Some(endpoint(tx,&j,&g,p)?),
+                        _=>return Err(invalid("open must have no endpoint; fixed requires endpoint and empty ongoing")),
+                    };
+                    let id=issued_id("journal-entry");let time=now();
+                    let position:u64=tx.query_row("SELECT COALESCE(MAX(position)+1,0) FROM journal_entries WHERE journal_id=?1",[&j.id],|r|r.get(0))?;
+                    tx.execute("INSERT INTO journal_entries(id,subject_id,journal_id,position,state,title,body,ongoing,source_end_json,source_order,created_at,updated_at,fixed_at) VALUES(?1,?2,?3,?4,?5,?6,?7,?8,?9,?10,?11,?11,?12)",params![id,j.subject_id,j.id,position,if *state==JournalEntryState::Open {"open"} else {"fixed"},title,body,ongoing,source_end.as_ref().map(serde_json::to_string).transpose()?,source_order,time,source_order.map(|_|time.clone())])?;
+                    for target in correction_refs {
+                        if !tx.query_row("SELECT EXISTS(SELECT 1 FROM journal_entries WHERE subject_id=?1 AND id=?2)",params![j.subject_id,target],|r|r.get::<_,bool>(0))? {return Err(JournalError::PermissionDenied("correction target is outside Subject or missing".into()));}
+                        tx.execute("INSERT INTO journal_corrections(subject_id,entry_id,target_id) VALUES(?1,?2,?3)",params![j.subject_id,id,target])?;
+                    }
+                    entry_id=Some(id);
+                }
+                JournalAction::Rewrite {entry_id:id,title,body,ongoing}=> {
+                    text(title,1024)?;text(body,MAX_JOURNAL_TEXT_BYTES)?;text(ongoing,MAX_JOURNAL_TEXT_BYTES)?;
+                    let e=entry(tx,&j.id,id)?;if e.state!=JournalEntryState::Open {return Err(conflict("fixed entry cannot be rewritten"));}
+                    tx.execute("UPDATE journal_entries SET title=?2,body=?3,ongoing=?4,updated_at=?5 WHERE id=?1",params![id,title,body,ongoing,now()])?;entry_id=Some(id.clone());
+                }
+                JournalAction::Fix {entry_id:id,source_end}=> {
+                    let e=entry(tx,&j.id,id)?;if e.state!=JournalEntryState::Open {return Err(conflict("only open entry can be fixed"));}
+                    let order=endpoint(tx,&j,&g,source_end)?;let time=now();
+                    tx.execute("UPDATE journal_entries SET state='fixed',ongoing='',source_end_json=?2,source_order=?3,updated_at=?4,fixed_at=?4 WHERE id=?1",params![id,serde_json::to_string(source_end)?,order,time])?;entry_id=Some(id.clone());
+                }
+                JournalAction::UpdatePosition {current_position}=> {
+                    if !operation.covered_ranges.is_empty() {return Err(invalid("position update cannot attest reflected history"));}
+                    text(current_position,MAX_JOURNAL_TEXT_BYTES)?;j.current_position=current_position.clone();
+                }
+                JournalAction::ConfirmUnchanged=> {if operation.covered_ranges.is_empty() {return Err(invalid("no-change confirmation must name actually checked input"));}}
+                JournalAction::Checkpoint {record_operation_ids}=> {
+                    check_ranges(tx,&j,&operation.covered_ranges,true)?;
+                    let (order,byte)=prefix(&j,&operation.covered_ranges)?;
+                    if record_operation_ids.is_empty() || record_operation_ids.len()>MAX_JOURNAL_RANGES {return Err(invalid("checkpoint requires 1..64 saved record receipts"));}
+                    let mut evidence:Vec<JournalSourceRange>=Vec::new();
+                    for id in record_operation_ids {
+                        label(id)?;
+                        let saved=saved(tx,&j.id,id)?.ok_or_else(||JournalError::NotFound(id.clone()))?;
+                        if matches!(saved.operation.action,JournalAction::UpdatePosition {..}|JournalAction::Checkpoint {..}) {return Err(invalid("receipt is not a saved record or no-change confirmation"));}
+                        evidence.extend(saved.operation.covered_ranges);
+                    }
+                    for r in &operation.covered_ranges {
+                        let merged=merge(evidence.iter().filter(|a|a.source==r.source).map(|a|a.bytes).collect())?;
+                        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])?;
+                    }
+                    j.process_order=order;j.process_byte=byte;
+                }
+            }
+            refresh_state(tx,&mut j)?;
+            let receipt=JournalReceipt {request_id:operation.request_id.clone(),generation_id:g.id,entry_id,state_token:j.state_token,process_order:j.process_order,process_byte:j.process_byte};
+            tx.execute("INSERT INTO journal_operations(journal_id,request_id,generation_id,request_json,receipt_json) VALUES(?1,?2,?3,?4,?5)",params![j.id,operation.request_id,receipt.generation_id,serde_json::to_string(&operation)?,serde_json::to_string(&receipt)?])?;
+            Ok(receipt)
+        })
+    }
+    /// The checkpoint has the same retry identity rules as every individual record operation.
+    pub fn checkpoint_journal(
+        &self,
+        ctx: &JournalJobContext,
+        operation: JournalOperation,
+    ) -> JournalResult<JournalReceipt> {
+        if !matches!(operation.action, JournalAction::Checkpoint { .. }) {
+            return Err(invalid("checkpoint API requires a checkpoint action"));
+        }
+        self.execute_journal(ctx, operation)
+    }
+    /// Bounded metadata about the saved effect. Coverage has its own immutable pages, so
+    /// restart recovery need not retrieve or retain the original model-generated body.
+    pub fn read_journal_operation_status(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        id: &str,
+    ) -> JournalResult<JournalOperationStatus> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            journal(tx, host, journal_id, false)?;
+            let saved =
+                saved(tx, journal_id, id)?.ok_or_else(|| JournalError::NotFound(id.into()))?;
+            let kind = match saved.operation.action {
+                JournalAction::Create { .. } => JournalOperationKind::Create,
+                JournalAction::Rewrite { .. } => JournalOperationKind::Rewrite,
+                JournalAction::Fix { .. } => JournalOperationKind::Fix,
+                JournalAction::UpdatePosition { .. } => JournalOperationKind::UpdatePosition,
+                JournalAction::ConfirmUnchanged => JournalOperationKind::ConfirmUnchanged,
+                JournalAction::Checkpoint { .. } => JournalOperationKind::Checkpoint,
+            };
+            Ok(JournalOperationStatus {
+                receipt: saved.receipt,
+                kind,
+                covered_range_count: saved.operation.covered_ranges.len(),
+            })
+        })
+    }
+    pub fn read_journal_operation_coverage(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        id: &str,
+        offset: usize,
+        limit: usize,
+    ) -> JournalResult<JournalPage<JournalSourceRange>> {
+        let limit = page_limit(limit)?;
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            journal(tx, host, journal_id, false)?;
+            let saved =
+                saved(tx, journal_id, id)?.ok_or_else(|| JournalError::NotFound(id.into()))?;
+            let len = saved.operation.covered_ranges.len();
+            if offset > len {
+                return Err(invalid("coverage offset exceeds saved operation"));
+            }
+            let rows = saved
+                .operation
+                .covered_ranges
+                .into_iter()
+                .enumerate()
+                .skip(offset)
+                .take(limit)
+                .map(|(i, r)| ((i + 1).to_string(), r))
+                .collect();
+            fit_page(rows, len - offset > limit)
+        })
+    }
+    /// Receipt lookup remains possible after cancellation and Session deletion. No new mutation is authorized by a receipt.
+    pub fn query_journal_operation(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        id: &str,
+    ) -> JournalResult<Option<JournalReceipt>> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            journal(tx, host, journal_id, false)?;
+            Ok(saved(tx, journal_id, id)?.map(|s| s.receipt))
+        })
+    }
+    pub fn list_journal_operations(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        after: Option<&str>,
+        limit: usize,
+    ) -> JournalResult<JournalPage<JournalReceipt>> {
+        let limit = page_limit(limit)?;
+        self.database.try_with_connection(|tx| {
+            check_host(self,tx,host,false)?;journal(tx,host,journal_id,false)?;
+            // Immutable request IDs are not chronological; the durable SQLite rowid provides keyset order.
+            let after=after.map(str::parse::<i64>).transpose().map_err(|_|invalid("invalid operation cursor"))?.unwrap_or(0);
+            let mut stmt=tx.prepare("SELECT rowid,receipt_json FROM journal_operations WHERE journal_id=?1 AND rowid>?2 ORDER BY rowid LIMIT ?3")?;
+            let rows=stmt.query_map(params![journal_id,after,limit+1],|r|Ok((r.get::<_,i64>(0)?,r.get::<_,String>(1)?)))?.collect::<Result<Vec<_>,_>>()?;
+            let more=rows.len()>limit;let mut items=Vec::new();
+            for (id,raw) in rows.into_iter().take(limit) {items.push((id.to_string(),serde_json::from_str::<JournalReceipt>(&raw)?));}
+            fit_page(items,more)
+        })
+    }
+    /// Coverage changes can fill an earlier gap or extend an already listed entry. Continuation
+    /// pins the complete ledger content condition, rejecting changes instead of losing those bytes.
+    pub fn list_journal_sources(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        cursor: Option<&str>,
+        limit: usize,
+    ) -> JournalResult<JournalPage<JournalSourceStatus>> {
+        let limit = page_limit(limit)?;
+        self.database.try_transaction(|tx| {
+            check_host(self,tx,host,false)?;journal(tx,host,journal_id,false)?;
+            let fingerprint=source_fingerprint(tx,journal_id)?;
+            let after=if let Some(cursor)=cursor {
+                if cursor.len()>2048 {return Err(invalid("source cursor exceeds byte limit"));}
+                let cursor:SourcePageCursor=serde_json::from_str(cursor).map_err(|_|invalid("invalid source cursor"))?;
+                if cursor.scope_id!=host.scope_id || cursor.subject_id!=host.subject_id || cursor.journal_id!=journal_id || cursor.after_order>i64::MAX as u64 {return Err(invalid("source cursor belongs to a different scope/Subject/Journal or invalid order"));}
+                if cursor.source_fingerprint!=fingerprint {return Err(conflict("source coverage changed; restart pagination from current state"));}
+                Some(cursor.after_order)
+            } else {None};
+            let mut stmt=tx.prepare("SELECT source_order FROM journal_sources WHERE journal_id=?1 AND (?2 IS NULL OR source_order>?2) ORDER BY source_order LIMIT ?3")?;
+            let orders=stmt.query_map(params![journal_id,after,limit+1],|r|r.get::<_,u64>(0))?.collect::<Result<Vec<_>,_>>()?;
+            let more=orders.len()>limit;let mut items=Vec::new();
+            for order in orders.into_iter().take(limit) {
+                let next=serde_json::to_string(&SourcePageCursor {scope_id:host.scope_id.clone(),subject_id:host.subject_id.clone(),journal_id:journal_id.into(),source_fingerprint:fingerprint.clone(),after_order:order})?;
+                items.push((next,source(tx,journal_id,order)?));
+            }
+            fit_page(items,more)
+        })
+    }
+    /// Bounded point lookup for Host resume from `process_order`, without scanning reflected pages.
+    pub fn read_journal_source(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        order: u64,
+    ) -> JournalResult<JournalSourceStatus> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            journal(tx, host, journal_id, false)?;
+            source(tx, journal_id, order)
+        })
+    }
+    /// Returns metadata only. IDs and entry positions are append-stable keysets, so appends/rewrites cannot shift unread pages.
+    pub fn list_journals(
+        &self,
+        host: &JournalHostContext,
+        after: Option<&str>,
+        limit: usize,
+    ) -> JournalResult<JournalPage<JournalMetadata>> {
+        let limit = page_limit(limit)?;
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            let mut stmt = tx.prepare(
+                "SELECT id FROM journals WHERE subject_id=?1 AND id>?2 ORDER BY id LIMIT ?3",
+            )?;
+            let ids = stmt
+                .query_map(
+                    params![host.subject_id, after.unwrap_or(""), limit + 1],
+                    |r| r.get::<_, String>(0),
+                )?
+                .collect::<Result<Vec<_>, _>>()?;
+            let more = ids.len() > limit;
+            let mut items = Vec::new();
+            for id in ids.into_iter().take(limit) {
+                items.push((id.clone(), metadata(tx, host, &id)?));
+            }
+            fit_page(items, more)
+        })
+    }
+    pub fn read_journal(
+        &self,
+        host: &JournalHostContext,
+        id: &str,
+    ) -> JournalResult<JournalMetadata> {
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            metadata(tx, host, id)
+        })
+    }
+    pub fn list_journal_entries(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        after: Option<u64>,
+        limit: usize,
+    ) -> JournalResult<JournalPage<JournalEntrySummary>> {
+        let limit = page_limit(limit)?;
+        self.database.try_with_connection(|tx| {
+            check_host(self,tx,host,false)?;journal(tx,host,journal_id,false)?;
+            let mut stmt=tx.prepare("SELECT id FROM journal_entries WHERE journal_id=?1 AND (?2 IS NULL OR position>?2) ORDER BY position LIMIT ?3")?;
+            let ids=stmt.query_map(params![journal_id,after,limit+1],|r|r.get::<_,String>(0))?.collect::<Result<Vec<_>,_>>()?;
+            let more=ids.len()>limit;let mut items=Vec::new();for id in ids.into_iter().take(limit) {let e=entry(tx,journal_id,&id)?;items.push((e.position.to_string(),summary(tx,e)?));}
+            fit_page(items,more)
+        })
+    }
+    /// Continuation must present the observed content token. A concurrent open/position rewrite
+    /// rejects stale offsets rather than returning a torn document or skipping unread bytes.
+    pub fn read_journal_text(
+        &self,
+        host: &JournalHostContext,
+        journal_id: &str,
+        entry_id: Option<&str>,
+        field: JournalTextField,
+        expected_token: Option<&str>,
+        byte_offset: usize,
+        max_bytes: usize,
+    ) -> JournalResult<JournalTextPage> {
+        if !(4..=MAX_JOURNAL_READ_BYTES).contains(&max_bytes) {
+            return Err(invalid("text page must be 4..16384 bytes"));
+        }
+        self.database.try_with_connection(|tx| {
+            check_host(self, tx, host, false)?;
+            let j = journal(tx, host, journal_id, false)?;
+            let (body, token) = match (field, entry_id) {
+                (JournalTextField::CurrentPosition, None) => (j.current_position, j.state_token),
+                (JournalTextField::Body, Some(id)) => {
+                    let e = entry(tx, journal_id, id)?;
+                    let token = hash(&e)?;
+                    (e.body, token)
+                }
+                (JournalTextField::Ongoing, Some(id)) => {
+                    let e = entry(tx, journal_id, id)?;
+                    let token = hash(&e)?;
+                    (e.ongoing, token)
+                }
+                _ => return Err(invalid("text field/entry selector mismatch")),
+            };
+            if (byte_offset > 0 && expected_token.is_none())
+                || expected_token.is_some_and(|t| t != token)
+            {
+                return Err(conflict(
+                    "text changed or continuation has no observed content token",
+                ));
+            }
+            if byte_offset > body.len() || !body.is_char_boundary(byte_offset) {
+                return Err(invalid("offset is not a UTF-8 byte boundary"));
+            }
+            let mut end = byte_offset.saturating_add(max_bytes).min(body.len());
+            while !body.is_char_boundary(end) {
+                end -= 1;
+            }
+            loop {
+                let page = JournalTextPage {
+                    text: body[byte_offset..end].into(),
+                    next_byte: (end < body.len()).then_some(end),
+                    content_token: token.clone(),
+                };
+                if serde_json::to_string_pretty(&page)?.len() <= MAX_JOURNAL_RESPONSE_BYTES {
+                    return Ok(page);
+                }
+                end = byte_offset + (end - byte_offset) / 2;
+                while !body.is_char_boundary(end) {
+                    end -= 1;
+                }
+                if end == byte_offset {
+                    return Err(invalid("text envelope exceeds response budget"));
+                }
+            }
+        })
+    }
+}
+fn metadata(
+    tx: &Connection,
+    host: &JournalHostContext,
+    id: &str,
+) -> JournalResult<JournalMetadata> {
+    let j = journal(tx, host, id, false)?;
+    Ok(JournalMetadata {
+        id: j.id,
+        subject_id: j.subject_id,
+        source_session_id: j.source_session_id,
+        title: j.title,
+        current_position_bytes: j.current_position.len(),
+        process_order: j.process_order,
+        process_byte: j.process_byte,
+        state_token: j.state_token,
+        created_at: j.created_at,
+        updated_at: j.updated_at,
+    })
+}
+fn summary(tx: &Connection, e: JournalEntry) -> JournalResult<JournalEntrySummary> {
+    let previous:Option<String>=tx.query_row("SELECT source_end_json FROM journal_entries WHERE journal_id=?1 AND position<?2 AND state='fixed' ORDER BY position DESC LIMIT 1",params![e.journal_id,e.position],|r|r.get(0)).optional()?;
+    let source_available = if e.source_end.is_some() {
+        let end: u64 = tx.query_row(
+            "SELECT source_order FROM journal_entries WHERE id=?1",
+            [&e.id],
+            |r| r.get(0),
+        )?;
+        let previous_order:Option<u64>=tx.query_row("SELECT source_order FROM journal_entries WHERE journal_id=?1 AND position<?2 AND state='fixed' ORDER BY position DESC LIMIT 1",params![e.journal_id,e.position],|r|r.get(0)).optional()?;
+        let start = previous_order.map_or(0, |o| o + 1);
+        let (count,missing):(u64,u64)=tx.query_row("SELECT COUNT(*),COALESCE(SUM(unavailable_json<>'[]'),0) FROM journal_sources WHERE journal_id=?1 AND source_order BETWEEN ?2 AND ?3",params![e.journal_id,start,end],|r|Ok((r.get(0)?,r.get(1)?)))?;
+        count == end - start + 1 && missing == 0
+    } else {
+        false
+    };
+    let token = hash(&e)?;
+    Ok(JournalEntrySummary {
+        id: e.id,
+        journal_id: e.journal_id,
+        position: e.position,
+        state: e.state,
+        title: e.title,
+        source_end: e.source_end,
+        previous_fixed_end: previous.map(|s| serde_json::from_str(&s)).transpose()?,
+        source_available,
+        correction_refs: e.correction_refs,
+        content_token: token,
+        created_at: e.created_at,
+        updated_at: e.updated_at,
+        fixed_at: e.fixed_at,
+    })
+}

crates/subjektiv/src/legacy_migrations.rs

@@ -953,6 +953,31 @@ mod tests {
             store.session_attribution("session-1").unwrap().unwrap(),
             stored_attribution
         );
+        let host = crate::journal::JournalHostContext::validated_session(
+            &store,
+            stored_attribution.clone(),
+        )
+        .unwrap();
+        let range = crate::journal::JournalSourceRange {
+            source: crate::journal::JournalSource {
+                session_id: stored_attribution.session_id.clone(),
+                pointer: crate::journal::JournalSourcePointer {
+                    segment_id: Uuid::now_v7(),
+                    entry_ref: "Eafter-v1-migration".into(),
+                },
+                order: 0,
+                content_sha256: "ab".repeat(32),
+                total_bytes: 8,
+            },
+            bytes: crate::journal::JournalByteRange { start: 0, end: 8 },
+        };
+        let journal_id = store
+            .accept_journal_sources(&host, "After oldest supported migration", &[range])
+            .unwrap();
+        assert_eq!(
+            store.read_journal(&host, &journal_id).unwrap().subject_id,
+            subject_id
+        );
     }
 
     #[test]

crates/subjektiv/src/lib.rs

@@ -2,6 +2,7 @@
 
 pub mod api;
 pub mod job;
+pub mod journal;
 mod legacy_migrations;
 mod schema;
 use legacy_migrations::{
@@ -602,6 +603,11 @@ static MIGRATIONS: &[FeatureMigration] = &[
         "replace counters with immutable changes and content conditions",
         schema::migrate,
     ),
+    FeatureMigration::new(
+        8,
+        "add durable journal operations and processing ranges",
+        journal::migrate,
+    ),
 ];
 
 pub const REGISTRATION: FeatureRegistration =