From 8f598da9312dc1e1960709c77c45916eaab679ca Mon Sep 17 00:00:00 2001 From: Hare Date: Sat, 17 Oct 2026 12:25:24 +0900 Subject: [PATCH] test: verify frozen Journal compaction across Hosts and restores --- .../standalone/src/subjektiv/journal/tests.rs | 60 ++ .../journal/tests/connected_compaction.rs | 171 ++++ crates/subjektiv/src/journal/tests.rs | 100 +++ crates/worker/tests/compact_events_test.rs | 777 +++++++++++++++++- .../subjektiv_journal_connected_tests.rs | 215 +++++ 5 files changed, 1322 insertions(+), 1 deletion(-) create mode 100644 crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs diff --git a/crates/standalone/src/subjektiv/journal/tests.rs b/crates/standalone/src/subjektiv/journal/tests.rs index 959eefb8c..3eacfeeb0 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 connected_compaction; mod handoff; mod memory_links; @@ -290,6 +291,65 @@ fn save( receipt } +#[test] +fn compaction_host_reads_pending_recording_without_advancing_or_accepting_foreign_sessions() { + let fixture = Fixture::new("partially performed; not confirmed"); + let connection = worker::subjektiv::SubjektivHostConnection::new( + fixture.host().clone(), + fixture.context.clone(), + ) + .unwrap(); + assert!(connection.journal_compaction_granted()); + assert!( + connection + .journal_compaction_context() + .unwrap() + .journals + .is_empty() + ); + let (adapter, _, _) = fixture.adapter(); + let before = adapter.state().unwrap(); + let snapshot = connection.journal_compaction_context().unwrap(); + assert_eq!( + snapshot.journals[0].source_session_id, + fixture.context.session_id + ); + assert_eq!( + snapshot.journals[0].process_order, + before.current_journal.process_order + ); + assert_eq!( + snapshot.journals[0].process_byte, + before.current_journal.process_byte + ); + assert_eq!(adapter.state().unwrap(), before); + assert!( + fixture + .host() + .catalog + .store + .active_journal_generation(&adapter.context, &adapter.grant.journal_id) + .unwrap() + .is_some() + ); + for context in [ + SubjektivHostContext { + worker_id: "foreign-worker".into(), + ..fixture.context.clone() + }, + SubjektivHostContext { + session_id: session_store::SessionId::now_v7().to_string(), + ..fixture.context.clone() + }, + ] { + assert!(fixture.host().journal_compaction_context(&context).is_err()); + } + fixture.host().live.store(false, Ordering::Release); + assert!(!connection.journal_compaction_granted()); + assert!(connection.journal_compaction_context().is_err()); + fixture.host().live.store(true, Ordering::Release); +} + #[test] fn duplicate_admission_reserves_one_job_and_excludes_private_prompt() { let fixture = Fixture::new("short request"); diff --git a/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs b/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs new file mode 100644 index 000000000..04032a8d7 --- /dev/null +++ b/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs @@ -0,0 +1,171 @@ +//! Real LocalSubjectHost/Journal storage -> PreparedWorker -> compactor Request -> durable seed. +//! Only external LLM responses and the fixture's Job execution fence are substituted. The +//! in-flight generation is retained while its real adapter updates storage during generation. +//! Temporary storage and the parent fixture own cleanup; ordinary standalone lib tests select +//! this suite. Predetermined summaries attest transport and persistence, not model understanding. + +use super::*; +use agen::llm_client::{ + ClientError, LlmClient, Request, + event::{Event, ResponseStatus, StatusEvent}, + types::Item, +}; +use futures::Stream; +use session_store::{CombinedStore, FsWorkerStore, LogEntry, LoggedSessionHistoryOrigin}; +use std::pin::Pin; + +const BEFORE: &str = "FROZEN_POSITION: B partially implemented; tests not run; result unconfirmed"; +const AFTER: &str = "NEXT_POSITION: background Journal changed during compactor generation"; +const COMPLEMENT: &str = + "Unrecorded request: keep the user's constraint and finish the remaining checks."; + +#[derive(Clone)] +struct MockClient { + adapter: Arc, + requests: Arc>>, + updated: Arc, +} + +#[async_trait::async_trait] +impl LlmClient for MockClient { + fn clone_boxed(&self) -> Box { + Box::new(self.clone()) + } + + async fn stream( + &self, + request: Request, + ) -> Result> + Send>>, ClientError> { + assert!( + request + .tools + .iter() + .any(|tool| tool.name == "write_summary") + ); + let first = self.requests.lock().unwrap().is_empty(); + self.requests.lock().unwrap().push(request); + let events = if first { + assert!(!self.updated.swap(true, Ordering::SeqCst)); + let state = self.adapter.state().unwrap(); + save( + &self.adapter, + "position-during-compaction", + &state.current_journal.state_token, + JournalAction::UpdatePosition { + current_position: AFTER.into(), + }, + vec![], + ); + vec![ + Event::tool_use_start(0, "summary", "write_summary"), + Event::tool_input_delta(0, serde_json::json!({ "text": COMPLEMENT }).to_string()), + Event::tool_use_stop(0), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ] + } else { + vec![ + Event::text_block_start(0), + Event::text_delta(0, "summary saved"), + Event::text_block_stop(0, None), + Event::Status(StatusEvent { + status: ResponseStatus::Completed, + }), + ] + }; + Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok)))) + } +} + +#[tokio::test] +async fn local_host_compaction_freezes_pending_journal_through_request_and_durable_seed_during_update() + { + tokio::time::timeout(std::time::Duration::from_secs(30), async { + let fixture = Fixture::new("Keep the user's constraint; work remains unfinished and unconfirmed."); + let (adapter, _, job) = fixture.adapter(); + let created = save( + &adapter, + "record-partial-work", + &adapter.generation.initial_journal.state_token, + JournalAction::Create { + title: "A corrected to B".into(), + body: "B partially implemented; tests not run; result unconfirmed".into(), + ongoing: "Finish remaining checks".into(), + state: JournalEntryState::Open, + source_end: None, + correction_refs: vec![], + }, + adapter.generation.selected_ranges.clone(), + ); + save(&adapter, "position-before-compaction", &created.state_token, + JournalAction::UpdatePosition { current_position: BEFORE.into() }, vec![]); + let before = adapter.state().unwrap(); + let expected = fixture.host().journal_compaction_context(&fixture.context).unwrap(); + let journal = &expected.journals[0]; + assert_eq!(journal.source_session_id, fixture.context.session_id); + assert_eq!(journal.current_position, BEFORE); + assert_eq!(journal.entries[0].state, JournalEntryState::Open); + assert!(!before.selected_ranges.is_empty()); + assert_eq!(journal.process_order, before.current_journal.process_order); + assert_eq!(journal.process_byte, before.current_journal.process_byte); + assert_eq!(journal.process_order, 0); + assert_eq!(journal.process_byte, 0, "saved open records do not advance processing checkpoint"); + let expected_json = serde_json::to_string(&expected).unwrap(); + let adapter = Arc::new(adapter); + let model = MockClient { adapter: adapter.clone(), requests: Arc::default(), updated: Arc::new(AtomicBool::new(false)) }; + let workers = StandaloneWorkerStore::open_existing(&fixture.host().catalog.state_root).unwrap(); + let worker_id = fixture._allocation.as_ref().unwrap().worker_id(); + let store = CombinedStore::new(fixture.sessions.clone(), FsWorkerStore::new(workers.worker_metadata_dir(worker_id)).unwrap()); + let mut manifest = manifest::WorkerManifest::from_toml(&format!("[worker]\nname='{}'\n[model]\nscheme='anthropic'\nmodel_id='scripted'\n[engine]\n[scope]\nallow=[]", fixture.context.worker_id)).unwrap(); + manifest.feature.subjektiv.profile.enabled = true; + manifest.feature.subjektiv.profile.extraction.enabled = false; + let mut prepared = worker::WorkerBootstrap::new( + manifest, + store, + worker::PromptCatalogSource::builtins_only(), + worker::WorkerWorkspaceContext::no_workspace(), + worker::WorkerFilesystemAuthority::None, + worker::WorkerBootstrapLayout::Direct { runtime_base: fixture._temp.path().join("runtime"), bash_output_dir: fixture._temp.path().join("bash") }, + worker::WorkerControllerTransport::InProcess, + ).with_model_client(model.clone()).with_subjektiv_host(fixture.host().clone()).prepare_restored("journal-test").await.unwrap(); + assert_eq!(prepared.worker().session_id().to_string(), fixture.context.session_id); + let source_segment = prepared.worker().segment_id(); + let original = fixture.sessions.read_all(prepared.worker().session_id(), source_segment).unwrap(); + let generation = fixture.host().catalog.store.active_journal_generation(&adapter.context, &adapter.grant.journal_id).unwrap().unwrap(); + assert!(!fixture.host().session_fully_reflected(&fixture.context.session_id).unwrap()); + prepared.worker_mut().compact(0).await.unwrap(); + assert!(model.updated.load(Ordering::SeqCst)); + let worker = prepared.worker(); + assert_ne!(worker.segment_id(), source_segment); + let seed = worker.store().read_all(worker.session_id(), worker.segment_id()).unwrap().into_iter().find_map(|entry| match entry { LogEntry::AnnotatedSegmentStart { history, .. } => Some(history), _ => None }).expect("durable compacted seed"); + let journal_items: Vec<_> = seed.iter().filter(|entry| entry.metadata.origin == LoggedSessionHistoryOrigin::JournalContext).collect(); + assert_eq!(journal_items.len(), 1); + let journal_item = Item::from(journal_items[0].item.clone()); + let body = journal_item.as_text().unwrap(); + assert!(body.contains(&expected_json), "durable seed contains the full selected JSON and processing checkpoint"); + assert!(body.contains(BEFORE)); + assert!(!body.contains(AFTER)); + assert!(seed.iter().any(|entry| entry.metadata.origin == LoggedSessionHistoryOrigin::CompactionSummary && Item::from(entry.item.clone()).as_text().is_some_and(|text| text.contains(COMPLEMENT)))); + let requests = model.requests.lock().unwrap(); + let input = requests[0].items.iter().find_map(|item| match item { + Item::Message { role: agen::llm_client::types::Role::User, .. } => item.as_text().filter(|text| text.contains("## Conversation overview/index")), + _ => None, + }).expect("actual compactor user request"); + assert!(input.ends_with(body), "exact same frozen reference goes to actual compactor and seed"); + assert_eq!(input.matches(&expected_json).count(), 1); + assert!(!input.contains(AFTER)); + drop(requests); + let latest = fixture.host().journal_compaction_context(&fixture.context).unwrap(); + assert_eq!(latest.journals[0].current_position, AFTER); + assert_ne!(latest.journals[0].state_token, expected.journals[0].state_token); + assert_eq!(latest.journals[0].process_order, expected.journals[0].process_order); + assert_eq!(latest.journals[0].process_byte, expected.journals[0].process_byte); + assert_eq!(latest.journals[0].entries, expected.journals[0].entries); + assert_eq!(fixture.host().catalog.store.active_journal_generation(&adapter.context, &adapter.grant.journal_id).unwrap(), Some(generation)); + assert_eq!(fixture.host().jobs().unwrap().get(&job.request.job_id).unwrap().state, job.state); + let unchanged: Vec<_> = fixture.sessions.read_all(worker.session_id(), source_segment).unwrap().into_iter().filter(|entry| !matches!(entry, LogEntry::Extension { domain, .. } if domain == "metrics")).collect(); + assert_eq!(serde_json::to_value(unchanged).unwrap(), serde_json::to_value(original).unwrap()); + assert!(!fixture.host().session_fully_reflected(&fixture.context.session_id).unwrap()); + }).await.expect("compaction must finish without recording Job completion"); +} diff --git a/crates/subjektiv/src/journal/tests.rs b/crates/subjektiv/src/journal/tests.rs index fa8ca515d..7fae087db 100644 --- a/crates/subjektiv/src/journal/tests.rs +++ b/crates/subjektiv/src/journal/tests.rs @@ -100,6 +100,10 @@ fn released_binding_preserves_open_and_receipts_but_never_revives_old_session_wr f.store().release_journal_writer(&binding).unwrap(); f.store().release_journal_writer(&binding).unwrap(); f.reopen(); + assert!(matches!( + f.store().journal_compaction_snapshot(&f.host), + Err(JournalError::PermissionDenied(_)) + )); let old = f .store() .query_journal_operation(&f.host, &f.journal_id, "create") @@ -409,6 +413,102 @@ fn release_and_late_admission_are_serialized_by_the_persisted_binding_check() { manager.shutdown().unwrap(); } +#[test] +fn compaction_snapshot_prioritizes_current_session_without_waiting_or_mutating_recording() { + let f = Fixture::new(); + let (generation, context) = f.generation("in-flight", f.ranges.clone()); + f.execute( + &context, + "open", + open_action(&"日本語\\\"\\n".repeat(6000)), + f.ranges.clone(), + ); + f.execute( + &context, + "position", + JournalAction::UpdatePosition { + current_position: "partial activity; checkpoint still behind".into(), + }, + vec![], + ); + let mut latest = String::new(); + for n in 0..5 { + let session = format!("supplement-{n}"); + let host = session_host(f.store(), &f.host.subject_id, &session); + f.store() + .accept_journal_sources(&host, "recent", &[source_range(&session, 0, 10)]) + .unwrap(); + latest = session; + } + let reader = JournalHostContext::validated_reader(f.store(), &f.host.subject_id).unwrap(); + let startup = f + .store() + .journal_startup_snapshot(&reader, Some(&latest)) + .unwrap(); + assert_eq!(startup.journals[0].source_session_id, latest); + let before = f.state(); + let snapshot = f.store().journal_compaction_snapshot(&f.host).unwrap(); + assert_eq!(snapshot.journals[0].source_session_id, "source-session"); + assert_eq!(snapshot.journals[0].process_order, before.0.process_order); + assert_eq!(snapshot.journals[0].process_byte, before.0.process_byte); + assert_eq!( + snapshot.journals[0].entries[0].state, + JournalEntryState::Open + ); + assert!(snapshot.journals[0].entries[0].omitted); + assert!(snapshot.omitted); + assert_eq!(snapshot.journals.len(), 3); + assert!(serde_json::to_vec(&snapshot).unwrap().len() <= startup::STARTUP_MAX_BYTES); + assert_eq!( + f.store().journal_compaction_snapshot(&f.host).unwrap(), + snapshot + ); + assert_eq!(f.state(), before); + assert_eq!( + f.store() + .active_journal_generation(&f.host, &f.journal_id) + .unwrap(), + Some(generation) + ); + assert!(matches!( + f.store().journal_compaction_snapshot(&reader), + Err(JournalError::PermissionDenied(_)) + )); + f.execute( + &context, + "new-position", + JournalAction::UpdatePosition { + current_position: "background recording progressed after snapshot".into(), + }, + vec![], + ); + let refreshed = f.store().journal_compaction_snapshot(&f.host).unwrap(); + assert_ne!( + refreshed.journals[0].state_token, + snapshot.journals[0].state_token + ); + assert_ne!( + refreshed.journals[0].current_position, + snapshot.journals[0].current_position + ); + let empty_host = session_host(f.store(), &f.host.subject_id, "not-yet-admitted"); + assert!( + !f.store() + .journal_compaction_snapshot(&empty_host) + .unwrap() + .journals + .is_empty() + ); + let other = f + .store() + .create_subject_with_behavior(SubjectRole::new("other").unwrap(), "unchanged".into()) + .unwrap(); + let other_host = session_host(f.store(), &other.id, "other-session"); + let empty = f.store().journal_compaction_snapshot(&other_host).unwrap(); + assert!(empty.journals.is_empty()); + assert!(!empty.omitted); +} + #[test] fn startup_snapshot_distinguishes_none_is_subject_local_and_bounds_utf8_at_storage() { let f = Fixture::new(); diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index 072f857c3..ad46564ab 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -2086,6 +2086,27 @@ struct SummaryInputFixture { impl SummaryInputFixture { async fn new(manifest_toml: &str, client: MockClient) -> Self { + Self::with_manifest( + worker::WorkerManifest::from_toml(manifest_toml).unwrap(), + client, + ) + .await + } + + async fn with_journal( + manifest_toml: &str, + client: MockClient, + host: Arc, + ) -> Self { + let mut manifest = worker::WorkerManifest::from_toml(manifest_toml).unwrap(); + manifest.feature.subjektiv.profile.enabled = true; + manifest.feature.subjektiv.profile.extraction.enabled = false; + let mut fixture = Self::with_manifest(manifest, client).await; + fixture.worker.bind_subjektiv_host(Some(host)); + fixture + } + + async fn with_manifest(manifest: worker::WorkerManifest, client: MockClient) -> Self { let store_dir = tempfile::tempdir().unwrap(); let store = CombinedStore::new( FsStore::new(store_dir.path()).unwrap(), @@ -2096,7 +2117,7 @@ impl SummaryInputFixture { client, ); let mut worker = Worker::new( - worker::WorkerManifest::from_toml(manifest_toml).unwrap(), + manifest, engine, store, worker::WorkerWorkspaceContext::no_workspace(), @@ -2120,6 +2141,16 @@ impl SummaryInputFixture { &mut self, history: Vec, compacted: bool, + ) { + self.restore_seed_with_system(history, compacted, vec![]) + .await; + } + + async fn restore_seed_with_system( + &mut self, + history: Vec, + compacted: bool, + systems: Vec, ) { self.worker .run_text("fixture rewind boundary") @@ -2150,6 +2181,16 @@ impl SummaryInputFixture { } _ => unreachable!(), } + records.splice( + 1..1, + systems + .into_iter() + .map(|entry| LogEntry::AnnotatedSystemItem { + ts: 1, + entry, + extensions: vec![], + }), + ); self.worker .store() .create_segment(session_id, segment_id, &records) @@ -2801,3 +2842,737 @@ async fn budget_warning_is_durable_and_keeps_full_summary_in_first_model_request if alert.message.contains("compact worker context budget is low"))) ); } + +const COMPACTION_HOST_WORKSPACE_ID: &str = "journal-transport-host-workspace"; + +// Public compact/rewind -> real internal Worker, prompt catalog, and FsStore. +// Only the authenticated Host and LLM provider are replaced. This does not test +// Host authorization/selection or live-model semantic continuation quality. +struct CompactionJournalHost { + granted: bool, + unavailable: bool, + snapshot: Mutex, + reads: Mutex>, + startup_reads: AtomicUsize, +} + +impl CompactionJournalHost { + fn new(snapshot: subjektiv::journal::startup::JournalStartupSnapshot) -> Self { + Self { + granted: true, + unavailable: false, + snapshot: Mutex::new(snapshot), + reads: Mutex::new(vec![]), + startup_reads: AtomicUsize::new(0), + } + } +} + +impl worker::subjektiv::SubjektivHost for CompactionJournalHost { + fn settings(&self) -> worker::subjektiv::SubjektivHostSettings { + worker::subjektiv::SubjektivHostSettings { + language: "English".into(), + } + } + + fn workspace_id(&self) -> Option { + Some(COMPACTION_HOST_WORKSPACE_ID.into()) + } + + fn journal_compaction_granted(&self) -> bool { + self.granted + } + + fn journal_compaction_context( + &self, + context: &worker::subjektiv::SubjektivHostContext, + ) -> Result< + subjektiv::journal::startup::JournalStartupSnapshot, + worker::subjektiv::SubjektivHostError, + > { + self.reads.lock().unwrap().push(context.clone()); + if self.unavailable { + return Err(worker::subjektiv::SubjektivHostError::Unavailable( + "injected Journal read failure".into(), + )); + } + Ok(self.snapshot.lock().unwrap().clone()) + } + + fn journal_startup_granted(&self) -> bool { + true + } + + fn journal_startup_context( + &self, + _: &worker::subjektiv::SubjektivHostContext, + ) -> Result< + subjektiv::journal::startup::JournalStartupSnapshot, + worker::subjektiv::SubjektivHostError, + > { + self.startup_reads.fetch_add(1, Ordering::SeqCst); + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "startup must not be read during compaction/rewind/normal turns".into(), + )) + } + + fn memory( + &self, + _: &worker::subjektiv::SubjektivHostContext, + _: server_api::SubjektivMemoryBackendOperation, + ) -> Result + { + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "not used".into(), + )) + } + fn session( + &self, + _: &worker::subjektiv::SubjektivHostContext, + _: server_api::SubjektivSessionBackendOperation, + ) -> Result + { + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "not used".into(), + )) + } + fn stage_candidate( + &self, + _: &worker::subjektiv::SubjektivHostContext, + _: memory::backend::MemoryStageCandidateOperation, + ) -> Result + { + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "not used".into(), + )) + } + fn record_session( + &self, + _: &worker::subjektiv::SubjektivHostContext, + _: bool, + ) -> Result + { + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "not used".into(), + )) + } + fn request_consolidation( + &self, + _: &worker::subjektiv::SubjektivHostContext, + _: memory::backend::MemoryConsolidateStagingOperation, + ) -> Result + { + Err(worker::subjektiv::SubjektivHostError::Unavailable( + "not used".into(), + )) + } +} + +fn compaction_snapshot(generation: usize) -> subjektiv::journal::startup::JournalStartupSnapshot { + use subjektiv::journal::startup::{JournalStartupSnapshot, StartupEntry, StartupJournal}; + JournalStartupSnapshot { + subject_id: "compaction-subject".into(), + omitted: true, + journals: vec![StartupJournal { + journal_id: format!("journal-reference-{generation}"), + source_session_id: "current-source-session".into(), + state_token: format!("state-token-{generation}"), + process_order: 7 + generation as u64, + process_byte: 23 + generation as u64, + current_position: format!( + "JOURNAL_POSITION_{generation}: work ongoing, result not verified\nExact JSON bytes: 日本語, \"quoted\", \\path" + ), + entries: vec![StartupEntry { + entry_id: format!("fixed-event-reference-{generation}"), + position: 2, + state: subjektiv::journal::JournalEntryState::Fixed, + title: "An event ended, not the entire conversation".into(), + body: format!("JOURNAL_BODY_{generation}"), + ongoing: "User instructions after the checkpoint are not recorded".into(), + source_end: Some(subjektiv::journal::JournalSourcePointer { + segment_id: session_store::new_segment_id(), + entry_ref: "E3".into(), + }), + correction_refs: vec![format!("correction-reference-{generation}")], + omitted: false, + }], + omitted: true, + }], + } +} + +fn journal_body(worker: &Worker) -> String { + let seed = replacement_seed(worker); + assert_eq!( + seed[1].metadata.origin, + session_store::LoggedSessionHistoryOrigin::JournalContext + ); + let item = Item::from(seed[1].item.clone()); + assert!(matches!( + item, + Item::Message { + role: agen::Role::System, + .. + } + )); + assert_eq!( + seed.iter() + .filter(|entry| entry.metadata.origin + == session_store::LoggedSessionHistoryOrigin::JournalContext) + .count(), + 1 + ); + item.as_text().unwrap().to_string() +} + +fn journal_source_records( + store: &TestStore, + session: session_store::SessionId, + segment: session_store::SegmentId, +) -> serde_json::Value { + // Compaction appends telemetry; the source material and typed startup + // records must remain equivalent, not rewritten or removed. + let records = store + .read_all(session, segment) + .unwrap() + .into_iter() + .filter( + |record| !matches!(record, LogEntry::Extension { domain, .. } if domain == "metrics"), + ) + .collect::>(); + serde_json::to_value(records).unwrap() +} + +fn assert_frozen_journal_transport(input: &str, body: &str) { + assert!( + input.ends_with(body), + "the actual compactor user input ends with the frozen reference" + ); + assert_eq!(input.matches(body).count(), 1); +} + +fn assert_journal_provenance( + body: &str, + expected_json: Option<&str>, + catalog: &worker::PromptCatalog, + workspace_id: Option<&str>, +) { + use sha2::{Digest, Sha256}; + let projection = catalog.projection(); + assert!(body.contains(&format!( + "Prompt provenance: `internal.subjektiv_journal_compaction`, source `{}`, projection `{}`.", + projection.source_digest, projection.catalog_digest, + ))); + if let Some(workspace_id) = workspace_id { + assert!(body.contains(&format!("Host Workspace: `{workspace_id}`."))); + } else { + assert!(!body.contains("Host Workspace:")); + } + if let Some(expected_json) = expected_json { + // Hash the actual rendered JSON, including escapes and UTF-8 bytes, + // not a parsed/re-serialized or pretty-printed approximation. + let rendered_json = body.lines().rfind(|line| !line.is_empty()).unwrap(); + assert_eq!(rendered_json.as_bytes(), expected_json.as_bytes()); + let expected_hash = Sha256::digest(rendered_json.as_bytes()) + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + let declared_hash = body + .lines() + .find_map(|line| { + line.strip_prefix("Snapshot SHA-256 (the exact JSON bytes below): `") + .and_then(|value| value.strip_suffix("`.")) + }) + .expect("snapshot SHA-256 must be persisted in the same rendered body"); + assert_eq!(declared_hash, expected_hash); + } else { + assert!( + !body.contains("Snapshot SHA-256"), + "failed/disabled reads must not invent a snapshot hash" + ); + } +} + +#[tokio::test] +async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tail_tasks_and_source() +{ + use session_store::LoggedSessionHistoryOrigin::{ + CompactionSummary, DerivedSummary, JournalContext, + }; + let initial = long_summary("journal-initial"); + let summaries = (0..3) + .map(|index| long_summary(&format!("journal-next-{index}"))) + .collect::>(); + let mut responses = vec![single_text_events("fixture")]; + for (index, summary) in summaries.iter().enumerate() { + responses.push(single_text_events(&format!("TAIL_ASSISTANT_{index}"))); + responses.push(write_summary_tool_use_events( + &format!("journal-summary-{index}"), + summary, + )); + responses.push(single_text_events("done")); + } + responses.extend([ + single_text_events("normal after compact"), + single_text_events("after rewind"), + ]); + let client = MockClient::new(responses); + let observed = client.clone(); + let host = Arc::new(CompactionJournalHost::new(compaction_snapshot(0))); + let mut fixture = + SummaryInputFixture::with_journal(MANUAL_ONLY_MANIFEST_TOML, client, host.clone()).await; + // A resolved nonempty source digest distinguishes provenance forwarding + // from silently substituting the built-in catalog's empty source digest. + let mut projection = fixture.worker.prompts().load_full().projection().clone(); + projection.source_digest = "a".repeat(64); + let initial_catalog = Arc::new(worker::PromptCatalog::from_projection(projection).unwrap()); + fixture.worker.prompts().store(initial_catalog.clone()); + let legacy_startup = session_store::LoggedSystemHistoryEntry { + item: session_store::SystemItem::JournalStartup { + body: "OLD_TYPED_STARTUP_JOURNAL".into(), + snapshot: serde_json::json!({"journals": []}), + snapshot_sha256: "legacy-startup-digest".into(), + prompt_provenance: session_store::PromptRenderProvenance { + workspace_id: None, + source_digest: "legacy-source".into(), + projection_digest: "legacy-projection".into(), + logical_name: "internal.subjektiv_journal_startup".into(), + }, + }, + metadata: summary_entry(Item::system_message(""), DerivedSummary).metadata, + }; + let legacy_id = legacy_startup.metadata.entry_id.clone(); + fixture + .restore_seed_with_system( + vec![ + summary_entry( + Item::system_message(format!("{SUMMARY_MARKER}{initial}")), + CompactionSummary, + ), + summary_entry( + Item::system_message("OLD_COMPACTION_JOURNAL"), + JournalContext, + ), + annotated(Item::tool_call( + "seed-task", + "TaskCreate", + r#"{"subject":"JOURNAL_TASK","description":"Keep the unfinished task"}"#, + )), + annotated(Item::tool_result("seed-task", "created")), + ], + true, + vec![legacy_startup], + ) + .await; + assert_eq!( + fixture + .worker + .session_history() + .entries() + .iter() + .find(|entry| entry.annotation.entry_id == legacy_id) + .unwrap() + .annotation + .origin, + JournalContext + ); + let session = fixture.worker.session_id(); + let original_segment = fixture.worker.segment_id(); + let mut historical_sources = Vec::new(); + let mut old_bodies = vec![ + "OLD_TYPED_STARTUP_JOURNAL".to_string(), + "OLD_COMPACTION_JOURNAL".to_string(), + ]; + for (index, retained) in [0, 1_000_000, 100].into_iter().enumerate() { + let directive = format!( + "UNRECORDED_USER_{index}: do not merge; original completion conditions still apply" + ); + fixture.worker.run_text(&directive).await.unwrap(); + if index == 0 { + assert_eq!( + fixture + .worker + .session_history() + .entries() + .iter() + .find(|entry| entry.annotation.entry_id == legacy_id) + .unwrap() + .annotation + .origin, + JournalContext, + "legacy startup classification survives a normal turn" + ); + } + assert_eq!( + host.reads.lock().unwrap().len(), + index, + "normal turns do not fetch compaction Journal" + ); + let snapshot = compaction_snapshot(index); + *host.snapshot.lock().unwrap() = snapshot.clone(); + let first_request = observed.requests().len(); + let updating_host = host.clone(); + observed.set_observer(move |call| { + if call == first_request { + // A background update at the provider boundary, after the + // request has been captured but before generation completes. + *updating_host.snapshot.lock().unwrap() = compaction_snapshot(100 + index); + } + }); + let source = fixture.worker.segment_id(); + let frozen_catalog = fixture.worker.prompts().load_full(); + historical_sources.push(( + source, + journal_source_records(fixture.worker.store(), session, source), + )); + fixture.worker.compact(retained).await.unwrap(); + let input = first_compactor_input(&observed.requests()[first_request]); + assert_full_initial_summary( + &input, + if index == 0 { + &initial + } else { + &summaries[index - 1] + }, + ); + let body = journal_body(&fixture.worker); + assert_frozen_journal_transport(&input, &body); + assert_journal_provenance( + &body, + Some(&serde_json::to_string(&snapshot).unwrap()), + &frozen_catalog, + Some(COMPACTION_HOST_WORKSPACE_ID), + ); + assert!( + body.contains(&serde_json::to_string(&snapshot).unwrap()), + "contents, references, fixed source_end and partial processing checkpoint must match" + ); + assert!(!body.contains(&format!("JOURNAL_POSITION_{}:", 100 + index))); + assert_eq!(host.reads.lock().unwrap().len(), index + 1); + let history = fixture.worker.history(); + let overview = input + .split("## Previous Journal reference (historical, not automatically retained)") + .next() + .unwrap(); + for old in &old_bodies { + assert!( + !overview.contains(old), + "superseded Journal is excluded from ordinary overview" + ); + assert!( + !body.contains(old), + "frozen latest reference does not incorporate old injections" + ); + assert!( + history + .iter() + .all(|item| !item.as_text().is_some_and(|text| text.contains(old))), + "superseded Journal is not retained" + ); + } + assert!( + input.contains(&directive) + || history + .iter() + .any(|item| item.as_text() == Some(directive.as_str())), + "unrecorded user instruction must reach summary input or retained tail" + ); + if retained == 1_000_000 { + assert!( + history + .iter() + .any(|item| item.as_text() == Some("TAIL_ASSISTANT_1")) + ); + } + let task = history + .iter() + .filter_map(Item::as_text) + .find(|text| text.starts_with("[Session TaskStore snapshot]")) + .unwrap(); + let json = task + .split("```json\n") + .nth(1) + .unwrap() + .split("\n```") + .next() + .unwrap(); + let tasks: serde_json::Value = serde_json::from_str(json).unwrap(); + assert_eq!( + tasks["tasks"].as_array().unwrap().len(), + 1, + "replayed TaskCreate must not duplicate across compactions" + ); + assert_eq!(tasks["tasks"][0]["subject"], "JOURNAL_TASK"); + assert_eq!(tasks["tasks"][0]["status"], "pending"); + assert!(input.contains("JOURNAL_TASK")); + old_bodies.push(body); + for (source, before) in &historical_sources { + assert_eq!( + journal_source_records(fixture.worker.store(), session, *source), + *before, + "original segments must remain unchanged" + ); + } + } + assert_ne!(original_segment, fixture.worker.segment_id()); + let reads = host.reads.lock().unwrap().clone(); + assert!( + reads + .iter() + .all(|context| context.worker_id == "test-worker" + && context.session_id == session.to_string()) + ); + // Reloaded prompt resources must not regenerate the persisted reference + // or relabel its original source/projection on normal turns or restore. + let mut templates = initial_catalog.projection().templates.clone(); + templates + .get_mut("internal.subjektiv_journal_compaction") + .unwrap() + .push_str("\nRELOADED_PROMPT_MUST_NOT_REPLACE_FROZEN_BODY"); + let mut reloaded_projection = + worker::EffectivePromptCatalog::new(templates, "test-schema", "test-toolchain").unwrap(); + reloaded_projection.source_digest = "b".repeat(64); + let reloaded_catalog = + Arc::new(worker::PromptCatalog::from_projection(reloaded_projection).unwrap()); + assert_ne!( + initial_catalog.projection().catalog_digest, + reloaded_catalog.projection().catalog_digest + ); + fixture.worker.prompts().store(reloaded_catalog); + fixture + .worker + .run_text("normal post-compaction turn") + .await + .unwrap(); + let records = fixture + .worker + .store() + .read_all(session, fixture.worker.segment_id()) + .unwrap(); + let snapshot = session_store::public_snapshot::project_current_session_snapshot(&records); + let target = snapshot + .entries + .iter() + .rev() + .find(|entry| { + matches!( + entry.data, + protocol::SessionSnapshotEntryData::UserInput { .. } + ) + }) + .unwrap(); + fixture + .worker + .rewind_to(target.entry_id.clone()) + .await + .unwrap(); + let restored_body = journal_body(&fixture.worker); + assert_eq!(restored_body, *old_bodies.last().unwrap()); + assert_journal_provenance( + &restored_body, + old_bodies + .last() + .unwrap() + .lines() + .rfind(|line| !line.is_empty()), + &initial_catalog, + Some(COMPACTION_HOST_WORKSPACE_ID), + ); + fixture + .worker + .run_text("normal turn after typed history restore") + .await + .unwrap(); + assert_eq!( + host.reads.lock().unwrap().len(), + 3, + "rewind uses the shared typed restore path without fresh Journal acquisition" + ); + assert_eq!(host.startup_reads.load(Ordering::SeqCst), 0); + let normal = observed.requests().last().unwrap().clone(); + assert_eq!( + normal + .items + .iter() + .filter(|item| item.as_text() == Some(old_bodies.last().unwrap().as_str())) + .count(), + 1 + ); +} + +#[tokio::test] +async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversation_compaction() { + for case in [ + "absent", + "disabled", + "unavailable", + "over-byte-budget", + "no-host", + ] { + let mut snapshot = compaction_snapshot(0); + if case == "absent" { + snapshot.journals.clear(); + } + if case == "over-byte-budget" { + snapshot.journals[0].current_position = + "x".repeat(subjektiv::journal::startup::STARTUP_MAX_BYTES); + } + let expected_json = (case == "absent").then(|| serde_json::to_string(&snapshot).unwrap()); + let mut host = CompactionJournalHost::new(snapshot); + host.granted = case != "disabled"; + host.unavailable = case == "unavailable"; + let host = Arc::new(host); + let client = MockClient::new(vec![ + single_text_events("seed"), + write_summary_tool_use_events("fallback", "fallback summary"), + single_text_events("done"), + ]); + let observed = client.clone(); + let mut fixture = if case == "no-host" { + SummaryInputFixture::new(MANUAL_ONLY_MANIFEST_TOML, client).await + } else { + SummaryInputFixture::with_journal(MANUAL_ONLY_MANIFEST_TOML, client, host.clone()).await + }; + let inherited = long_summary("fallback-predecessor"); + fixture + .restore_seed( + vec![ + summary_entry( + Item::system_message(format!("{SUMMARY_MARKER}{inherited}")), + session_store::LoggedSessionHistoryOrigin::CompactionSummary, + ), + summary_entry( + Item::system_message("PREVIOUS_JOURNAL_DELEGATED_CONTEXT"), + session_store::LoggedSessionHistoryOrigin::JournalContext, + ), + annotated(Item::user_message( + "UNRECORDED_FALLBACK_GOAL: implementation still unfinished", + )), + ], + true, + ) + .await; + fixture.worker.compact(0).await.unwrap(); + let body = journal_body(&fixture.worker); + let expected = match case { + "no-host" | "disabled" => "disabled", + "absent" => "absent", + _ => "unavailable", + }; + assert!( + body.contains(&format!("Journal read status: {expected}.")), + "{case}" + ); + let input = first_compactor_input(&observed.requests()[1]); + assert_frozen_journal_transport(&input, &body); + assert_journal_provenance( + &body, + expected_json.as_deref(), + &fixture.worker.prompts().load_full(), + (case != "no-host").then_some(COMPACTION_HOST_WORKSPACE_ID), + ); + assert_full_initial_summary(&input, &inherited); + assert!(input.contains("UNRECORDED_FALLBACK_GOAL"), "{case}"); + assert!(input.contains("## Previous Journal reference (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT")); + assert!( + fixture.worker.history().iter().all(|item| !item + .as_text() + .is_some_and(|text| text.contains("PREVIOUS_JOURNAL_DELEGATED_CONTEXT"))), + "fallback retains only the newly frozen reference, not an old snapshot as latest" + ); + assert!(!body.contains("JOURNAL_POSITION_0")); + assert!(!body.contains("journal-reference-0")); + assert_eq!( + host.reads.lock().unwrap().len(), + usize::from(!matches!(case, "disabled" | "no-host")), + "{case}" + ); + assert!( + fixture.worker.history()[0] + .as_text() + .unwrap() + .contains("fallback summary") + ); + } +} + +#[tokio::test] +async fn journal_snapshot_counts_against_compactor_hard_input_cap_without_mutating_source() { + let manifest = MANUAL_ONLY_MANIFEST_TOML.replace("[compaction]", "[compaction]\nworker_context_max_tokens = 7000\nfinish_warning_remaining_tokens = 0\nfinal_reserve_tokens = 0\noverview_target_tokens = 1"); + for large in [false, true] { + let mut snapshot = compaction_snapshot(0); + if large { + snapshot.journals[0].current_position = " x".repeat(15_000); + } + assert!( + serde_json::to_vec(&snapshot).unwrap().len() + <= subjektiv::journal::startup::STARTUP_MAX_BYTES + ); + if large { + assert!( + agen::token_counter::total_tokens( + &[Item::user_message(&snapshot.journals[0].current_position)], + &[] + ) + .tokens + > 7_000 + ); + } + let host = Arc::new(CompactionJournalHost::new(snapshot)); + let client = MockClient::new(vec![ + single_text_events("seed"), + write_summary_tool_use_events("summary", "new summary"), + single_text_events("done"), + ]); + let observed = client.clone(); + let mut fixture = SummaryInputFixture::with_journal(&manifest, client, host.clone()).await; + fixture.worker.run_text("unfinished goal").await.unwrap(); + let session = fixture.worker.session_id(); + let source = fixture.worker.segment_id(); + let before = journal_source_records(fixture.worker.store(), session, source); + let history = fixture.worker.history(); + let result = fixture.worker.compact(0).await; + if large { + let error = result.expect_err( + "mandatory frozen Journal must not be truncated to evade hard input cap", + ); + assert!( + matches!(error, WorkerError::CompactCancelled) + || error.to_string().contains("input occupancy"), + "{error:?}" + ); + assert_eq!( + observed.requests().len(), + 1, + "over-budget input must not reach the provider" + ); + assert_eq!(fixture.worker.segment_id(), source); + assert_eq!(fixture.worker.history(), history); + assert_eq!( + fixture.worker.store().list_segments(session).unwrap(), + vec![source] + ); + assert_eq!( + fixture + .worker + .store() + .read_by_name("test-worker") + .unwrap() + .unwrap() + .active + .unwrap() + .segment_id, + Some(source) + ); + } else { + result.unwrap(); + assert_eq!(observed.requests().len(), 3); + assert_ne!(fixture.worker.segment_id(), source); + } + assert_eq!(host.reads.lock().unwrap().len(), 1); + assert_eq!( + journal_source_records(fixture.worker.store(), session, source), + before + ); + } +} diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs index aea08e5ae..c3aca207b 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_connected_tests.rs @@ -619,5 +619,220 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() ) ); drop(requests); + let connection = connected_host_reader(&api, &next, &next_attr.session_id); + let before = store + .journal_resume_state( + &JournalHostContext::validated_session(&store, next_attr.clone()).unwrap(), + &store + .journal_for_session( + &JournalHostContext::validated_session(&store, next_attr.clone()).unwrap(), + ) + .unwrap() + .unwrap(), + ) + .unwrap(); + let (startup, compacted) = tokio::task::spawn_blocking(move || { + ( + connection.journal_startup_context().unwrap(), + connection.journal_compaction_context().unwrap(), + ) + }) + .await + .unwrap(); + assert_eq!(startup.journals[0].source_session_id, old_attr.session_id); + assert_eq!( + compacted.journals[0].source_session_id, + next_attr.session_id + ); + assert_eq!(compacted.journals[0].state_token, before.0.state_token); + assert_eq!(compacted.journals[0].process_order, before.0.process_order); + assert_eq!(compacted.journals[0].process_byte, before.0.process_byte); + assert_eq!( + store + .journal_resume_state( + &JournalHostContext::validated_session(&store, next_attr.clone()).unwrap(), + &before.0.id + ) + .unwrap(), + before + ); + for (worker, session) in [ + (&old, old_attr.session_id.as_str()), + (&next, old_attr.session_id.as_str()), + (&next, "unattributed-session"), + ] { + let connection = connected_host_reader(&api, worker, session); + assert!( + tokio::task::spawn_blocking(move || connection.journal_compaction_context()) + .await + .unwrap() + .is_err() + ); + } server.abort(); } + +#[tokio::test] +async fn signed_compaction_read_does_not_contact_runtime_or_wait_for_active_generation_and_rejects_jobs() + { + let dir = tempfile::tempdir().unwrap(); + let mut api = test_api(dir.path()).await; + let identity = worker_runtime::auth::RuntimeIdentityMaterial::generate("runtime-test").unwrap(); + configure_runtime_request_auth(&mut api, &identity, "runtime-test"); + let store = open_subjektiv_store(&api).unwrap(); + let subject = store + .create_subject(crate::subjektiv::SubjectRole::new("snapshot").unwrap()) + .unwrap(); + let settings = api + .config_store + .get_workspace_memory_settings(TEST_WORKSPACE_ID) + .unwrap(); + let reserve = |key: String, operation: &str| { + api.config_store + .reserve_worker_create( + TEST_WORKSPACE_ID, + "runtime-test", + operation, + &"f".repeat(64), + Some(&key), + &settings, + ) + .unwrap() + .worker_id + .to_string() + }; + let body = reserve( + crate::subjektiv::subject_worker_singleton_key(&subject.id).unwrap(), + "body", + ); + let job = reserve( + format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{}", subject.id), + "job", + ); + let attr = store + .record_session_attribution( + crate::subjektiv::SubjectSessionAttribution::new( + &subject.id, + "runtime-test", + &body, + "committed-session", + ) + .unwrap(), + ) + .unwrap(); + let host = bind_fixture_session(&store, attr); + let range = JournalSourceRange { + source: JournalSource { + session_id: "committed-session".into(), + pointer: JournalSourcePointer { + segment_id: uuid::Uuid::now_v7(), + entry_ref: "E0".into(), + }, + order: 0, + content_sha256: "f".repeat(64), + total_bytes: 10, + }, + bytes: JournalByteRange { start: 0, end: 10 }, + }; + let id = store + .accept_journal_sources(&host, "pending", &[range.clone()]) + .unwrap(); + let generation = store + .prepare_journal_generation( + &host, + &id, + subjektiv::api::JobAttemptBinding { + job_id: "active-job".into(), + attempt_id: "attempt".into(), + }, + vec![range], + ) + .unwrap(); + let expected = store.journal_compaction_snapshot(&host).unwrap(); + let path = format!("/api/w/{TEST_WORKSPACE_ID}/subjektiv/journal/compaction"); + let app = build_router(api.clone()); + for (worker, session, status) in [ + (&body, "committed-session", StatusCode::OK), + (&job, "committed-session", StatusCode::FORBIDDEN), + (&body, "foreign-session", StatusCode::FORBIDDEN), + ] { + let response = tokio::time::timeout( + std::time::Duration::from_secs(2), + app.clone().oneshot(runtime_source_request( + &identity, + Some(worker), + "POST", + &path, + serde_json::to_vec(&JournalCompactionRequest { + session_id: session.into(), + }) + .unwrap(), + )), + ) + .await + .expect("snapshot must not wait for an unavailable Runtime or active generation") + .unwrap(); + assert_eq!(response.status(), status); + if status == StatusCode::OK { + let snapshot: subjektiv::journal::startup::JournalStartupSnapshot = + serde_json::from_slice(&to_bytes(response.into_body(), usize::MAX).await.unwrap()) + .unwrap(); + assert_eq!(snapshot, expected); + } + } + let response = app + .oneshot( + axum::http::Request::builder() + .method("POST") + .uri(&path) + .header(CONTENT_TYPE, "application/json") + .body(Body::from( + serde_json::to_vec(&JournalCompactionRequest { + session_id: "committed-session".into(), + }) + .unwrap(), + )) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::UNAUTHORIZED); + assert_eq!(store.journal_compaction_snapshot(&host).unwrap(), expected); + assert_eq!( + store.active_journal_generation(&host, &id).unwrap(), + Some(generation) + ); +} + +fn connected_host_reader( + api: &WorkspaceApi, + worker: &RuntimeWorkerRef, + session: &str, +) -> worker::subjektiv::SubjektivHostConnection { + let audience = embedded_runtime_request_audience(&api.config).unwrap(); + let client = worker_runtime::worker_source::RuntimeOwnedWorkspaceClient::new( + TEST_WORKSPACE_ID, + api.config.backend_base_url.as_ref().unwrap(), + &worker.runtime_id, + &worker.worker_id, + ) + .with_runtime_request_source(&EMBEDDED_RUNTIME_REQUEST_IDENTITY, audience); + let mut config = manifest::ResolvedSubjektivFeatureConfig::default(); + config.profile.enabled = true; + config + .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: TEST_WORKSPACE_ID.into(), + language: "English".into(), + }) + .unwrap(); + worker::subjektiv::SubjektivHostConnection::new( + worker::subjektiv::BackendSubjektivHost::from_resolved_config(&config, Arc::new(client)) + .unwrap() + .unwrap(), + worker::subjektiv::SubjektivHostContext { + worker_id: worker.worker_id.clone(), + session_id: session.into(), + }, + ) + .unwrap() +}