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

fix: preserve feature metadata for concise Journal startup context

Browse Source
This commit is contained in:
Hare
2026-10-19 18:45:32 +09:00
parent cea0225a6e
commit c961f244ab
6 changed files with 219 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/src/feature/builtin/journal.rs
+25 -13
View File
@@ -47,12 +47,15 @@ impl JournalStartupContextSource {
}
#[async_trait]
impl crate::worker::InitialContextSource for JournalStartupContextSource {
async fn load(&self) -> Result<Vec<session_store::SystemItem>, String> {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
use sha2::{Digest, Sha256};
let snapshot = self
.host
.journal_startup_context()
.map_err(|e| e.to_string())?;
// A successful empty read contributes no model context. Keep an omitted
// snapshot even when empty: truncation is not evidence of absence.
let empty = snapshot.journals.is_empty() && !snapshot.omitted;
// Hash the same canonical JSON value retained by SystemItem, so replay
// can verify provenance without depending on domain struct field order.
let snapshot = serde_json::to_value(snapshot).map_err(|e| e.to_string())?;
@@ -60,6 +63,9 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
if snapshot_json.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES {
return Err("Host Journal startup snapshot exceeds independent byte budget".into());
}
if empty {
return Ok(Vec::new());
}
let prompts = self.prompts.load();
let body = prompts
.subjektiv_journal_startup(&snapshot_json)
@@ -68,19 +74,25 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource {
return Err("Journal startup prompt exceeds rendered byte budget".into());
}
let projection = prompts.projection();
Ok(vec![session_store::SystemItem::JournalStartup {
body,
snapshot,
snapshot_sha256: Sha256::digest(snapshot_json.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect(),
prompt_provenance: session_store::PromptRenderProvenance {
workspace_id: self.host.host.workspace_id(),
source_digest: projection.source_digest.clone(),
projection_digest: projection.catalog_digest.clone(),
logical_name: "internal.subjektiv_journal_startup".into(),
Ok(vec![session_store::LoggedSystemHistoryEntry {
item: session_store::SystemItem::JournalStartup {
body,
snapshot,
snapshot_sha256: Sha256::digest(snapshot_json.as_bytes())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect(),
prompt_provenance: session_store::PromptRenderProvenance {
workspace_id: self.host.host.workspace_id(),
source_digest: projection.source_digest.clone(),
projection_digest: projection.catalog_digest.clone(),
logical_name: "internal.subjektiv_journal_startup".into(),
},
},
metadata: crate::session_history::metadata(
session_store::LoggedSessionHistoryOrigin::JournalContext,
None,
),
}])
}
}
crates/worker/src/session_capture.rs
+6 -1
View File
@@ -902,6 +902,11 @@ mod tests {
Item::system_message("other-derived-secret"),
WorkerHistoryProvenance::DerivedSummary,
),
(
// Exclusion is driven by origin, not role or instructions in the body.
Item::user_message("prior-journal-secret"),
WorkerHistoryProvenance::JournalContext,
),
(
Item::system_message("raw-system-secret"),
WorkerHistoryProvenance::LegacyUnknown,
@@ -930,7 +935,7 @@ mod tests {
);
assert!(
capture
.read(ReadSelector::EntryRange([0, 3]), ReadOptions::default())
.read(ReadSelector::EntryRange([0, 4]), ReadOptions::default())
.entries
.is_empty()
);
crates/worker/src/session_history.rs
+8 -3
View File
@@ -133,9 +133,14 @@ pub(crate) fn restore_history_entries(
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 { .. }) {
// Old typed startup records used generic placeholder origins.
// Preserve explicitly supplied metadata on current contributions.
if matches!(
metadata.origin,
WorkerHistoryProvenance::DerivedSummary
| WorkerHistoryProvenance::LegacyUnknown
) && matches!(entry.item, session_store::SystemItem::JournalStartup { .. })
{
metadata.origin = WorkerHistoryProvenance::JournalContext;
}
metadata
crates/worker/src/subjektiv.rs
+80 -57
View File
@@ -1068,7 +1068,8 @@ enabled = true
async fn direct_startup_journal_is_durable_before_first_user_and_provider_exactly_once() {
use session_store::{LogEntry, LoggedSessionHistoryOrigin, Store, SystemItem};
use std::sync::atomic::Ordering;
for empty in [false, true] {
for (empty, omitted) in [(false, false), (true, false), (true, true)] {
let contributes_context = !empty || omitted;
let temp = tempfile::tempdir().unwrap();
let _sandbox =
crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
@@ -1078,7 +1079,7 @@ enabled = true
manifest.feature.subjektiv.profile.extraction.enabled = false;
let snapshot = JournalStartupSnapshot {
subject_id: "subject".into(),
omitted: false,
omitted,
journals: if empty {
vec![]
} else {
@@ -1127,43 +1128,51 @@ enabled = true
);
let controller = prepared.start().await.unwrap();
let initial = store.read_all(session, segment).unwrap();
let journal = initial
.iter()
.find_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
{
Some(entry)
}
_ => None,
})
.unwrap();
assert_eq!(
journal.metadata.origin,
LoggedSessionHistoryOrigin::DerivedSummary
);
let SystemItem::JournalStartup {
body,
snapshot: saved,
snapshot_sha256,
prompt_provenance,
} = &journal.item
else {
unreachable!()
};
assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap());
use sha2::{Digest, Sha256};
assert_eq!(
*snapshot_sha256,
Sha256::digest(serde_json::to_vec(saved).unwrap())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
);
assert_eq!(
prompt_provenance.logical_name,
"internal.subjektiv_journal_startup"
);
let journal = initial.iter().find_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
{
Some(entry)
}
_ => None,
});
assert_eq!(journal.is_some(), contributes_context);
let body = journal.map(|journal| {
assert_eq!(
journal.metadata.origin,
LoggedSessionHistoryOrigin::JournalContext
);
let SystemItem::JournalStartup {
body,
snapshot: saved,
snapshot_sha256,
prompt_provenance,
} = &journal.item
else {
unreachable!()
};
assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap());
use sha2::{Digest, Sha256};
assert_eq!(
*snapshot_sha256,
Sha256::digest(serde_json::to_vec(saved).unwrap())
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
);
assert_eq!(
prompt_provenance.logical_name,
"internal.subjektiv_journal_startup"
);
assert_eq!(
body.trim_end(),
format!(
"## Prior Journal context\n\n```json\n{}\n```",
serde_json::to_string(saved).unwrap()
)
);
body.as_str()
});
assert!(
!initial
.iter()
@@ -1183,25 +1192,39 @@ enabled = true
.await
.unwrap()
.unwrap();
let index = request
.items
.iter()
.position(|item| item.as_text() == Some(body.as_str()))
.unwrap();
let user_index = request
.items
.iter()
.position(|item| item.as_text() == Some("FIRST_USER_REQUEST"))
.unwrap();
assert!(index < user_index);
assert_eq!(
request
if let Some(body) = body {
let index = request
.items
.iter()
.filter(|item| item.as_text() == Some(body.as_str()))
.count(),
1
);
.position(|item| item.as_text() == Some(body))
.unwrap();
let user_index = request
.items
.iter()
.position(|item| item.as_text() == Some("FIRST_USER_REQUEST"))
.unwrap();
assert!(index < user_index);
assert!(matches!(
request.items[index],
agen::Item::Message {
role: agen::llm_client::types::Role::System,
..
}
));
assert_eq!(
request
.items
.iter()
.filter(|item| item.as_text() == Some(body))
.count(),
1
);
} else {
assert!(!request.items.iter().any(|item| {
item.as_text()
.is_some_and(|text| text.contains("Prior Journal context"))
}));
}
tokio::time::timeout(Duration::from_secs(5), async {
loop {
if matches!(events.recv().await.unwrap(), protocol::Event::RunEnd { .. }) {
@@ -1214,7 +1237,7 @@ enabled = true
}
assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
let logs = store.read_all(session, segment).unwrap();
assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).count(), 1);
assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).count(), usize::from(contributes_context));
controller
.handle
.send(Method::Shutdown {
crates/worker/src/worker.rs
+97 -22
View File
@@ -2917,10 +2917,12 @@ pub(crate) struct ResidentContextRefresh {
pub behavior_md: String,
}
/// Initial-only Feature contributions; prepared with the head, never at restore/request hooks.
/// Initial-only System messages and Feature-owned metadata. Core persists and
/// replays each pair unchanged; it does not infer provenance from the payload.
/// Prepared with the head, never at restore/request hooks.
#[async_trait::async_trait]
pub(crate) trait InitialContextSource: Send + Sync {
async fn load(&self) -> Result<Vec<SystemItem>, String>;
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String>;
}
#[async_trait::async_trait]
@@ -5878,16 +5880,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
forked_from: None,
compacted_from: None,
}];
for item in items {
for entry in items {
seed.push(LogEntry::AnnotatedSystemItem {
ts: segment_log::now_millis(),
entry: session_store::LoggedSystemHistoryEntry {
item,
metadata: new_history_metadata(
WorkerHistoryProvenance::DerivedSummary,
None,
),
},
entry,
extensions: Vec::new(),
});
}
@@ -5914,12 +5910,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
));
};
if *session_id != loc.session_id
|| seed.iter().skip(1).any(|entry| {
!matches!(entry,
LogEntry::AnnotatedSystemItem { entry, .. }
if entry.metadata.origin == WorkerHistoryProvenance::DerivedSummary
)
})
|| seed
.iter()
.skip(1)
.any(|entry| !matches!(entry, LogEntry::AnnotatedSystemItem { .. }))
{
return Err(WorkerError::InvalidState(
"unexpected records in initial segment seed".into(),
@@ -16738,12 +16732,92 @@ permission = "write"
}
#[async_trait]
impl InitialContextSource for StartupTestSource {
async fn load(&self) -> Result<Vec<SystemItem>, String> {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
self.calls.fetch_add(1, Ordering::SeqCst);
if self.fail.load(Ordering::SeqCst) {
return Err("startup read failed".into());
}
Ok(vec![startup_test_item()])
Ok(vec![session_store::LoggedSystemHistoryEntry {
item: startup_test_item(),
metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None),
}])
}
}
#[tokio::test]
async fn initial_system_messages_preserve_feature_metadata_without_payload_classification() {
struct Source(Vec<session_store::LoggedSystemHistoryEntry>);
#[async_trait]
impl InitialContextSource for Source {
async fn load(&self) -> Result<Vec<session_store::LoggedSystemHistoryEntry>, String> {
Ok(self.0.clone())
}
}
let temp = tempfile::tempdir().unwrap();
let store = session_store::FsStore::new(temp.path()).unwrap();
let expected = vec![
session_store::LoggedSystemHistoryEntry {
item: SystemItem::Notification {
message: "reference".into(),
body: "reference".into(),
prompt_provenance: None,
},
metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None),
},
session_store::LoggedSystemHistoryEntry {
item: startup_test_item(),
metadata: new_history_metadata(
WorkerHistoryProvenance::BackendInstruction {
operation_id: Some("feature-selected-origin".into()),
},
Some(SessionHistoryDerivation {
sources: vec![SessionHistoryEntryId::new()],
}),
),
},
];
let mut worker = Worker::new(
minimal_manifest(),
Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(NoopClient),
store.clone(),
WorkerWorkspaceContext::no_workspace(),
WorkerFilesystemAuthority::None,
Scope::empty(),
)
.await
.unwrap();
worker.install_initial_context_source(Arc::new(Source(expected.clone())));
worker.materialize_durable_session_head().await.unwrap();
let log = store
.read_all(worker.session_id(), worker.segment_id())
.unwrap();
let saved: Vec<_> = log
.iter()
.filter_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. } => Some(entry),
_ => None,
})
.collect();
assert_eq!(
serde_json::to_value(saved).unwrap(),
serde_json::to_value(&expected).unwrap()
);
let replay =
restore_history_entries(worker.session_id(), worker.segment_id(), &log).unwrap();
assert_eq!(replay.len(), expected.len());
for (entry, expected) in replay.iter().zip(&expected) {
assert_eq!(entry.annotation, expected.metadata);
assert_eq!(
entry.item.as_text(),
Some(expected.item.history_text().as_str())
);
assert!(matches!(
entry.item,
Item::Message {
role: agen::llm_client::types::Role::System,
..
}
));
}
}
@@ -17240,7 +17314,7 @@ permission = "write"
}
#[tokio::test]
async fn normal_restore_revalidates_current_body_authority_without_startup_fetch() {
async fn normal_restore_revalidates_current_body_authority_without_refetching_empty_startup() {
let temp = tempfile::tempdir().unwrap();
let _sandbox = worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
let (manifest, store) = body_publication_fixture(&temp);
@@ -17258,7 +17332,7 @@ permission = "write"
let initial = store.read_all(session, segment).unwrap();
let saved_startup = initial
.iter()
.find_map(|entry| match entry {
.filter_map(|entry| match entry {
LogEntry::AnnotatedSystemItem { entry, .. }
if matches!(entry.item, SystemItem::JournalStartup { .. }) =>
{
@@ -17266,7 +17340,8 @@ permission = "write"
}
_ => None,
})
.unwrap();
.collect::<Vec<_>>();
assert!(saved_startup.is_empty());
stop_body_publication_controller(controller).await;
assert_eq!(host.startup_calls.load(Ordering::SeqCst), 1);
host.startup_unavailable.store(true, Ordering::SeqCst);
@@ -17319,7 +17394,7 @@ permission = "write"
.collect::<Vec<_>>();
assert_eq!(
serde_json::to_value(saved).unwrap(),
serde_json::to_value(vec![&saved_startup]).unwrap()
serde_json::to_value(&saved_startup).unwrap()
);
stop_body_publication_controller(controller).await;
} else {