diff --git a/crates/subjektiv/README.md b/crates/subjektiv/README.md index b120b95e8..8c97b2eb7 100644 --- a/crates/subjektiv/README.md +++ b/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 diff --git a/crates/subjektiv/src/api.rs b/crates/subjektiv/src/api.rs index 185d55b5d..e8785e549 100644 --- a/crates/subjektiv/src/api.rs +++ b/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; diff --git a/crates/subjektiv/src/api/context.rs b/crates/subjektiv/src/api/context.rs index af88e3581..bfea8ad1f 100644 --- a/crates/subjektiv/src/api/context.rs +++ b/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, diff --git a/crates/subjektiv/src/api/operation.rs b/crates/subjektiv/src/api/operation.rs index 6b48371d1..834b322d4 100644 --- a/crates/subjektiv/src/api/operation.rs +++ b/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 { + 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 { + store.checkpoint_journal(context, operation) +} use server_api::{ SubjektivMemoryBackendOperation as Op, SubjektivMemoryBackendResponse as Response, }; diff --git a/crates/subjektiv/src/journal/mod.rs b/crates/subjektiv/src/journal/mod.rs new file mode 100644 index 000000000..d3cd57a80 --- /dev/null +++ b/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 = Result; +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 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, +} +impl JournalHostContext { + pub fn validated_reader(store: &SubjektivStore, subject_id: &str) -> JournalResult { + 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 { + 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 { + 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, + pub created_at: String, + pub updated_at: String, + pub fixed_at: Option, + pub correction_refs: Vec, +} +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JournalGeneration { + pub id: String, + pub journal_id: JournalId, + pub job: JobAttemptBinding, + pub selected_ranges: Vec, + pub initial_journal: Journal, + pub initial_open: Option, + pub previous_fixed_end: Option, +} +#[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, + correction_refs: Vec, + }, + 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, + }, +} +#[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, +} +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JournalReceipt { + pub request_id: String, + pub generation_id: String, + pub entry_id: Option, + 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, + pub unavailable: Vec, + pub reflected_bytes: u64, +} +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct JournalPage { + pub items: Vec, + pub next: Option, +} +#[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, + /// Exclusive start; None means Session start. The end is inclusive. + pub previous_fixed_end: Option, + pub source_available: bool, + pub correction_refs: Vec, + pub content_token: String, + pub created_at: String, + pub updated_at: String, + pub fixed_at: Option, +} +#[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, + pub content_token: String, +} + +pub(super) fn migrate(tx: &Transaction<'_>) -> feature_storage::Result<()> { + tx.execute_batch(include_str!("schema.sql"))?; + Ok(()) +} diff --git a/crates/subjektiv/src/journal/schema.sql b/crates/subjektiv/src/journal/schema.sql new file mode 100644 index 000000000..9ca59efdd --- /dev/null +++ b/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; diff --git a/crates/subjektiv/src/journal/store.rs b/crates/subjektiv/src/journal/store.rs new file mode 100644 index 000000000..a5b8d7dfa --- /dev/null +++ b/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 { + 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(value: &T) -> JournalResult { + Ok(Sha256::digest(serde_json::to_vec(value)?) + .iter() + .map(|b| format!("{b:02x}")) + .collect()) +} +fn page_limit(limit: usize) -> JournalResult { + if !(1..=MAX_JOURNAL_PAGE).contains(&limit) { + return Err(invalid("page limit must be 1..32")); + } + Ok(limit) +} +fn fit_page(rows: Vec<(String, T)>, mut more: bool) -> JournalResult> { + 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 = 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::(&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 { + 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 { + 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>(7)?,r.get::<_,String>(8)?,r.get::<_,String>(9)?,r.get::<_,Option>(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::, _>>()?; + 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> { + let id: Option = 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> { + 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 = 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) -> JournalResult> { + ranges.sort_by_key(|r| r.start); + let mut out: Vec = 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 { + 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 { + 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> { + 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 { + 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 { + 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=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=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::(&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.startr.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> { + self.database.try_with_connection(|tx| { + check_host(self, tx, host, true)?; + journal(tx, host, id, true)?; + let generation_id: Option = 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, + ) -> JournalResult { + 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 { + 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)> { + 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 { + 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=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 { + 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 { + 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> { + 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> { + 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> { + 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::).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::,_>>()?; + 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::(&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> { + 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::,_>>()?; + 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 { + 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> { + 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::, _>>()?; + 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 { + 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, + limit: usize, + ) -> JournalResult> { + 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::,_>>()?; + 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 { + 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 { + 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 { + let previous:Option=tx.query_row("SELECT source_end_json FROM journal_entries WHERE journal_id=?1 AND position=tx.query_row("SELECT source_order FROM journal_entries WHERE journal_id=?1 AND position'[]'),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, + }) +} diff --git a/crates/subjektiv/src/journal/tests.rs b/crates/subjektiv/src/journal/tests.rs new file mode 100644 index 000000000..433b44bf7 --- /dev/null +++ b/crates/subjektiv/src/journal/tests.rs @@ -0,0 +1,2623 @@ +//! Component tests through the Journal store boundary and production FeatureStorage migrations. +//! Only Host-attested Session fragments and Job bindings are supplied by the fixture; no Session +//! runtime or provider is started. Temporary storage owns cleanup. Receipt insertion faults are +//! deterministic SQLite triggers, not a claim of arbitrary process-crash durability. + +use super::*; +use crate::SubjectRole; +use feature_storage::{FeatureStorage, ScopedFeatureStorage}; +use sha2::{Digest, Sha256}; +use std::sync::{Arc, Barrier}; +use std::thread; + +struct Fixture { + store: Option, + workspace: Option, + manager: Option, + temp: tempfile::TempDir, + host: JournalHostContext, + journal_id: String, + ranges: Vec, +} + +impl Fixture { + fn new() -> Self { + Self::with_sizes(&[256, 256]) + } + + fn with_sizes(sizes: &[u64]) -> Self { + Self::with_schema(sizes, false) + } + + fn with_schema(sizes: &[u64], upgrade: bool) -> Self { + let temp = tempfile::tempdir().unwrap(); + let manager = FeatureStorage::new(temp.path().join("storage")); + let workspace = manager.workspace("journal-tests").unwrap(); + let registration = if upgrade { + workspace + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + &crate::MIGRATIONS[..crate::MIGRATIONS.len() - 1], + )) + .unwrap() + } else { + SubjektivStore::register(&workspace).unwrap() + }; + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + let subject = store + .create_subject_with_behavior( + SubjectRole::new("companion").unwrap(), + if upgrade { + "Preserve existing Subject behavior".into() + } else { + String::new() + }, + ) + .unwrap(); + let host = session_host(&store, &subject.id, "source-session"); + let (manager, workspace, store) = if upgrade { + let has_journals: bool = store.database.try_with_connection::<_, JournalError>(|connection| { + Ok(connection.query_row("SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='journals')", [], |row| row.get(0))?) + }).unwrap(); + assert!(!has_journals); + drop(store); + drop(workspace); + manager.shutdown().unwrap(); + drop(manager); + let manager = FeatureStorage::new(temp.path().join("storage")); + let workspace = manager.workspace("journal-tests").unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + assert_eq!(store.subject(&subject.id).unwrap(), Some(subject)); + assert_eq!( + store.session_attribution("source-session").unwrap(), + host.attribution + ); + (manager, workspace, store) + } else { + (manager, workspace, store) + }; + let ranges = sizes + .iter() + .enumerate() + .map(|(order, size)| source_range("source-session", order as u64, *size)) + .collect::>(); + let journal_id = store + .accept_journal_sources(&host, "Development journal", &ranges) + .unwrap(); + Self { + store: Some(store), + workspace: Some(workspace), + manager: Some(manager), + temp, + host, + journal_id, + ranges, + } + } + + fn store(&self) -> &SubjektivStore { + self.store.as_ref().unwrap() + } + + fn reopen(&mut self) { + self.store.take(); + self.workspace.take(); + self.manager.take().unwrap().shutdown().unwrap(); + let manager = FeatureStorage::new(self.temp.path().join("storage")); + let workspace = manager.workspace("journal-tests").unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + self.store = Some(SubjektivStore::open(&workspace, ®istration).unwrap()); + self.workspace = Some(workspace); + self.manager = Some(manager); + } + + fn state(&self) -> (Journal, Option) { + self.store() + .journal_resume_state(&self.host, &self.journal_id) + .unwrap() + } + + fn generation( + &self, + name: &str, + ranges: Vec, + ) -> (JournalGeneration, JournalJobContext) { + let binding = JobAttemptBinding { + job_id: name.into(), + attempt_id: "attempt-1".into(), + }; + let generation = self + .store() + .prepare_journal_generation(&self.host, &self.journal_id, binding.clone(), ranges) + .unwrap(); + let context = + JournalJobContext::validated(self.store(), &self.host, &generation.id, binding) + .unwrap(); + (generation, context) + } + + fn operation( + &self, + id: &str, + action: JournalAction, + ranges: Vec, + ) -> JournalOperation { + JournalOperation { + request_id: id.into(), + expected_state: self.state().0.state_token, + action, + covered_ranges: ranges, + } + } + + fn execute( + &self, + ctx: &JournalJobContext, + id: &str, + action: JournalAction, + ranges: Vec, + ) -> JournalReceipt { + self.store() + .execute_journal(ctx, self.operation(id, action, ranges)) + .unwrap() + } + + fn close(&self, generation: &JournalGeneration) { + self.store() + .close_journal_generation(&self.host, &generation.id, &generation.job) + .unwrap(); + } + + fn sources(&self) -> Vec { + self.store() + .list_journal_sources(&self.host, &self.journal_id, None, MAX_JOURNAL_PAGE) + .unwrap() + .items + } + + fn entries(&self) -> Vec { + self.store() + .list_journal_entries(&self.host, &self.journal_id, None, MAX_JOURNAL_PAGE) + .unwrap() + .items + } + + fn receipt(&self, id: &str) -> Option { + self.store() + .query_journal_operation(&self.host, &self.journal_id, id) + .unwrap() + } + + fn sql(&self, sql: &str) -> JournalResult<()> { + self.store().database.try_transaction(|tx| { + tx.execute_batch(sql)?; + Ok(()) + }) + } + + fn execute_and_reopen( + &mut self, + ctx: &JournalJobContext, + id: &str, + action: JournalAction, + ranges: Vec, + ) -> JournalReceipt { + let operation = self.operation(id, action, ranges); + let receipt = self + .store() + .execute_journal(ctx, operation.clone()) + .unwrap(); + let state = self.state(); + self.reopen(); + assert_eq!(self.state(), state); + assert_eq!(self.receipt(id), Some(receipt.clone())); + assert_eq!( + self.store().execute_journal(ctx, operation).unwrap(), + receipt + ); + receipt + } +} + +impl Drop for Fixture { + fn drop(&mut self) { + self.store.take(); + self.workspace.take(); + if let Some(manager) = self.manager.take() { + let _ = manager.shutdown(); + } + } +} + +fn session_host(store: &SubjektivStore, subject: &str, session: &str) -> JournalHostContext { + let attribution = + SubjectSessionAttribution::new(subject, "runtime", "worker", session).unwrap(); + store + .record_session_attribution(attribution.clone()) + .unwrap(); + JournalHostContext::validated_session(store, attribution).unwrap() +} + +fn source_range(session: &str, order: u64, total_bytes: u64) -> JournalSourceRange { + let digest = Sha256::digest(format!( + "canonical-public-entry:{session}:{order}:{total_bytes}" + )); + JournalSourceRange { + source: JournalSource { + session_id: session.into(), + pointer: JournalSourcePointer { + segment_id: Uuid::now_v7(), + entry_ref: format!("E{order}"), + }, + order, + content_sha256: digest.iter().map(|byte| format!("{byte:02x}")).collect(), + total_bytes, + }, + bytes: JournalByteRange { + start: 0, + end: total_bytes, + }, + } +} + +fn fragment(range: &JournalSourceRange, start: u64, end: u64) -> JournalSourceRange { + JournalSourceRange { + source: range.source.clone(), + bytes: JournalByteRange { start, end }, + } +} + +fn open_action(body: &str) -> JournalAction { + JournalAction::Create { + title: "Investigation".into(), + body: body.into(), + ongoing: "Verify the next boundary".into(), + state: JournalEntryState::Open, + source_end: None, + correction_refs: vec![], + } +} + +fn fixed_action(body: &str, range: &JournalSourceRange, corrections: Vec) -> JournalAction { + JournalAction::Create { + title: "Verified outcome".into(), + body: body.into(), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(range.source.pointer.clone()), + correction_refs: corrections, + } +} + +fn checkpoint(ids: &[&str]) -> JournalAction { + JournalAction::Checkpoint { + record_operation_ids: ids.iter().map(|id| (*id).into()).collect(), + } +} + +#[test] +fn sequential_open_rewrite_fix_and_correction_preserve_fixed_history() { + let fixture = Fixture::new(); + let (_, ctx) = fixture.generation("sequence", fixture.ranges.clone()); + let first = fixture.execute( + &ctx, + "create", + open_action("Initial finding"), + vec![fixture.ranges[0].clone()], + ); + let first_id = first.entry_id.unwrap(); + let before = fixture.state(); + let blocked = fixture.operation( + "append-before-fix", + fixed_action("Must not append", &fixture.ranges[1], vec![]), + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, blocked), + Err(JournalError::Conflict(_)) + )); + assert_eq!(fixture.state(), before); + assert!(fixture.receipt("append-before-fix").is_none()); + + fixture.execute( + &ctx, + "rewrite", + JournalAction::Rewrite { + entry_id: first_id.clone(), + title: "Evidence checked".into(), + body: "Revised finding".into(), + ongoing: "One final check".into(), + }, + vec![fixture.ranges[0].clone()], + ); + let rewritten = fixture.state().1.unwrap(); + assert_eq!(rewritten.id, first_id); + assert_eq!(rewritten.position, 0); + assert_eq!(rewritten.body, "Revised finding"); + fixture.execute( + &ctx, + "fix", + JournalAction::Fix { + entry_id: first_id.clone(), + source_end: fixture.ranges[0].source.pointer.clone(), + }, + vec![fixture.ranges[0].clone()], + ); + assert!(fixture.state().1.is_none()); + let fixed = fixture.entries()[0].clone(); + assert_eq!(fixed.state, JournalEntryState::Fixed); + assert!(fixed.fixed_at.is_some()); + assert!(fixed.previous_fixed_end.is_none()); + assert!(fixed.source_available); + assert_eq!(fixture.state().0.process_order, 0); + + let immutable = fixture.operation( + "rewrite-fixed", + JournalAction::Rewrite { + entry_id: first_id.clone(), + title: "Tampered".into(), + body: "Tampered".into(), + ongoing: String::new(), + }, + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, immutable), + Err(JournalError::Conflict(_)) + )); + let repeated_end = fixture.operation( + "non-advancing-end", + fixed_action( + "Cannot reuse the earlier endpoint", + &fixture.ranges[0], + vec![], + ), + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, repeated_end), + Err(JournalError::InvalidInput(_)) + )); + let missing_correction = fixture.operation( + "missing-correction", + fixed_action( + "No such target", + &fixture.ranges[1], + vec!["missing-entry".into()], + ), + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, missing_correction), + Err(JournalError::PermissionDenied(_)) + )); + assert_eq!(fixture.entries(), vec![fixed.clone()]); + assert!(fixture.receipt("missing-correction").is_none()); + let second = fixture.execute( + &ctx, + "correction", + fixed_action( + "The earlier finding needs qualification", + &fixture.ranges[1], + vec![first_id.clone()], + ), + vec![fixture.ranges[1].clone()], + ); + let entries = fixture.entries(); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0], fixed); + assert_eq!(entries[1].id, second.entry_id.unwrap()); + assert_eq!(entries[1].position, 1); + assert_eq!(entries[1].correction_refs, vec![first_id]); + assert_eq!(entries[1].previous_fixed_end, fixed.source_end); + assert_eq!(fixture.state().0.process_order, 0); + let cp = fixture.execute( + &ctx, + "checkpoint", + checkpoint(&["rewrite", "correction"]), + fixture.ranges.clone(), + ); + assert_eq!((cp.process_order, cp.process_byte), (2, 0)); +} + +#[test] +fn same_attempt_keeps_immutable_input_while_next_attempt_carries_current_state() { + let fixture = Fixture::new(); + let selected = vec![fixture.ranges[0].clone()]; + let (generation, ctx) = fixture.generation("first-job", selected.clone()); + fixture.execute( + &ctx, + "create", + open_action("Pending investigation"), + selected.clone(), + ); + fixture.execute( + &ctx, + "position", + JournalAction::UpdatePosition { + current_position: "Investigating receipt recovery".into(), + }, + vec![], + ); + fixture.execute( + &ctx, + "checkpoint", + checkpoint(&["create"]), + selected.clone(), + ); + let current = fixture.state(); + let new_source = source_range("source-session", 2, 400); + assert_eq!( + fixture + .store() + .accept_journal_sources( + &fixture.host, + "Ignored replacement title", + &[new_source.clone()] + ) + .unwrap(), + fixture.journal_id + ); + assert_eq!(fixture.state(), current); + assert_eq!( + fixture + .store() + .read_journal_generation(&fixture.host, &generation.id) + .unwrap(), + generation + ); + assert_eq!( + fixture + .store() + .prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + generation.job.clone(), + selected + ) + .unwrap(), + generation + ); + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + generation.job.clone(), + fixture.ranges.clone() + ), + Err(JournalError::Conflict(_)) + )); + let changed_input = fixture.operation( + "outside-snapshot", + JournalAction::ConfirmUnchanged, + vec![fixture.ranges[1].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, changed_input), + Err(JournalError::InvalidInput(_)) + )); + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + JobAttemptBinding { + job_id: "another-job".into(), + attempt_id: "attempt-1".into() + }, + vec![fixture.ranges[1].clone()] + ), + Err(JournalError::Conflict(_)) + )); + fixture.close(&generation); + let (next, _) = fixture.generation("next-job", vec![fixture.ranges[1].clone(), new_source]); + assert_eq!(next.initial_journal, current.0); + assert_eq!(next.initial_open, current.1); + assert!(next.previous_fixed_end.is_none()); + assert_eq!( + fixture + .store() + .read_journal_generation(&fixture.host, &generation.id) + .unwrap(), + generation + ); +} + +#[test] +fn lost_response_reopen_returns_exact_receipt_and_rejects_changed_retry() { + let mut fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("response-loss", fixture.ranges.clone()); + let operation = fixture.operation( + "stable-request", + open_action("Saved before response loss"), + vec![fixture.ranges[0].clone()], + ); + // The caller discards the successful response and recovers from the durable receipt. + fixture + .store() + .execute_journal(&ctx, operation.clone()) + .unwrap(); + fixture.reopen(); + let recovered = fixture.receipt("stable-request").unwrap(); + let state = fixture.state(); + assert_eq!( + fixture + .store() + .execute_journal(&ctx, operation.clone()) + .unwrap(), + recovered + ); + assert_eq!(fixture.state(), state); + assert_eq!(fixture.entries().len(), 1); + let mut changed = operation.clone(); + changed.action = open_action("Different body using the same request ID"); + assert!(matches!( + fixture.store().execute_journal(&ctx, changed), + Err(JournalError::Conflict(_)) + )); + let mut changed = operation.clone(); + changed.expected_state = recovered.state_token.clone(); + assert!(matches!( + fixture.store().execute_journal(&ctx, changed), + Err(JournalError::Conflict(_)) + )); + let mut changed = operation.clone(); + changed.covered_ranges = vec![fixture.ranges[1].clone()]; + assert!(matches!( + fixture.store().execute_journal(&ctx, changed), + Err(JournalError::Conflict(_)) + )); + fixture.close(&generation); + fixture.reopen(); + assert_eq!( + fixture + .store() + .execute_journal(&ctx, operation.clone()) + .unwrap(), + recovered + ); + let (_, replacement) = fixture.generation("replacement-job", fixture.ranges.clone()); + assert!(matches!( + fixture.store().execute_journal(&replacement, operation), + Err(JournalError::Conflict(_)) + )); + assert_eq!(fixture.receipt("stable-request"), Some(recovered)); + assert_eq!(fixture.state(), state); +} + +#[test] +fn cancellation_preserves_open_and_receipts_but_fences_late_writes_and_checkpoint() { + let mut fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("cancelled", fixture.ranges.clone()); + let saved = fixture.execute( + &ctx, + "create", + open_action("Saved but not reflected"), + fixture.ranges.clone(), + ); + let late = fixture.operation( + "late-checkpoint", + checkpoint(&["create"]), + fixture.ranges.clone(), + ); + let before = fixture.state(); + fixture.close(&generation); + fixture.close(&generation); + fixture.reopen(); + assert!(matches!( + fixture.store().execute_journal(&ctx, late), + Err(JournalError::Conflict(_)) + )); + let late_rewrite = fixture.operation( + "late-rewrite", + JournalAction::Rewrite { + entry_id: saved.entry_id.clone().unwrap(), + title: "Late".into(), + body: "Late".into(), + ongoing: String::new(), + }, + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, late_rewrite), + Err(JournalError::Conflict(_)) + )); + assert!(matches!( + JournalJobContext::validated( + fixture.store(), + &fixture.host, + &generation.id, + generation.job.clone() + ), + Err(JournalError::Conflict(_)) + )); + assert_eq!(fixture.state(), before); + assert_eq!(fixture.receipt("create"), Some(saved)); + assert!(fixture.receipt("late-checkpoint").is_none()); + assert!( + fixture + .sources() + .iter() + .all(|source| source.reflected_bytes == 0) + ); +} + +#[test] +fn stale_cas_is_rejected_even_after_position_returns_to_same_text() { + let mut fixture = Fixture::new(); + let (_, ctx) = fixture.generation("causal-state", fixture.ranges.clone()); + fixture.execute( + &ctx, + "position-a", + JournalAction::UpdatePosition { + current_position: "A".into(), + }, + vec![], + ); + let at_a = fixture.state().0; + let stale = fixture.operation( + "stale-create", + open_action("Must not apply against a later A"), + vec![], + ); + fixture.execute( + &ctx, + "position-b", + JournalAction::UpdatePosition { + current_position: "B".into(), + }, + vec![], + ); + fixture.execute( + &ctx, + "position-a-again", + JournalAction::UpdatePosition { + current_position: "A".into(), + }, + vec![], + ); + fixture.reopen(); + let current = fixture.state().0; + assert_eq!(current.current_position, at_a.current_position); + assert_ne!(current.state_token, at_a.state_token); + assert!(matches!( + fixture.store().execute_journal(&ctx, stale), + Err(JournalError::Conflict(_)) + )); + assert!(fixture.entries().is_empty()); + assert!(fixture.receipt("stale-create").is_none()); + assert_eq!(fixture.state().0, current); +} + +#[test] +fn foreign_scope_subject_session_job_and_journal_cannot_mutate_or_read_receipts() { + let fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("authorized-job", fixture.ranges.clone()); + let created = fixture.execute(&ctx, "create", open_action("Subject-owned record"), vec![]); + let entry_id = created.entry_id.unwrap(); + let before = fixture.state(); + let foreign_subject = fixture + .store() + .create_subject(SubjectRole::new("other").unwrap()) + .unwrap(); + let foreign_host = session_host(fixture.store(), &foreign_subject.id, "foreign-session"); + assert!(matches!( + fixture + .store() + .read_journal(&foreign_host, &fixture.journal_id), + Err(JournalError::NotFound(_)) + )); + assert!(matches!( + fixture + .store() + .query_journal_operation(&foreign_host, &fixture.journal_id, "create"), + Err(JournalError::NotFound(_)) + )); + assert!(matches!( + JournalJobContext::validated( + fixture.store(), + &foreign_host, + &generation.id, + generation.job.clone() + ), + Err(JournalError::NotFound(_)) + )); + let other_session = session_host(fixture.store(), &fixture.host.subject_id, "other-session"); + assert!(matches!( + fixture + .store() + .journal_resume_state(&other_session, &fixture.journal_id), + Err(JournalError::PermissionDenied(_)) + )); + assert!(matches!( + JournalJobContext::validated( + fixture.store(), + &other_session, + &generation.id, + generation.job.clone() + ), + Err(JournalError::PermissionDenied(_)) + )); + let wrong_attempt = JobAttemptBinding { + job_id: generation.job.job_id.clone(), + attempt_id: "attempt-2".into(), + }; + assert!(matches!( + JournalJobContext::validated( + fixture.store(), + &fixture.host, + &generation.id, + wrong_attempt.clone() + ), + Err(JournalError::PermissionDenied(_)) + )); + assert!(matches!( + fixture + .store() + .close_journal_generation(&fixture.host, &generation.id, &wrong_attempt), + Err(JournalError::PermissionDenied(_)) + )); + + let other_manager = FeatureStorage::new(fixture.temp.path().join("foreign-storage")); + let other_workspace = other_manager.workspace("foreign-scope").unwrap(); + let registration = SubjektivStore::register(&other_workspace).unwrap(); + let other_store = SubjektivStore::open(&other_workspace, ®istration).unwrap(); + assert!(matches!( + other_store.read_journal(&fixture.host, &fixture.journal_id), + Err(JournalError::PermissionDenied(_)) + )); + assert!(matches!( + other_store.execute_journal( + &ctx, + fixture.operation("wrong-scope", open_action("No"), vec![]) + ), + Err(JournalError::PermissionDenied(_)) + )); + let reader = + JournalHostContext::validated_reader(fixture.store(), &fixture.host.subject_id).unwrap(); + assert!(matches!( + fixture + .store() + .accept_journal_sources(&reader, "No authority", &fixture.ranges), + Err(JournalError::PermissionDenied(_)) + )); + let mut forged_attribution = fixture.host.attribution.clone().unwrap(); + forged_attribution.worker_id = "different-worker".into(); + assert!(matches!( + JournalHostContext::validated_session(fixture.store(), forged_attribution), + Err(JournalError::PermissionDenied(_)) + )); + + let other_range = source_range("other-session", 0, 256); + let other_id = fixture + .store() + .accept_journal_sources(&other_session, "Another journal", &[other_range.clone()]) + .unwrap(); + let other_generation = fixture + .store() + .prepare_journal_generation( + &other_session, + &other_id, + JobAttemptBinding { + job_id: "other-journal-job".into(), + attempt_id: "attempt-1".into(), + }, + vec![other_range], + ) + .unwrap(); + let other_ctx = JournalJobContext::validated( + fixture.store(), + &other_session, + &other_generation.id, + other_generation.job, + ) + .unwrap(); + let other_state = fixture + .store() + .journal_resume_state(&other_session, &other_id) + .unwrap(); + let cross_journal = JournalOperation { + request_id: "cross-journal-rewrite".into(), + expected_state: other_state.0.state_token.clone(), + action: JournalAction::Rewrite { + entry_id, + title: "No".into(), + body: "No".into(), + ongoing: "No".into(), + }, + covered_ranges: vec![], + }; + assert!(matches!( + fixture.store().execute_journal(&other_ctx, cross_journal), + Err(JournalError::NotFound(_)) + )); + assert_eq!( + fixture + .store() + .journal_resume_state(&other_session, &other_id) + .unwrap(), + other_state + ); + assert_eq!(fixture.state(), before); +} + +#[test] +fn corrections_allow_same_subject_other_journal_but_reject_foreign_subject_atomically() { + let fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("first", fixture.ranges.clone()); + let target = fixture + .execute( + &ctx, + "fixed", + fixed_action("Original record", &fixture.ranges[0], vec![]), + vec![], + ) + .entry_id + .unwrap(); + fixture.close(&generation); + let same_subject = session_host(fixture.store(), &fixture.host.subject_id, "later-session"); + let later_source = source_range("later-session", 0, 128); + let later_id = fixture + .store() + .accept_journal_sources(&same_subject, "Later journal", &[later_source.clone()]) + .unwrap(); + let later_generation = fixture + .store() + .prepare_journal_generation( + &same_subject, + &later_id, + JobAttemptBinding { + job_id: "later".into(), + attempt_id: "one".into(), + }, + vec![later_source.clone()], + ) + .unwrap(); + let later_ctx = JournalJobContext::validated( + fixture.store(), + &same_subject, + &later_generation.id, + later_generation.job, + ) + .unwrap(); + let operation = JournalOperation { + request_id: "cross-journal-correction".into(), + expected_state: later_generation.initial_journal.state_token, + action: fixed_action( + "Qualification from a later Session", + &later_source, + vec![target.clone()], + ), + covered_ranges: vec![later_source], + }; + fixture + .store() + .execute_journal(&later_ctx, operation) + .unwrap(); + let later_entries = fixture + .store() + .list_journal_entries(&same_subject, &later_id, None, 1) + .unwrap(); + assert_eq!(later_entries.items[0].correction_refs, vec![target.clone()]); + + let foreign_subject = fixture + .store() + .create_subject(SubjectRole::new("foreign").unwrap()) + .unwrap(); + let foreign_host = session_host(fixture.store(), &foreign_subject.id, "foreign-session"); + let foreign_source = source_range("foreign-session", 0, 128); + let foreign_id = fixture + .store() + .accept_journal_sources(&foreign_host, "Foreign", &[foreign_source.clone()]) + .unwrap(); + let foreign_generation = fixture + .store() + .prepare_journal_generation( + &foreign_host, + &foreign_id, + JobAttemptBinding { + job_id: "foreign".into(), + attempt_id: "one".into(), + }, + vec![foreign_source.clone()], + ) + .unwrap(); + let foreign_ctx = JournalJobContext::validated( + fixture.store(), + &foreign_host, + &foreign_generation.id, + foreign_generation.job, + ) + .unwrap(); + let rejected = JournalOperation { + request_id: "foreign-correction".into(), + expected_state: foreign_generation.initial_journal.state_token.clone(), + action: fixed_action("Forbidden correction", &foreign_source, vec![target]), + covered_ranges: vec![foreign_source], + }; + assert!(matches!( + fixture.store().execute_journal(&foreign_ctx, rejected), + Err(JournalError::PermissionDenied(_)) + )); + assert!( + fixture + .store() + .list_journal_entries(&foreign_host, &foreign_id, None, 1) + .unwrap() + .items + .is_empty() + ); + assert!( + fixture + .store() + .query_journal_operation(&foreign_host, &foreign_id, "foreign-correction") + .unwrap() + .is_none() + ); + assert_eq!( + fixture + .store() + .read_journal(&foreign_host, &foreign_id) + .unwrap() + .state_token, + foreign_generation.initial_journal.state_token + ); +} + +#[test] +fn retired_subject_keeps_reader_access_but_rejects_existing_write_context() { + let fixture = Fixture::new(); + let (_, ctx) = fixture.generation("retired-subject", fixture.ranges.clone()); + let receipt = fixture.execute(&ctx, "saved", open_action("Historical record"), vec![]); + let reader = + JournalHostContext::validated_reader(fixture.store(), &fixture.host.subject_id).unwrap(); + let late = fixture.operation( + "late", + JournalAction::UpdatePosition { + current_position: "No longer authorized".into(), + }, + vec![], + ); + fixture + .store() + .retire_subject(&fixture.host.subject_id) + .unwrap(); + assert!(matches!( + fixture.store().execute_journal(&ctx, late), + Err(JournalError::PermissionDenied(_)) + )); + assert_eq!( + fixture + .store() + .query_journal_operation(&reader, &fixture.journal_id, "saved") + .unwrap(), + Some(receipt) + ); + assert_eq!( + fixture + .store() + .list_journal_entries(&reader, &fixture.journal_id, None, 1) + .unwrap() + .items + .len(), + 1 + ); +} + +#[test] +fn concurrent_creates_from_independent_connections_save_exactly_one_open_and_receipt() { + let fixture = Fixture::new(); + let (_, ctx) = fixture.generation("concurrent", fixture.ranges.clone()); + let manager = FeatureStorage::new(fixture.temp.path().join("storage")); + let workspace = manager.workspace("journal-tests").unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + let second_store = SubjektivStore::open(&workspace, ®istration).unwrap(); + let barrier = Arc::new(Barrier::new(3)); + let first_op = fixture.operation( + "writer-a", + open_action("A"), + vec![fixture.ranges[0].clone()], + ); + let second_op = fixture.operation( + "writer-b", + open_action("B"), + vec![fixture.ranges[0].clone()], + ); + let spawn = |store: SubjektivStore, operation: JournalOperation| { + let barrier = barrier.clone(); + let ctx = ctx.clone(); + thread::spawn(move || { + barrier.wait(); + store.execute_journal(&ctx, operation) + }) + }; + let first = spawn(fixture.store().clone(), first_op); + let second = spawn(second_store, second_op); + barrier.wait(); + let results = [first.join().unwrap(), second.join().unwrap()]; + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!( + results + .iter() + .filter(|result| matches!(result, Err(JournalError::Conflict(_)))) + .count(), + 1 + ); + let winner = results.into_iter().find_map(Result::ok).unwrap(); + assert_eq!(fixture.entries().len(), 1); + assert_eq!(fixture.state().1.unwrap().id, winner.entry_id.unwrap()); + let receipts = fixture + .store() + .list_journal_operations(&fixture.host, &fixture.journal_id, None, 2) + .unwrap(); + assert_eq!(receipts.items.len(), 1); + assert_eq!(receipts.items[0].request_id, winner.request_id); + assert_eq!(fixture.state().0.process_order, 0); + drop(workspace); + manager.shutdown().unwrap(); +} + +#[test] +fn checkpoint_requires_saved_record_correspondence_and_never_position_only_evidence() { + let fixture = Fixture::new(); + let (_, ctx) = fixture.generation("checkpoint-evidence", fixture.ranges.clone()); + fixture.execute( + &ctx, + "position", + JournalAction::UpdatePosition { + current_position: "A summary is not a receipt for source reflection".into(), + }, + vec![], + ); + let before = fixture.state(); + let position_only = fixture.operation( + "position-checkpoint", + checkpoint(&["position"]), + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, position_only), + Err(JournalError::InvalidInput(_)) + )); + let missing = fixture.operation( + "missing-checkpoint", + checkpoint(&["missing-record"]), + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, missing), + Err(JournalError::NotFound(_)) + )); + let false_position = fixture.operation( + "false-position", + JournalAction::UpdatePosition { + current_position: "Pretend processed".into(), + }, + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, false_position), + Err(JournalError::InvalidInput(_)) + )); + assert_eq!(fixture.state(), before); + fixture.execute( + &ctx, + "record", + open_action("Only the first source is represented"), + vec![fixture.ranges[0].clone()], + ); + let after_record = fixture.state(); + let excess = fixture.operation( + "excess-checkpoint", + checkpoint(&["record"]), + fixture.ranges.clone(), + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, excess), + Err(JournalError::InvalidInput(_)) + )); + assert_eq!(fixture.state(), after_record); + assert!( + fixture + .sources() + .iter() + .all(|source| source.reflected_bytes == 0) + ); + fixture.execute( + &ctx, + "checked-unchanged", + JournalAction::ConfirmUnchanged, + vec![fixture.ranges[1].clone()], + ); + let receipt = fixture.execute( + &ctx, + "valid-checkpoint", + checkpoint(&["record", "checked-unchanged"]), + fixture.ranges.clone(), + ); + assert_eq!((receipt.process_order, receipt.process_byte), (2, 0)); + assert!( + fixture + .sources() + .iter() + .all(|source| source.reflected_bytes == source.source.total_bytes) + ); + let before = fixture.state(); + let replay_prefix = fixture.operation( + "rewind-checkpoint", + checkpoint(&["record"]), + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, replay_prefix), + Err(JournalError::InvalidInput(_)) + )); + assert_eq!(fixture.state(), before); +} + +#[test] +fn huge_source_partial_generations_checkpoint_only_read_prefix_and_fix_only_at_entry_end() { + let mut fixture = Fixture::with_sizes(&[8 * 1024 * 1024]); + let source = fixture.ranges[0].clone(); + let head = fragment(&source, 0, 16 * 1024); + let rest = fragment(&source, head.bytes.end, source.source.total_bytes); + let (first_generation, first_ctx) = fixture.generation("huge-head", vec![head.clone()]); + let created = fixture.execute( + &first_ctx, + "partial-record", + open_action("Only the initial fragment was checked"), + vec![head.clone()], + ); + let unread_end = fixture.operation( + "premature-fix", + JournalAction::Fix { + entry_id: created.entry_id.clone().unwrap(), + source_end: source.source.pointer.clone(), + }, + vec![], + ); + assert!(matches!( + fixture.store().execute_journal(&first_ctx, unread_end), + Err(JournalError::InvalidInput(_)) + )); + let cp = fixture.execute( + &first_ctx, + "partial-checkpoint", + checkpoint(&["partial-record"]), + vec![head.clone()], + ); + assert_eq!((cp.process_order, cp.process_byte), (0, head.bytes.end)); + assert_eq!(fixture.sources()[0].reflected_bytes, head.bytes.end); + fixture.close(&first_generation); + fixture.reopen(); + let gap = fragment(&source, head.bytes.end + 1, source.source.total_bytes); + let binding = JobAttemptBinding { + job_id: "gap-job".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + binding, + vec![gap] + ), + Err(JournalError::InvalidInput(_)) + )); + let overlap = fragment(&source, head.bytes.end - 1, source.source.total_bytes); + let binding = JobAttemptBinding { + job_id: "overlap-job".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + binding, + vec![overlap] + ), + Err(JournalError::InvalidInput(_)) + )); + let (next, next_ctx) = fixture.generation("huge-tail", vec![rest.clone()]); + assert_eq!(next.initial_journal.process_byte, head.bytes.end); + assert_eq!( + next.initial_open.as_ref().unwrap().id, + created.entry_id.clone().unwrap() + ); + fixture.execute( + &next_ctx, + "tail-record", + JournalAction::Rewrite { + entry_id: created.entry_id.clone().unwrap(), + title: "Complete input checked".into(), + body: "A bounded record of a huge source".into(), + ongoing: "Ready to fix".into(), + }, + vec![rest.clone()], + ); + fixture.execute( + &next_ctx, + "fix", + JournalAction::Fix { + entry_id: created.entry_id.unwrap(), + source_end: source.source.pointer.clone(), + }, + vec![], + ); + assert_eq!(fixture.state().0.process_byte, head.bytes.end); + let cp = fixture.execute( + &next_ctx, + "tail-checkpoint", + checkpoint(&["tail-record"]), + vec![rest], + ); + assert_eq!((cp.process_order, cp.process_byte), (1, 0)); + assert_eq!( + fixture.sources()[0].reflected_bytes, + source.source.total_bytes + ); +} + +#[test] +fn accepted_overlap_merges_but_gaps_and_overlapping_selected_prefixes_cannot_skip_input() { + let fixture = Fixture::new(); + let ranges = vec![ + fragment(&fixture.ranges[0], 0, 150), + fragment(&fixture.ranges[0], 100, 256), + ]; + fixture + .store() + .accept_journal_sources(&fixture.host, "Same journal", &ranges) + .unwrap(); + assert_eq!( + fixture.sources()[0].accepted, + vec![JournalByteRange { start: 0, end: 256 }] + ); + let binding = JobAttemptBinding { + job_id: "overlapping-selection".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + binding, + ranges + ), + Err(JournalError::InvalidInput(_)) + )); + let binding = JobAttemptBinding { + job_id: "missing-order-zero".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + binding, + vec![fixture.ranges[1].clone()] + ), + Err(JournalError::InvalidInput(_)) + )); + + let host = session_host(fixture.store(), &fixture.host.subject_id, "gapped-session"); + let full = source_range("gapped-session", 0, 1000); + let head = fragment(&full, 0, 100); + let tail = fragment(&full, 200, 1000); + let id = fixture + .store() + .accept_journal_sources(&host, "Gapped admission", &[head.clone(), tail.clone()]) + .unwrap(); + let binding = JobAttemptBinding { + job_id: "gapped-source".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture + .store() + .prepare_journal_generation(&host, &id, binding.clone(), vec![full.clone()]), + Err(JournalError::InvalidInput(_)) + )); + assert!(matches!( + fixture.store().prepare_journal_generation( + &host, + &id, + binding.clone(), + vec![head.clone(), tail.clone()] + ), + Err(JournalError::InvalidInput(_)) + )); + fixture + .store() + .accept_journal_sources(&host, "Fill only the gap", &[fragment(&full, 100, 200)]) + .unwrap(); + let generation = fixture + .store() + .prepare_journal_generation(&host, &id, binding, vec![full]) + .unwrap(); + assert_eq!(generation.initial_journal.process_byte, 0); + assert!(fixture.entries().is_empty()); +} + +#[test] +fn canonical_compaction_identity_is_idempotent_and_rewind_replacement_rolls_back_admission() { + let fixture = Fixture::new(); + let before = fixture.state(); + let sources = fixture.sources(); + // A retained compaction copy supplies the original canonical identity, not a new source. + assert_eq!( + fixture + .store() + .accept_journal_sources(&fixture.host, "Compacted copy", &fixture.ranges) + .unwrap(), + fixture.journal_id + ); + assert_eq!(fixture.sources(), sources); + assert_eq!(fixture.state(), before); + let original = fixture.ranges[0].clone(); + let mut replacements = Vec::new(); + let mut changed = original.clone(); + changed.source.content_sha256 = "f".repeat(64); + replacements.push(changed); + let mut changed = original.clone(); + changed.source.pointer.segment_id = Uuid::now_v7(); + replacements.push(changed); + let mut changed = original.clone(); + changed.source.pointer.entry_ref = "Ereplacement".into(); + replacements.push(changed); + let mut changed = original.clone(); + changed.source.total_bytes += 1; + replacements.push(changed); + let mut changed = original.clone(); + changed.source.order = 2; + replacements.push(changed); + for replacement in replacements { + let new_source = source_range("source-session", 3, 64); + let result = fixture.store().accept_journal_sources( + &fixture.host, + "Must roll back both", + &[new_source, replacement], + ); + assert!( + matches!(result, Err(JournalError::Conflict(_))), + "{result:?}" + ); + assert_eq!(fixture.sources(), sources); + assert_eq!(fixture.state(), before); + } + let wrong_session = source_range("another-session", 2, 100); + assert!(matches!( + fixture + .store() + .accept_journal_sources(&fixture.host, "No", &[wrong_session]), + Err(JournalError::PermissionDenied(_)) + )); + let mut derived_summary = source_range("source-session", 2, 100); + derived_summary.source.pointer.entry_ref = "summary-2".into(); + assert!(matches!( + fixture.store().accept_journal_sources( + &fixture.host, + "No derived source", + &[derived_summary] + ), + Err(JournalError::InvalidInput(_)) + )); + assert_eq!(fixture.sources(), sources); +} + +#[test] +fn unavailable_source_preserves_records_and_receipts_without_advancing_processing() { + let mut fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("retention", fixture.ranges.clone()); + let record = fixture.execute( + &ctx, + "record", + fixed_action("Retained outcome", &fixture.ranges[0], vec![]), + vec![fixture.ranges[0].clone()], + ); + let before = fixture.state(); + fixture + .store() + .mark_journal_sources_unavailable( + &fixture.host, + &fixture.journal_id, + &[fixture.ranges[0].clone()], + ) + .unwrap(); + fixture.reopen(); + assert_eq!(fixture.state(), before); + assert_eq!(fixture.receipt("record"), Some(record)); + assert_eq!( + fixture.entries()[0].source_end, + Some(fixture.ranges[0].source.pointer.clone()) + ); + assert!(!fixture.entries()[0].source_available); + assert_eq!(fixture.sources()[0].reflected_bytes, 0); + let unread = fixture.operation( + "unavailable-checkpoint", + checkpoint(&["record"]), + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, unread), + Err(JournalError::Conflict(_)) + )); + let unavailable = fixture.operation( + "unavailable-confirmation", + JournalAction::ConfirmUnchanged, + vec![fixture.ranges[0].clone()], + ); + assert!(matches!( + fixture.store().execute_journal(&ctx, unavailable), + Err(JournalError::Conflict(_)) + )); + fixture.close(&generation); + let binding = JobAttemptBinding { + job_id: "retention-replacement".into(), + attempt_id: "one".into(), + }; + assert!(matches!( + fixture.store().prepare_journal_generation( + &fixture.host, + &fixture.journal_id, + binding, + fixture.ranges.clone() + ), + Err(JournalError::Conflict(_)) + )); + let reader = + JournalHostContext::validated_reader(fixture.store(), &fixture.host.subject_id).unwrap(); + let page = fixture + .store() + .read_journal_text( + &reader, + &fixture.journal_id, + Some(fixture.entries()[0].id.as_str()), + JournalTextField::Body, + None, + 0, + 100, + ) + .unwrap(); + assert_eq!(page.text, "Retained outcome"); + assert!(fixture.receipt("unavailable-checkpoint").is_none()); +} + +#[test] +fn unicode_and_escaped_long_text_pages_reconstruct_exactly_and_reject_stale_continuation() { + let fixture = Fixture::new(); + let (_, ctx) = fixture.generation("text-pages", fixture.ranges.clone()); + let body = "็•Œ๐Ÿ™‚\"\\\n\t".repeat(12_000); + let ongoing = "ๆฌกใฎ็ขบ่ช๐Ÿ™‚\n".repeat(3000); + let action = JournalAction::Create { + title: "Unicode".into(), + body: body.clone(), + ongoing: ongoing.clone(), + state: JournalEntryState::Open, + source_end: None, + correction_refs: vec![], + }; + let id = fixture + .execute(&ctx, "create", action, vec![]) + .entry_id + .unwrap(); + fixture.execute( + &ctx, + "position", + JournalAction::UpdatePosition { + current_position: body.clone(), + }, + vec![], + ); + for (field, entry, expected) in [ + (JournalTextField::Body, Some(id.as_str()), body.as_str()), + ( + JournalTextField::Ongoing, + Some(id.as_str()), + ongoing.as_str(), + ), + (JournalTextField::CurrentPosition, None, body.as_str()), + ] { + let mut offset = 0; + let mut token = None::; + let mut reconstructed = String::new(); + loop { + let page = fixture + .store() + .read_journal_text( + &fixture.host, + &fixture.journal_id, + entry, + field, + token.as_deref(), + offset, + 257, + ) + .unwrap(); + assert!(page.text.len() <= 257); + assert!(!page.text.is_empty()); + assert!(serde_json::to_vec(&page).unwrap().len() < MAX_JOURNAL_READ_BYTES); + if let Some(token) = &token { + assert_eq!(&page.content_token, token); + } + reconstructed.push_str(&page.text); + token = Some(page.content_token); + match page.next_byte { + Some(next) => { + assert!(next > offset); + assert!(expected.is_char_boundary(next)); + offset = next; + } + None => break, + } + } + assert_eq!(reconstructed, expected); + } + let first = fixture + .store() + .read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&id), + JournalTextField::Body, + None, + 0, + 257, + ) + .unwrap(); + let position = fixture + .store() + .read_journal_text( + &fixture.host, + &fixture.journal_id, + None, + JournalTextField::CurrentPosition, + None, + 0, + 257, + ) + .unwrap(); + assert!(matches!( + fixture.store().read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&id), + JournalTextField::Body, + None, + first.next_byte.unwrap(), + 257 + ), + Err(JournalError::Conflict(_)) + )); + assert!(matches!( + fixture.store().read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&id), + JournalTextField::Body, + Some(&first.content_token), + 1, + 257 + ), + Err(JournalError::InvalidInput(_)) + )); + fixture.execute( + &ctx, + "rewrite", + JournalAction::Rewrite { + entry_id: id.clone(), + title: "Changed".into(), + body: "Different body".into(), + ongoing: "Changed ongoing".into(), + }, + vec![], + ); + assert!(matches!( + fixture.store().read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&id), + JournalTextField::Body, + Some(&first.content_token), + first.next_byte.unwrap(), + 257 + ), + Err(JournalError::Conflict(_)) + )); + fixture.execute( + &ctx, + "position-rewrite", + JournalAction::UpdatePosition { + current_position: "Different position".into(), + }, + vec![], + ); + assert!(matches!( + fixture.store().read_journal_text( + &fixture.host, + &fixture.journal_id, + None, + JournalTextField::CurrentPosition, + Some(&position.content_token), + position.next_byte.unwrap(), + 257 + ), + Err(JournalError::Conflict(_)) + )); + for invalid_limit in [0, 3, MAX_JOURNAL_READ_BYTES + 1] { + assert!(matches!( + fixture.store().read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&id), + JournalTextField::Body, + None, + 0, + invalid_limit + ), + Err(JournalError::InvalidInput(_)) + )); + } + let summary = &fixture.entries()[0]; + let json = serde_json::to_value(summary).unwrap(); + assert!(json.get("body").is_none()); + assert!(json.get("ongoing").is_none()); +} + +#[test] +fn metadata_and_operation_keysets_do_not_skip_appends_or_sort_by_request_id() { + let fixture = Fixture::with_sizes(&[100, 100, 100]); + let (_, ctx) = fixture.generation("metadata-pages", fixture.ranges.clone()); + fixture.execute( + &ctx, + "z-request", + fixed_action("First", &fixture.ranges[0], vec![]), + vec![], + ); + fixture.execute( + &ctx, + "a-request", + fixed_action("Second", &fixture.ranges[1], vec![]), + vec![], + ); + let first = fixture + .store() + .list_journal_entries(&fixture.host, &fixture.journal_id, None, 1) + .unwrap(); + let receipts = fixture + .store() + .list_journal_operations(&fixture.host, &fixture.journal_id, None, 1) + .unwrap(); + assert_eq!(first.items[0].position, 0); + assert_eq!(receipts.items[0].request_id, "z-request"); + fixture.execute( + &ctx, + "m-request", + fixed_action("Appended between pages", &fixture.ranges[2], vec![]), + vec![], + ); + let second = fixture + .store() + .list_journal_entries( + &fixture.host, + &fixture.journal_id, + Some(first.next.unwrap().parse().unwrap()), + 1, + ) + .unwrap(); + let third = fixture + .store() + .list_journal_entries( + &fixture.host, + &fixture.journal_id, + Some(second.next.unwrap().parse().unwrap()), + 1, + ) + .unwrap(); + assert_eq!(second.items[0].position, 1); + assert_eq!(third.items[0].position, 2); + assert!(third.next.is_none()); + let second_receipt = fixture + .store() + .list_journal_operations( + &fixture.host, + &fixture.journal_id, + receipts.next.as_deref(), + 1, + ) + .unwrap(); + let third_receipt = fixture + .store() + .list_journal_operations( + &fixture.host, + &fixture.journal_id, + second_receipt.next.as_deref(), + 1, + ) + .unwrap(); + assert_eq!(second_receipt.items[0].request_id, "a-request"); + assert_eq!(third_receipt.items[0].request_id, "m-request"); + assert!(third_receipt.next.is_none()); + let source_page = fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, None, 1) + .unwrap(); + let next_source = fixture + .store() + .list_journal_sources( + &fixture.host, + &fixture.journal_id, + source_page.next.as_deref(), + 2, + ) + .unwrap(); + assert_eq!( + next_source + .items + .iter() + .map(|s| s.source.order) + .collect::>(), + vec![1, 2] + ); + for limit in [0, MAX_JOURNAL_PAGE + 1] { + assert!(matches!( + fixture + .store() + .list_journal_entries(&fixture.host, &fixture.journal_id, None, limit), + Err(JournalError::InvalidInput(_)) + )); + assert!(matches!( + fixture.store().list_journal_operations( + &fixture.host, + &fixture.journal_id, + None, + limit + ), + Err(JournalError::InvalidInput(_)) + )); + assert!(matches!( + fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, None, limit), + Err(JournalError::InvalidInput(_)) + )); + assert!(matches!( + fixture.store().list_journals(&fixture.host, None, limit), + Err(JournalError::InvalidInput(_)) + )); + } +} + +#[test] +fn journal_metadata_pages_are_subject_scoped_and_reader_context_needs_no_session() { + let fixture = Fixture::new(); + let later_host = session_host( + fixture.store(), + &fixture.host.subject_id, + "metadata-later-session", + ); + let later_source = source_range("metadata-later-session", 0, 100); + let later_id = fixture + .store() + .accept_journal_sources(&later_host, "Later", &[later_source]) + .unwrap(); + let foreign_subject = fixture + .store() + .create_subject(SubjectRole::new("foreign").unwrap()) + .unwrap(); + let foreign_host = session_host( + fixture.store(), + &foreign_subject.id, + "metadata-foreign-session", + ); + let foreign_source = source_range("metadata-foreign-session", 0, 100); + let foreign_id = fixture + .store() + .accept_journal_sources(&foreign_host, "Not visible", &[foreign_source]) + .unwrap(); + let reader = + JournalHostContext::validated_reader(fixture.store(), &fixture.host.subject_id).unwrap(); + let first = fixture.store().list_journals(&reader, None, 1).unwrap(); + assert_eq!(first.items.len(), 1); + let second = fixture + .store() + .list_journals(&reader, first.next.as_deref(), 1) + .unwrap(); + assert_eq!(second.items.len(), 1); + assert!(second.next.is_none()); + let mut expected = vec![fixture.journal_id.clone(), later_id]; + expected.sort(); + assert_eq!( + vec![first.items[0].id.clone(), second.items[0].id.clone()], + expected + ); + assert!( + first + .items + .iter() + .chain(&second.items) + .all(|item| item.subject_id == fixture.host.subject_id) + ); + assert!(matches!( + fixture.store().read_journal(&reader, &foreign_id), + Err(JournalError::NotFound(_)) + )); + let json = serde_json::to_value(&first.items[0]).unwrap(); + assert!(json.get("current_position").is_none()); + assert!(json.get("entries").is_none()); + assert!(json.get("body").is_none()); +} + +#[test] +fn backup_restore_preserves_generation_records_corrections_sources_position_and_receipts() { + let mut fixture = Fixture::new(); + let (generation, ctx) = fixture.generation("backup", fixture.ranges.clone()); + let first = fixture + .execute( + &ctx, + "fixed", + fixed_action("Durable outcome", &fixture.ranges[0], vec![]), + vec![fixture.ranges[0].clone()], + ) + .entry_id + .unwrap(); + let mut correction = open_action("Further qualification is ongoing"); + if let JournalAction::Create { + correction_refs, .. + } = &mut correction + { + correction_refs.push(first); + } + fixture.execute( + &ctx, + "correction", + correction, + vec![fixture.ranges[1].clone()], + ); + fixture.execute( + &ctx, + "position", + JournalAction::UpdatePosition { + current_position: "Correction awaits verification".into(), + }, + vec![], + ); + fixture.execute( + &ctx, + "checkpoint", + checkpoint(&["fixed"]), + vec![fixture.ranges[0].clone()], + ); + fixture + .store() + .mark_journal_sources_unavailable( + &fixture.host, + &fixture.journal_id, + &[fragment(&fixture.ranges[0], 0, 10)], + ) + .unwrap(); + let expected_state = fixture.state(); + let expected_entries = fixture.entries(); + let expected_sources = fixture.sources(); + let expected_receipts = fixture + .store() + .list_journal_operations(&fixture.host, &fixture.journal_id, None, 32) + .unwrap(); + let snapshot = fixture.temp.path().join("snapshot"); + fixture + .workspace + .as_ref() + .unwrap() + .backup(&snapshot) + .unwrap(); + fixture.execute( + &ctx, + "after-backup", + JournalAction::UpdatePosition { + current_position: "Not part of snapshot".into(), + }, + vec![], + ); + fixture.close(&generation); + fixture.workspace.as_ref().unwrap().delete().unwrap(); + fixture.store.take(); + fixture.workspace.take(); + fixture.manager.take().unwrap().shutdown().unwrap(); + let manager = FeatureStorage::new(fixture.temp.path().join("storage")); + let workspace = manager.workspace("journal-tests").unwrap(); + workspace.restore(&snapshot).unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + fixture.store = Some(SubjektivStore::open(&workspace, ®istration).unwrap()); + fixture.workspace = Some(workspace); + fixture.manager = Some(manager); + assert_eq!(fixture.state(), expected_state); + assert_eq!(fixture.entries(), expected_entries); + assert_eq!(fixture.sources(), expected_sources); + assert_eq!( + fixture + .store() + .read_journal_generation(&fixture.host, &generation.id) + .unwrap(), + generation + ); + assert_eq!( + fixture + .store() + .list_journal_operations(&fixture.host, &fixture.journal_id, None, 32) + .unwrap(), + expected_receipts + ); + assert!(fixture.receipt("after-backup").is_none()); + assert!( + JournalJobContext::validated( + fixture.store(), + &fixture.host, + &generation.id, + generation.job + ) + .is_ok() + ); + fixture.reopen(); + assert_eq!(fixture.state(), expected_state); + assert_eq!(fixture.entries(), expected_entries); +} + +#[test] +fn migrated_schema_rejects_null_fixed_order_open_tail_mutation_and_receipt_tampering() { + let mut fixture = Fixture::with_schema(&[256, 256], true); + let (_, ctx) = fixture.generation("schema", fixture.ranges.clone()); + let fixed_id = fixture + .execute( + &ctx, + "fixed", + fixed_action("Immutable record", &fixture.ranges[0], vec![]), + vec![], + ) + .entry_id + .unwrap(); + let open_id = fixture + .execute(&ctx, "open", open_action("Mutable tail"), vec![]) + .entry_id + .unwrap(); + let foreign_subject = fixture + .store() + .create_subject(SubjectRole::new("foreign").unwrap()) + .unwrap(); + let foreign_host = session_host( + fixture.store(), + &foreign_subject.id, + "schema-foreign-session", + ); + let foreign_source = source_range("schema-foreign-session", 0, 100); + let foreign_journal = fixture + .store() + .accept_journal_sources(&foreign_host, "Foreign journal", &[foreign_source]) + .unwrap(); + let before = fixture.state(); + let entries = fixture.entries(); + let journal = &fixture.journal_id; + let subject = &fixture.host.subject_id; + let invalid_sql = [ + format!( + "INSERT INTO journal_entries(id,subject_id,journal_id,position,state,title,body,ongoing,created_at,updated_at) VALUES('foreign-owner','{subject}','{foreign_journal}',0,'open','','','','now','now')" + ), + format!("UPDATE journal_entries SET body='changed' WHERE id='{fixed_id}'"), + format!("UPDATE journal_entries SET position=10 WHERE id='{open_id}'"), + format!("UPDATE journal_entries SET subject_id='missing-subject' WHERE id='{open_id}'"), + format!( + "UPDATE journal_entries SET state='fixed', source_end_json='{{}}', source_order=NULL, fixed_at='now', ongoing='' WHERE id='{open_id}'" + ), + format!( + "UPDATE journal_entries SET state='fixed', source_end_json='{{}}', source_order=-1, fixed_at='now', ongoing='' WHERE id='{open_id}'" + ), + format!( + "UPDATE journal_entries SET state='fixed', source_end_json='{{}}', source_order=1, fixed_at='now', ongoing='still open' WHERE id='{open_id}'" + ), + format!("UPDATE journal_entries SET source_end_json='{{}}' WHERE id='{open_id}'"), + format!( + "INSERT INTO journal_entries(id,subject_id,journal_id,position,state,title,body,ongoing,created_at,updated_at) VALUES('second-open','{subject}','{journal}',2,'open','','','','now','now')" + ), + format!( + "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('after-open','{subject}','{journal}',2,'fixed','','','','{{}}',1,'now','now','now')" + ), + format!( + "UPDATE journal_operations SET receipt_json='{{}}' WHERE journal_id='{journal}' AND request_id='fixed'" + ), + format!( + "DELETE FROM journal_operations WHERE journal_id='{journal}' AND request_id='fixed'" + ), + format!("DELETE FROM subjects WHERE subject_id='{subject}'"), + format!( + "INSERT INTO journal_corrections(subject_id,entry_id,target_id) VALUES('{subject}','{open_id}','{open_id}')" + ), + ]; + for sql in invalid_sql { + assert!( + matches!(fixture.sql(&sql), Err(JournalError::Storage(_))), + "schema accepted {sql}" + ); + assert_eq!(fixture.state(), before); + assert_eq!(fixture.entries(), entries); + } + fixture.reopen(); + assert_eq!(fixture.state(), before); + assert_eq!(fixture.entries(), entries); + assert!(fixture.receipt("fixed").is_some()); + assert!(fixture.receipt("open").is_some()); +} + +#[test] +fn failure_before_receipt_rolls_back_record_and_checkpoint_then_exact_retry_succeeds() { + let mut fixture = Fixture::new(); + let (_, ctx) = fixture.generation("transaction-fault", fixture.ranges.clone()); + fixture.sql("CREATE TRIGGER fail_test_receipt BEFORE INSERT ON journal_operations WHEN NEW.request_id IN ('fault-record','fault-checkpoint') BEGIN SELECT RAISE(ABORT, 'injected before receipt'); END;").unwrap(); + let record = fixture.operation( + "fault-record", + open_action("Must commit with its receipt"), + fixture.ranges.clone(), + ); + let before = fixture.state(); + let sources = fixture.sources(); + assert!(matches!( + fixture.store().execute_journal(&ctx, record.clone()), + Err(JournalError::Storage(_)) + )); + fixture.reopen(); + assert_eq!(fixture.state(), before); + assert_eq!(fixture.sources(), sources); + assert!(fixture.entries().is_empty()); + assert!(fixture.receipt("fault-record").is_none()); + fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap(); + let saved = fixture + .store() + .execute_journal(&ctx, record.clone()) + .unwrap(); + assert_eq!( + fixture.store().execute_journal(&ctx, record).unwrap(), + saved + ); + let cp = fixture.operation( + "fault-checkpoint", + checkpoint(&["fault-record"]), + fixture.ranges.clone(), + ); + fixture.sql("CREATE TRIGGER fail_test_receipt BEFORE INSERT ON journal_operations WHEN NEW.request_id='fault-checkpoint' BEGIN SELECT RAISE(ABORT, 'injected before receipt'); END;").unwrap(); + let before_checkpoint = fixture.state(); + assert!(matches!( + fixture.store().execute_journal(&ctx, cp.clone()), + Err(JournalError::Storage(_)) + )); + fixture.reopen(); + assert_eq!(fixture.state(), before_checkpoint); + assert_eq!(fixture.sources(), sources); + assert_eq!(fixture.receipt("fault-record"), Some(saved)); + assert!(fixture.receipt("fault-checkpoint").is_none()); + fixture.sql("DROP TRIGGER fail_test_receipt;").unwrap(); + let reflected = fixture.store().execute_journal(&ctx, cp.clone()).unwrap(); + assert_eq!((reflected.process_order, reflected.process_byte), (2, 0)); + fixture.reopen(); + assert_eq!( + fixture.store().execute_journal(&ctx, cp).unwrap(), + reflected + ); + assert_eq!(fixture.entries().len(), 1); + assert!( + fixture + .sources() + .iter() + .all(|source| source.reflected_bytes == source.source.total_bytes) + ); +} + +#[test] +fn per_operation_reopen_recovers_a_realistic_investigation_and_later_correction_without_job_result() +{ + let mut fixture = Fixture::new(); + let first_range = fixture.ranges[0].clone(); + let second_range = fixture.ranges[1].clone(); + let (first_generation, first_ctx) = + fixture.generation("investigation-job", vec![first_range.clone()]); + let created = fixture.execute_and_reopen( + &first_ctx, + "investigation-create", + open_action("Observed a gap in checkpoint evidence; reproducing it"), + vec![first_range.clone()], + ); + let first_id = created.entry_id.unwrap(); + fixture.execute_and_reopen( + &first_ctx, + "investigation-rewrite", + JournalAction::Rewrite { + entry_id: first_id.clone(), + title: "Checkpoint evidence verified".into(), + body: "Position-only updates cannot prove source reflection; record receipts can." + .into(), + ongoing: "Verify reopen before fixing the outcome".into(), + }, + vec![first_range.clone()], + ); + fixture.execute_and_reopen( + &first_ctx, + "investigation-position", + JournalAction::UpdatePosition { + current_position: "Record saved; checkpoint not yet committed".into(), + }, + vec![], + ); + assert_eq!(fixture.state().0.process_order, 0); + fixture.execute_and_reopen( + &first_ctx, + "investigation-fix", + JournalAction::Fix { + entry_id: first_id.clone(), + source_end: first_range.source.pointer.clone(), + }, + vec![], + ); + assert!(fixture.state().1.is_none()); + assert_eq!(fixture.state().0.process_order, 0); + let fixed_before_correction = fixture.entries()[0].clone(); + fixture.execute_and_reopen( + &first_ctx, + "investigation-checkpoint", + checkpoint(&["investigation-rewrite"]), + vec![first_range.clone()], + ); + fixture.close(&first_generation); + fixture.reopen(); + let (second_generation, second_ctx) = + fixture.generation("followup-job", vec![second_range.clone()]); + assert_eq!( + second_generation.previous_fixed_end, + Some(first_range.source.pointer) + ); + assert_eq!(second_generation.initial_journal.process_order, 1); + assert_eq!( + second_generation.initial_journal.current_position, + "Record saved; checkpoint not yet committed" + ); + assert!(second_generation.initial_open.is_none()); + let mut correction = + open_action("Clarify that explicit no-change receipts also count as evidence"); + if let JournalAction::Create { + correction_refs, .. + } = &mut correction + { + correction_refs.push(first_id); + } + let correction_id = fixture + .execute_and_reopen( + &second_ctx, + "followup-create", + correction, + vec![second_range.clone()], + ) + .entry_id + .unwrap(); + fixture.execute_and_reopen( + &second_ctx, + "followup-fix", + JournalAction::Fix { + entry_id: correction_id, + source_end: second_range.source.pointer.clone(), + }, + vec![], + ); + fixture.execute_and_reopen( + &second_ctx, + "followup-confirm", + JournalAction::ConfirmUnchanged, + vec![second_range.clone()], + ); + fixture.execute_and_reopen( + &second_ctx, + "followup-position", + JournalAction::UpdatePosition { + current_position: "Follow-up recorded; all admitted input has been checked".into(), + }, + vec![], + ); + fixture.execute_and_reopen( + &second_ctx, + "followup-checkpoint", + checkpoint(&["followup-create", "followup-confirm"]), + vec![second_range], + ); + fixture.close(&second_generation); + fixture.reopen(); + let entries = fixture.entries(); + assert_eq!(entries.len(), 2); + assert_eq!(entries[0], fixed_before_correction); + assert_eq!(entries[1].correction_refs, vec![entries[0].id.clone()]); + assert!( + entries + .iter() + .all(|entry| entry.state == JournalEntryState::Fixed) + ); + assert_eq!( + ( + fixture.state().0.process_order, + fixture.state().0.process_byte + ), + (2, 0) + ); + assert!(fixture.state().1.is_none()); +} + +#[test] +fn split_input_and_replacement_job_reconcile_old_record_coverage_without_duplicate_create() { + let mut fixture = Fixture::with_sizes(&[8]); + let whole = fixture.ranges[0].clone(); + let (generation, ctx) = fixture.generation( + "split-job", + vec![fragment(&whole, 0, 4), fragment(&whole, 4, 8)], + ); + let saved = fixture.execute( + &ctx, + "record", + fixed_action("Confirmed from both UTF-8 input fragments", &whole, vec![]), + vec![whole.clone()], + ); + fixture.close(&generation); + fixture.reopen(); + assert!( + fixture + .store() + .active_journal_generation(&fixture.host, &fixture.journal_id) + .unwrap() + .is_none() + ); + let status = fixture + .store() + .read_journal_operation_status(&fixture.host, &fixture.journal_id, "record") + .unwrap(); + assert_eq!(status.kind, JournalOperationKind::Create); + assert_eq!(status.receipt, saved); + assert_eq!(status.covered_range_count, 1); + let coverage = fixture + .store() + .read_journal_operation_coverage(&fixture.host, &fixture.journal_id, "record", 0, 1) + .unwrap(); + assert_eq!(coverage.items, vec![whole.clone()]); + let (_, replacement) = fixture.generation("replacement-job", vec![whole.clone()]); + let cp = fixture.operation("recover-reflection", checkpoint(&["record"]), vec![whole]); + let result = fixture + .store() + .checkpoint_journal(&replacement, cp) + .unwrap(); + assert_eq!((result.process_order, result.process_byte), (1, 0)); + assert_eq!(fixture.entries().len(), 1); + assert_eq!(fixture.entries()[0].id, saved.entry_id.unwrap()); + assert!(fixture.state().1.is_none()); +} + +#[test] +fn serialized_metadata_and_text_budgets_preserve_continuation_and_availability_recovery() { + let fixture = Fixture::with_sizes(&[8; 64]); + let (_, ctx) = fixture.generation("large-pages", fixture.ranges.clone()); + for (i, r) in fixture.ranges.iter().enumerate() { + let action = JournalAction::Create { + title: "\u{0001}".repeat(1024), + body: "\u{0001}".repeat(65536), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(r.source.pointer.clone()), + correction_refs: vec![], + }; + fixture.execute(&ctx, &format!("record-{i}"), action, vec![r.clone()]); + } + let mut after = None; + let mut ids = Vec::new(); + loop { + let page = fixture + .store() + .list_journal_entries(&fixture.host, &fixture.journal_id, after, 32) + .unwrap(); + assert!(serde_json::to_string_pretty(&page).unwrap().len() <= MAX_JOURNAL_RESPONSE_BYTES); + assert!(!page.items.is_empty()); + ids.extend(page.items.iter().map(|e| e.id.clone())); + let Some(next) = page.next else { break }; + after = Some(next.parse().unwrap()); + } + assert_eq!(ids.len(), 64); + assert_eq!( + ids.iter().collect::>().len(), + 64 + ); + let mut offset = 0; + let mut token = None; + let mut body = String::new(); + loop { + let page = fixture + .store() + .read_journal_text( + &fixture.host, + &fixture.journal_id, + Some(&ids[0]), + JournalTextField::Body, + token.as_deref(), + offset, + MAX_JOURNAL_READ_BYTES, + ) + .unwrap(); + assert!(serde_json::to_string_pretty(&page).unwrap().len() <= MAX_JOURNAL_RESPONSE_BYTES); + body.push_str(&page.text); + token = Some(page.content_token); + let Some(next) = page.next_byte else { break }; + assert!(next > offset); + offset = next; + } + assert_eq!(body, "\u{0001}".repeat(65536)); + fixture + .store() + .mark_journal_sources_unavailable(&fixture.host, &fixture.journal_id, &fixture.ranges[..1]) + .unwrap(); + assert!(!fixture.entries()[0].source_available); + fixture + .store() + .mark_journal_sources_available(&fixture.host, &fixture.journal_id, &fixture.ranges[..1]) + .unwrap(); + assert!(fixture.entries()[0].source_available); + assert_eq!(fixture.state().0.process_order, 0); +} + +#[test] +fn journal_migration_failure_rolls_back_ddl_and_retries_without_changing_existing_subject() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let manager = FeatureStorage::new(&root); + let scope = manager.scope("migration-rollback").unwrap(); + let registration = scope + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + &crate::MIGRATIONS[..7], + )) + .unwrap(); + let store = SubjektivStore::open(&scope, ®istration).unwrap(); + let subject = store + .create_subject_with_behavior( + SubjectRole::new("preserved").unwrap(), + "Original behavior".into(), + ) + .unwrap(); + store + .database + .try_transaction::<_, JournalError>(|tx| { + tx.execute_batch("CREATE TABLE journal_entries(injected INTEGER)")?; + Ok(()) + }) + .unwrap(); + drop(store); + drop(scope); + manager.shutdown().unwrap(); + drop(manager); + let manager = FeatureStorage::new(&root); + let scope = manager.scope("migration-rollback").unwrap(); + let registration = SubjektivStore::register(&scope).unwrap(); + assert!(matches!( + SubjektivStore::open(&scope, ®istration), + Err(crate::SubjektivError::Storage( + feature_storage::FeatureStorageError::MigrationFailed { .. } + )) + )); + drop(scope); + manager.shutdown().unwrap(); + drop(manager); + let manager = FeatureStorage::new(&root); + let scope = manager.scope("migration-rollback").unwrap(); + let registration = scope + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + &crate::MIGRATIONS[..7], + )) + .unwrap(); + let store = SubjektivStore::open(&scope, ®istration).unwrap(); + assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone())); + store + .database + .try_transaction::<_, JournalError>(|tx| { + assert!(!tx.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE name='journals')", + [], + |r| r.get::<_, bool>(0) + )?); + tx.execute_batch("DROP TABLE journal_entries")?; + Ok(()) + }) + .unwrap(); + drop(store); + drop(scope); + manager.shutdown().unwrap(); + drop(manager); + let manager = FeatureStorage::new(&root); + let scope = manager.scope("migration-rollback").unwrap(); + let registration = SubjektivStore::register(&scope).unwrap(); + let store = SubjektivStore::open(&scope, ®istration).unwrap(); + assert_eq!(store.subject(&subject.id).unwrap(), Some(subject.clone())); + let host = session_host(&store, &subject.id, "post-migration-session"); + let range = source_range("post-migration-session", 0, 8); + let id = store + .accept_journal_sources(&host, "Recovered migration", &[range]) + .unwrap(); + assert_eq!( + store.read_journal(&host, &id).unwrap().source_session_id, + "post-migration-session" + ); +} + +#[test] +fn late_gap_admission_and_partial_updates_reject_stale_source_page_cursors() { + let fixture = Fixture::with_sizes(&[8]); + let second = source_range("source-session", 1, 8); + let third = source_range("source-session", 2, 8); + let fourth = source_range("source-session", 3, 8); + fixture + .store() + .accept_journal_sources(&fixture.host, "", &[third, fourth]) + .unwrap(); + let first = fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, None, 2) + .unwrap(); + assert_eq!( + first + .items + .iter() + .map(|s| s.source.order) + .collect::>(), + vec![0, 2] + ); + let cursor = first.next.unwrap(); + fixture + .store() + .accept_journal_sources(&fixture.host, "", &[second]) + .unwrap(); + let next = + fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, Some(&cursor), 2); + assert!( + matches!(next, Err(JournalError::Conflict(_))), + "gap completion must not silently skip newly admitted order 1: {next:?}" + ); + let fresh = fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, None, 2) + .unwrap(); + assert_eq!( + fresh + .items + .iter() + .map(|s| s.source.order) + .collect::>(), + vec![0, 1] + ); + let next = fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2) + .unwrap(); + assert_eq!( + next.items + .iter() + .map(|s| s.source.order) + .collect::>(), + vec![2, 3] + ); + // Duplicate notifications do not change the content condition. + fixture + .store() + .accept_journal_sources(&fixture.host, "", &fixture.ranges) + .unwrap(); + assert!( + fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2) + .is_ok() + ); + let selected = fixture.ranges[0].clone(); + let (_, job) = fixture.generation("reflect-during-list", vec![selected.clone()]); + fixture.execute( + &job, + "no-change", + JournalAction::ConfirmUnchanged, + vec![selected.clone()], + ); + assert!( + fixture + .store() + .list_journal_sources(&fixture.host, &fixture.journal_id, fresh.next.as_deref(), 2) + .is_ok() + ); + fixture.execute( + &job, + "checkpoint", + checkpoint(&["no-change"]), + vec![selected], + ); + assert!(matches!( + fixture.store().list_journal_sources( + &fixture.host, + &fixture.journal_id, + fresh.next.as_deref(), + 2 + ), + Err(JournalError::Conflict(_)) + )); + assert_eq!( + fixture + .store() + .read_journal_source(&fixture.host, &fixture.journal_id, 0) + .unwrap() + .reflected_bytes, + 8 + ); + + let host = session_host(fixture.store(), &fixture.host.subject_id, "partial-session"); + let range = source_range("partial-session", 0, 16); + let later = source_range("partial-session", 1, 8); + let id = fixture + .store() + .accept_journal_sources(&host, "Partial source", &[fragment(&range, 0, 4), later]) + .unwrap(); + let page = fixture + .store() + .list_journal_sources(&host, &id, None, 1) + .unwrap(); + fixture + .store() + .accept_journal_sources(&host, "", &[fragment(&range, 4, 8)]) + .unwrap(); + assert!(matches!( + fixture + .store() + .list_journal_sources(&host, &id, page.next.as_deref(), 1), + Err(JournalError::Conflict(_)) + )); + let page = fixture + .store() + .list_journal_sources(&host, &id, None, 1) + .unwrap(); + fixture + .store() + .mark_journal_sources_unavailable(&host, &id, &[fragment(&range, 0, 8)]) + .unwrap(); + assert!(matches!( + fixture + .store() + .list_journal_sources(&host, &id, page.next.as_deref(), 1), + Err(JournalError::Conflict(_)) + )); + let page = fixture + .store() + .list_journal_sources(&host, &id, None, 1) + .unwrap(); + fixture + .store() + .mark_journal_sources_available(&host, &id, &[fragment(&range, 0, 4)]) + .unwrap(); + assert!(matches!( + fixture + .store() + .list_journal_sources(&host, &id, page.next.as_deref(), 1), + Err(JournalError::Conflict(_)) + )); + let page = fixture + .store() + .list_journal_sources(&host, &id, None, 1) + .unwrap(); + assert!(matches!( + fixture + .store() + .list_journal_sources(&host, &fixture.journal_id, page.next.as_deref(), 1), + Err(JournalError::InvalidInput(_)) + )); + let mut cursor: serde_json::Value = + serde_json::from_str(page.next.as_deref().unwrap()).unwrap(); + cursor["unexpected_field"] = true.into(); + assert!(matches!( + fixture + .store() + .list_journal_sources(&host, &id, Some(&cursor.to_string()), 1), + Err(JournalError::InvalidInput(_)) + )); +} diff --git a/crates/subjektiv/src/legacy_migrations.rs b/crates/subjektiv/src/legacy_migrations.rs index 614f79e05..e68fe5bdc 100644 --- a/crates/subjektiv/src/legacy_migrations.rs +++ b/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] diff --git a/crates/subjektiv/src/lib.rs b/crates/subjektiv/src/lib.rs index b8e176940..ba86ac4fd 100644 --- a/crates/subjektiv/src/lib.rs +++ b/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 =