Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

chore: merge develop into hare/develop

Browse Source
This commit is contained in:
Hare
2026-10-19 01:31:23 +09:00
parent b02360fd75 41c9d2d5c7
commit 24bc45a31e
66 changed files with 8656 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/feature/builtin/journal.rs
+60
View File
@@ -3,6 +3,12 @@
//! Hooks request Host reconciliation, never infer source ranges from in-memory history.
//! Hosts retain exact operation inputs/receipts and validate live Job authority on every
//! operation. Successful operations are never rolled back by Job finalization.
//!
//! Compaction freezes a read-only Host snapshot independently of admission/Job progress.
//! Its rendered reference is shared by mandatory compactor input and the replacement seed;
//! no restore/turn hook refreshes it. Previous references are only historical compactor
//! material so fallback can recover information intentionally omitted from an earlier
//! complementary summary, without accumulating stale references in the next segment.
use crate::feature::{
FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule,
@@ -79,6 +85,60 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
}
}
/// Read once without waiting for recording. The returned body is immutable mandatory
/// compactor input and the exact replacement-segment reference, including fallback status.
/// Selection and authorization belong to the Host, not the core compaction engine.
pub(crate) fn freeze_compaction_journal(
connection: std::io::Result<Option<SubjektivHostConnection>>,
prompts: &crate::prompt::catalog::PromptCatalog,
) -> Result<String, crate::prompt::catalog::CatalogError> {
let workspace_id = connection.as_ref().ok().and_then(|connection| {
connection
.as_ref()
.and_then(|connection| connection.host.workspace_id())
});
let read = match connection {
Ok(Some(connection)) if connection.host.journal_compaction_granted() => {
Some(connection.journal_compaction_context().map_err(|_| ()))
}
Ok(_) => None,
Err(_) => Some(Err(())),
};
let (status, json) = match read {
None => ("disabled", String::new()),
Some(Ok(snapshot)) => {
let status = if snapshot.journals.is_empty() {
"absent"
} else {
"available"
};
match serde_json::to_string(&snapshot) {
Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => {
(status, json)
}
_ => ("unavailable", String::new()),
}
}
Some(Err(())) => ("unavailable", String::new()),
};
if status == "unavailable" {
tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal");
}
use sha2::{Digest, Sha256};
let snapshot_sha256: Option<String> = (!json.is_empty()).then(|| {
Sha256::digest(json.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect()
});
prompts.subjektiv_journal_compaction(
status,
&json,
snapshot_sha256.as_deref(),
workspace_id.as_deref(),
)
}
#[derive(Clone)]
pub struct JournalRecordingFeature {
host: SubjektivHostConnection,
crates/worker/src/feature/builtin/manage_worker.rs
+21 -3
View File
@@ -175,6 +175,26 @@ impl WorkerControlService for WorkspaceWorkerControlService {
"invalid Workspace control response: missing items".to_string(),
)
})?;
for item in items.iter_mut() {
let subject = item.get_mut("subject").ok_or_else(|| {
WorkspaceClientError::Request(
"invalid Workspace control response: missing subject".to_string(),
)
})?;
let runtime: server_api::WorkerControlSubject = serde_json::from_value(subject.take())
.map_err(|error| {
WorkspaceClientError::Request(format!(
"invalid Workspace control subject: {error}"
))
})?;
// Backend lists only Runtime subjects; Tools use a tagged union so
// callers can pass the observed subject unchanged to control tools.
*subject = serde_json::json!({
"kind": "runtime_worker",
"runtime_id": runtime.runtime_id,
"worker_id": runtime.worker_id,
});
}
let mut seen = std::collections::HashSet::new();
items.retain(|item| {
item.get("subject")
@@ -1318,7 +1338,6 @@ mod tests {
"items": [
{
"subject": {
"kind": "runtime_worker",
"runtime_id": "runtime-a",
"worker_id": "worker-a"
},
@@ -1326,7 +1345,6 @@ mod tests {
},
{
"subject": {
"kind": "runtime_worker",
"runtime_id": "runtime-a",
"worker_id": "worker-a"
},
@@ -1664,7 +1682,7 @@ mod tests {
}
#[tokio::test]
async fn control_service_deduplicates_worker_list_by_stable_subject() {
async fn control_service_projects_runtime_subject_and_deduplicates_worker_list() {
let client = Arc::new(RecordingWorkspaceClient::default());
let control = WorkspaceWorkerControlService {
client: client.clone(),
crates/worker/src/prompt/catalog.rs
+22
View File
@@ -103,6 +103,7 @@ pub enum WorkerPrompt {
ResidentMemoryRestoreSection,
SubjektivResidentContext,
SubjektivJournalStartup,
SubjektivJournalCompaction,
WorkerOrchestrationGuidanceSection,
SubWorkerSpawnToolDescription,
}
@@ -128,6 +129,7 @@ impl WorkerPrompt {
Self::ResidentMemorySummarySection => "internal.resident_memory_summary_section",
Self::ResidentMemoryRestoreSection => "internal.resident_memory_restore_section",
Self::SubjektivJournalStartup => "internal.subjektiv_journal_startup",
Self::SubjektivJournalCompaction => "internal.subjektiv_journal_compaction",
Self::SubjektivResidentContext => "internal.subjektiv_resident_context",
Self::WorkerOrchestrationGuidanceSection => {
"internal.worker_orchestration_guidance_section"
@@ -154,6 +156,7 @@ impl WorkerPrompt {
WorkerPrompt::ResidentMemoryRestoreSection,
WorkerPrompt::SubjektivResidentContext,
WorkerPrompt::SubjektivJournalStartup,
WorkerPrompt::SubjektivJournalCompaction,
WorkerPrompt::WorkerOrchestrationGuidanceSection,
WorkerPrompt::SubWorkerSpawnToolDescription,
];
@@ -543,6 +546,25 @@ impl PromptCatalog {
},
)
}
pub(crate) fn subjektiv_journal_compaction(
&self,
status: &str,
snapshot_json: &str,
snapshot_sha256: Option<&str>,
workspace_id: Option<&str>,
) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::SubjektivJournalCompaction,
minijinja::context! {
status => status,
snapshot_json => snapshot_json,
snapshot_sha256 => snapshot_sha256,
workspace_id => workspace_id,
source_digest => self.projection.source_digest,
projection_digest => self.projection.catalog_digest,
},
)
}
pub fn subjektiv_journal_startup(&self, snapshot_json: &str) -> Result<String, CatalogError> {
self.render(
WorkerPrompt::SubjektivJournalStartup,
crates/worker/src/session_capture.rs
+3 -3
View File
@@ -736,9 +736,9 @@ fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance
}
WorkerHistoryProvenance::ModelOutput { .. } => SessionEntryProvenance::ModelOutput,
WorkerHistoryProvenance::ToolOutput { .. } => SessionEntryProvenance::ToolOutput,
WorkerHistoryProvenance::CompactionSummary | WorkerHistoryProvenance::DerivedSummary => {
SessionEntryProvenance::DerivedSummary
}
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::DerivedSummary
| WorkerHistoryProvenance::JournalContext => SessionEntryProvenance::DerivedSummary,
WorkerHistoryProvenance::LegacyUnknown => SessionEntryProvenance::LegacyUnknown,
}
}
crates/worker/src/session_history.rs
+11 -4
View File
@@ -156,10 +156,17 @@ pub(crate) fn restore_history_entries(
}
*target = from_logged(entry);
}
LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new(
entry.item.to_history_item(),
entry.metadata.clone(),
)),
LogEntry::AnnotatedSystemItem { entry, .. } => {
history.push(HistoryEntry::new(entry.item.to_history_item(), {
let mut metadata = entry.metadata.clone();
// Typed startup records predate the dedicated Journal origin.
// Classify only this trusted event, never a plaintext marker.
if matches!(entry.item, session_store::SystemItem::JournalStartup { .. }) {
metadata.origin = WorkerHistoryProvenance::JournalContext;
}
metadata
}))
}
_ => {}
}
}
crates/worker/src/subjektiv.rs
+47
View File
@@ -80,6 +80,17 @@ pub trait SubjektivHost: Send + Sync + 'static {
"Journal startup not granted".into(),
))
}
fn journal_compaction_granted(&self) -> bool {
false
}
fn journal_compaction_context(
&self,
_context: &SubjektivHostContext,
) -> Result<JournalStartupSnapshot, SubjektivHostError> {
Err(SubjektivHostError::Unavailable(
"Journal compaction not granted".into(),
))
}
/// Recording admission is separate from both Memory and Journal Job operation grants.
fn journal_recording_granted(&self) -> bool {
false
@@ -267,6 +278,29 @@ impl SubjektivHost for BackendSubjektivHost {
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
)
}
fn journal_compaction_granted(&self) -> bool {
true
}
fn journal_compaction_context(
&self,
context: &SubjektivHostContext,
) -> Result<JournalStartupSnapshot, SubjektivHostError> {
decode(
self.client
.execute_with_timeout(
WorkspaceRequest::json(
WorkspaceRequestMethod::Post,
format!("/api/w/{}/subjektiv/journal/compaction", self.workspace_id),
serde_json::to_string(&transport::JournalCompactionRequest {
session_id: context.session_id.clone(),
})
.map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
),
Duration::from_secs(5),
)
.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.
@@ -488,6 +522,19 @@ impl SubjektivHostConnection {
}
self.host.journal_startup_context(&self.context)
}
pub fn journal_compaction_granted(&self) -> bool {
self.host.journal_compaction_granted()
}
pub fn journal_compaction_context(&self) -> Result<JournalStartupSnapshot, SubjektivHostError> {
if !self.host.journal_compaction_granted() {
return Err(SubjektivHostError::Unavailable(
"Journal compaction not granted".into(),
));
}
self.host.journal_compaction_context(&self.context)
}
pub fn new(
host: Arc<dyn SubjektivHost>,
context: SubjektivHostContext,
crates/worker/src/worker.rs
+58 -4
View File
@@ -6918,6 +6918,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// within `retained_tokens`. Item-granular, turn boundaries ignored.
let cut = self.split_for_retained(retained_tokens);
// Freeze one Host-owned read before starting the compactor. Background recording
// remains independent; never re-read when building the replacement segment.
let journal_context = crate::feature::builtin::journal::freeze_compaction_journal(
self.subjektiv_host_connection(),
&self.prompts.load_full(),
)
.map_err(WorkerError::PromptCatalog)?;
let history_entries = self.session.history().entries();
let retain_from = cut.index.min(history_entries.len());
let initial_summary = initial_compaction_summary(history_entries)?;
@@ -6926,7 +6933,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// it as well would duplicate inherited context on every compact.
let retained_history_entries = history_entries[retain_from..]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
})
.cloned()
.collect::<Vec<_>>();
let mut retained_items = retained_history_entries
@@ -6943,7 +6956,13 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// never needs durable image bodies; relevant images are nominated with add_reference.
let mut entries_to_summarise = history_entries[..retain_from]
.iter()
.filter(|entry| entry.annotation.origin != WorkerHistoryProvenance::CompactionSummary)
.filter(|entry| {
!matches!(
entry.annotation.origin,
WorkerHistoryProvenance::CompactionSummary
| WorkerHistoryProvenance::JournalContext
)
})
.cloned()
.collect::<Vec<_>>();
project_compactor_capture_attachments(&mut entries_to_summarise);
@@ -7013,7 +7032,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
// The full Host summary is mandatory input, separate from the bounded
// prefix overview and current TaskStore snapshot.
let task_snapshot_text = self.task_feature.snapshot_text();
let summary_input = build_summary_input(
let mut summary_input = build_summary_input(
&items_to_summarise,
initial_summary.as_deref(),
&default_refs,
@@ -7025,6 +7044,28 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
summary_target_tokens,
},
);
// Previous references are historical source material, not additional latest
// Journals. In particular, a failed/empty read must not silently lose context
// that the previous summary intentionally delegated to its injected Journal.
// Carry them only into the compactor, never alongside the new reference seed.
for entry in history_entries
.iter()
.filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::JournalContext)
{
if let Some(body) = entry.item.as_text() {
summary_input.text.push_str(
"\n\n## Previous Journal reference (historical, not automatically retained)\n",
);
summary_input.text.push_str(body);
summary_input
.text
.push_str("\n## End previous Journal reference\n");
}
}
// Mandatory context, outside overview truncation. The internal Worker's normal
// request budget checks include this exact body and the full previous summary.
summary_input.text.push_str("\n\n");
summary_input.text.push_str(&journal_context);
if summary_input.warning_exceeded {
self.alert(
AlertLevel::Warn,
@@ -7340,6 +7381,8 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
crate::session_history::COMPACTION_SUMMARY_HEADER
));
compact_introduced_system_messages.push(summary_message.clone());
let journal_message = Item::system_message(journal_context);
compact_introduced_system_messages.push(journal_message.clone());
compact_introduced_system_messages.extend(auto_read_messages.iter().cloned());
if let Some(msg) = reference_message.as_ref() {
compact_introduced_system_messages.push(msg.clone());
@@ -7352,6 +7395,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
compact_introduced_system_messages.push(task_snapshot_message.clone());
new_history.push(summary_message);
new_history.push(journal_message);
new_history.extend(auto_read_messages);
if let Some(msg) = reference_message {
new_history.push(msg);
@@ -7407,6 +7451,16 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
}),
),
)
} else if index == 1 {
HistoryEntry::new(
item,
new_history_metadata(
WorkerHistoryProvenance::JournalContext,
Some(SessionHistoryDerivation {
sources: derived_sources.clone(),
}),
),
)
} else if let Some((offset, original)) = original_entries[original_cursor..]
.iter()
.enumerate()
@@ -16732,7 +16786,7 @@ permission = "write"
assert_eq!(replay[0].item.as_text(), Some("PRIOR_JOURNAL_REFERENCE"));
assert_eq!(
replay[0].annotation.origin,
WorkerHistoryProvenance::DerivedSummary
WorkerHistoryProvenance::JournalContext
);
let view =
crate::session_capture::SessionCapture::from_history_entries("startup", replay);
crates/worker/tests/compact_events_test.rs
+776 -1
View File
@@ -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<CompactionJournalHost>,
) -> 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<session_store::LoggedHistoryEntry>,
compacted: bool,
) {
self.restore_seed_with_system(history, compacted, vec![])
.await;
}
async fn restore_seed_with_system(
&mut self,
history: Vec<session_store::LoggedHistoryEntry>,
compacted: bool,
systems: Vec<session_store::LoggedSystemHistoryEntry>,
) {
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<subjektiv::journal::startup::JournalStartupSnapshot>,
reads: Mutex<Vec<worker::subjektiv::SubjektivHostContext>>,
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<String> {
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<server_api::SubjektivMemoryBackendResponse, worker::subjektiv::SubjektivHostError>
{
Err(worker::subjektiv::SubjektivHostError::Unavailable(
"not used".into(),
))
}
fn session(
&self,
_: &worker::subjektiv::SubjektivHostContext,
_: server_api::SubjektivSessionBackendOperation,
) -> Result<server_api::SubjektivSessionBackendResponse, worker::subjektiv::SubjektivHostError>
{
Err(worker::subjektiv::SubjektivHostError::Unavailable(
"not used".into(),
))
}
fn stage_candidate(
&self,
_: &worker::subjektiv::SubjektivHostContext,
_: memory::backend::MemoryStageCandidateOperation,
) -> Result<server_api::SubjektivStageCandidateResponse, worker::subjektiv::SubjektivHostError>
{
Err(worker::subjektiv::SubjektivHostError::Unavailable(
"not used".into(),
))
}
fn record_session(
&self,
_: &worker::subjektiv::SubjektivHostContext,
_: bool,
) -> Result<server_api::SubjektivRecordSessionResponse, worker::subjektiv::SubjektivHostError>
{
Err(worker::subjektiv::SubjektivHostError::Unavailable(
"not used".into(),
))
}
fn request_consolidation(
&self,
_: &worker::subjektiv::SubjektivHostContext,
_: memory::backend::MemoryConsolidateStagingOperation,
) -> Result<memory::backend::MemoryConsolidationOutput, worker::subjektiv::SubjektivHostError>
{
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<MockClient, TestStore>) -> 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::<Vec<_>>();
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::<String>();
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::<Vec<_>>();
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
);
}
}