diff --git a/crates/standalone/src/subjektiv.rs b/crates/standalone/src/subjektiv.rs index 4b6260b94..d9f31c100 100644 --- a/crates/standalone/src/subjektiv.rs +++ b/crates/standalone/src/subjektiv.rs @@ -698,6 +698,30 @@ impl LocalSubjectHost { } } impl SubjektivHost for LocalSubjectHost { + fn memory_journal( + &self, + context: &SubjektivHostContext, + request: subjektiv::journal::memory_links::MemoryJournalRequest, + ) -> Result { + self.body(context)?; + let reader = subjektiv::journal::JournalHostContext::validated_reader( + &self.catalog.store, + &self.binding.subject_id, + ) + .map_err(journal::journal_error)?; + self.catalog + .store + .execute_memory_journal_with_source_presence(&reader, request, |session| { + crate::subjektiv_sessions::original_session_presence( + &self.catalog.state_root, + &self.binding.scope_id, + &self.binding.subject_id, + &self.catalog.store, + session, + ) + }) + .map_err(journal::journal_error) + } fn journal_recording_granted(&self) -> bool { self.live.load(Ordering::Acquire) } diff --git a/crates/standalone/src/subjektiv/journal.rs b/crates/standalone/src/subjektiv/journal.rs index 87849a17f..c089aefae 100644 --- a/crates/standalone/src/subjektiv/journal.rs +++ b/crates/standalone/src/subjektiv/journal.rs @@ -233,7 +233,7 @@ pub(crate) fn prepare_worker_deletion( .map_err(|error| crate::store::StandaloneStoreError::JournalRetention(error.to_string())) } -fn journal_error(error: JournalError) -> SubjektivHostError { +pub(super) fn journal_error(error: JournalError) -> SubjektivHostError { match error { JournalError::Conflict(message) => SubjektivHostError::Conflict { code: "journal_conflict".into(), diff --git a/crates/standalone/src/subjektiv/journal/tests.rs b/crates/standalone/src/subjektiv/journal/tests.rs index 6ddcaa6bc..5b13425ca 100644 --- a/crates/standalone/src/subjektiv/journal/tests.rs +++ b/crates/standalone/src/subjektiv/journal/tests.rs @@ -8,6 +8,7 @@ use session_store::{Store, WorkerMetadataStore}; use std::sync::atomic::AtomicBool; mod compaction; +mod memory_links; struct Fixture { _temp: tempfile::TempDir, diff --git a/crates/standalone/src/subjektiv/journal/tests/memory_links.rs b/crates/standalone/src/subjektiv/journal/tests/memory_links.rs new file mode 100644 index 000000000..a858a4aa6 --- /dev/null +++ b/crates/standalone/src/subjektiv/journal/tests/memory_links.rs @@ -0,0 +1,565 @@ +//! Connected standalone Host -> Subjektiv Memory/Journal -> real FsStore coverage. +//! Candidate and Journal decisions are explicit test inputs; no model, Runtime or live DB +//! is used. The parent Journal fixture owns isolated temporary storage and binding cleanup. +//! Selected by the ordinary standalone lib tests, with no additional feature or service. + +use super::*; +use memory::{ + backend::MemoryStageCandidateOperation, + extract::{CandidateKind, ExtractedCandidate, StagingEvidence}, + schema::{EvidenceKind, EvidenceOriginKind, SourceEvidenceRef, SourceRef}, +}; +use subjektiv::{ + CandidateDecision, CandidateDecisionRequest, MemoryChangeTarget, MemoryDraft, MemoryRecord, + StagingResolution, SubjectStagingRecord, + journal::memory_links::{ + JournalTargetAvailability, MemoryJournalLinks, MemoryJournalRequest, MemoryJournalResponse, + MemoryJournalSelector, SourceConnection, UnconnectedReason, + }, +}; + +fn adopt_original(fixture: &Fixture) -> (MemoryRecord, SubjectStagingRecord, StagingResolution) { + let evidence_index = session_store::public_index::read_fs_session_evidence_index( + fixture.sessions.root_dir(), + fixture.context.session_id.parse().unwrap(), + session_store::SessionPublicIndexLimits { + max_bytes: 8 * 1024 * 1024, + max_entries: 20_000, + max_segments: 64, + }, + ) + .unwrap(); + let range = *evidence_index + .ranges + .get(&(fixture.segment.to_string(), "Efirst".into())) + .unwrap() + .first() + .unwrap(); + let staged = fixture + .host() + .stage_candidate( + &fixture.context, + MemoryStageCandidateOperation { + source: SourceRef { + segment_id: fixture.segment.to_string(), + range, + }, + extract_run_id: "explicit-test-extraction-input".into(), + candidate: ExtractedCandidate { + kind: CandidateKind::Preference, + claim: "Keep correction history".into(), + why_useful: "Preserves the reason for later decisions".into(), + staleness: None, + evidence_ids: vec!["Efirst".into()], + }, + evidence: vec![StagingEvidence { + id: "Efirst".into(), + kind: EvidenceKind::new(EvidenceKind::MESSAGE), + entry_range: Some(range), + origin: None, + excerpt: None, + summary: None, + }], + source_refs: vec![SourceEvidenceRef { + session_id: Some(fixture.context.session_id.clone()), + segment_id: Some(fixture.segment.to_string()), + entry_range: Some(range), + evidence_id: Some("Efirst".into()), + ..Default::default() + }], + }, + ) + .unwrap(); + let store = &fixture.host().catalog.store; + let subject = &fixture.connection.binding.subject_id; + let candidate = store + .staging_candidate(subject, &staged.staging_id) + .unwrap() + .unwrap(); + let receipt = store + .decide_candidate( + subject, + CandidateDecisionRequest { + request_id: "explicit-test-adoption-input".into(), + candidate_id: staged.staging_id, + reason: "Adopt the committed original evidence, not the Journal summary".into(), + decision: CandidateDecision::Apply { + target: MemoryChangeTarget::Create, + draft: MemoryDraft::active( + CandidateKind::Preference, + "Keep correction history", + "Preserve earlier decisions and their corrections", + "Preserves the reason for later decisions", + "Adopt original HumanInput", + ), + }, + }, + ) + .unwrap(); + (receipt.memory.unwrap(), candidate, receipt.resolution) +} + +fn create_entry(adapter: &LocalJournalHost, state: JournalEntryState) -> JournalReceipt { + save( + adapter, + "record-original-context", + &adapter.generation.initial_journal.state_token, + JournalAction::Create { + title: "A proposal was corrected".into(), + body: "Proposal A changed to B; only part was performed, outcome unconfirmed.".into(), + ongoing: if state == JournalEntryState::Open { + "Remaining work is not complete".into() + } else { + String::new() + }, + state, + source_end: (state == JournalEntryState::Fixed).then(|| { + adapter + .generation + .selected_ranges + .last() + .unwrap() + .source + .pointer + .clone() + }), + correction_refs: vec![], + }, + adapter.generation.selected_ranges.clone(), + ) +} + +fn assert_originals_unchanged( + fixture: &Fixture, + memory: &MemoryRecord, + candidate: &SubjectStagingRecord, + resolution: &StagingResolution, +) { + let catalog = StandaloneSubjects::open_existing(&fixture.host().catalog.state_root).unwrap(); + let store = &catalog.store; + let subject = &fixture.connection.binding.subject_id; + assert_eq!( + serde_json::to_value(store.memory(subject, &memory.id).unwrap().unwrap()).unwrap(), + serde_json::to_value(memory).unwrap(), + ); + assert_eq!( + serde_json::to_value( + store + .staging_candidate(subject, &candidate.id) + .unwrap() + .unwrap() + ) + .unwrap(), + serde_json::to_value(candidate).unwrap(), + ); + assert_eq!( + serde_json::to_value( + store + .staging_resolution(subject, &candidate.id) + .unwrap() + .unwrap() + ) + .unwrap(), + serde_json::to_value(resolution).unwrap(), + ); + assert_eq!(store.list_memories(subject).unwrap().len(), 1); +} + +fn selector(memory: &MemoryRecord, candidate: &SubjectStagingRecord) -> MemoryJournalSelector { + MemoryJournalSelector { + memory_id: memory.id.clone(), + change_id: memory.change_id.clone(), + candidate_id: candidate.id.clone(), + } +} + +fn links( + fixture: &Fixture, + selector: &MemoryJournalSelector, + reconcile: bool, +) -> MemoryJournalLinks { + let request = if reconcile { + MemoryJournalRequest::Reconcile { + selector: selector.clone(), + } + } else { + MemoryJournalRequest::Links { + selector: selector.clone(), + } + }; + let MemoryJournalResponse::Links(links) = fixture + .host() + .memory_journal(&fixture.context, request) + .unwrap() + else { + panic!("expected Memory Journal links") + }; + links +} + +fn read_request( + selector: &MemoryJournalSelector, + journal: &str, + entry: &str, +) -> MemoryJournalRequest { + MemoryJournalRequest::Read { + selector: selector.clone(), + source_index: 0, + journal_id: journal.into(), + entry_id: entry.into(), + expected_token: None, + byte_offset: 0, + max_bytes: MAX_JOURNAL_READ_BYTES, + } +} + +fn assert_link(links: &MemoryJournalLinks, journal: &str, entry: &str, reconsideration: bool) { + assert_eq!(links.sources.len(), 1); + assert_eq!(links.sources[0].source_index, 0); + assert_eq!( + links.sources[0].connection, + SourceConnection::Linked { + journal_id: journal.into(), + entry_id: entry.into(), + target: JournalTargetAvailability::Available, + reconsideration, + }, + ); +} + +#[test] +fn saved_link_reads_after_original_worker_and_session_deletion_from_replacement_host() { + let mut fixture = Fixture::new("Please preserve the reasons for my corrected decisions"); + let (memory, candidate, resolution) = adopt_original(&fixture); + let selector = selector(&memory, &candidate); + assert_eq!( + candidate.evidence[0].origin.as_ref().unwrap().kind, + EvidenceOriginKind::HumanInput + ); + assert_eq!( + candidate.source_refs[0].origin, + candidate.evidence[0].origin + ); + assert!( + candidate.evidence[0] + .excerpt + .as_ref() + .unwrap() + .contains("corrected decisions") + ); + let (adapter, _, _) = fixture.adapter(); + let recorded = create_entry(&adapter, JournalEntryState::Fixed); + let entry = recorded.entry_id.clone().unwrap(); + let journal = adapter.grant.journal_id.clone(); + let connected = links(&fixture, &selector, true); + assert_link(&connected, &journal, &entry, false); + let request = read_request(&selector, &journal, &entry); + let before = fixture + .host() + .memory_journal(&fixture.context, request.clone()) + .unwrap(); + save( + &adapter, + "checkpoint-original", + &recorded.state_token, + JournalAction::Checkpoint { + record_operation_ids: vec![recorded.request_id], + }, + adapter.generation.selected_ranges.clone(), + ); + adapter.finish().unwrap(); + fixture + .host() + .catalog + .store + .close_journal_generation( + &adapter.context, + &adapter.generation.id, + &adapter.generation.job, + ) + .unwrap(); + let state = fixture.host().catalog.state_root.clone(); + let subject = fixture.connection.binding.subject_id.clone(); + let original_session = fixture.context.session_id.clone(); + let (worker, workers) = fixture.stop_worker(); + workers.delete(worker).unwrap(); + assert!(matches!( + workers.load(worker), + Err(crate::store::StandaloneStoreError::WorkerNotFound(_)) + )); + assert!(!fixture.sessions.root_dir().join(&original_session).exists()); + assert!( + !retained_workers_root(&state) + .join(worker.to_string()) + .exists() + ); + + let replacement_worker = protocol::WorkerId::now_v7(); + let mut replacement = StandaloneSubjects::open_existing(&state) + .unwrap() + .connect(&subject, replacement_worker) + .unwrap(); + let replacement_context = SubjektivHostContext { + worker_id: replacement_worker.to_string(), + session_id: session_store::SessionId::now_v7().to_string(), + }; + let MemoryJournalResponse::Links(saved) = replacement + .host + .memory_journal( + &replacement_context, + MemoryJournalRequest::Links { + selector: selector.clone(), + }, + ) + .unwrap() + else { + panic!("expected saved links") + }; + assert_link(&saved, &journal, &entry, false); + let after = replacement + .host + .memory_journal(&replacement_context, request) + .unwrap(); + let MemoryJournalResponse::Read(read) = &after else { + panic!("expected saved Journal text") + }; + assert!( + read.page + .as_ref() + .unwrap() + .text + .contains("outcome unconfirmed") + ); + assert_eq!( + read.source.original_source, + subjektiv::journal::memory_links::OriginalSourceAvailability::Missing + ); + // Reading the saved edge does not depend on a live Session fetch or a replacement Job. + let MemoryJournalResponse::Read(before) = before else { + panic!("expected Journal text") + }; + assert_eq!(before.page, read.page); + assert!( + replacement.host.jobs().is_err(), + "the reader has no model/Job executor" + ); + assert_originals_unchanged(&fixture, &memory, &candidate, &resolution); + replacement.close().unwrap(); +} + +#[test] +fn late_journal_fix_connects_existing_memory_without_using_checkpoint_as_event_end() { + let fixture = Fixture::new("Keep the reasons behind corrected decisions"); + let (memory, candidate, resolution) = adopt_original(&fixture); + let selector = selector(&memory, &candidate); + assert_eq!( + links(&fixture, &selector, false).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled + } + ); + assert_eq!( + links(&fixture, &selector, true).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::SourceNotAdmitted + } + ); + let (adapter, _, _) = fixture.adapter(); + let recorded = create_entry(&adapter, JournalEntryState::Open); + let entry = recorded.entry_id.clone().unwrap(); + let checkpoint = save( + &adapter, + "checkpoint-open", + &recorded.state_token, + JournalAction::Checkpoint { + record_operation_ids: vec![recorded.request_id], + }, + adapter.generation.selected_ranges.clone(), + ); + let unconnected = links(&fixture, &selector, true); + assert_eq!( + unconnected.sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::JournalNotFixed + } + ); + assert_eq!(links(&fixture, &selector, true), unconnected); + let state = adapter.state().unwrap(); + let open = state.current_open.unwrap(); + assert_eq!(open.id, entry); + assert_eq!(open.state, JournalEntryState::Open); + assert!(open.source_end.is_none()); + let last = adapter.generation.selected_ranges.last().unwrap(); + assert_eq!(state.current_journal.process_order, last.source.order + 1); + assert_eq!(state.current_journal.process_byte, 0); + save( + &adapter, + "fix-later", + &checkpoint.state_token, + JournalAction::Fix { + entry_id: entry.clone(), + source_end: adapter + .generation + .selected_ranges + .last() + .unwrap() + .source + .pointer + .clone(), + }, + adapter.generation.selected_ranges.clone(), + ); + // A plain saved-state read must not silently perform reconciliation. + assert_eq!(links(&fixture, &selector, false), unconnected); + let connected = links(&fixture, &selector, true); + assert_link(&connected, &adapter.grant.journal_id, &entry, false); + assert_eq!(links(&fixture, &selector, true), connected); + assert_eq!(links(&fixture, &selector, false), connected); + assert_originals_unchanged(&fixture, &memory, &candidate, &resolution); +} + +#[test] +fn host_rejects_foreign_subject_worker_context_and_closed_connection_for_all_link_operations() { + let mut fixture = Fixture::new("Keep correction history"); + let (memory, candidate, resolution) = adopt_original(&fixture); + let selector = selector(&memory, &candidate); + let (adapter, _, _) = fixture.adapter(); + let recorded = create_entry(&adapter, JournalEntryState::Fixed); + let journal = adapter.grant.journal_id.clone(); + let entry = recorded.entry_id.unwrap(); + assert_link(&links(&fixture, &selector, true), &journal, &entry, false); + let requests = [ + MemoryJournalRequest::Reconcile { + selector: selector.clone(), + }, + MemoryJournalRequest::Links { + selector: selector.clone(), + }, + read_request(&selector, &journal, &entry), + MemoryJournalRequest::Related { + journal_id: journal, + entry_id: entry, + after: None, + limit: 32, + }, + ]; + let catalog = StandaloneSubjects::open_existing(&fixture.host().catalog.state_root).unwrap(); + let foreign_subject = catalog.create("different-reader", None).unwrap(); + let foreign_worker = protocol::WorkerId::now_v7(); + let mut foreign = catalog + .connect(&foreign_subject.id, foreign_worker) + .unwrap(); + let foreign_context = SubjektivHostContext { + worker_id: foreign_worker.to_string(), + session_id: session_store::SessionId::now_v7().to_string(), + }; + let wrong_worker = SubjektivHostContext { + worker_id: foreign_worker.to_string(), + ..fixture.context.clone() + }; + for request in &requests { + assert!( + foreign + .host + .memory_journal(&foreign_context, request.clone()) + .is_err() + ); + assert!( + fixture + .host() + .memory_journal(&wrong_worker, request.clone()) + .is_err() + ); + } + foreign.close().unwrap(); + fixture.connection.close().unwrap(); + for request in requests { + assert!( + fixture + .host() + .memory_journal(&fixture.context, request) + .is_err() + ); + } + assert_originals_unchanged(&fixture, &memory, &candidate, &resolution); +} + +#[test] +fn journal_correction_finds_related_memory_without_revising_or_retargeting_its_evidence() { + let fixture = Fixture::new("Keep correction history"); + let (memory, candidate, resolution) = adopt_original(&fixture); + let selector = selector(&memory, &candidate); + fixture.append("correction", "The earlier result was not confirmed"); + let (adapter, _, _) = fixture.adapter(); + let ranges = &adapter.generation.selected_ranges; + assert_eq!(ranges.len(), 2); + let original = save( + &adapter, + "original-event", + &adapter.generation.initial_journal.state_token, + JournalAction::Create { + title: "Original context".into(), + body: "A change was proposed; execution remained uncertain".into(), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(ranges[0].source.pointer.clone()), + correction_refs: vec![], + }, + vec![ranges[0].clone()], + ); + let entry = original.entry_id.unwrap(); + let journal = &adapter.grant.journal_id; + assert_link(&links(&fixture, &selector, true), journal, &entry, false); + let corrected = save( + &adapter, + "correction-event", + &original.state_token, + JournalAction::Create { + title: "Correction".into(), + body: "The earlier result was unconfirmed; reconsider its associated Memory".into(), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(ranges[1].source.pointer.clone()), + correction_refs: vec![entry.clone()], + }, + vec![ranges[1].clone()], + ); + assert_link(&links(&fixture, &selector, false), journal, &entry, true); + let correction_entry = corrected.entry_id.unwrap(); + for target in [&entry, &correction_entry] { + let MemoryJournalResponse::Related(related) = fixture + .host() + .memory_journal( + &fixture.context, + MemoryJournalRequest::Related { + journal_id: journal.clone(), + entry_id: target.clone(), + after: None, + limit: 32, + }, + ) + .unwrap() + else { + panic!("expected related Memory") + }; + assert_eq!(related.items.len(), 1); + assert_eq!(related.items[0].selector, selector); + assert_eq!(related.items[0].source_index, 0); + assert!(related.items[0].reconsideration); + assert!(related.next.is_none()); + } + let MemoryJournalResponse::Read(read) = fixture + .host() + .memory_journal(&fixture.context, read_request(&selector, journal, &entry)) + .unwrap() + else { + panic!("expected original context") + }; + assert!( + read.page + .unwrap() + .text + .contains("execution remained uncertain") + ); + assert_originals_unchanged(&fixture, &memory, &candidate, &resolution); +} diff --git a/crates/standalone/src/subjektiv_sessions.rs b/crates/standalone/src/subjektiv_sessions.rs index df496e335..92b0b5d87 100644 --- a/crates/standalone/src/subjektiv_sessions.rs +++ b/crates/standalone/src/subjektiv_sessions.rs @@ -178,6 +178,58 @@ pub(crate) fn authorized_sessions_path( .map_err(|error: Failure| SubjektivHostError::Unavailable(error.message)) } +/// Checks only saved ownership and filesystem metadata, never Session bytes or a Worker restore. +/// Inaccessible/corrupt storage stays unknown rather than being reported as absent. +pub(crate) fn original_session_presence( + state_root: &Path, + scope_id: &str, + subject_id: &str, + store: &SubjektivStore, + session_id: &str, +) -> Option { + let attribution = store.session_attribution(session_id).ok()??; + if store.scope_id() != scope_id + || attribution.subject_id != subject_id + || attribution.runtime_id != scope_id + { + return None; + } + let worker: WorkerId = attribution.worker_id.parse().ok()?; + let session: SessionId = session_id.parse().ok()?; + let active = StandaloneWorkerStore::open_existing(state_root).ok()?; + let (workers, record) = match active.load(worker) { + Ok(record) => (active, record), + Err(StandaloneStoreError::WorkerNotFound(_)) => { + let root = state_root.join("subjektiv").join("retained-workers"); + match std::fs::metadata(&root) { + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Some(false), + Err(_) => return None, + Ok(_) => {} + } + let retained = StandaloneWorkerStore::open_existing(root).ok()?; + match retained.load(worker) { + Ok(record) => (retained, record), + Err(StandaloneStoreError::WorkerNotFound(_)) => return Some(false), + Err(_) => return None, + } + } + Err(_) => return None, + }; + if record.worker_id != worker + || !record + .subject + .as_ref() + .is_some_and(|binding| binding.scope_id == scope_id && binding.subject_id == subject_id) + { + return None; + } + match std::fs::metadata(workers.sessions_dir(worker).join(session.to_string())) { + Ok(metadata) if metadata.is_dir() => Some(true), + Err(error) if error.kind() == std::io::ErrorKind::NotFound => Some(false), + _ => None, + } +} + fn authorized_index_with_reader( state_root: &Path, scope_id: &str, diff --git a/crates/subjektiv/src/journal/memory_links/mod.rs b/crates/subjektiv/src/journal/memory_links/mod.rs new file mode 100644 index 000000000..c32675a3c --- /dev/null +++ b/crates/subjektiv/src/journal/memory_links/mod.rs @@ -0,0 +1,668 @@ +//! Additive, immutable provenance-to-context links, not additional evidence for a Memory claim. +//! +//! Reconciliation reads only saved candidate anchors and the Host-attested Journal source ledger. +//! It never fetches history, starts Workers, changes extraction/consolidation, fixes an open event, +//! or mutates Memory. A stable evidence reference identifies a public entry, not its legacy numeric +//! display range. Missing stable identity stays unconnected; numeric ranges are not guessed to be +//! canonical order. Saved edges remain usable without the original runtime and cannot be retargeted. +//! Journal content is never copied here. Explicit erasure retains edges and exposes the tombstone. + +use super::{ + JournalError, JournalHostContext, JournalResult, JournalSourcePointer, JournalTextField, + JournalTextPage, MAX_JOURNAL_PAGE, MAX_JOURNAL_RESPONSE_BYTES, +}; +use crate::{MAX_STAGING_ANCHORS, SubjectStagingRecord, SubjektivStore}; +use memory::schema::SourceEvidenceRef; +use rusqlite::{Connection, OptionalExtension, Transaction, params}; +use serde::{Deserialize, Serialize}; + +#[cfg(test)] +mod tests; + +pub(crate) fn migrate(tx: &Transaction<'_>) -> feature_storage::Result<()> { + tx.execute_batch(include_str!("schema.sql"))?; + Ok(()) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MemoryJournalSelector { + pub memory_id: String, + pub change_id: String, + pub candidate_id: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "action", rename_all = "snake_case", deny_unknown_fields)] +pub enum MemoryJournalRequest { + Reconcile { + selector: MemoryJournalSelector, + }, + /// Reads saved link state only; does not implicitly retry reconciliation. + Links { + selector: MemoryJournalSelector, + }, + /// Continues reconciliation over immutable source indices, at most MAX_STAGING_ANCHORS per call. + ReconcilePage { + selector: MemoryJournalSelector, + source_offset: usize, + limit: usize, + }, + /// Reads another saved source page without reconciliation. + LinksPage { + selector: MemoryJournalSelector, + source_offset: usize, + limit: usize, + }, + Read { + selector: MemoryJournalSelector, + source_index: usize, + journal_id: String, + entry_id: String, + expected_token: Option, + byte_offset: usize, + max_bytes: usize, + }, + Related { + journal_id: String, + entry_id: String, + after: Option, + limit: usize, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde( + tag = "kind", + content = "value", + rename_all = "snake_case", + deny_unknown_fields +)] +pub enum MemoryJournalResponse { + Links(MemoryJournalLinks), + Read(MemoryJournalRead), + Related(RelatedMemories), +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum OriginalSourceAvailability { + /// No saved Host attestation. None of these values imply a live history fetch. + Unknown, + Missing, + Available, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum JournalTargetAvailability { + Available, + Unavailable, + Erased, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum UnconnectedReason { + NotReconciled, + InsufficientSource, + SourceNotAdmitted, + AmbiguousSource, + JournalNotFixed, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", rename_all = "snake_case", deny_unknown_fields)] +pub enum SourceConnection { + Unconnected { + reason: UnconnectedReason, + }, + Linked { + journal_id: String, + entry_id: String, + target: JournalTargetAvailability, + /// A saved correction invites reconsideration, never an automatic Memory edit. + reconsideration: bool, + }, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MemoryJournalSourceLink { + pub source_index: usize, + pub original_source: OriginalSourceAvailability, + pub connection: SourceConnection, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MemoryJournalLinks { + pub selector: MemoryJournalSelector, + pub sources: Vec, + pub total_sources: usize, + pub next_source_index: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct MemoryJournalRead { + pub source: MemoryJournalSourceLink, + /// Absent for an unavailable or explicitly erased target; never falls back to another entry. + pub page: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RelatedMemoryCursor { + pub memory_id: String, + pub change_id: String, + pub candidate_id: String, + pub source_index: usize, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RelatedMemory { + pub selector: MemoryJournalSelector, + pub source_index: usize, + pub reconsideration: bool, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct RelatedMemories { + pub items: Vec, + pub next: Option, +} + +fn invalid(message: &str) -> JournalError { + JournalError::InvalidInput(message.into()) +} + +fn identifier(value: &str) -> JournalResult<()> { + if value.is_empty() || value.len() > 256 || value.chars().any(char::is_control) { + return Err(invalid("invalid Memory/Journal selector identifier")); + } + Ok(()) +} + +fn selector_valid(selector: &MemoryJournalSelector) -> JournalResult<()> { + identifier(&selector.memory_id)?; + identifier(&selector.change_id)?; + identifier(&selector.candidate_id) +} + +fn candidate( + store: &SubjektivStore, + tx: &Connection, + host: &JournalHostContext, + selector: &MemoryJournalSelector, +) -> JournalResult { + super::store::check_host(store, tx, host, false)?; + selector_valid(selector)?; + // The immutable change-candidate edge, not the current Memory projection or a supplied + // candidate ID alone, authorizes this provenance selector. + let raw: Option = tx + .query_row( + "SELECT s.record_json FROM staging_records s JOIN memory_change_candidates c + ON c.subject_id=s.subject_id AND c.candidate_id=s.candidate_id + WHERE c.subject_id=?1 AND c.memory_id=?2 AND c.change_id=?3 AND c.candidate_id=?4", + params![ + host.subject_id, + selector.memory_id, + selector.change_id, + selector.candidate_id + ], + |r| r.get(0), + ) + .optional()?; + let record: SubjectStagingRecord = serde_json::from_str( + &raw.ok_or_else(|| JournalError::NotFound("Memory change candidate membership".into()))?, + )?; + Ok(record) +} + +#[derive(Debug)] +struct SavedLink { + reason: String, + journal_id: Option, + entry_id: Option, + session_id: Option, + order: Option, +} + +fn saved( + tx: &Connection, + host: &JournalHostContext, + candidate_id: &str, + index: usize, +) -> JournalResult> { + Ok(tx.query_row( + "SELECT reason,journal_id,entry_id,source_session_id,source_order FROM memory_journal_links + WHERE subject_id=?1 AND candidate_id=?2 AND source_index=?3", + params![host.subject_id, candidate_id, index], + |r| Ok(SavedLink { reason:r.get(0)?,journal_id:r.get(1)?,entry_id:r.get(2)?,session_id:r.get(3)?,order:r.get(4)? }), + ).optional()?) +} + +fn original_availability( + tx: &Connection, + journal: &str, + order: u64, +) -> JournalResult { + let unavailable: Option = tx + .query_row( + "SELECT unavailable_json FROM journal_sources WHERE journal_id=?1 AND source_order=?2", + params![journal, order], + |r| r.get(0), + ) + .optional()?; + Ok(match unavailable { + None => OriginalSourceAvailability::Unknown, + Some(raw) => { + let ranges: Vec = serde_json::from_str(&raw)?; + if ranges.is_empty() { + OriginalSourceAvailability::Available + } else { + OriginalSourceAvailability::Missing + } + } + }) +} + +fn project( + tx: &Connection, + host: &JournalHostContext, + candidate_id: &str, + index: usize, +) -> JournalResult { + let Some(link) = saved(tx, host, candidate_id, index)? else { + return Ok(MemoryJournalSourceLink { + source_index: index, + original_source: OriginalSourceAvailability::Unknown, + connection: SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled, + }, + }); + }; + if link.reason != "linked" { + let reason = match link.reason.as_str() { + "insufficient_source" => UnconnectedReason::InsufficientSource, + "source_not_admitted" => UnconnectedReason::SourceNotAdmitted, + "ambiguous_source" => UnconnectedReason::AmbiguousSource, + "journal_not_fixed" => UnconnectedReason::JournalNotFixed, + _ => return Err(invalid("invalid saved connection reason")), + }; + return Ok(MemoryJournalSourceLink { + source_index: index, + original_source: OriginalSourceAvailability::Unknown, + connection: SourceConnection::Unconnected { reason }, + }); + } + let journal_id = link + .journal_id + .ok_or_else(|| invalid("saved link lacks Journal"))?; + let entry_id = link + .entry_id + .ok_or_else(|| invalid("saved link lacks entry"))?; + let session = link + .session_id + .ok_or_else(|| invalid("saved link lacks Session"))?; + let order = link + .order + .ok_or_else(|| invalid("saved link lacks source order"))?; + let erased: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM journal_erased_sessions WHERE subject_id=?1 AND session_id=?2)", + params![host.subject_id, session], |r| r.get(0), + )?; + let exists: bool = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM journal_entries e JOIN journals j ON j.id=e.journal_id + WHERE e.subject_id=?1 AND e.journal_id=?2 AND e.id=?3 AND j.source_session_id=?4)", + params![host.subject_id, journal_id, entry_id, session], + |r| r.get(0), + )?; + let reconsideration = tx.query_row( + "SELECT EXISTS(SELECT 1 FROM journal_corrections WHERE subject_id=?1 AND target_id=?2)", + params![host.subject_id, entry_id], + |r| r.get(0), + )?; + Ok(MemoryJournalSourceLink { + source_index: index, + original_source: original_availability(tx, &journal_id, order)?, + connection: SourceConnection::Linked { + journal_id, + entry_id, + target: if erased { + JournalTargetAvailability::Erased + } else if exists { + JournalTargetAvailability::Available + } else { + JournalTargetAvailability::Unavailable + }, + reconsideration, + }, + }) +} + +struct Resolved { + journal: String, + entry: String, + session: String, + order: u64, +} + +fn resolve( + tx: &Connection, + host: &JournalHostContext, + source: &SourceEvidenceRef, +) -> JournalResult> { + let Some(evidence) = source.evidence_id.as_deref() else { + return Ok(Err("insufficient_source")); + }; + if !evidence.starts_with('E') + || evidence.len() < 2 + || evidence.len() > 129 + || !evidence[1..] + .bytes() + .all(|c| c.is_ascii_alphanumeric() || c == b'-') + || (source.session_id.is_none() && source.segment_id.is_none()) + { + return Ok(Err("insufficient_source")); + } + // Segment identity may be omitted only when an immutable Session attribution disambiguates it. + // A candidate's primary legacy source range is not substituted for an individual anchor. + let segment = match source.segment_id.as_deref() { + Some(raw) => match uuid::Uuid::parse_str(raw) { + Ok(id) => Some(id), + Err(_) => return Ok(Err("insufficient_source")), + }, + None => None, + }; + let mut stmt = tx.prepare( + "SELECT j.id,j.source_session_id,s.source_order,s.pointer_json FROM journal_sources s + JOIN journals j ON j.id=s.journal_id + JOIN subject_session_attributions a ON a.subject_id=j.subject_id AND a.session_id=j.source_session_id + WHERE j.subject_id=?1 AND s.entry_ref=?2 + AND (?3 IS NULL OR j.source_session_id=?3) + AND (?4 IS NULL OR json_extract(s.pointer_json,'$.segment_id')=?4) + LIMIT 2", + )?; + let found = stmt + .query_map( + params![ + host.subject_id, + evidence, + source.session_id, + segment.map(|s| s.to_string()) + ], + |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, String>(1)?, + r.get::<_, u64>(2)?, + r.get::<_, String>(3)?, + )) + }, + )? + .collect::, _>>()?; + if found.is_empty() { + return Ok(Err("source_not_admitted")); + } + if found.len() != 1 { + return Ok(Err("ambiguous_source")); + } + let (journal, session, order, pointer) = &found[0]; + let pointer: JournalSourcePointer = serde_json::from_str(pointer)?; + if pointer.entry_ref != evidence || segment.is_some_and(|s| s != pointer.segment_id) { + return Ok(Err("ambiguous_source")); + } + // A fixed endpoint covers (previous_fixed_end, source_end]. The Session start is the first + // interval's lower boundary. Processing checkpoints and open entries cannot supply an end. + let mut stmt = tx.prepare( + "SELECT e.id FROM journal_entries e WHERE e.subject_id=?1 AND e.journal_id=?2 + AND e.state='fixed' AND e.source_order>=?3 + AND NOT EXISTS(SELECT 1 FROM journal_entries p WHERE p.journal_id=e.journal_id + AND p.position=?3) + ORDER BY e.position LIMIT 2", + )?; + let entries = stmt + .query_map(params![host.subject_id, journal, order], |r| { + r.get::<_, String>(0) + })? + .collect::, _>>()?; + match entries.as_slice() { + [entry] => Ok(Ok(Resolved { + journal: journal.clone(), + entry: entry.clone(), + session: session.clone(), + order: *order, + })), + [] => Ok(Err("journal_not_fixed")), + _ => Ok(Err("ambiguous_source")), + } +} + +fn reconcile( + tx: &Connection, + host: &JournalHostContext, + record: &SubjectStagingRecord, + indices: std::ops::Range, +) -> JournalResult<()> { + for index in indices { + let source = &record.source_refs[index]; + if saved(tx, host, &record.id, index)?.is_some_and(|s| s.reason == "linked") { + continue; + } + let (reason, resolved) = match resolve(tx, host, source)? { + Ok(link) => ("linked", Some(link)), + Err(reason) => (reason, None), + }; + tx.execute( + "INSERT INTO memory_journal_links(subject_id,candidate_id,source_index,reason, + journal_id,entry_id,source_session_id,source_order) VALUES(?1,?2,?3,?4,?5,?6,?7,?8) + ON CONFLICT(subject_id,candidate_id,source_index) DO UPDATE SET + reason=excluded.reason,journal_id=excluded.journal_id,entry_id=excluded.entry_id, + source_session_id=excluded.source_session_id,source_order=excluded.source_order", + params![ + host.subject_id, + record.id, + index, + reason, + resolved.as_ref().map(|r| r.journal.as_str()), + resolved.as_ref().map(|r| r.entry.as_str()), + resolved.as_ref().map(|r| r.session.as_str()), + resolved.as_ref().map(|r| r.order) + ], + )?; + } + Ok(()) +} + +impl SubjektivStore { + /// Scoped Memory provenance bridge. Reconcile is additive and idempotent, so a lost response + /// needs no operation ledger: retry the same selector and reread saved edges. Reads never reconcile. + pub fn execute_memory_journal( + &self, + host: &JournalHostContext, + request: MemoryJournalRequest, + ) -> JournalResult { + let response = match request { + MemoryJournalRequest::Reconcile { selector } => { + self.memory_journal_links(host, selector, true, 0, MAX_STAGING_ANCHORS)? + } + MemoryJournalRequest::Links { selector } => { + self.memory_journal_links(host, selector, false, 0, MAX_STAGING_ANCHORS)? + } + MemoryJournalRequest::ReconcilePage { + selector, + source_offset, + limit, + } => self.memory_journal_links(host, selector, true, source_offset, limit)?, + MemoryJournalRequest::LinksPage { + selector, + source_offset, + limit, + } => self.memory_journal_links(host, selector, false, source_offset, limit)?, + MemoryJournalRequest::Read { + selector, + source_index, + journal_id, + entry_id, + expected_token, + byte_offset, + max_bytes, + } => { + identifier(&journal_id)?; + identifier(&entry_id)?; + if !(4..=super::MAX_JOURNAL_READ_BYTES).contains(&max_bytes) { + return Err(invalid("text page must be 4..16384 bytes")); + } + if let Some(token) = expected_token.as_deref() { + identifier(token)?; + } + let mut source = self.database.try_with_connection(|tx| { + let record = candidate(self, tx, host, &selector)?; + if source_index >= record.source_refs.len() { + return Err(invalid("source index outside candidate anchors")); + } + let link = project(tx, host, &selector.candidate_id, source_index)?; + match &link.connection { + SourceConnection::Linked { + journal_id: saved_journal, + entry_id: saved_entry, + .. + } if saved_journal == &journal_id && saved_entry == &entry_id => Ok(link), + _ => Err(JournalError::PermissionDenied( + "read requires the exact saved Journal link".into(), + )), + } + })?; + let page = match &source.connection { + SourceConnection::Linked { + target: JournalTargetAvailability::Available, + .. + } => { + match self.read_journal_text( + host, + &journal_id, + Some(&entry_id), + JournalTextField::Body, + expected_token.as_deref(), + byte_offset, + max_bytes, + ) { + Ok(page) => Some(page), + // Erasure between the saved-edge check and shared text read must not + // fall back, expose cached text, or pretend the link never existed. + Err(JournalError::NotFound(_)) => { + source = self.database.try_with_connection(|tx| { + project(tx, host, &selector.candidate_id, source_index) + })?; + None + } + Err(error) => return Err(error), + } + } + _ => None, + }; + MemoryJournalResponse::Read(MemoryJournalRead { source, page }) + } + MemoryJournalRequest::Related { + journal_id, + entry_id, + after, + limit, + } => self.related_memories(host, &journal_id, &entry_id, after.as_ref(), limit)?, + }; + if serde_json::to_string_pretty(&response)?.len() > MAX_JOURNAL_RESPONSE_BYTES { + return Err(invalid("Memory Journal response exceeds byte budget")); + } + Ok(response) + } + + fn memory_journal_links( + &self, + host: &JournalHostContext, + selector: MemoryJournalSelector, + do_reconcile: bool, + source_offset: usize, + limit: usize, + ) -> JournalResult { + if !(1..=MAX_STAGING_ANCHORS).contains(&limit) { + return Err(invalid("source page exceeds staging anchor bound")); + } + self.database.try_transaction(|tx| { + let record = candidate(self, tx, host, &selector)?; + let total_sources = record.source_refs.len(); + if source_offset > total_sources { + return Err(invalid("source offset outside immutable candidate anchors")); + } + let end = source_offset.saturating_add(limit).min(total_sources); + if do_reconcile { + reconcile(tx, host, &record, source_offset..end)?; + } + let sources = (source_offset..end) + .map(|index| project(tx, host, &selector.candidate_id, index)) + .collect::>>()?; + Ok(MemoryJournalResponse::Links(MemoryJournalLinks { + selector, + sources, + total_sources, + next_source_index: (end < total_sources).then_some(end), + })) + }) + } + + fn related_memories( + &self, + host: &JournalHostContext, + journal_id: &str, + entry_id: &str, + after: Option<&RelatedMemoryCursor>, + limit: usize, + ) -> JournalResult { + identifier(journal_id)?; + identifier(entry_id)?; + if !(1..=MAX_JOURNAL_PAGE).contains(&limit) { + return Err(invalid("related page must be 1..32")); + } + if let Some(cursor) = after { + identifier(&cursor.memory_id)?; + identifier(&cursor.change_id)?; + identifier(&cursor.candidate_id)?; + if i64::try_from(cursor.source_index).is_err() { + return Err(invalid("invalid related source index")); + } + } + self.database.try_with_connection(|tx| { + super::store::check_host(self,tx,host,false)?; + // Saved target identity authorizes reverse lookup after erasure as well as before it. + let known:bool=tx.query_row( + "SELECT EXISTS(SELECT 1 FROM journal_entries WHERE subject_id=?1 AND journal_id=?2 AND id=?3) + OR EXISTS(SELECT 1 FROM memory_journal_links WHERE subject_id=?1 AND journal_id=?2 AND entry_id=?3)", + params![host.subject_id,journal_id,entry_id],|r|r.get(0))?; + if !known { return Err(JournalError::NotFound("Journal entry".into())); } + let mut stmt=tx.prepare( + "SELECT m.memory_id,m.change_id,l.candidate_id,l.source_index, + EXISTS(SELECT 1 FROM journal_corrections c WHERE c.subject_id=l.subject_id AND c.target_id=l.entry_id) + FROM memory_journal_links l JOIN memory_change_candidates m + ON m.subject_id=l.subject_id AND m.candidate_id=l.candidate_id + WHERE l.subject_id=?1 AND l.reason='linked' + AND ((l.journal_id=?2 AND l.entry_id=?3) + OR EXISTS(SELECT 1 FROM journal_corrections c WHERE c.subject_id=?1 AND c.entry_id=?3 AND c.target_id=l.entry_id)) + AND (?4 IS NULL OR (m.memory_id,m.change_id,l.candidate_id,l.source_index)>(?4,?5,?6,?7)) + ORDER BY m.memory_id,m.change_id,l.candidate_id,l.source_index LIMIT ?8")?; + let mut items=stmt.query_map(params![host.subject_id,journal_id,entry_id, + after.map(|c|c.memory_id.as_str()),after.map(|c|c.change_id.as_str()), + after.map(|c|c.candidate_id.as_str()),after.map(|c|c.source_index),limit+1], + |r|Ok(RelatedMemory {selector:MemoryJournalSelector {memory_id:r.get(0)?,change_id:r.get(1)?,candidate_id:r.get(2)?}, + source_index:r.get(3)?,reconsideration:r.get(4)?}))?.collect::,_>>()?; + let more=items.len()>limit; + items.truncate(limit); + let next=if more {items.last().map(|last|RelatedMemoryCursor { + memory_id:last.selector.memory_id.clone(),change_id:last.selector.change_id.clone(), + candidate_id:last.selector.candidate_id.clone(),source_index:last.source_index })} else {None}; + Ok(MemoryJournalResponse::Related(RelatedMemories {items,next})) + }) + } +} diff --git a/crates/subjektiv/src/journal/memory_links/schema.sql b/crates/subjektiv/src/journal/memory_links/schema.sql new file mode 100644 index 000000000..f29110c8a --- /dev/null +++ b/crates/subjektiv/src/journal/memory_links/schema.sql @@ -0,0 +1,33 @@ +-- References outlive explicit Journal erasure. Do not add target-row foreign keys: +-- lifecycle erasure deletes Journal content but retains its Session tombstone. +CREATE TABLE memory_journal_links ( + subject_id TEXT NOT NULL, + candidate_id TEXT NOT NULL, + source_index INTEGER NOT NULL CHECK(source_index >= 0), + reason TEXT NOT NULL CHECK(reason IN ( + 'insufficient_source', 'source_not_admitted', 'ambiguous_source', + 'journal_not_fixed', 'linked' + )), + journal_id TEXT, + entry_id TEXT, + source_session_id TEXT, + source_order INTEGER, + PRIMARY KEY(subject_id, candidate_id, source_index), + FOREIGN KEY(subject_id, candidate_id) + REFERENCES staging_records(subject_id, candidate_id) ON DELETE RESTRICT, + CHECK((reason = 'linked' AND journal_id IS NOT NULL AND entry_id IS NOT NULL + AND source_session_id IS NOT NULL AND source_order IS NOT NULL AND source_order >= 0) + OR (reason <> 'linked' AND journal_id IS NULL AND entry_id IS NULL + AND source_session_id IS NULL AND source_order IS NULL)) +); +CREATE INDEX memory_journal_links_reverse + ON memory_journal_links(subject_id, journal_id, entry_id, candidate_id, source_index); +CREATE TRIGGER memory_journal_links_identity BEFORE UPDATE ON memory_journal_links + WHEN OLD.subject_id <> NEW.subject_id OR OLD.candidate_id <> NEW.candidate_id + OR OLD.source_index <> NEW.source_index + BEGIN SELECT RAISE(ABORT, 'Journal source reference identity is immutable'); END; +CREATE TRIGGER memory_journal_links_connected_immutable BEFORE UPDATE ON memory_journal_links + WHEN OLD.reason = 'linked' + BEGIN SELECT RAISE(ABORT, 'connected Journal references are immutable'); END; +CREATE TRIGGER memory_journal_links_retained BEFORE DELETE ON memory_journal_links + BEGIN SELECT RAISE(ABORT, 'Journal references are retained'); END; diff --git a/crates/subjektiv/src/journal/memory_links/tests.rs b/crates/subjektiv/src/journal/memory_links/tests.rs new file mode 100644 index 000000000..439a111b3 --- /dev/null +++ b/crates/subjektiv/src/journal/memory_links/tests.rs @@ -0,0 +1,1085 @@ +//! Component coverage through real FeatureStorage migrations, staging/application and Journal APIs. +//! Only extraction's Host-resolved evidence metadata and Journal source bytes/Job binding are fixture +//! inputs. No history provider or Worker is present: direct reads cannot fetch/restore either. +//! Host lifecycle/transport integrations are tested by their owning suites, not simulated here. + +use super::*; +use crate::journal::*; +use crate::{MemoryChangeTarget, MemoryDraft, SubjectRole, SubjectSessionAttribution}; +use feature_storage::{FeatureStorage, ScopedFeatureStorage}; +use uuid::Uuid; + +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_migrations(crate::MIGRATIONS) + } + fn with_migrations(migrations: &'static [feature_storage::FeatureMigration]) -> Self { + let temp = tempfile::tempdir().unwrap(); + let manager = FeatureStorage::new(temp.path().join("storage")); + let workspace = manager + .workspace(&format!("memory-journal-test-{}", Uuid::now_v7())) + .unwrap(); + let registration = workspace + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + migrations, + )) + .unwrap(); + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + let subject = store + .create_subject(SubjectRole::new("companion").unwrap()) + .unwrap(); + let attr = + SubjectSessionAttribution::new(&subject.id, "runtime", "worker", "session").unwrap(); + store.record_session_attribution(attr.clone()).unwrap(); + let host = JournalHostContext::validated_session(&store, attr).unwrap(); + // Canonical order crosses two Segments; legacy numeric ranges below intentionally do not + // agree with this order. Stable E identities, not display positions, are the join contract. + let segments = [Uuid::now_v7(), Uuid::now_v7()]; + let ranges = (0..4) + .map(|order| JournalSourceRange { + source: JournalSource { + session_id: "session".into(), + pointer: JournalSourcePointer { + segment_id: segments[(order / 2) as usize], + entry_ref: format!("Eanchor-{order}"), + }, + order, + content_sha256: "a".repeat(64), + total_bytes: 32, + }, + bytes: JournalByteRange { start: 0, end: 32 }, + }) + .collect::>(); + let journal_id = store + .accept_journal_sources(&host, "Activity", &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(&self.host.scope_id).unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + self.store = Some(SubjektivStore::open(&workspace, ®istration).unwrap()); + self.workspace = Some(workspace); + self.manager = Some(manager); + assert_eq!(self.store().scope_id(), self.host.scope_id); + assert_eq!( + self.store().session_attribution("session").unwrap(), + self.host.attribution + ); + } + fn candidate(&self, id: &str, orders: &[usize]) -> SubjectStagingRecord { + // Reuse the existing Memory suite's valid extraction fixture, including HumanInput origin. + let mut record = + crate::tests::candidate(&self.host.subject_id, id, self.store().workspace_id()); + let evidence = record.evidence[0].clone(); + let source = record.source_refs[0].clone(); + record.evidence = orders + .iter() + .map(|order| { + let mut e = evidence.clone(); + e.id = self.ranges[*order].source.pointer.entry_ref.clone(); + e.entry_range = Some([100 + *order as u64, 100 + *order as u64]); + e + }) + .collect(); + record.source_refs = orders + .iter() + .enumerate() + .map(|(i, order)| { + let mut s = source.clone(); + s.session_id = Some("session".into()); + s.segment_id = Some(self.ranges[*order].source.pointer.segment_id.to_string()); + s.evidence_id = Some(record.evidence[i].id.clone()); + s.entry_range = record.evidence[i].entry_range; + s + }) + .collect(); + record.source.segment_id = self.ranges[0].source.pointer.segment_id.to_string(); + record + } + fn stage(&self, id: &str, orders: &[usize]) -> SubjectStagingRecord { + self.store() + .stage_candidate(self.candidate(id, orders)) + .unwrap() + } + fn apply(&self, record: &SubjectStagingRecord) -> MemoryJournalSelector { + let draft = MemoryDraft::active( + record.kind.clone(), + record.claim.clone(), + "Retain original evidence.", + record.why_useful.clone(), + "Reviewed original extraction.", + ); + let (memory, _) = self + .store() + .apply_candidate( + &self.host.subject_id, + &record.id, + MemoryChangeTarget::Create, + draft, + "adopt original extraction", + ) + .unwrap(); + MemoryJournalSelector { + memory_id: memory.id, + change_id: memory.change_id, + candidate_id: record.id.clone(), + } + } + fn action(&self, id: &str, action: JournalAction) -> JournalReceipt { + let binding = crate::api::JobAttemptBinding { + job_id: "journal-recording".into(), + attempt_id: "attempt".into(), + }; + let generation = self + .store() + .prepare_journal_generation( + &self.host, + &self.journal_id, + binding.clone(), + self.ranges.clone(), + ) + .unwrap(); + let context = + JournalJobContext::validated(self.store(), &self.host, &generation.id, binding) + .unwrap(); + let receipt = self + .store() + .execute_journal( + &context, + JournalOperation { + request_id: id.into(), + expected_state: self + .store() + .read_journal(&self.host, &self.journal_id) + .unwrap() + .state_token, + action, + covered_ranges: self.ranges.clone(), + }, + ) + .unwrap(); + receipt + } + fn fixed(&self, id: &str, end: usize, corrections: Vec) -> String { + self.action( + id, + JournalAction::Create { + title: id.into(), + body: format!("Journal context for {id}"), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(self.ranges[end].source.pointer.clone()), + correction_refs: corrections, + }, + ) + .entry_id + .unwrap() + } + fn links(&self, selector: &MemoryJournalSelector, reconcile: bool) -> MemoryJournalLinks { + let request = if reconcile { + MemoryJournalRequest::Reconcile { + selector: selector.clone(), + } + } else { + MemoryJournalRequest::Links { + selector: selector.clone(), + } + }; + let MemoryJournalResponse::Links(links) = self + .store() + .execute_memory_journal(&self.host, request) + .unwrap() + else { + panic!("links response") + }; + links + } + fn read( + &self, + selector: &MemoryJournalSelector, + index: usize, + entry: &str, + ) -> MemoryJournalRead { + let MemoryJournalResponse::Read(read) = self + .store() + .execute_memory_journal( + &self.host, + MemoryJournalRequest::Read { + selector: selector.clone(), + source_index: index, + journal_id: self.journal_id.clone(), + entry_id: entry.into(), + expected_token: None, + byte_offset: 0, + max_bytes: 4096, + }, + ) + .unwrap() + else { + panic!("read response") + }; + read + } + fn related( + &self, + entry: &str, + after: Option, + limit: usize, + ) -> RelatedMemories { + let MemoryJournalResponse::Related(page) = self + .store() + .execute_memory_journal( + &self.host, + MemoryJournalRequest::Related { + journal_id: self.journal_id.clone(), + entry_id: entry.into(), + after, + limit, + }, + ) + .unwrap() + else { + panic!("related response") + }; + page + } + fn snapshot(&self) -> Vec { + self.store() + .database + .try_with_connection(|tx| { + let mut values = Vec::new(); + for table in ["staging_records", "memory_changes", "memory_records"] { + let mut stmt = tx.prepare(&format!( + "SELECT record_json FROM {table} ORDER BY record_json" + ))?; + values.extend( + stmt.query_map([], |r| r.get::<_, String>(0))? + .collect::, _>>()?, + ); + } + Ok::<_, JournalError>(values) + }) + .unwrap() + } +} +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 linked(source: &MemoryJournalSourceLink) -> (&str, &str, JournalTargetAvailability, bool) { + let SourceConnection::Linked { + journal_id, + entry_id, + target, + reconsideration, + } = &source.connection + else { + panic!("expected linked source: {source:?}") + }; + (journal_id, entry_id, *target, *reconsideration) +} + +#[test] +fn existing_applied_candidate_links_preserve_original_evidence_and_memory_history() { + let mut f = Fixture::new(); + let staged = f.stage("existing", &[0]); + let selector = f.apply(&staged); + let before = f.snapshot(); + let entry = f.fixed("first", 1, vec![]); + let initial = f.links(&selector, false); + assert!(matches!( + initial.sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled + } + )); + let reconciled = f.links(&selector, true); + assert_eq!(linked(&reconciled.sources[0]).1, entry); + assert_eq!(f.snapshot(), before); + f.reopen(); + assert_eq!(f.links(&selector, false), reconciled); + assert_eq!(f.links(&selector, true), reconciled); + let retained = f + .store() + .staging_candidate(&f.host.subject_id, &staged.id) + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value(retained).unwrap(), + serde_json::to_value(staged).unwrap() + ); + assert_eq!(f.snapshot(), before); +} + +#[test] +fn missing_original_history_does_not_prevent_saved_link_body_read_after_reopen() { + let mut f = Fixture::new(); + let selector = f.apply(&f.stage("saved", &[0])); + let entry = f.fixed("first", 1, vec![]); + f.links(&selector, true); + // This is the durable Host availability attestation. The fixture contains no Session runtime + // or history backend at all; no read below has an opportunity to fetch original history. + f.store() + .mark_journal_sources_unavailable(&f.host, &f.journal_id, &[f.ranges[0].clone()]) + .unwrap(); + f.reopen(); + let read = f.read(&selector, 0, &entry); + assert_eq!(read.page.unwrap().text, "Journal context for first"); + assert_eq!( + read.source.original_source, + OriginalSourceAvailability::Missing + ); + assert_eq!(linked(&read.source).2, JournalTargetAvailability::Available); +} + +#[test] +fn original_source_metadata_failure_keeps_saved_journal_readable_and_does_not_claim_absence() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("metadata", &[0])); + let entry = f.fixed("first", 1, vec![]); + f.links(&selector, true); + for (observation, expected) in [ + (None, OriginalSourceAvailability::Unknown), + (Some(true), OriginalSourceAvailability::Unknown), + (Some(false), OriginalSourceAvailability::Missing), + ] { + let mut calls = Vec::new(); + let response = f + .store() + .execute_memory_journal_with_source_presence( + &f.host, + MemoryJournalRequest::Read { + selector: selector.clone(), + source_index: 0, + journal_id: f.journal_id.clone(), + entry_id: entry.clone(), + expected_token: None, + byte_offset: 0, + max_bytes: 4096, + }, + |session| { + calls.push(session.to_owned()); + observation + }, + ) + .unwrap(); + let MemoryJournalResponse::Read(read) = response else { + panic!("read required") + }; + assert_eq!(calls, [f.ranges[0].source.session_id.clone()]); + assert_eq!(read.source.original_source, expected); + assert_eq!(read.page.unwrap().text, "Journal context for first"); + } +} + +#[test] +fn multiple_segment_anchors_map_to_fixed_intervals_and_subject_crossing_is_denied() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("multi", &[0, 1, 2, 3])); + let first = f.fixed("first", 1, vec![]); + let second = f.fixed("second", 3, vec![]); + let links = f.links(&selector, true); + assert_eq!( + links + .sources + .iter() + .map(|s| linked(s).1) + .collect::>(), + vec![ + first.as_str(), + first.as_str(), + second.as_str(), + second.as_str() + ] + ); + let page = f.related(&first, None, 1); + assert_eq!(page.items.len(), 1); + let next = f.related(&first, page.next, 1); + assert_eq!(next.items.len(), 1); + assert!(next.next.is_none()); + assert_ne!(page.items[0].source_index, next.items[0].source_index); + let other = f + .store() + .create_subject(SubjectRole::new("other").unwrap()) + .unwrap(); + let host = JournalHostContext::validated_reader(f.store(), &other.id).unwrap(); + for request in [ + MemoryJournalRequest::Links { + selector: selector.clone(), + }, + MemoryJournalRequest::Reconcile { + selector: selector.clone(), + }, + MemoryJournalRequest::Read { + selector: selector.clone(), + source_index: 0, + journal_id: f.journal_id.clone(), + entry_id: first.clone(), + expected_token: None, + byte_offset: 0, + max_bytes: 100, + }, + MemoryJournalRequest::Related { + journal_id: f.journal_id.clone(), + entry_id: first.clone(), + after: None, + limit: 1, + }, + ] { + assert!(f.store().execute_memory_journal(&host, request).is_err()); + } + let mut wrong = selector.clone(); + wrong.change_id = "not-the-adopting-change".into(); + assert!( + f.store() + .execute_memory_journal(&f.host, MemoryJournalRequest::Reconcile { selector: wrong }) + .is_err() + ); + assert!( + f.store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::Read { + selector, + source_index: 0, + journal_id: f.journal_id.clone(), + entry_id: second, + expected_token: None, + byte_offset: 0, + max_bytes: 100 + } + ) + .is_err() + ); +} + +#[test] +fn delayed_fixed_endpoint_connects_only_on_explicit_reconcile_without_using_checkpoint() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("delayed", &[2])); + assert!(matches!( + f.links(&selector, true).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::JournalNotFixed + } + )); + let open = f + .action( + "open", + JournalAction::Create { + title: "In progress".into(), + body: "Not final".into(), + ongoing: "Still working".into(), + state: JournalEntryState::Open, + source_end: None, + correction_refs: vec![], + }, + ) + .entry_id + .unwrap(); + f.action( + "checkpoint", + JournalAction::Checkpoint { + record_operation_ids: vec!["open".into()], + }, + ); + assert!(matches!( + f.links(&selector, true).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::JournalNotFixed + } + )); + assert_eq!( + f.store() + .list_journal_entries(&f.host, &f.journal_id, None, 32) + .unwrap() + .items[0] + .state, + JournalEntryState::Open + ); + f.action( + "fix", + JournalAction::Fix { + entry_id: open.clone(), + source_end: f.ranges[3].source.pointer.clone(), + }, + ); + assert!(matches!( + f.links(&selector, false).sources[0].connection, + SourceConnection::Unconnected { .. } + )); + let connected = f.links(&selector, true); + assert_eq!(linked(&connected.sources[0]).1, open); + assert_eq!(f.links(&selector, true), connected); + assert_eq!( + f.store() + .list_journal_entries(&f.host, &f.journal_id, None, 32) + .unwrap() + .items + .len(), + 1 + ); +} + +#[test] +fn legacy_numeric_range_is_not_guessed_and_scope_context_cannot_cross_stores() { + let f = Fixture::new(); + let mut record = + crate::tests::candidate(&f.host.subject_id, "legacy", f.store().workspace_id()); + record.source_refs[0].session_id = Some("session".into()); + record.source_refs[0].segment_id = Some(f.ranges[0].source.pointer.segment_id.to_string()); + record.source_refs[0].evidence_id = None; + record.source_refs[0].entry_range = Some([0, 0]); + let record = f.store().stage_candidate(record).unwrap(); + let selector = f.apply(&record); + f.fixed("first", 1, vec![]); + let before = f.snapshot(); + assert!(matches!( + f.links(&selector, true).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::InsufficientSource + } + )); + assert_eq!(before, f.snapshot()); + let other = Fixture::new(); + assert!( + other + .store() + .execute_memory_journal(&f.host, MemoryJournalRequest::Links { selector }) + .is_err() + ); +} + +#[test] +fn correction_identifies_reconsideration_without_mutation_and_erasure_retains_reference_identity() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("corrected", &[0])); + let original = f.fixed("original", 1, vec![]); + f.links(&selector, true); + let before = f.snapshot(); + let correction = f.fixed("correction", 3, vec![original.clone()]); + let related = f.related(&correction, None, 32); + assert_eq!(related.items[0].selector, selector); + assert!(related.items[0].reconsideration); + assert!(linked(&f.links(&selector, false).sources[0]).3); + assert_eq!(before, f.snapshot()); + f.store() + .erase_journal_session(&f.host, "explicit-forgetting") + .unwrap(); + let erased = f.read(&selector, 0, &original); + assert!(erased.page.is_none()); + assert_eq!(linked(&erased.source).2, JournalTargetAvailability::Erased); + let again = f.links(&selector, true); + assert_eq!(linked(&again.sources[0]).1, original); + assert_eq!( + linked(&again.sources[0]).2, + JournalTargetAvailability::Erased + ); + assert_eq!(f.related(&original, None, 32).items[0].selector, selector); + assert_eq!(before, f.snapshot()); +} + +#[test] +fn malformed_wire_selectors_and_page_limits_are_rejected_without_side_effects() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("wire", &[0])); + assert!( + serde_json::from_value::(serde_json::json!({ + "action":"reconcile","selector":selector,"subject_id":"another-subject"})) + .is_err() + ); + let first = f.fixed("first", 1, vec![]); + f.links(&selector, true); + for limit in [0, MAX_JOURNAL_PAGE + 1] { + assert!( + f.store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::Related { + journal_id: f.journal_id.clone(), + entry_id: first.clone(), + after: None, + limit + } + ) + .is_err() + ); + } + let response = f.store().execute_memory_journal( + &f.host, + MemoryJournalRequest::Read { + selector, + source_index: 0, + journal_id: f.journal_id.clone(), + entry_id: first, + expected_token: Some("stale".into()), + byte_offset: 1, + max_bytes: 32, + }, + ); + assert!(matches!(response, Err(JournalError::Conflict(_)))); +} + +#[test] +fn missing_target_never_retargets_saved_edge_to_a_later_event() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("unavailable", &[0])); + let first = f.fixed("first", 1, vec![]); + f.links(&selector, true); + let later = f.fixed("later", 3, vec![]); + // Simulate unavailable storage rather than explicit forgetting. This must remain distinct + // from the erasure tombstone, and a subsequent reconciliation must not choose `later`. + f.store() + .database + .try_transaction(|tx| { + tx.execute("DELETE FROM journal_entries WHERE id=?1", [&first])?; + Ok::<_, JournalError>(()) + }) + .unwrap(); + let read = f.read(&selector, 0, &first); + assert!(read.page.is_none()); + assert_eq!( + linked(&read.source).2, + JournalTargetAvailability::Unavailable + ); + let saved = f.links(&selector, true); + assert_eq!(linked(&saved.sources[0]).1, first); + assert_ne!(linked(&saved.sources[0]).1, later); + assert_eq!(f.related(&first, None, 32).items[0].selector, selector); +} + +#[test] +fn ambiguous_saved_anchor_does_not_choose_between_attributed_sessions() { + let f = Fixture::new(); + f.fixed("first", 1, vec![]); + let attr = SubjectSessionAttribution::new( + &f.host.subject_id, + "runtime", + "other-worker", + "other-session", + ) + .unwrap(); + f.store().record_session_attribution(attr.clone()).unwrap(); + let other_host = JournalHostContext::validated_session(f.store(), attr).unwrap(); + let mut range = f.ranges[0].clone(); + range.source.session_id = "other-session".into(); + f.store() + .accept_journal_sources(&other_host, "Other", &[range]) + .unwrap(); + let mut candidate = + crate::tests::candidate(&f.host.subject_id, "ambiguous", f.store().workspace_id()); + candidate.evidence[0].id = f.ranges[0].source.pointer.entry_ref.clone(); + candidate.source_refs[0].evidence_id = Some(candidate.evidence[0].id.clone()); + candidate.source_refs[0].segment_id = Some(f.ranges[0].source.pointer.segment_id.to_string()); + candidate.source_refs[0].session_id = None; + let staged = f.store().stage_candidate(candidate).unwrap(); + let selector = f.apply(&staged); + assert!(matches!( + f.links(&selector, true).sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::AmbiguousSource + } + )); +} + +#[test] +fn reconcile_transaction_failure_rolls_back_all_sources_and_connected_edges_are_retained() { + let f = Fixture::new(); + let selector = f.apply(&f.stage("atomic", &[0, 2])); + f.fixed("first", 1, vec![]); + f.fixed("second", 3, vec![]); + let before = f.snapshot(); + f.store() + .database + .try_transaction(|tx| { + tx.execute_batch( + "CREATE TRIGGER reject_second_link BEFORE INSERT ON memory_journal_links + WHEN NEW.source_index=1 BEGIN SELECT RAISE(ABORT, 'injected link fault'); END;", + )?; + Ok::<_, JournalError>(()) + }) + .unwrap(); + assert!( + f.store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::Reconcile { + selector: selector.clone() + } + ) + .is_err() + ); + assert!(f.links(&selector, false).sources.iter().all(|s| matches!( + s.connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled + } + ))); + assert_eq!(before, f.snapshot()); + f.store() + .database + .try_transaction(|tx| { + tx.execute_batch("DROP TRIGGER reject_second_link")?; + Ok::<_, JournalError>(()) + }) + .unwrap(); + let links = f.links(&selector, true); + f.store() + .database + .try_with_connection(|tx| { + assert!( + tx.execute("UPDATE memory_journal_links SET entry_id='replacement'", []) + .is_err() + ); + assert!(tx.execute("DELETE FROM memory_journal_links", []).is_err()); + Ok::<_, JournalError>(()) + }) + .unwrap(); + assert_eq!(f.links(&selector, false), links); + assert_eq!(before, f.snapshot()); +} + +#[test] +fn historical_candidates_page_all_immutable_sources_without_reextracting_or_losing_late_indices() { + let mut f = Fixture::new(); + for order in 4..12 { + let mut range = f.ranges[3].clone(); + range.source.order = order; + range.source.pointer.entry_ref = format!("Eanchor-{order}"); + f.ranges.push(range); + } + f.store() + .accept_journal_sources(&f.host, "Activity", &f.ranges) + .unwrap(); + let staged = f.candidate("historical-many", &(0..12).collect::>()); + // Seed a historical saved record directly: new staging admission is capped at ten, but + // reading/adopting a preserved older record is not. This does not bypass link authorization. + f.store() + .database + .try_transaction(|tx| { + tx.execute( + "INSERT INTO staging_records(subject_id,candidate_id,kind,record_json,created_at) + VALUES(?1,?2,'constraint',?3,?4)", + params![ + staged.subject_id, + staged.id, + serde_json::to_string(&staged)?, + staged.created_at + ], + )?; + Ok::<_, JournalError>(()) + }) + .unwrap(); + let selector = f.apply(&staged); + let before = f.snapshot(); + let entry = f.fixed("all-anchors", 11, vec![]); + let first = f.links(&selector, true); + assert_eq!(first.total_sources, 12); + assert_eq!(first.sources.len(), MAX_STAGING_ANCHORS); + assert_eq!(first.next_source_index, Some(10)); + let MemoryJournalResponse::Links(unreconciled) = f + .store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::LinksPage { + selector: selector.clone(), + source_offset: 10, + limit: 2, + }, + ) + .unwrap() + else { + panic!("links required") + }; + assert!(unreconciled.sources.iter().all(|s| matches!( + s.connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled + } + ))); + let MemoryJournalResponse::Links(last) = f + .store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::ReconcilePage { + selector: selector.clone(), + source_offset: 10, + limit: 2, + }, + ) + .unwrap() + else { + panic!("links required") + }; + assert_eq!( + last.sources + .iter() + .map(|s| s.source_index) + .collect::>(), + vec![10, 11] + ); + assert!(last.next_source_index.is_none()); + assert_eq!(linked(&last.sources[1]).1, entry); + assert!(f.read(&selector, 11, &entry).page.is_some()); + let page = f.related(&entry, None, 11); + assert_eq!(page.next.as_ref().unwrap().source_index, 10); + assert_eq!(f.related(&entry, page.next, 11).items[0].source_index, 11); + f.reopen(); + assert_eq!(f.snapshot(), before); + for (offset, limit) in [(0, 0), (0, MAX_STAGING_ANCHORS + 1), (13, 1)] { + assert!( + f.store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::ReconcilePage { + selector: selector.clone(), + source_offset: offset, + limit + } + ) + .is_err() + ); + } + let MemoryJournalResponse::Links(saved) = f + .store() + .execute_memory_journal( + &f.host, + MemoryJournalRequest::LinksPage { + selector, + source_offset: 10, + limit: 2, + }, + ) + .unwrap() + else { + panic!("links required") + }; + assert_eq!(saved, last); +} + +fn before_links() -> &'static [feature_storage::FeatureMigration] { + &crate::MIGRATIONS[..crate::MIGRATIONS + .iter() + .position(|m| m.version() == 10) + .unwrap()] +} + +fn schema_objects(store: &SubjektivStore) -> Vec<(String, String, Option)> { + store.database.try_with_connection(|tx| { + let mut stmt=tx.prepare("SELECT type,name,sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%' ORDER BY type,name")?; + Ok::<_,JournalError>(stmt.query_map([], |r|Ok((r.get(0)?,r.get(1)?,r.get(2)?)))? + .collect::,_>>()?) + }).unwrap() +} + +#[test] +fn populated_pre_link_schema_upgrade_preserves_history_and_matches_fresh_schema() { + let mut old = Fixture::with_migrations(before_links()); + let staged = old.stage("pre-link-candidate", &[0, 2]); + let selector = old.apply(&staged); + let first = old.fixed("old-first", 1, vec![]); + old.fixed("old-second", 3, vec![]); + let before = old.snapshot(); + let prior_entries = old + .store() + .list_journal_entries(&old.host, &old.journal_id, None, 32) + .unwrap(); + assert!( + !schema_objects(old.store()) + .iter() + .any(|(_, name, _)| name == "memory_journal_links") + ); + old.reopen(); + assert_eq!(old.snapshot(), before); + assert_eq!( + old.store() + .list_journal_entries(&old.host, &old.journal_id, None, 32) + .unwrap(), + prior_entries + ); + assert_eq!(linked(&old.links(&selector, true).sources[0]).1, first); + assert_eq!(old.snapshot(), before); + let fresh = Fixture::new(); + assert_eq!(schema_objects(old.store()), schema_objects(fresh.store())); + // Verify the introduced constraints, not merely migration history numbers. + old.store() + .database + .try_with_connection(|tx| { + assert!( + tx.execute( + "INSERT INTO memory_journal_links(subject_id,candidate_id,source_index,reason) + VALUES(?1,?2,-1,'insufficient_source')", + params![old.host.subject_id, staged.id] + ) + .is_err() + ); + assert!( + tx.execute( + "INSERT INTO memory_journal_links(subject_id,candidate_id,source_index,reason) + VALUES(?1,'absent-candidate',0,'insufficient_source')", + [&old.host.subject_id] + ) + .is_err() + ); + assert!( + tx.execute("UPDATE memory_journal_links SET entry_id='retarget'", []) + .is_err() + ); + Ok::<_, JournalError>(()) + }) + .unwrap(); +} + +#[test] +fn oldest_supported_schema_upgrade_matches_fresh_and_preserves_staged_original_evidence() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let subject = "oldest-subject"; + let candidate = crate::tests::candidate(subject, "oldest-candidate", "oldest-scope"); + let raw = serde_json::to_string(&candidate).unwrap(); + { + let manager = FeatureStorage::new(&root); + let workspace = manager.workspace("oldest-scope").unwrap(); + let registration = workspace + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + &crate::MIGRATIONS[..1], + )) + .unwrap(); + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + let timestamp = crate::now(); + let subject_json = serde_json::json!({"schema_version":crate::SUBJEKTIV_SCHEMA_VERSION, + "id":subject,"role":"companion","state":"active","store_revision":0, + "created_at":timestamp,"updated_at":timestamp}); + store.database.try_transaction(|tx| { + tx.execute("INSERT INTO subjects(subject_id,role,state,store_revision,record_json,created_at,updated_at) + VALUES(?1,'companion','active',0,?2,?3,?3)",params![subject,subject_json.to_string(),timestamp])?; + tx.execute("INSERT INTO staging_records(subject_id,candidate_id,kind,record_json,created_at) + VALUES(?1,?2,'constraint',?3,?4)",params![subject,candidate.id,raw,timestamp])?; + Ok::<_,JournalError>(()) + }).unwrap(); + drop(store); + drop(workspace); + manager.shutdown().unwrap(); + } + let manager = FeatureStorage::new(&root); + let workspace = manager.workspace("oldest-scope").unwrap(); + let registration = SubjektivStore::register(&workspace).unwrap(); + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + let preserved = store + .staging_candidate(subject, &candidate.id) + .unwrap() + .unwrap(); + assert_eq!( + serde_json::to_value(preserved).unwrap(), + serde_json::to_value(candidate).unwrap() + ); + assert_eq!(store.subject(subject).unwrap().unwrap().id, subject); + let fresh = Fixture::new(); + assert_eq!(schema_objects(&store), schema_objects(fresh.store())); + drop(store); + drop(workspace); + manager.shutdown().unwrap(); +} + +fn fail_after_link_schema(tx: &Transaction<'_>) -> feature_storage::Result<()> { + super::migrate(tx)?; + Err(feature_storage::FeatureStorageError::operation( + "injected after link DDL", + )) +} + +#[test] +fn failed_link_migration_rolls_back_ddl_and_preserves_existing_data_for_retry() { + static FAILED_PLAN: std::sync::LazyLock> = + std::sync::LazyLock::new(|| { + crate::MIGRATIONS + .iter() + .map(|migration| { + if migration.version() == 10 { + feature_storage::FeatureMigration::new( + 10, + migration.name(), + fail_after_link_schema, + ) + } else { + *migration + } + }) + .collect() + }); + let mut f = Fixture::with_migrations(before_links()); + let selector = f.apply(&f.stage("rollback-candidate", &[0])); + let entry = f.fixed("before-failed-upgrade", 1, vec![]); + let before = f.snapshot(); + let before_schema = schema_objects(f.store()); + f.store.take(); + f.workspace.take(); + f.manager.take().unwrap().shutdown().unwrap(); + { + let manager = FeatureStorage::new(f.temp.path().join("storage")); + let workspace = manager.workspace(&f.host.scope_id).unwrap(); + let registration = workspace + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + FAILED_PLAN.as_slice(), + )) + .unwrap(); + assert!(matches!( + SubjektivStore::open(&workspace, ®istration), + Err(crate::SubjektivError::Storage( + feature_storage::FeatureStorageError::MigrationFailed { version: 10, .. } + )) + )); + drop(workspace); + manager.shutdown().unwrap(); + } + let manager = FeatureStorage::new(f.temp.path().join("storage")); + let workspace = manager.workspace(&f.host.scope_id).unwrap(); + let registration = workspace + .register(feature_storage::FeatureRegistration::new( + crate::SUBJEKTIV_FEATURE_ID, + before_links(), + )) + .unwrap(); + f.store = Some(SubjektivStore::open(&workspace, ®istration).unwrap()); + f.workspace = Some(workspace); + f.manager = Some(manager); + assert_eq!(f.snapshot(), before); + assert_eq!(schema_objects(f.store()), before_schema); + f.reopen(); + assert_eq!(linked(&f.links(&selector, true).sources[0]).1, entry); + assert_eq!(f.snapshot(), before); +} diff --git a/crates/subjektiv/src/journal/mod.rs b/crates/subjektiv/src/journal/mod.rs index c8d25c7cf..1a4a0f76a 100644 --- a/crates/subjektiv/src/journal/mod.rs +++ b/crates/subjektiv/src/journal/mod.rs @@ -38,9 +38,11 @@ //! traversing already reflected pages. Receipt/entry keysets remain append-stable. mod lifecycle; +pub mod memory_links; pub(crate) use lifecycle::migrate as migrate_lifecycle; pub mod recording; pub mod service; +mod source_status; mod store; #[cfg(test)] mod tests; diff --git a/crates/subjektiv/src/journal/source_status.rs b/crates/subjektiv/src/journal/source_status.rs new file mode 100644 index 000000000..58e5f322e --- /dev/null +++ b/crates/subjektiv/src/journal/source_status.rs @@ -0,0 +1,79 @@ +//! Optional Host metadata observations for original material, independent of saved Journal reading. +//! +//! Presence checks must not fetch original history or restore a Worker. A failed/unauthorized +//! metadata lookup is unknown, not missing, and cannot prevent a saved Journal body from being read. + +use super::memory_links::{ + MemoryJournalRequest, MemoryJournalResponse, OriginalSourceAvailability, +}; +use super::{JournalHostContext, JournalResult, SubjektivStore}; + +impl SubjektivStore { + /// Executes the saved-reference operation first, then annotates original-source absence using + /// Host-owned metadata only. `Some(false)` is confirmed absence; `None` is an unavailable lookup. + /// Presence alone cannot attest that a particular historical entry is still readable. + pub fn execute_memory_journal_with_source_presence( + &self, + host: &JournalHostContext, + request: MemoryJournalRequest, + mut presence: impl FnMut(&str) -> Option, + ) -> JournalResult { + let selector = match &request { + MemoryJournalRequest::Reconcile { selector } + | MemoryJournalRequest::Links { selector } + | MemoryJournalRequest::ReconcilePage { selector, .. } + | MemoryJournalRequest::LinksPage { selector, .. } + | MemoryJournalRequest::Read { selector, .. } => Some(selector.clone()), + MemoryJournalRequest::Related { .. } => None, + }; + let mut response = self.execute_memory_journal(host, request)?; + let Some(selector) = selector else { + return Ok(response); + }; + // Membership and scoped reader authority were checked by the operation. Candidate JSON is + // immutable; never infer source ownership from its presence without checking attribution. + let candidate = self.staging_candidate(&host.subject_id, &selector.candidate_id)?; + let mut observations = std::collections::HashMap::new(); + let mut annotate = + |link: &mut super::memory_links::MemoryJournalSourceLink| -> JournalResult<()> { + let source = candidate + .as_ref() + .and_then(|c| c.source_refs.get(link.source_index)); + let session = source.and_then(|s| s.session_id.as_deref()); + let observed = if let Some(session) = session { + if let Some(value) = observations.get(session) { + *value + } else { + let owned = self + .session_attribution(session)? + .is_some_and(|a| a.subject_id == host.subject_id); + let value = if owned { presence(session) } else { None }; + observations.insert(session.to_owned(), value); + value + } + } else { + None + }; + link.original_source = match observed { + Some(false) => OriginalSourceAvailability::Missing, + // Preserve explicit saved missing-byte attestations. Directory/registry presence + // does not prove the referenced entry survived rewind or storage corruption. + _ if link.original_source == OriginalSourceAvailability::Missing => { + OriginalSourceAvailability::Missing + } + _ => OriginalSourceAvailability::Unknown, + }; + Ok(()) + }; + match &mut response { + MemoryJournalResponse::Links(page) => { + for source in &mut page.sources { + annotate(source)?; + } + } + MemoryJournalResponse::Read(read) => annotate(&mut read.source)?, + MemoryJournalResponse::Related(_) => {} + } + Ok(response) + } +} diff --git a/crates/subjektiv/src/lib.rs b/crates/subjektiv/src/lib.rs index 473017cef..8a1a2887c 100644 --- a/crates/subjektiv/src/lib.rs +++ b/crates/subjektiv/src/lib.rs @@ -613,6 +613,11 @@ static MIGRATIONS: &[FeatureMigration] = &[ "fence erased journal Sessions and permit explicit content erasure", journal::migrate_lifecycle, ), + FeatureMigration::new( + 10, + "add immutable candidate Journal context links", + journal::memory_links::migrate, + ), ]; pub const REGISTRATION: FeatureRegistration = diff --git a/crates/worker/src/subjektiv.rs b/crates/worker/src/subjektiv.rs index b0db8159a..4f3cfe884 100644 --- a/crates/worker/src/subjektiv.rs +++ b/crates/worker/src/subjektiv.rs @@ -10,6 +10,7 @@ pub use crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifec pub use crate::feature::builtin::journal::{ JournalJobFinalizer, JournalJobHost, JournalJobToolsFeature, JournalRecordingFeature, }; +pub use subjektiv::journal::memory_links::{MemoryJournalRequest, MemoryJournalResponse}; pub use subjektiv::journal::transport; use transport::{JournalRecordingReason, JournalRecordingRequest, JournalRecordingResponse}; @@ -79,6 +80,17 @@ pub trait SubjektivHost: Send + Sync + 'static { "Journal recording not granted".into(), )) } + /// Additive provenance links and direct Journal reads, separate from recording/Memory mutation. + /// Restricted Job Hosts inherit denial rather than the main Subject reader capability. + fn memory_journal( + &self, + _context: &SubjektivHostContext, + _request: MemoryJournalRequest, + ) -> Result { + Err(SubjektivHostError::Unavailable( + "Memory Journal references not granted".into(), + )) + } /// Separate explicit grant for decision/surface execution; Profile policy /// alone must never authorize consolidation tools. fn consolidation_granted(&self) -> bool { @@ -207,6 +219,22 @@ fn decode( } impl SubjektivHost for BackendSubjektivHost { + fn memory_journal( + &self, + _context: &SubjektivHostContext, + request: MemoryJournalRequest, + ) -> Result { + decode( + self.client + .execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!("/api/w/{}/subjektiv/journal/memory", self.workspace_id), + serde_json::to_string(&request) + .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?, + )) + .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?, + ) + } fn journal_recording_granted(&self) -> bool { // This adapter exists only for explicitly enabled, authenticated Host connections. // The Backend still verifies the main Worker/Subject binding on every request. @@ -438,6 +466,12 @@ impl SubjektivHostConnection { } Ok(Self { host, context }) } + pub fn memory_journal( + &self, + request: MemoryJournalRequest, + ) -> Result { + self.host.memory_journal(&self.context, request) + } pub fn memory( &self, operation: SubjektivMemoryBackendOperation, diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index fa96254bb..216ef232b 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -28,6 +28,7 @@ use subjektiv::api::*; mod backend_job_profile; mod subjektiv_job; mod subjektiv_journal; +mod subjektiv_memory_journal; mod worker_operations; mod worker_protocol; pub(crate) use worker_operations::{ @@ -13123,6 +13124,7 @@ fn build_inner_router(api: WorkspaceApi) -> Router { post(scoped_submit_backend_job_result), ) .route("/api/w/{workspace_id}/subjektiv/journal/recording", post(subjektiv_journal::recording_route)) + .route("/api/w/{workspace_id}/subjektiv/journal/memory", post(subjektiv_memory_journal::route)) .route("/api/w/{workspace_id}/subjektiv/journal/job", post(subjektiv_journal::job_route)) .route("/api/w/{workspace_id}/subjektiv/journal/job/finish", post(subjektiv_journal::finish_route)) .fallback(get(static_or_spa_fallback)) @@ -36928,6 +36930,7 @@ mod tests { mod subject_spawn_tests; mod subjektiv_jobs_tests; mod subjektiv_journal_tests; + mod subjektiv_memory_journal_tests; mod ticket_evidence_tests; mod ticket_worker_claim_tests; mod ticket_worker_launch_tests; diff --git a/crates/workspace-server/src/server/subjektiv_memory_journal.rs b/crates/workspace-server/src/server/subjektiv_memory_journal.rs new file mode 100644 index 000000000..96c1de16b --- /dev/null +++ b/crates/workspace-server/src/server/subjektiv_memory_journal.rs @@ -0,0 +1,107 @@ +//! Authenticated Subject-body access to durable Memory/Journal links. +//! +//! The current singleton grants access; identifiers in the request do not. Saved +//! Journal reads deliberately use a reader context, not a source Session, Runtime +//! lookup or restore. Domain storage owns link matching, persistence and paging. +use super::*; +use subjektiv::journal::memory_links::{MemoryJournalRequest, MemoryJournalResponse}; +use subjektiv::journal::{JournalError, JournalHostContext}; + +fn domain(error: JournalError) -> Error { + match error { + JournalError::PermissionDenied(message) => Error::WorkspacePermissionDenied(message), + JournalError::Conflict(message) => Error::RepositoryConflict(message), + JournalError::InvalidInput(message) => Error::InvalidInput(message), + JournalError::NotFound(id) => { + Error::InvalidInput(format!("Journal record not found: {id}")) + } + JournalError::Subject(error) => subjektiv_store_error(error), + other => Error::Store(other.to_string()), + } +} + +pub(super) async fn route( + State(api): State, + AxumPath(path): AxumPath, + source: Option>, + Json(request): Json, +) -> ApiResult> { + let source = source + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "Memory Journal requires signed Runtime request identity".into(), + ) + })? + .0; + let worker_id = source.worker_id.ok_or_else(|| { + Error::WorkspacePermissionDenied("Memory Journal requires bound Worker identity".into()) + })?; + let worker = RuntimeWorkerRef::new(source.runtime_id, worker_id); + // This is entirely store-owned, including callbacks while a Host parent waits + // on Runtime. Do not put it behind Host or consult live Runtime/config state. + api.store + .db_executor() + .run(move || { + validate_workspace_scope(&api, &path.workspace_id)?; + let projection = api + .store + .worker_registry_projection(&path.workspace_id, &worker)? + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "Memory Journal Worker registration missing".into(), + ) + })?; + // No Job grant implicitly expands to all of a Subject's saved Journal. + if projection.job.is_some() { + return Err(Error::WorkspacePermissionDenied( + "Memory Journal is not authorized for Job Workers".into(), + ) + .into()); + } + let lease = api + .store + .require_current_worker_singleton_owner(&path.workspace_id, &worker)? + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "Memory Journal requires the current Subject Worker".into(), + ) + })?; + let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?; + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Subject, + "Memory Journal access", + )?; + let store = open_subjektiv_store(&api)?; + let host = JournalHostContext::validated_reader(&store, subject_id).map_err(domain)?; + store + .execute_memory_journal_with_source_presence(&host, request, |session| { + // Absence is known only when neither retained Worker registration nor an + // archive location remains. Runtime disconnection is not material deletion. + // These SQL lookups never fetch Session bytes or restore the old Worker. + let attribution = store.session_attribution(session).ok()??; + if attribution.subject_id != subject_id { + return None; + } + let target = + RuntimeWorkerRef::new(&attribution.runtime_id, &attribution.worker_id); + let archive = api + .config_store + .worker_session_archive(&path.workspace_id, &target, session) + .ok()?; + let registered = api + .store + .get_worker_registry(&path.workspace_id, &target) + .ok()?; + if archive.is_none() && registered.is_none() { + Some(false) + } else { + None + } + }) + .map(Json) + .map_err(domain) + .map_err(ApiError::from) + }) + .await +} diff --git a/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs new file mode 100644 index 000000000..f462f7eea --- /dev/null +++ b/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs @@ -0,0 +1,728 @@ +//! Connected ordinary library tests: production router, signed Runtime authentication, +//! singleton/Job authority and Subjektiv FeatureStorage/SQLite. Extraction output and +//! Host admission/generation are fixture inputs; staging/adoption and canonical Session +//! serialization are real. No Runtime execution, model, external config or network is used. +use super::*; +use subjektiv::journal::memory_links::*; +use subjektiv::journal::transport::{JournalJobRequest, JournalTextRead, JournalTextTarget}; +use subjektiv::journal::*; + +const RUNTIME: &str = "runtime-memory-journal"; +const JOURNAL_BODY: &str = + "A was proposed; B was chosen after checking the constraint. Work remains."; + +fn route_path() -> String { + format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/journal/memory") +} + +async fn response(app: Router, request: Request) -> (StatusCode, Vec) { + let response = app.oneshot(request).await.unwrap(); + let status = response.status(); + let bytes = to_bytes(response.into_body(), 128 * 1024).await.unwrap(); + (status, bytes.to_vec()) +} + +struct Fixture { + _dir: tempfile::TempDir, + api: WorkspaceApi, + identity: worker_runtime::auth::RuntimeIdentityMaterial, + store: crate::subjektiv::SubjektivStore, + host: JournalHostContext, + job: JournalJobContext, + worker: RuntimeWorkerRef, + source_worker: RuntimeWorkerRef, + source_root: PathBuf, + selector: MemoryJournalSelector, + journal_id: String, + entry_id: String, + ranges: Vec, +} + +impl Fixture { + async fn new() -> Self { + use session_store::{LogEntry, Store as _, WorkerSessionStore}; + let dir = tempfile::tempdir().unwrap(); + let mut api = test_api(dir.path()).await; + let identity = worker_runtime::auth::RuntimeIdentityMaterial::generate(RUNTIME).unwrap(); + configure_runtime_request_auth(&mut api, &identity, RUNTIME); + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("memory-journal").unwrap()) + .unwrap(); + let worker = Self::body_worker(&api, &subject.id); + let source_worker = RuntimeWorkerRef::new(RUNTIME, Uuid::new_v4().to_string()); + seed_worker_source_member(&api, RUNTIME, &source_worker.worker_id); + let session = session_store::new_session_id(); + let segment = Uuid::new_v4(); + let source_root = dir.path().join("old-worker/session"); + let history = [ + "Choose B after checking the constraint", + "Correction: B still needs validation", + ] + .into_iter() + .enumerate() + .map(|(index, text)| session_store::LoggedHistoryEntry { + item: session_store::LoggedItem::Message { + role: session_store::LoggedRole::User, + content: vec![session_store::LoggedContentPart::Text { text: text.into() }], + }, + metadata: session_store::LoggedSessionHistoryMetadata { + entry_id: session_store::LoggedSessionHistoryEntryId(format!("original-{index}")), + origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, + derivation: None, + }, + }) + .collect(); + WorkerSessionStore::new(&source_root) + .unwrap() + .create_segment( + session, + segment, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: None, + config: agen::llm_client::RequestConfig::default(), + history, + forked_from: None, + compacted_from: None, + }], + ) + .unwrap(); + let reader = session_store::public_material::SessionPublicMaterialReader::open_worker( + &source_root, + segment, + Default::default(), + ) + .unwrap(); + let ranges: Vec<_> = reader + .descriptors() + .map(|d| JournalSourceRange { + source: JournalSource { + session_id: session.to_string(), + pointer: JournalSourcePointer { + segment_id: d.identity.segment_id, + entry_ref: d.identity.entry_ref.clone(), + }, + order: d.canonical_order, + content_sha256: d.sha256.clone(), + total_bytes: d.total_bytes, + }, + bytes: JournalByteRange { + start: 0, + end: d.total_bytes, + }, + }) + .collect(); + assert_eq!(ranges.len(), 2); + let attr = store + .record_session_attribution( + crate::subjektiv::SubjectSessionAttribution::new( + &subject.id, + RUNTIME, + &source_worker.worker_id, + session.to_string(), + ) + .unwrap(), + ) + .unwrap(); + let host = JournalHostContext::validated_session(&store, attr.clone()).unwrap(); + let journal_id = store + .accept_journal_sources(&host, "original activity", &ranges) + .unwrap(); + let binding = subjektiv::api::JobAttemptBinding { + job_id: "fixture-journal-generation".into(), + attempt_id: "fixture-attempt".into(), + }; + let generation = store + .prepare_journal_generation(&host, &journal_id, binding.clone(), ranges.clone()) + .unwrap(); + let job = JournalJobContext::validated(&store, &host, &generation.id, binding).unwrap(); + for (descriptor, range) in reader.descriptors().zip(&ranges) { + subjektiv::journal::service::execute_bounded( + &store, + &job, + JournalJobRequest::ReadText { + read: JournalTextRead { + target: JournalTextTarget::Source { + pointer: range.source.pointer.clone(), + }, + offset: 0, + limit: MAX_JOURNAL_READ_BYTES, + content_token: None, + }, + }, + |source, start, limit| { + assert_eq!(source, &range.source); + let page = reader.read_content(descriptor, start, limit).unwrap(); + assert_eq!(page.next_offset, range.bytes.end); + Ok(subjektiv::journal::recording::JournalMaterial { + range: range.clone(), + text: page.content, + }) + }, + ) + .unwrap(); + } + let receipt = store + .execute_journal( + &job, + JournalOperation { + request_id: "initial-open".into(), + expected_state: generation.initial_journal.state_token, + action: JournalAction::Create { + title: "Choosing the approach".into(), + body: JOURNAL_BODY.into(), + ongoing: "Validation pending".into(), + state: JournalEntryState::Open, + source_end: None, + correction_refs: vec![], + }, + covered_ranges: vec![ranges[0].clone()], + }, + ) + .unwrap(); + let entry_id = receipt.entry_id.unwrap(); + let kind = memory::extract::CandidateKind::Lesson; + let evidence_id = ranges[0].source.pointer.entry_ref.clone(); + let candidate = memory::extract::StagingRecord::from_candidate( + "extracted-candidate", + "extraction-run", + memory::schema::SourceRef { + segment_id: segment.to_string(), + range: [0, 0], + }, + memory::extract::ExtractedCandidate { + kind: kind.clone(), + claim: "Check the constraint before choosing".into(), + why_useful: "Avoid repeating the rejected approach".into(), + staleness: None, + evidence_ids: vec![evidence_id.clone()], + }, + vec![memory::extract::StagingEvidence { + id: evidence_id.clone(), + kind: memory::schema::EvidenceKind::new("message"), + entry_range: Some([0, 0]), + origin: Some(serde_json::from_value(json!({"kind":"human_input"})).unwrap()), + excerpt: Some("Choose B after checking the constraint".into()), + summary: None, + }], + vec![memory::schema::SourceEvidenceRef { + session_id: Some(attr.session_id), + segment_id: Some(segment.to_string()), + entry_range: Some([0, 0]), + evidence_id: Some(evidence_id), + origin: Some(serde_json::from_value(json!({"kind":"human_input"})).unwrap()), + evidence_kind: Some(memory::schema::EvidenceKind::new("message")), + ..Default::default() + }], + ); + let (staged, _) = store + .stage_candidate_with_attribution( + crate::subjektiv::SubjectStagingRecord::attach(&subject.id, candidate), + store + .session_attribution(&session.to_string()) + .unwrap() + .unwrap(), + ) + .unwrap(); + let (memory, _) = store + .apply_candidate( + &subject.id, + &staged.id, + crate::subjektiv::MemoryChangeTarget::Create, + crate::subjektiv::MemoryDraft::active( + kind, + &staged.claim, + "Check before choosing.", + &staged.why_useful, + "Adopt the extracted lesson", + ), + "Accepted after comparison", + ) + .unwrap(); + let selector = MemoryJournalSelector { + memory_id: memory.id, + change_id: memory.change_id, + candidate_id: staged.id, + }; + Self { + _dir: dir, + api, + identity, + store, + host, + job, + worker, + source_worker, + source_root, + selector, + journal_id, + entry_id, + ranges, + } + } + + fn body_worker(api: &WorkspaceApi, subject: &str) -> RuntimeWorkerRef { + let memory = api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let key = crate::subjektiv::subject_worker_singleton_key(subject).unwrap(); + let reservation = api + .config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + RUNTIME, + &format!("body-{subject}"), + &format!("body-input-{subject}"), + Some(&key), + &memory, + ) + .unwrap(); + let worker = RuntimeWorkerRef::new(RUNTIME, reservation.worker_id.to_string()); + seed_worker_source_member(api, RUNTIME, &worker.worker_id); + api.config_store + .complete_worker_create_reservation(TEST_WORKSPACE_ID, reservation.worker_id) + .unwrap(); + worker + } + + fn request(&self, worker: Option<&str>, request: &MemoryJournalRequest) -> Request { + runtime_source_request( + &self.identity, + worker, + "POST", + &route_path(), + serde_json::to_vec(request).unwrap(), + ) + } + + async fn call_as( + &self, + worker: &RuntimeWorkerRef, + request: MemoryJournalRequest, + ) -> (StatusCode, Vec) { + response( + build_router(self.api.clone()), + self.request(Some(&worker.worker_id), &request), + ) + .await + } + + async fn call(&self, request: MemoryJournalRequest) -> MemoryJournalResponse { + let (status, bytes) = self.call_as(&self.worker, request).await; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + serde_json::from_slice(&bytes).unwrap() + } + + fn fix_initial(&self) { + self.store + .execute_journal( + &self.job, + JournalOperation { + request_id: "fix-initial".into(), + expected_state: self + .store + .read_journal(&self.host, &self.journal_id) + .unwrap() + .state_token, + action: JournalAction::Fix { + entry_id: self.entry_id.clone(), + source_end: self.ranges[0].source.pointer.clone(), + }, + covered_ranges: vec![self.ranges[0].clone()], + }, + ) + .unwrap(); + } + + fn add_correction(&self) -> String { + self.store + .execute_journal( + &self.job, + JournalOperation { + request_id: "later-correction".into(), + expected_state: self + .store + .read_journal(&self.host, &self.journal_id) + .unwrap() + .state_token, + action: JournalAction::Create { + title: "Correction".into(), + body: "The chosen approach still needs validation.".into(), + ongoing: String::new(), + state: JournalEntryState::Fixed, + source_end: Some(self.ranges[1].source.pointer.clone()), + correction_refs: vec![self.entry_id.clone()], + }, + covered_ranges: vec![self.ranges[1].clone()], + }, + ) + .unwrap() + .entry_id + .unwrap() + } + + fn read_request(&self) -> MemoryJournalRequest { + MemoryJournalRequest::Read { + selector: self.selector.clone(), + source_index: 0, + journal_id: self.journal_id.clone(), + entry_id: self.entry_id.clone(), + expected_token: None, + byte_offset: 0, + max_bytes: 4096, + } + } +} + +#[tokio::test] +async fn memory_journal_router_requires_signed_bound_worker_not_user_or_forged_identity() { + let f = Fixture::new().await; + let request = MemoryJournalRequest::Links { + selector: f.selector.clone(), + }; + let body = serde_json::to_vec(&request).unwrap(); + let app = build_router(f.api.clone()); + let anonymous = Request::builder() + .method("POST") + .uri(route_path()) + .header(CONTENT_TYPE, "application/json") + .body(Body::from(body.clone())) + .unwrap(); + assert_eq!( + response(app.clone(), anonymous).await.0, + StatusCode::UNAUTHORIZED + ); + let forged = Request::builder() + .method("POST") + .uri(route_path()) + .header(CONTENT_TYPE, "application/json") + .header( + worker_runtime::auth::RUNTIME_REQUEST_SOURCE_PROOF_HEADER, + "forged", + ) + .body(Body::from(body.clone())) + .unwrap(); + assert_eq!( + response(app.clone(), forged).await.0, + StatusCode::UNAUTHORIZED + ); + let token = seed_test_workspace_owner_token(f.api.store.as_ref(), "memory-journal-user").await; + let user = Request::builder() + .method("POST") + .uri(route_path()) + .header(CONTENT_TYPE, "application/json") + .header("authorization", format!("Bearer {token}")) + .body(Body::from(body)) + .unwrap(); + assert_eq!(response(app.clone(), user).await.0, StatusCode::FORBIDDEN); + assert_eq!( + response(app, f.request(None, &request)).await.0, + StatusCode::FORBIDDEN + ); +} + +#[tokio::test] +async fn memory_journal_router_default_worker_and_job_have_no_subject_capability() { + let f = Fixture::new().await; + let request = MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }; + let ordinary = RuntimeWorkerRef::new(RUNTIME, Uuid::new_v4().to_string()); + seed_worker_source_member(&f.api, RUNTIME, &ordinary.worker_id); + let (status, bytes) = f.call_as(&ordinary, request.clone()).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "{}", + String::from_utf8_lossy(&bytes) + ); + // Even a singleton owner cannot reuse its body authority through a restricted Job binding. + let job = BackendJobRequest { + job_id: "restricted-job".into(), + purpose: "test".into(), + input_ref: "fixture://job".into(), + input: json!({}), + instruction: "No Memory Journal grant".into(), + profile: "builtin:backend-job".into(), + grants: Default::default(), + serialization_key: None, + source_worker: None, + notification_target: None, + limits: Default::default(), + }; + let now = Utc::now().to_rfc3339(); + let reserved = f + .api + .store + .reserve_backend_job(TEST_WORKSPACE_ID, &job, &now) + .unwrap(); + f.api + .store + .claim_backend_job_attempt_dispatch( + TEST_WORKSPACE_ID, + &job.job_id, + &reserved.attempt.attempt_id, + &now, + ) + .unwrap(); + f.api + .store + .bind_backend_job_attempt_worker( + TEST_WORKSPACE_ID, + &job.job_id, + &reserved.attempt.attempt_id, + &f.worker, + None, + &now, + ) + .unwrap(); + let (status, bytes) = f.call_as(&f.worker, request).await; + assert_eq!( + status, + StatusCode::FORBIDDEN, + "{}", + String::from_utf8_lossy(&bytes) + ); + let reader = JournalHostContext::validated_reader( + &f.store, + &f.store + .read_journal(&f.host, &f.journal_id) + .unwrap() + .subject_id, + ) + .unwrap(); + let MemoryJournalResponse::Links(links) = f + .store + .execute_memory_journal( + &reader, + MemoryJournalRequest::Links { + selector: f.selector.clone(), + }, + ) + .unwrap() + else { + panic!("links required") + }; + assert_eq!( + links.sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::NotReconciled + } + ); +} + +#[tokio::test] +async fn memory_journal_router_reconciles_only_fixed_events_and_retries_without_duplication() { + let f = Fixture::new().await; + let reconcile = MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }; + let MemoryJournalResponse::Links(open) = f.call(reconcile.clone()).await else { + panic!("links required") + }; + assert_eq!( + open.sources[0].connection, + SourceConnection::Unconnected { + reason: UnconnectedReason::JournalNotFixed + } + ); + assert_eq!( + f.store + .list_journal_entries(&f.host, &f.journal_id, None, 32) + .unwrap() + .items[0] + .state, + JournalEntryState::Open + ); + f.fix_initial(); + // Reading stored state alone must not silently trigger reconciliation. + assert_eq!( + f.call(MemoryJournalRequest::Links { + selector: f.selector.clone() + }) + .await, + MemoryJournalResponse::Links(open) + ); + let linked = f.call(reconcile.clone()).await; + assert_eq!(f.call(reconcile).await, linked); + assert_eq!( + f.call(MemoryJournalRequest::Links { + selector: f.selector.clone() + }) + .await, + linked + ); + let MemoryJournalResponse::Links(links) = linked else { + panic!("links required") + }; + assert_eq!(links.sources.len(), 1); + assert_eq!( + links.sources[0].connection, + SourceConnection::Linked { + journal_id: f.journal_id.clone(), + entry_id: f.entry_id.clone(), + target: JournalTargetAvailability::Available, + reconsideration: false, + } + ); + assert_eq!( + f.store + .list_journal_entries(&f.host, &f.journal_id, None, 32) + .unwrap() + .items + .len(), + 1 + ); +} + +#[tokio::test] +async fn memory_journal_router_reads_saved_text_after_original_session_and_source_worker_removal() { + let f = Fixture::new().await; + f.fix_initial(); + f.call(MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }) + .await; + assert!(f.source_root.exists()); + fs::remove_dir_all(&f.source_root).unwrap(); + f.api + .store + .delete_worker_registry(TEST_WORKSPACE_ID, &f.source_worker) + .unwrap(); + assert!(!f.source_root.exists()); + assert!( + f.api + .store + .get_worker_registry(TEST_WORKSPACE_ID, &f.source_worker) + .unwrap() + .is_none() + ); + // No Runtime adapter exists for RUNTIME, so a restore/history dependency would fail. + let MemoryJournalResponse::Read(read) = f.call(f.read_request()).await else { + panic!("read required") + }; + assert_eq!( + read.source.original_source, + OriginalSourceAvailability::Missing + ); + assert_eq!(read.page.unwrap().text, JOURNAL_BODY); + let MemoryJournalResponse::Related(related) = f + .call(MemoryJournalRequest::Related { + journal_id: f.journal_id.clone(), + entry_id: f.entry_id.clone(), + after: None, + limit: 32, + }) + .await + else { + panic!("related required") + }; + assert_eq!(related.items.len(), 1); + assert_eq!(related.items[0].selector, f.selector); + assert!(!related.items[0].reconsideration); + assert!(related.next.is_none()); +} + +#[tokio::test] +async fn memory_journal_router_ids_do_not_grant_cross_subject_reads_or_reconciliation() { + let f = Fixture::new().await; + f.fix_initial(); + f.call(MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }) + .await; + let other = f + .store + .create_subject(crate::subjektiv::SubjectRole::new("other").unwrap()) + .unwrap(); + let other_worker = Fixture::body_worker(&f.api, &other.id); + for request in [ + MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }, + MemoryJournalRequest::Links { + selector: f.selector.clone(), + }, + f.read_request(), + MemoryJournalRequest::Related { + journal_id: f.journal_id.clone(), + entry_id: f.entry_id.clone(), + after: None, + limit: 32, + }, + ] { + let (status, bytes) = f.call_as(&other_worker, request).await; + assert_eq!( + status, + StatusCode::BAD_REQUEST, + "{}", + String::from_utf8_lossy(&bytes) + ); + assert!(!String::from_utf8_lossy(&bytes).contains(JOURNAL_BODY)); + } + // A model cannot override the authenticated Subject with an additional JSON field. + let mut body = serde_json::to_value(f.read_request()).unwrap(); + body["subject_id"] = json!(other.id); + let request = runtime_source_request( + &f.identity, + Some(&f.worker.worker_id), + "POST", + &route_path(), + serde_json::to_vec(&body).unwrap(), + ); + assert_eq!( + response(build_router(f.api.clone()), request).await.0, + StatusCode::UNPROCESSABLE_ENTITY + ); +} + +#[tokio::test] +async fn memory_journal_router_correction_reverse_lookup_preserves_memory_and_original_link() { + let f = Fixture::new().await; + f.fix_initial(); + f.call(MemoryJournalRequest::Reconcile { + selector: f.selector.clone(), + }) + .await; + let subject = f + .store + .read_journal(&f.host, &f.journal_id) + .unwrap() + .subject_id; + let before = + serde_json::to_value(f.store.memory(&subject, &f.selector.memory_id).unwrap()).unwrap(); + let correction = f.add_correction(); + let MemoryJournalResponse::Related(related) = f + .call(MemoryJournalRequest::Related { + journal_id: f.journal_id.clone(), + entry_id: correction, + after: None, + limit: 32, + }) + .await + else { + panic!("related required") + }; + assert_eq!(related.items.len(), 1); + assert_eq!(related.items[0].selector, f.selector); + assert!(related.items[0].reconsideration); + let MemoryJournalResponse::Read(read) = f.call(f.read_request()).await else { + panic!("read required") + }; + assert_eq!(read.page.unwrap().text, JOURNAL_BODY); + assert!(matches!(read.source.connection, SourceConnection::Linked { + entry_id, reconsideration:true, .. + } if entry_id == f.entry_id)); + assert_eq!( + serde_json::to_value(f.store.memory(&subject, &f.selector.memory_id).unwrap()).unwrap(), + before + ); +} diff --git a/tools/state-contract-name-exceptions.json b/tools/state-contract-name-exceptions.json index e571712ea..07fa26137 100644 --- a/tools/state-contract-name-exceptions.json +++ b/tools/state-contract-name-exceptions.json @@ -1,4 +1,13 @@ [ + { + "path": "crates/subjektiv/src/journal/memory_links/tests.rs", + "kind": "saved_data", + "reason": "最古対応v1 schemaの保存済Subject JSONとSQLをそのまま作成し、最新migrationが元出典を保持する検証に必要な固定綴り。現行writerや状態判断には使わない。", + "lines": [ + { "sha256": "e30d8ddbb500aa10a4a23c52a1f34abbf1b6a8f3692501bb3b74084b63db911c", "count": 1 }, + { "sha256": "ed4d9edc0c05bf5ab788da52e8b1e44f4543c57d13b9a002144dbdc4430b5837", "count": 1 } + ] + }, { "path": "crates/manifest/src/lib.rs", "kind": "saved_data",