From 6e729ed06c01595ba8e3736e5741d850162257f0 Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 03:21:22 +0900 Subject: [PATCH] refactor: isolate Feature System input migration and compaction policy --- crates/session-store/src/history.rs | 2 +- .../session-store/src/legacy_session_log.rs | 316 +++++++++- .../src/legacy_session_log/tests.rs | 2 + .../tests/fixtures/v1-system-inputs.jsonl | 3 + .../tests/fixtures/v4-system-inputs.jsonl | 8 + .../src/legacy_session_log/tests/v5.rs | 596 ++++++++++++++++++ crates/session-store/src/public_material.rs | 23 +- .../src/public_material/tests.rs | 170 +++-- crates/session-store/src/system_item.rs | 134 +--- .../session-store/src/worker_session_store.rs | 206 +++++- .../src/subjektiv/journal/tests/handoff.rs | 2 +- crates/worker/src/controller.rs | 6 +- crates/worker/src/feature/builtin/journal.rs | 169 ++++- crates/worker/src/feature/builtin/memory.rs | 192 ++++-- crates/worker/src/feature/builtin/task/mod.rs | 10 +- crates/worker/src/feature/invocation.rs | 39 ++ crates/worker/src/hook.rs | 80 ++- crates/worker/src/ipc/interceptor.rs | 90 ++- crates/worker/src/ipc/protocol_session.rs | 48 +- crates/worker/src/session_history.rs | 30 +- crates/worker/src/skill.rs | 44 ++ crates/worker/src/subjektiv.rs | 66 +- crates/worker/src/worker.rs | 394 +++++++----- crates/worker/tests/compact_events_test.rs | 180 ++++-- .../worker/tests/feature_invocation_test.rs | 20 +- .../subjektiv_journal_connected_tests.rs | 36 +- tools/state-contract-name-exceptions.json | 6 +- 27 files changed, 2226 insertions(+), 646 deletions(-) create mode 100644 crates/session-store/src/legacy_session_log/tests/fixtures/v1-system-inputs.jsonl create mode 100644 crates/session-store/src/legacy_session_log/tests/fixtures/v4-system-inputs.jsonl create mode 100644 crates/session-store/src/legacy_session_log/tests/v5.rs diff --git a/crates/session-store/src/history.rs b/crates/session-store/src/history.rs index f25617b1d..67d753c13 100644 --- a/crates/session-store/src/history.rs +++ b/crates/session-store/src/history.rs @@ -95,7 +95,7 @@ pub struct LoggedSessionHistoryMetadata { } impl LoggedSessionHistoryMetadata { - fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { + pub(crate) fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { if matches!(self.origin, LoggedSessionHistoryOrigin::Provided { .. }) && self.system_input.is_none() { diff --git a/crates/session-store/src/legacy_session_log.rs b/crates/session-store/src/legacy_session_log.rs index 987a53984..10e247539 100644 --- a/crates/session-store/src/legacy_session_log.rs +++ b/crates/session-store/src/legacy_session_log.rs @@ -1,12 +1,12 @@ -//! Versioned decoder for Session schemas that predate canonical annotated history. +//! Versioned decoder for Session schemas v1-4, including frozen Feature wire. //! //! These types are intentionally private to `session-store`. Current writers, //! replay, and public projections use [`crate::LogEntry`] exclusively; only the -//! Worker Session schema migration is allowed to deserialize these shapes. +//! explicit Session schema migration is allowed to deserialize these shapes. use agen::llm_client::types::RequestConfig; use protocol::Segment; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; #[cfg(test)] mod tests; @@ -17,6 +17,212 @@ use crate::{ SegmentOrigin, SessionExtension, SessionId, SystemItem, }; +// Frozen v1-4 Feature wire. These types are deliberately not exported and are +// never used by LogEntry/SystemItem's current deserializers. Payload structs +// retain the owner's v1 shape rather than erasing typed fields into arbitrary JSON. +#[derive(Debug, Serialize, Deserialize)] +struct JournalStartupPayload { + snapshot: serde_json::Value, + snapshot_sha256: String, + prompt_provenance: crate::PromptRenderProvenance, +} + +#[derive(Debug, Serialize, Deserialize)] +struct ResidentSummaryPayload { + #[serde(default)] + prompt_provenance: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +struct SubjectBehaviorPayload { + #[serde(default)] + behavior_md: String, + #[serde(default)] + prompt_provenance: Option, +} + +#[derive(Debug, Serialize, Deserialize)] +struct SkillActivationPayload { + name: String, +} + +#[derive(Debug, Serialize, Deserialize)] +struct InvocationResultPayload { + result: protocol::FeatureInvocationResult, +} + +#[derive(Debug, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum LegacyFeatureSystemItem { + JournalStartup { + body: String, + #[serde(flatten)] + payload: JournalStartupPayload, + }, + ResidentSummaryRefresh { + body: String, + #[serde(flatten)] + payload: ResidentSummaryPayload, + }, + SubjectBehaviorRefresh { + body: String, + #[serde(flatten)] + payload: SubjectBehaviorPayload, + }, + SkillActivation { + body: String, + #[serde(flatten)] + payload: SkillActivationPayload, + }, + FeatureInvocationResult { + body: String, + #[serde(flatten)] + payload: InvocationResultPayload, + }, +} + +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum LegacySystemItem { + Feature(LegacyFeatureSystemItem), + // Includes already-migrated Input records for interrupted migration retry. + Core(SystemItem), +} + +impl LegacySystemItem { + fn into_entry( + self, + mut metadata: LoggedSessionHistoryMetadata, + ) -> Result { + use crate::InputCompactionUse::{ReferenceInput, Retain}; + let item = match self { + Self::Core(item) => item, + Self::Feature(feature) => { + metadata.validate_system_input()?; + if metadata.system_input.is_some() { + return Err(crate::SystemInputError::InconsistentRecord); + } + let (body, input_metadata) = match feature { + LegacyFeatureSystemItem::JournalStartup { body, payload } => { + migrated_feature_input( + body, + "journal_startup", + "journal:startup", + payload, + ReferenceInput, + )? + } + LegacyFeatureSystemItem::ResidentSummaryRefresh { body, payload } => { + migrated_feature_input( + body, + "resident_summary_refresh", + "memory:resident-summary", + payload, + Retain, + )? + } + LegacyFeatureSystemItem::SubjectBehaviorRefresh { body, payload } => { + migrated_feature_input( + body, + "subject_behavior_refresh", + "subject:behavior", + payload, + Retain, + )? + } + LegacyFeatureSystemItem::SkillActivation { body, payload } => { + migrated_feature_input( + body, + "skill_activation", + "skill:activation", + payload, + Retain, + )? + } + LegacyFeatureSystemItem::FeatureInvocationResult { body, payload } => { + migrated_feature_input( + body, + "feature_invocation_result", + "feature:invocation-result", + payload, + Retain, + )? + } + }; + // Identity/origin/derivation are retained, never re-inferred. + metadata.system_input = Some(input_metadata); + SystemItem::Input { body } + } + }; + let entry = LoggedSystemHistoryEntry { item, metadata }; + entry.validate_system_input()?; + Ok(entry) + } +} + +fn migrated_feature_input( + body: String, + label: &str, + schema: &str, + payload: T, + compaction: crate::InputCompactionUse, +) -> Result<(String, crate::SystemInputMetadata), crate::SystemInputError> { + let metadata = crate::SystemInputMetadata::new( + migration_policy(compaction), + Some(crate::FeatureInputPayload::encode(schema, 1, &payload)?), + Some(crate::PublicSystemItem { + kind: label.into(), + body: body.clone(), + presentation: crate::PublicSystemPresentation::Context, + }), + )?; + Ok((body, metadata)) +} + +/// Annotated old wire used by v3/v4, and by already-converted files when an +/// older marker survived an interrupted migration. No ordinary decoder uses it. +#[derive(Debug, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum LegacyAnnotatedLogEntry { + AnnotatedSystemItem { + ts: u64, + entry: LegacySystemHistoryEntry, + #[serde(default)] + extensions: Vec, + }, +} + +#[derive(Debug, Deserialize)] +#[serde(deny_unknown_fields)] +struct LegacySystemHistoryEntry { + item: LegacySystemItem, + metadata: LoggedSessionHistoryMetadata, +} + +#[derive(Debug, Deserialize)] +#[serde(untagged)] +enum LegacyAnnotatedOrCurrent { + Annotated(LegacyAnnotatedLogEntry), + Current(LogEntry), +} + +impl LegacyAnnotatedOrCurrent { + fn into_current(self) -> Result { + match self { + Self::Annotated(LegacyAnnotatedLogEntry::AnnotatedSystemItem { + ts, + entry, + extensions, + }) => Ok(LogEntry::AnnotatedSystemItem { + ts, + entry: entry.item.into_entry(entry.metadata)?, + extensions, + }), + Self::Current(entry) => Ok(entry), + } + } +} + #[derive(Debug, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] enum LegacyHistoryLogEntry { @@ -47,7 +253,9 @@ enum LegacyHistoryLogEntry { }, SystemItem { ts: u64, - item: SystemItem, + item: LegacySystemItem, + #[serde(default)] + extensions: Vec, }, } @@ -57,7 +265,7 @@ enum LegacyHistoryLogEntry { #[serde(untagged)] enum LegacySessionLogEntryV1 { History(LegacyHistoryLogEntry), - Current(LogEntry), + Current(LegacyAnnotatedOrCurrent), } /// Schema v2 retained the v1 history shapes while adding non-history records. @@ -67,7 +275,7 @@ enum LegacySessionLogEntryV1 { #[serde(untagged)] enum LegacySessionLogEntryV2 { History(LegacyHistoryLogEntry), - Current(LogEntry), + Current(LegacyAnnotatedOrCurrent), } pub(crate) fn decode_entry( @@ -86,15 +294,16 @@ pub(crate) fn decode_entry( LegacySessionLogEntryV2::History(entry) => Entry::History(entry), LegacySessionLogEntryV2::Current(entry) => Entry::Current(entry), }, - 3 => Entry::Current(serde_json::from_str(line)?), + 3 | 4 => Entry::Current(serde_json::from_str(line)?), _ => unreachable!("legacy decoder called for unsupported schema {schema_version}"), }; - Ok(match entry { + match entry { Entry::History(entry) => { canonicalize_history_entry(session_id, segment_id, line_index, entry) } - Entry::Current(entry) => entry, - }) + Entry::Current(entry) => entry.into_current(), + } + .map_err(serde::de::Error::custom) } /// Schema 3 used DerivedSummary for Host compaction seeds. Only the trusted @@ -152,9 +361,75 @@ pub(crate) fn migrate_compaction_origins( } } +/// Explicit v1-4 -> v5 boundary. No Feature inference belongs in ordinary +/// readers/restoration. Invalid or oversized payloads fail migration; the generic +/// payload/body/public limits are deliberately not relaxed for legacy data. +pub(crate) fn migrate_system_inputs(entry: &mut LogEntry) -> Result<(), crate::SystemInputError> { + entry.validate_system_inputs()?; + match entry { + // Feature wire has already been converted by the explicit old-schema + // decoder. Ordinary LogEntry can contain only generic Input/core events. + LogEntry::AnnotatedSystemItem { entry, .. } => entry.validate_system_input()?, + LogEntry::AnnotatedSegmentStart { history, .. } + | LogEntry::AnnotatedUserInput { history, .. } => { + for entry in history { + migrate_flattened_system_input(entry)?; + } + } + LogEntry::AnnotatedAssistantItem { entry, .. } + | LogEntry::AnnotatedToolResult { entry, .. } + | LogEntry::ToolResultCorrected { entry, .. } => { + migrate_flattened_system_input(entry)?; + } + _ => {} + } + entry.validate_system_inputs() +} + +fn migration_policy(compaction: crate::InputCompactionUse) -> crate::SystemInputUsePolicy { + crate::SystemInputUsePolicy { + recording_material: crate::InputEvidenceUse::Exclude, + memory_evidence: crate::InputEvidenceUse::Exclude, + observation: crate::InputObservationUse::Hidden, + compaction, + } +} + +fn migrate_flattened_system_input( + entry: &mut LoggedHistoryEntry, +) -> Result<(), crate::SystemInputError> { + // Check before supplying legacy policy: Provided without metadata is a + // malformed NEW record, not permission to synthesize a producer policy. + entry.validate_system_input()?; + if entry.metadata.system_input.is_none() + && matches!( + entry.item, + LoggedItem::Message { + role: crate::LoggedRole::System, + .. + } + ) + { + let compaction = match entry.metadata.origin { + LoggedSessionHistoryOrigin::JournalContext => crate::InputCompactionUse::ReferenceInput, + LoggedSessionHistoryOrigin::CompactionSummary => { + crate::InputCompactionUse::OpeningSummary + } + _ => crate::InputCompactionUse::Retain, + }; + // No Feature payload or public projection can be recovered from text. + entry.metadata.system_input = Some(crate::SystemInputMetadata::new( + migration_policy(compaction), + None, + None, + )?); + } + entry.validate_system_input() +} + enum Entry { History(LegacyHistoryLogEntry), - Current(LogEntry), + Current(LegacyAnnotatedOrCurrent), } fn legacy_metadata( @@ -182,8 +457,8 @@ fn canonicalize_history_entry( segment_id: SegmentId, line_index: usize, entry: LegacyHistoryLogEntry, -) -> LogEntry { - match entry { +) -> Result { + Ok(match entry { LegacyHistoryLogEntry::SegmentStart { ts, session_id, @@ -237,13 +512,14 @@ fn canonicalize_history_entry( metadata: legacy_metadata(segment_id, line_index, 0), }, }, - LegacyHistoryLogEntry::SystemItem { ts, item } => LogEntry::AnnotatedSystemItem { + LegacyHistoryLogEntry::SystemItem { ts, - entry: LoggedSystemHistoryEntry { - item, - metadata: legacy_metadata(segment_id, line_index, 0), - }, - extensions: Vec::new(), + item, + extensions, + } => LogEntry::AnnotatedSystemItem { + ts, + entry: item.into_entry(legacy_metadata(segment_id, line_index, 0))?, + extensions, }, - } + }) } diff --git a/crates/session-store/src/legacy_session_log/tests.rs b/crates/session-store/src/legacy_session_log/tests.rs index 314ee6d5f..d7914c194 100644 --- a/crates/session-store/src/legacy_session_log/tests.rs +++ b/crates/session-store/src/legacy_session_log/tests.rs @@ -1,5 +1,7 @@ use super::*; +mod v5; + fn seed( origin: LoggedSessionHistoryOrigin, compacted: bool, diff --git a/crates/session-store/src/legacy_session_log/tests/fixtures/v1-system-inputs.jsonl b/crates/session-store/src/legacy_session_log/tests/fixtures/v1-system-inputs.jsonl new file mode 100644 index 000000000..742ab5557 --- /dev/null +++ b/crates/session-store/src/legacy_session_log/tests/fixtures/v1-system-inputs.jsonl @@ -0,0 +1,3 @@ +{"kind":"system_item","ts":1,"item":{"kind":"skill_activation","name":"legacy-skill","body":"legacy skill body"}} +{"kind":"user_input","ts":2,"segments":[{"kind":"text","content":"legacy user body"}],"extensions":[{"domain":"old","payload":{"kept":true}}]} +{"kind":"assistant_item","ts":3,"item":{"kind":"message","role":"assistant","content":[{"kind":"text","text":"legacy reply"}]}} diff --git a/crates/session-store/src/legacy_session_log/tests/fixtures/v4-system-inputs.jsonl b/crates/session-store/src/legacy_session_log/tests/fixtures/v4-system-inputs.jsonl new file mode 100644 index 000000000..59a391d01 --- /dev/null +++ b/crates/session-store/src/legacy_session_log/tests/fixtures/v4-system-inputs.jsonl @@ -0,0 +1,8 @@ +{"kind":"annotated_system_item","ts":101,"entry":{"item":{"kind":"journal_startup","body":"journal body\nunchanged","snapshot":{"entries":[{"text":"private snapshot"}]},"snapshot_sha256":"snapshot-digest","prompt_provenance":{"workspace_id":"workspace","source_digest":"source","projection_digest":"projection","logical_name":"journal_startup"}},"metadata":{"entry_id":"journal-id","origin":{"kind":"legacy_unknown"},"derivation":{"sources":["source-id"]}}},"extensions":[{"domain":"migration-fixture","payload":{"kept":true}}]} +{"kind":"annotated_system_item","ts":102,"entry":{"item":{"kind":"resident_summary_refresh","body":"resident body"},"metadata":{"entry_id":"resident-id","origin":{"kind":"derived_summary"}}}} +{"kind":"annotated_system_item","ts":103,"entry":{"item":{"kind":"subject_behavior_refresh","body":"rendered subject body","behavior_md":"exact behavior","prompt_provenance":{"source_digest":"subject-source","projection_digest":"subject-projection","logical_name":"subject_behavior"}},"metadata":{"entry_id":"subject-id","origin":{"kind":"backend_instruction","operation_id":"op"}}}} +{"kind":"annotated_system_item","ts":104,"entry":{"item":{"kind":"skill_activation","body":"skill body","name":"example-skill"},"metadata":{"entry_id":"skill-id","origin":{"kind":"legacy_unknown"}}}} +{"kind":"annotated_system_item","ts":105,"entry":{"item":{"kind":"feature_invocation_result","body":"result body","result":{"invocation_id":"invoke-id","identity":"builtin:test/prepare","status":"outcome_unknown","message":"may have committed","context":null}},"metadata":{"entry_id":"result-id","origin":{"kind":"legacy_unknown"}}}} +{"kind":"annotated_user_input","ts":106,"segments":[],"history":[{"item":{"kind":"message","role":"system","content":[{"kind":"text","text":"[Journal] prefix is not Feature evidence"}]},"metadata":{"entry_id":"flat-journal","origin":{"kind":"journal_context"},"derivation":{"sources":["journal-id"]}}},{"item":{"kind":"message","role":"system","content":[{"kind":"text","text":"summary body"}]},"metadata":{"entry_id":"flat-summary","origin":{"kind":"compaction_summary"}}},{"item":{"kind":"message","role":"system","content":[{"kind":"text","text":"[Skill: example] never infer Feature"}]},"metadata":{"entry_id":"flat-other","origin":{"kind":"legacy_unknown"}}}]} +{"kind":"annotated_system_item","ts":107,"entry":{"item":{"kind":"notification","message":"raw message","body":"wrapped message"},"metadata":{"entry_id":"core-id","origin":{"kind":"legacy_unknown"}}}} +{"kind":"annotated_system_item","ts":108,"entry":{"item":{"kind":"input","body":"already generic body"},"metadata":{"entry_id":"generic-id","origin":{"kind":"provided","producer":"future-owner"},"derivation":{"sources":["source-id"]},"system_input":{"usage":{"recording_material":"include","memory_evidence":"exclude","observation":"readable","compaction":"retain"},"feature_payload":{"schema":"future:unknown","version":99,"value":{"opaque":"retained"}},"public":null}}}} diff --git a/crates/session-store/src/legacy_session_log/tests/v5.rs b/crates/session-store/src/legacy_session_log/tests/v5.rs new file mode 100644 index 000000000..f197eeec9 --- /dev/null +++ b/crates/session-store/src/legacy_session_log/tests/v5.rs @@ -0,0 +1,596 @@ +//! Fixed old wire bytes, not fixtures generated by the current serializer. +use super::*; +use crate::{FsStore, InputCompactionUse, Store, WorkerSessionStore}; +use serde_json::{Value, json}; +use std::{fs, path::Path}; + +const V4: &str = include_str!("fixtures/v4-system-inputs.jsonl"); +const V1: &str = include_str!("fixtures/v1-system-inputs.jsonl"); + +fn decode_fixture(version: u32, source: &str, segment: SegmentId) -> Vec { + source + .lines() + .enumerate() + .map(|(index, line)| decode_entry(version, line, SessionId::nil(), segment, index).unwrap()) + .collect() +} + +fn encode(entries: &[LogEntry]) -> Vec { + let mut bytes = Vec::new(); + for entry in entries { + serde_json::to_writer(&mut bytes, entry).unwrap(); + bytes.push(b'\n'); + } + bytes +} + +fn migrate(entries: &mut [LogEntry]) { + for entry in entries { + migrate_system_inputs(entry).unwrap(); + entry.validate_system_inputs().unwrap(); + } +} + +fn write_worker(root: &Path, version: u32, source: &[u8]) -> (SessionId, SegmentId) { + let session = SessionId::now_v7(); + let segment = SegmentId::from_u128(1); + fs::create_dir_all(root.join("segments")).unwrap(); + fs::write( + root.join("session.json"), + serde_json::to_vec(&json!({ + "schema_version": version, "session_id": session, + })) + .unwrap(), + ) + .unwrap(); + fs::write( + root.join("segments").join(format!("{segment}.jsonl")), + source, + ) + .unwrap(); + (session, segment) +} + +fn assert_marker(root: &Path, version: u32) { + let manifest: Value = + serde_json::from_slice(&fs::read(root.join("session.json")).unwrap()).unwrap(); + assert_eq!(manifest["schema_version"], version); +} + +#[test] +fn v4_fixed_wire_preserves_identity_body_payload_public_and_independent_policy() { + let mut entries = decode_fixture(4, V4, SegmentId::nil()); + let before = entries.clone(); + migrate(&mut entries); + let schemas = [ + "journal:startup", + "memory:resident-summary", + "subject:behavior", + "skill:activation", + "feature:invocation-result", + ]; + let old_wire: Vec = V4 + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + for index in 0..5 { + let LogEntry::AnnotatedSystemItem { + ts, + entry, + extensions, + } = &entries[index] + else { + panic!() + }; + let old = &old_wire[index]; + let old_item = &old["entry"]["item"]; + let old_metadata: LoggedSessionHistoryMetadata = + serde_json::from_value(old["entry"]["metadata"].clone()).unwrap(); + let old_ext: Vec = old + .get("extensions") + .map(|value| serde_json::from_value(value.clone()).unwrap()) + .unwrap_or_default(); + assert_eq!(*ts, old["ts"].as_u64().unwrap()); + assert_eq!(extensions, &old_ext); + assert!(matches!(entry.item, SystemItem::Input { .. })); + assert_eq!( + entry.item.history_text(), + old_item["body"].as_str().unwrap() + ); + assert_eq!(entry.metadata.entry_id, old_metadata.entry_id); + assert_eq!(entry.metadata.origin, old_metadata.origin); + assert_eq!(entry.metadata.derivation, old_metadata.derivation); + let metadata = entry.metadata.system_input.as_ref().unwrap(); + let public = metadata.public_item().unwrap(); + assert_eq!(public.kind, old_item["kind"].as_str().unwrap()); + assert_eq!(public.body, old_item["body"].as_str().unwrap()); + assert_eq!( + public.presentation, + crate::PublicSystemPresentation::Context + ); + assert_eq!( + metadata.usage(), + migration_policy(if index == 0 { + InputCompactionUse::ReferenceInput + } else { + InputCompactionUse::Retain + }) + ); + let payload = metadata.feature_payload().unwrap(); + assert_eq!(payload.schema(), schemas[index]); + assert_eq!(payload.version(), 1); + let value: Value = payload.decode(schemas[index], 1).unwrap(); + // Independently require each frozen owner's typed v1 payload contract. + match index { + 0 => { + let _: JournalStartupPayload = payload.decode(schemas[index], 1).unwrap(); + } + 1 => { + let _: ResidentSummaryPayload = payload.decode(schemas[index], 1).unwrap(); + } + 2 => { + let _: SubjectBehaviorPayload = payload.decode(schemas[index], 1).unwrap(); + } + 3 => { + let _: SkillActivationPayload = payload.decode(schemas[index], 1).unwrap(); + } + 4 => { + let _: InvocationResultPayload = payload.decode(schemas[index], 1).unwrap(); + } + _ => unreachable!(), + } + let mut expected = old_item.clone(); + expected.as_object_mut().unwrap().remove("kind"); + expected.as_object_mut().unwrap().remove("body"); + // Optional provenance is an explicit Option in the v1 payload contract. + if index == 1 { + expected["prompt_provenance"] = Value::Null; + } + if index == 4 { + let result: protocol::FeatureInvocationResult = + serde_json::from_value(old_item["result"].clone()).unwrap(); + let typed: InvocationResultPayload = payload.decode(schemas[index], 1).unwrap(); + assert_eq!(typed.result, result); + // Preserve the protocol's typed serialization contract, including + // its canonical representation of optional result context. + expected["result"] = serde_json::to_value(result).unwrap(); + } + assert_eq!(value, expected); + assert!(value.get("body").is_none()); + } + let LogEntry::AnnotatedUserInput { history, .. } = &entries[5] else { + panic!() + }; + let LogEntry::AnnotatedUserInput { history: old, .. } = &before[5] else { + panic!() + }; + for (index, compaction) in [ + InputCompactionUse::ReferenceInput, + InputCompactionUse::OpeningSummary, + InputCompactionUse::Retain, + ] + .into_iter() + .enumerate() + { + assert_eq!(history[index].item, old[index].item); + assert_eq!( + history[index].metadata.entry_id, + old[index].metadata.entry_id + ); + assert_eq!(history[index].metadata.origin, old[index].metadata.origin); + assert_eq!( + history[index].metadata.derivation, + old[index].metadata.derivation + ); + let metadata = history[index].metadata.system_input.as_ref().unwrap(); + assert_eq!(metadata.usage(), migration_policy(compaction)); + assert!(metadata.public_item().is_none()); + assert!(metadata.feature_payload().is_none()); + } + // Core notification is unchanged; existing generic Input and unknown valid + // owner payload are retained without decoding/reclassifying a Feature. + assert_eq!(encode(&entries[6..]), encode(&before[6..])); + let once = encode(&entries); + migrate(&mut entries); + assert_eq!(encode(&entries), once); +} + +#[test] +fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() { + for version in 1..=4 { + let source = if version <= 2 { V1 } else { V4 }; + let root = tempfile::tempdir().unwrap(); + let (session, segment) = write_worker(root.path(), version, source.as_bytes()); + let path = root + .path() + .join("segments") + .join(format!("{segment}.jsonl")); + // Simulate an interrupted publication: v5 records precede legacy records + // in one file, and a fully new file coexists with an untouched old file. + let mut converted = decode_fixture(version, source, segment); + migrate(&mut converted); + let mut mixed = encode(&converted[..1]); + mixed.extend_from_slice(source.split_once('\n').unwrap().1.as_bytes()); + fs::write(&path, mixed).unwrap(); + let new_path = root + .path() + .join("segments") + .join(format!("{}.jsonl", SegmentId::from_u128(2))); + let new_bytes = encode(&converted); + fs::write(&new_path, &new_bytes).unwrap(); + let old_path = root + .path() + .join("segments") + .join(format!("{}.jsonl", SegmentId::from_u128(3))); + fs::write(&old_path, source).unwrap(); + + let store = WorkerSessionStore::new(root.path()).unwrap(); + assert_marker(root.path(), 5); + let restored = store.read_all(session, segment).unwrap(); + assert_eq!(encode(&restored), new_bytes); + assert_eq!(fs::read(&new_path).unwrap(), new_bytes); + let once = fs::read(&path).unwrap(); + let old_once = fs::read(&old_path).unwrap(); + drop(store); + let reopened = WorkerSessionStore::new(root.path()).unwrap(); + assert_eq!(encode(&reopened.read_all(session, segment).unwrap()), once); + assert_eq!(fs::read(&path).unwrap(), once); + assert_eq!(fs::read(&old_path).unwrap(), old_once); + } +} + +#[test] +fn all_segments_preflight_keeps_old_bytes_on_decode_or_conversion_failure() { + let mut oversized: Value = serde_json::from_str(V4.lines().next().unwrap()).unwrap(); + oversized["entry"]["item"]["snapshot"] = + json!({"large": "x".repeat(crate::system_input::MAX_FEATURE_PAYLOAD_BYTES + 1)}); + let mut unknown: Value = serde_json::from_str(V4.lines().next().unwrap()).unwrap(); + unknown["entry"]["item"]["kind"] = json!("unsupported_future_feature"); + let mut malformed = unknown.clone(); + malformed["entry"]["item"] = json!({"kind":"journal_startup", "body":"missing snapshot"}); + let mut missing_metadata: Value = serde_json::from_str(V4.lines().last().unwrap()).unwrap(); + missing_metadata["entry"]["metadata"] + .as_object_mut() + .unwrap() + .remove("system_input"); + let mut missing_input = missing_metadata.clone(); + missing_input["entry"]["metadata"]["origin"] = json!({"kind":"legacy_unknown"}); + let mut invalid_policy: Value = serde_json::from_str(V4.lines().last().unwrap()).unwrap(); + invalid_policy["entry"]["metadata"]["system_input"]["usage"]["compaction"] = json!("unknown"); + let mut invalid_payload = invalid_policy.clone(); + invalid_payload["entry"]["metadata"]["system_input"]["usage"]["compaction"] = json!("retain"); + invalid_payload["entry"]["metadata"]["system_input"]["feature_payload"]["version"] = json!(0); + let mut flat_provided: Value = serde_json::from_str(V4.lines().nth(5).unwrap()).unwrap(); + flat_provided["history"][0]["metadata"]["origin"] = + json!({"kind":"provided", "producer":"owner"}); + for bad in [ + "{broken-json}\n".to_owned(), + format!("{}\n", oversized), + format!("{}\n", unknown), + format!("{}\n", malformed), + format!("{}\n", missing_metadata), + format!("{}\n", invalid_policy), + format!("{}\n", invalid_payload), + format!("{}\n", flat_provided), + format!("{}\n", missing_input), + ] { + let root = tempfile::tempdir().unwrap(); + let (_, segment) = write_worker(root.path(), 4, V4.as_bytes()); + let first = root + .path() + .join("segments") + .join(format!("{segment}.jsonl")); + let later = root + .path() + .join("segments") + .join(format!("{}.jsonl", SegmentId::from_u128(2))); + fs::write(&later, &bad).unwrap(); + let marker = fs::read(root.path().join("session.json")).unwrap(); + assert!(WorkerSessionStore::new(root.path()).is_err()); + assert_eq!(fs::read(&first).unwrap(), V4.as_bytes()); + assert_eq!(fs::read(&later).unwrap(), bad.as_bytes()); + assert_eq!(fs::read(root.path().join("session.json")).unwrap(), marker); + } +} + +#[test] +fn unsupported_schema_preserves_all_old_bytes() { + for version in [0, 6, 99] { + let root = tempfile::tempdir().unwrap(); + let (_, segment) = write_worker(root.path(), version, V4.as_bytes()); + let path = root + .path() + .join("segments") + .join(format!("{segment}.jsonl")); + let marker = fs::read(root.path().join("session.json")).unwrap(); + assert!(WorkerSessionStore::new(root.path()).is_err()); + assert_eq!(fs::read(&path).unwrap(), V4.as_bytes()); + assert_eq!(fs::read(root.path().join("session.json")).unwrap(), marker); + } +} + +#[test] +fn standalone_store_uses_same_v1_through_v4_migration_and_reopens() { + // None represents the existing unmarked annotated standalone layout (v3). + for version in [None, Some(1), Some(2), Some(3), Some(4)] { + let root = tempfile::tempdir().unwrap(); + let session = SessionId::now_v7(); + let segment = SegmentId::from_u128(1); + let dir = root.path().join(session.to_string()); + fs::create_dir_all(&dir).unwrap(); + let source = if version.is_some_and(|v| v <= 2) { + V1 + } else { + V4 + }; + let path = dir.join(format!("{segment}.jsonl")); + fs::write(&path, source).unwrap(); + if let Some(version) = version { + fs::write(dir.join("session-schema.json"), version.to_string()).unwrap(); + } + let store = FsStore::new(root.path()).unwrap(); + let entries = store.read_all(session, segment).unwrap(); + for entry in entries { + entry.validate_system_inputs().unwrap(); + } + assert_eq!( + fs::read_to_string(dir.join("session-schema.json")) + .unwrap() + .trim(), + "5" + ); + let once = fs::read(&path).unwrap(); + drop(store); + FsStore::new(root.path()).unwrap(); + assert_eq!(fs::read(&path).unwrap(), once); + } +} + +#[test] +fn compaction_origin_upgrade_is_retained_only_for_pre_v4_sources() { + for version in [3, 4] { + let root = tempfile::tempdir().unwrap(); + let creation = seed( + LoggedSessionHistoryOrigin::DerivedSummary, + true, + crate::LoggedRole::System, + "[Compacted context summary]\n\nold summary", + ); + let inherited = seed( + LoggedSessionHistoryOrigin::DerivedSummary, + false, + crate::LoggedRole::System, + "[Compacted context summary]\n\nold summary", + ); + let (session, segment) = + write_worker(root.path(), version, &encode(&[creation, inherited])); + let store = WorkerSessionStore::new(root.path()).unwrap(); + for record in store.read_all(session, segment).unwrap() { + let LogEntry::AnnotatedSegmentStart { history, .. } = record else { + panic!() + }; + let metadata = &history[0].metadata; + assert_eq!(metadata.entry_id.0, "summary"); + assert_eq!(metadata.derivation.as_ref().unwrap().sources[0].0, "source"); + assert_eq!( + metadata.origin, + if version == 3 { + LoggedSessionHistoryOrigin::CompactionSummary + } else { + LoggedSessionHistoryOrigin::DerivedSummary + } + ); + assert_eq!( + metadata.system_input.as_ref().unwrap().usage().compaction, + if version == 3 { + InputCompactionUse::OpeningSummary + } else { + InputCompactionUse::Retain + } + ); + } + } +} + +#[test] +fn standalone_preflight_and_unsupported_marker_leave_sources_unchanged() { + for version in [4, 99] { + let root = tempfile::tempdir().unwrap(); + let session = SessionId::now_v7(); + let dir = root.path().join(session.to_string()); + fs::create_dir_all(&dir).unwrap(); + let first = dir.join(format!("{}.jsonl", SegmentId::from_u128(1))); + let later = dir.join(format!("{}.jsonl", SegmentId::from_u128(2))); + fs::write(&first, V4).unwrap(); + fs::write(&later, "{broken-json}\n").unwrap(); + let marker = version.to_string(); + fs::write(dir.join("session-schema.json"), &marker).unwrap(); + assert!(FsStore::new(root.path()).is_err()); + assert_eq!(fs::read(&first).unwrap(), V4.as_bytes()); + assert_eq!(fs::read_to_string(&later).unwrap(), "{broken-json}\n"); + assert_eq!( + fs::read_to_string(dir.join("session-schema.json")).unwrap(), + marker + ); + } +} + +#[test] +fn current_v5_rejects_all_legacy_feature_kinds_without_rewriting_sources() { + for line in V4.lines().take(5) { + // Current serde, writable open and read-only restore share the same + // current wire contract. None may dispatch to the old Feature decoder. + assert!(serde_json::from_str::(line).is_err()); + let bytes = format!("{line}\n"); + let root = tempfile::tempdir().unwrap(); + let (session, segment) = write_worker(root.path(), 5, bytes.as_bytes()); + let path = root + .path() + .join("segments") + .join(format!("{segment}.jsonl")); + let marker = fs::read(root.path().join("session.json")).unwrap(); + let readonly = WorkerSessionStore::open_read_only(root.path()).unwrap(); + assert!(readonly.read_all_read_only(session, segment).is_err()); + assert!(WorkerSessionStore::new(root.path()).is_err()); + assert_eq!(fs::read(&path).unwrap(), bytes.as_bytes()); + assert_eq!(fs::read(root.path().join("session.json")).unwrap(), marker); + + let standalone = tempfile::tempdir().unwrap(); + let dir = standalone.path().join(session.to_string()); + fs::create_dir_all(&dir).unwrap(); + let path = dir.join(format!("{segment}.jsonl")); + fs::write(&path, &bytes).unwrap(); + fs::write(dir.join("session-schema.json"), "5").unwrap(); + assert!(FsStore::new(standalone.path()).is_err()); + assert_eq!(fs::read(&path).unwrap(), bytes.as_bytes()); + assert_eq!( + fs::read_to_string(dir.join("session-schema.json")).unwrap(), + "5" + ); + } +} + +#[test] +fn v1_v2_feature_wire_uses_the_same_typed_payloads_and_stable_legacy_identity() { + let segment = SegmentId::from_u128(42); + for version in [1, 2] { + for (index, line) in V4.lines().take(5).enumerate() { + let annotated: Value = serde_json::from_str(line).unwrap(); + let wire = json!({ + "kind": "system_item", "ts": annotated["ts"], + "item": annotated["entry"]["item"], + "extensions": [{"domain":"old-owner", "payload":{"kept":true}}], + }) + .to_string(); + let entry = decode_entry(version, &wire, SessionId::nil(), segment, index).unwrap(); + let retry = decode_entry(version, &wire, SessionId::nil(), segment, index).unwrap(); + assert_eq!(encode(&[entry.clone()]), encode(&[retry])); + let LogEntry::AnnotatedSystemItem { + ts, + entry, + extensions, + } = entry + else { + panic!() + }; + assert_eq!(ts, annotated["ts"].as_u64().unwrap()); + assert!(matches!(entry.item, SystemItem::Input { .. })); + assert_eq!( + entry.item.history_text(), + annotated["entry"]["item"]["body"].as_str().unwrap() + ); + assert_eq!( + entry.metadata.entry_id, + legacy_metadata(segment, index, 0).entry_id + ); + assert_eq!( + entry.metadata.origin, + LoggedSessionHistoryOrigin::LegacyUnknown + ); + assert!(entry.metadata.derivation.is_none()); + assert_eq!(extensions[0].domain, "old-owner"); + assert_eq!(extensions[0].payload, json!({"kept":true})); + let LogEntry::AnnotatedSystemItem { entry: v4, .. } = + decode_entry(4, line, SessionId::nil(), segment, index).unwrap() + else { + panic!() + }; + assert_eq!(entry.metadata.system_input, v4.metadata.system_input); + } + } +} + +#[test] +fn migration_rejects_malformed_typed_feature_fields_instead_of_retaining_unchecked_json() { + let mut items: Vec = V4 + .lines() + .take(5) + .map(|line| serde_json::from_str::(line).unwrap()["entry"]["item"].clone()) + .collect(); + items[0]["snapshot_sha256"] = json!(17); + items[1]["prompt_provenance"] = json!({"logical_name":"missing digests"}); + items[2]["behavior_md"] = json!(["not a string"]); + items[3]["name"] = json!({"not":"a name"}); + items[4]["result"]["status"] = json!("unsupported_status"); + for version in 1..=4 { + for item in &items { + let wire = if version <= 2 { + json!({"kind":"system_item", "ts":1, "item":item}) + } else { + json!({"kind":"annotated_system_item", "ts":1, "entry":{ + "item":item, "metadata":{ + "entry_id":"malformed-id", "origin":{"kind":"legacy_unknown"}, + }, + }}) + } + .to_string(); + assert!(decode_entry(version, &wire, SessionId::nil(), SegmentId::nil(), 0).is_err()); + } + } +} + +#[test] +fn frozen_behavior_counter_is_accepted_only_at_the_old_schema_boundary() { + for version in 1..=4 { + let item = json!({ + "kind":"subject_behavior_refresh", "body":"exact archived body", + "behavior_revision":7, + }); + let wire = if version <= 2 { + json!({"kind":"system_item", "ts":1, "item":item}) + } else { + json!({"kind":"annotated_system_item", "ts":1, "entry":{ + "item":item, "metadata":{ + "entry_id":"behavior-id", "origin":{"kind":"legacy_unknown"}, + }, + }}) + } + .to_string(); + assert!(serde_json::from_str::(&wire).is_err()); + let LogEntry::AnnotatedSystemItem { entry, .. } = + decode_entry(version, &wire, SessionId::nil(), SegmentId::nil(), 0).unwrap() + else { + panic!() + }; + assert_eq!(entry.item.history_text(), "exact archived body"); + let payload = entry + .metadata + .system_input + .as_ref() + .unwrap() + .feature_payload() + .unwrap(); + let typed: SubjectBehaviorPayload = payload.decode("subject:behavior", 1).unwrap(); + assert_eq!(typed.behavior_md, ""); + assert!(typed.prompt_provenance.is_none()); + assert_eq!( + serde_json::to_value(typed).unwrap(), + json!({ + "behavior_md":"", "prompt_provenance":null, + }) + ); + } +} + +#[test] +fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() { + let mut missing: Value = serde_json::from_str(V4.lines().last().unwrap()).unwrap(); + missing["entry"]["metadata"] + .as_object_mut() + .unwrap() + .remove("system_input"); + let bytes = format!("{}\n", missing); + let root = tempfile::tempdir().unwrap(); + let (_, segment) = write_worker(root.path(), 5, bytes.as_bytes()); + assert!(WorkerSessionStore::new(root.path()).is_err()); + assert_eq!( + fs::read( + root.path() + .join("segments") + .join(format!("{segment}.jsonl")) + ) + .unwrap(), + bytes.as_bytes() + ); + assert_marker(root.path(), 5); +} diff --git a/crates/session-store/src/public_material.rs b/crates/session-store/src/public_material.rs index f27795928..2c7d66293 100644 --- a/crates/session-store/src/public_material.rs +++ b/crates/session-store/src/public_material.rs @@ -157,8 +157,8 @@ pub enum PublicMaterialUse { /// A fixed material catalog. Opening a new catalog observes newly appended /// records; reading a body through this catalog verifies its saved digest and /// byte length again. Missing/corrected/deleted sources are explicit failures. -/// Explicit evidence permission applies to model text, not the private payload; -/// Web display and observation instead use the producer's public projection. +/// Evidence eligibility never publishes private model text or Feature payload: +/// System inputs also require an explicit public projection for every public use. pub struct SessionPublicMaterialReader { session_id: SessionId, active_segment_id: SegmentId, @@ -828,7 +828,7 @@ enum CanonicalBody<'a> { Message { origin: &'a LoggedSessionHistoryOrigin, role: LoggedRole, - content: &'a [crate::LoggedContentPart], + content: std::borrow::Cow<'a, [crate::LoggedContentPart]>, }, ToolCall { origin: &'a LoggedSessionHistoryOrigin, @@ -904,27 +904,22 @@ fn canonical_body( if allowed == crate::InputEvidenceUse::Exclude { return None; } + let public = input.public_item()?; if let LoggedItem::Message { role: LoggedRole::System, - content, + .. } = &entry.item { return Some(CanonicalBody::Message { origin: &entry.metadata.origin, role: LoggedRole::System, - content, + content: std::borrow::Cow::Owned(vec![crate::LoggedContentPart::Text { + text: public.body.clone(), + }]), }); } return None; } - if matches!( - entry.metadata.origin, - LoggedSessionHistoryOrigin::DerivedSummary - | LoggedSessionHistoryOrigin::CompactionSummary - | LoggedSessionHistoryOrigin::JournalContext - ) { - return None; - } match &entry.item { LoggedItem::Message { role: LoggedRole::System, @@ -935,7 +930,7 @@ fn canonical_body( LoggedItem::Message { role, content } => Some(CanonicalBody::Message { origin: &entry.metadata.origin, role: *role, - content, + content: std::borrow::Cow::Borrowed(content), }), LoggedItem::ToolCall { call_id, diff --git a/crates/session-store/src/public_material/tests.rs b/crates/session-store/src/public_material/tests.rs index 0376af0bc..5bf7ce7ea 100644 --- a/crates/session-store/src/public_material/tests.rs +++ b/crates/session-store/src/public_material/tests.rs @@ -25,6 +25,27 @@ fn message(id: &str, role: LoggedRole, text: &str) -> LoggedHistoryEntry { }, ) } +fn system_context( + id: &str, + text: &str, + compaction: crate::InputCompactionUse, +) -> LoggedHistoryEntry { + let mut entry = message(id, LoggedRole::System, text); + entry.metadata.system_input = Some( + crate::SystemInputMetadata::new( + crate::SystemInputUsePolicy { + recording_material: crate::InputEvidenceUse::Exclude, + memory_evidence: crate::InputEvidenceUse::Exclude, + observation: crate::InputObservationUse::Hidden, + compaction, + }, + None, + None, + ) + .unwrap(), + ); + entry +} fn seed(session: SessionId, history: Vec) -> LogEntry { LogEntry::AnnotatedSegmentStart { ts: 1, @@ -348,7 +369,11 @@ fn yield_resume_cancel_keep_original_boundary_identity_across_compaction() { .descriptors() .cloned() .collect(); - let mut summary = message("summary", LoggedRole::User, "must not replace originals"); + let mut summary = system_context( + "summary", + "must not replace originals", + crate::InputCompactionUse::OpeningSummary, + ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { sources: vec![ @@ -544,14 +569,18 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi let store = FsStore::new(temp.path()).unwrap(); let session = crate::new_session_id(); let segment = crate::new_segment_id(); - let mut derived = message("derived", LoggedRole::User, "summary"); + let mut derived = system_context("derived", "summary", crate::InputCompactionUse::Retain); derived.metadata.origin = LoggedSessionHistoryOrigin::DerivedSummary; - let mut journal = message("journal", LoggedRole::User, "frozen Journal reference"); + let mut journal = system_context( + "journal", + "frozen Journal reference", + crate::InputCompactionUse::ReferenceInput, + ); journal.metadata.origin = LoggedSessionHistoryOrigin::JournalContext; - let mut summary = message( + let mut summary = system_context( "compaction-summary", - LoggedRole::User, "complementary summary", + crate::InputCompactionUse::OpeningSummary, ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; let reasoning = entry( @@ -580,19 +609,23 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi LogEntry::AnnotatedSystemItem { ts: 1, entry: crate::LoggedSystemHistoryEntry { - item: crate::SystemItem::JournalStartup { + item: crate::SystemItem::Input { body: "previous Journal must not become new recording material".into(), - snapshot: serde_json::json!({"journals": []}), - snapshot_sha256: "retained-digest".into(), - prompt_provenance: crate::PromptRenderProvenance { - workspace_id: None, - source_digest: "source".into(), - projection_digest: "projection".into(), - logical_name: "internal.subjektiv_journal_startup".into(), - }, }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + system_input: Some( + crate::SystemInputMetadata::new( + crate::SystemInputUsePolicy { + recording_material: crate::InputEvidenceUse::Exclude, + memory_evidence: crate::InputEvidenceUse::Exclude, + observation: crate::InputObservationUse::Hidden, + compaction: crate::InputCompactionUse::ReferenceInput, + }, + None, + None, + ) + .unwrap(), + ), entry_id: LoggedSessionHistoryEntryId("startup-journal".into()), origin: LoggedSessionHistoryOrigin::DerivedSummary, derivation: None, @@ -779,6 +812,19 @@ fn schema_v3_compaction_migrates_persistently_and_strict_reader_keeps_original_i panic!() }; summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; + summary.metadata.system_input = Some( + crate::SystemInputMetadata::new( + crate::SystemInputUsePolicy { + recording_material: crate::InputEvidenceUse::Exclude, + memory_evidence: crate::InputEvidenceUse::Exclude, + observation: crate::InputObservationUse::Hidden, + compaction: crate::InputCompactionUse::OpeningSummary, + }, + None, + None, + ) + .unwrap(), + ); assert_eq!(history, &[summary]); // Run-transition ids synthesized from byte offsets must be frozen before // the changed summary record shifts every subsequent record offset. @@ -827,7 +873,11 @@ fn midturn_compaction_uses_exact_source_ids_and_preserves_discarded_originals() ], ) .unwrap(); - let mut summary = message("summary", LoggedRole::User, "never use as original"); + let mut summary = system_context( + "summary", + "never use as original", + crate::InputCompactionUse::OpeningSummary, + ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { sources: vec![ @@ -1032,7 +1082,11 @@ fn zero_turn_fork_from_seed_and_consecutive_compactions_preserve_original_identi for index in 0..3 { let compact = crate::new_segment_id(); let summary_id = format!("summary-{index}"); - let mut summary = message(&summary_id, LoggedRole::User, "not original"); + let mut summary = system_context( + &summary_id, + "not original", + crate::InputCompactionUse::OpeningSummary, + ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { sources: source_ids @@ -1180,7 +1234,11 @@ fn correction_preserves_original_identity_across_compaction_without_summary_fall ], ) .unwrap(); - let mut summary = message("summary", LoggedRole::User, "not original"); + let mut summary = system_context( + "summary", + "not original", + crate::InputCompactionUse::OpeningSummary, + ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { sources: vec![LoggedSessionHistoryEntryId("result".into())], @@ -1228,7 +1286,11 @@ fn compaction_without_exact_derivation_or_with_invalid_turn_is_rejected() { ), ] { let segment = crate::new_segment_id(); - let mut summary = message("summary", LoggedRole::User, "not original"); + let mut summary = system_context( + "summary", + "not original", + crate::InputCompactionUse::OpeningSummary, + ); summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; summary.metadata.derivation = derivation; let mut start = seed(session, vec![summary]); @@ -1257,54 +1319,69 @@ fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() let session = crate::new_session_id(); let segment = crate::new_segment_id(); let fork = crate::new_segment_id(); - let make = |producer: &str, recording_material, memory_evidence, body: &str| { - LoggedSystemHistoryEntry::from_input( - LoggedSessionHistoryOrigin::Provided { - producer: producer.into(), - }, - None, - SystemInputEnvelope::new( - body, - SystemInputMetadata::new( - SystemInputUsePolicy { - recording_material, - memory_evidence, - observation: InputObservationUse::Hidden, - compaction: InputCompactionUse::Retain, - }, - Some( - FeatureInputPayload::encode( - "unknown:future", - 99, - &serde_json::json!({"private": "not-public"}), - ) - .unwrap(), - ), - None, + let make = + |producer: &str, recording_material, memory_evidence, body: &str, public: bool| { + LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: producer.into(), + }, + None, + SystemInputEnvelope::new( + format!("private model text: {body}"), + SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material, + memory_evidence, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + Some( + FeatureInputPayload::encode( + "unknown:future", + 99, + &serde_json::json!({"private": "not-public"}), + ) + .unwrap(), + ), + public.then(|| crate::PublicSystemItem { + kind: "context".into(), + body: body.into(), + presentation: crate::PublicSystemPresentation::Context, + }), + ) + .unwrap(), ) .unwrap(), ) - .unwrap(), - ) - }; + }; let inputs = [ make( "same-owner", InputEvidenceUse::Include, InputEvidenceUse::Exclude, "do not extract this; original material", + true, ), make( "same-owner", InputEvidenceUse::Exclude, InputEvidenceUse::Include, "memory-only original", + true, ), make( "renamed-owner", InputEvidenceUse::Exclude, InputEvidenceUse::Exclude, "reference without warning text", + true, + ), + make( + "private-only", + InputEvidenceUse::Include, + InputEvidenceUse::Include, + "must not publish without projection", + false, ), ]; let root = if worker_layout { @@ -1403,6 +1480,7 @@ fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() assert_eq!(value["role"], "system"); assert_eq!(value["content"][0]["text"], expected_text); assert!(!body.contains("not-public")); + assert!(!body.contains("private model text")); assert!(!body.contains("unknown:future")); } } diff --git a/crates/session-store/src/system_item.rs b/crates/session-store/src/system_item.rs index 6d3d6ce43..c7bc6a3b9 100644 --- a/crates/session-store/src/system_item.rs +++ b/crates/session-store/src/system_item.rs @@ -86,23 +86,16 @@ pub struct PromptRenderProvenance { /// `notify_wrapper` template) re-shape the live rendering on a later /// resume. /// -/// New variants get added here as fresh injection kinds come online -/// (e.g. `TaskReminder`). The `kind` JSON tag is the snake_case form of -/// the variant name. +/// Feature-owned inputs use `Input` plus explicit annotated metadata; adding a +/// Feature must not extend this enum or ordinary replay. Only core events have +/// dedicated variants. Pre-v5 Feature kinds exist exclusively in the migration +/// decoder, not in this current wire contract. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum SystemItem { /// Generic System body; private metadata and public presentation are supplied /// independently by the producer in the same annotated history record. Input { body: String }, - /// One-time derived Journal reference context. The bounded Host snapshot is - /// retained as JSON independently of its rendered body and prompt projection. - JournalStartup { - body: String, - snapshot: serde_json::Value, - snapshot_sha256: String, - prompt_provenance: PromptRenderProvenance, - }, /// Free-form notification sent in by an external caller via /// `Method::Notify`. `message` is the raw caller-supplied text; /// `body` is the wrapped LLM-context form (Worker renders it via @@ -125,14 +118,6 @@ pub enum SystemItem { prompt_provenance: Option, }, - /// Result of an explicitly selected chat Feature invocation. The typed - /// result is retained for clients; `body` is the exact bounded context made - /// visible to the Worker model. - FeatureInvocationResult { - result: protocol::FeatureInvocationResult, - body: String, - }, - /// `@` file reference resolution. `body` is the rendered /// LLM-context text (`[File: ]\n…` for regular files, /// `[Dir: ]\n…` for directory listings, possibly with a @@ -140,10 +125,6 @@ pub enum SystemItem { /// byte-identical to what was sent. FileAttachment { path: String, body: String }, - /// Explicit Agent Skill activation. `body` is the exact LLM-facing - /// Skill text committed before any subsequent LLM request can use it. - SkillActivation { name: String, body: String }, - /// Historical persisted Knowledge reference resolution. Knowledge is no /// longer active, so restored sessions ignore this item instead of replaying /// archived record text into model context. @@ -167,29 +148,6 @@ pub enum SystemItem { prompt_provenance: Option, }, - /// Restore-boundary refresh of the current resident Memory projection. - /// The exact rendered body is retained so replay is byte-identical; a - /// refresh with no ready summary explicitly supersedes older resident - /// context without deleting or rewriting history. - ResidentSummaryRefresh { - body: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - prompt_provenance: Option, - }, - - /// A Host-fetched change to user-managed Subject behavior. Each changed - /// body is appended rather than rewriting prior prompt/session content. - SubjectBehaviorRefresh { - body: String, - /// Exact user-managed behavior acknowledged after durable commit. Old - /// logs omit this value; an empty value is not an acknowledgement of a - /// nonempty behavior and does not replace their stored rendered body. - #[serde(default)] - behavior_md: String, - #[serde(default, skip_serializing_if = "Option::is_none")] - prompt_provenance: Option, - }, - /// Synthetic note inserted after an interrupted turn before the next /// user input. `body` is the exact LLM-context text explaining that the /// previous turn was cut short. @@ -233,13 +191,9 @@ impl SystemItem { Self::Input { .. } | Self::LegacyKnowledgeIgnored { .. } | Self::LegacyIgnored { .. } => PublicSystemPresentation::Ignored, - Self::JournalStartup { .. } - | Self::ResidentSummaryRefresh { .. } - | Self::SubjectBehaviorRefresh { .. } - | Self::SkillActivation { .. } - | Self::FeatureInvocationResult { .. } - | Self::FileAttachment { .. } - | Self::Interrupt { .. } => PublicSystemPresentation::Context, + Self::FileAttachment { .. } | Self::Interrupt { .. } => { + PublicSystemPresentation::Context + } }; PublicSystemItem { kind: self.kind_label().to_owned(), @@ -257,19 +211,14 @@ impl SystemItem { pub fn history_text(&self) -> String { match self { SystemItem::Input { body } => body.clone(), - SystemItem::JournalStartup { body, .. } => body.clone(), SystemItem::Notification { body, .. } => body.clone(), SystemItem::WorkerEvent { body, .. } => body.clone(), - SystemItem::FeatureInvocationResult { body, .. } => body.clone(), SystemItem::FileAttachment { body, .. } => body.clone(), - SystemItem::SkillActivation { body, .. } => body.clone(), SystemItem::LegacyKnowledgeIgnored { .. } => String::new(), SystemItem::LegacyIgnored { slug } => { format!("Ignored legacy procedure item: /{slug}") } SystemItem::TaskReminder { body, .. } => body.clone(), - SystemItem::ResidentSummaryRefresh { body, .. } => body.clone(), - SystemItem::SubjectBehaviorRefresh { body, .. } => body.clone(), SystemItem::Interrupt { body, .. } => body.clone(), } } @@ -285,17 +234,12 @@ impl SystemItem { pub fn kind_label(&self) -> &'static str { match self { SystemItem::Input { .. } => "input", - SystemItem::JournalStartup { .. } => "journal_startup", SystemItem::Notification { .. } => "notification", SystemItem::WorkerEvent { .. } => "worker_event", - SystemItem::FeatureInvocationResult { .. } => "feature_invocation_result", SystemItem::FileAttachment { .. } => "file_attachment", - SystemItem::SkillActivation { .. } => "skill_activation", SystemItem::LegacyKnowledgeIgnored { .. } => "legacy_knowledge_ignored", SystemItem::LegacyIgnored { .. } => "legacy_ignored", SystemItem::TaskReminder { .. } => "task_reminder", - SystemItem::ResidentSummaryRefresh { .. } => "resident_summary_refresh", - SystemItem::SubjectBehaviorRefresh { .. } => "subject_behavior_refresh", SystemItem::Interrupt { .. } => "interrupt", } } @@ -353,18 +297,20 @@ mod tests { } #[test] - fn subject_behavior_refresh_round_trips_with_exact_body() { - let item = SystemItem::SubjectBehaviorRefresh { - body: "current Subject context".to_string(), - behavior_md: "exact behavior".to_string(), - prompt_provenance: None, - }; - let raw = serde_json::to_string(&item).unwrap(); - let parsed: SystemItem = serde_json::from_str(&raw).unwrap(); - assert_eq!(parsed.history_text(), "current Subject context"); - assert!( - matches!(parsed, SystemItem::SubjectBehaviorRefresh { behavior_md, .. } if behavior_md == "exact behavior") - ); + fn current_system_item_rejects_frozen_feature_kinds() { + // Even well-formed old payloads cannot enter normal v5 decode/replay. + for line in include_str!("legacy_session_log/tests/fixtures/v4-system-inputs.jsonl") + .lines() + .take(5) + { + let record: serde_json::Value = serde_json::from_str(line).unwrap(); + assert!(serde_json::from_value::(record["entry"]["item"].clone()).is_err()); + assert!( + serde_json::from_value::(record["entry"].clone()) + .is_err() + ); + assert!(serde_json::from_str::(line).is_err()); + } } #[test] @@ -392,21 +338,6 @@ mod tests { assert!(serde_json::from_value::(unknown).is_err()); } - #[test] - fn frozen_subject_behavior_log_reads_old_counter_without_republishing_it() { - let old = r#"{"kind":"subject_behavior_refresh","body":"exact archived body","behavior_revision":7}"#; - let item: SystemItem = serde_json::from_str(old).unwrap(); - assert_eq!(item.history_text(), "exact archived body"); - assert_eq!( - serde_json::to_value(item).unwrap(), - serde_json::json!({ - "kind": "subject_behavior_refresh", - "body": "exact archived body", - "behavior_md": "" - }) - ); - } - #[test] fn notification_history_text_returns_stored_body() { let item = SystemItem::Notification { @@ -500,29 +431,6 @@ mod tests { } } - #[test] - fn invocation_result_retains_typed_outcome_and_exact_context() { - let result = protocol::FeatureInvocationResult { - invocation_id: "stable-invoke".into(), - identity: protocol::FeatureInvocationIdentity("builtin:test/prepare".into()), - status: protocol::FeatureInvocationStatus::OutcomeUnknown, - message: "operation may have committed".into(), - context: None, - }; - let body = "Outcome unknown; do not retry automatically"; - let item = SystemItem::FeatureInvocationResult { - result: result.clone(), - body: body.into(), - }; - let restored: SystemItem = - serde_json::from_str(&serde_json::to_string(&item).unwrap()).unwrap(); - assert_eq!(restored.history_text(), body); - assert_eq!(restored.kind_label(), "feature_invocation_result"); - assert!( - matches!(restored, SystemItem::FeatureInvocationResult { result: saved, .. } if saved == result) - ); - } - #[test] fn round_trip_worker_event() { let item = SystemItem::WorkerEvent { diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index bb6ffe34f..17a2691d1 100644 --- a/crates/session-store/src/worker_session_store.rs +++ b/crates/session-store/src/worker_session_store.rs @@ -36,7 +36,8 @@ use std::path::{Path, PathBuf}; use std::sync::{Arc, Mutex}; use std::time::SystemTime; -const SESSION_SCHEMA_VERSION: u32 = 4; +const SESSION_SCHEMA_VERSION: u32 = 5; +const COMPACTION_SESSION_SCHEMA_VERSION: u32 = 4; const ANNOTATED_SESSION_SCHEMA_VERSION: u32 = 3; const PREVIOUS_SESSION_SCHEMA_VERSION: u32 = 2; const LEGACY_SESSION_SCHEMA_VERSION: u32 = 1; @@ -204,7 +205,15 @@ impl WorkerSessionStore { /// Open the Session store rooted at `/session`. pub fn new(root: impl Into) -> Result { - let root = root.into(); + Self::new_with_migration_writer(root.into(), atomic_write_bytes) + } + + // Keep fault injection at the storage publication boundary, without global + // flags or a public alternate restore API. Production always uses atomic writes. + fn new_with_migration_writer( + root: PathBuf, + write_segment: impl FnMut(&Path, &[u8]) -> Result<(), StoreError>, + ) -> Result { fs::create_dir_all(root.join(SEGMENTS_DIR))?; let session_id = match fs::read(root.join(SESSION_FILE)) { Ok(bytes) => { @@ -215,11 +224,13 @@ impl WorkerSessionStore { } LEGACY_SESSION_SCHEMA_VERSION | PREVIOUS_SESSION_SCHEMA_VERSION - | ANNOTATED_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v4( + | ANNOTATED_SESSION_SCHEMA_VERSION + | COMPACTION_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_v5( segment_log_paths(&root)?, manifest.session_id, manifest.schema_version, + write_segment, )?; manifest.schema_version = SESSION_SCHEMA_VERSION; atomic_write_json(&root.join(SESSION_FILE), &manifest)?; @@ -1162,10 +1173,11 @@ fn segment_log_paths_in_directory( Ok(paths) } -fn migrate_segment_logs_to_v4( +fn migrate_segment_logs_to_v5( paths: Vec<(SegmentId, PathBuf)>, session_id: SessionId, source_schema_version: u32, + mut write_segment: impl FnMut(&Path, &[u8]) -> Result<(), StoreError>, ) -> Result<(), StoreError> { struct MigrationPlan { path: PathBuf, @@ -1196,12 +1208,19 @@ fn migrate_segment_logs_to_v4( } // Resolve trusted creation boundaries first, then update inherited copies by // logical identity too (fork/rewind seeds retain the same entry id). - let summaries = crate::legacy_session_log::compaction_summary_ids( - plans.iter().flat_map(|plan| plan.entries.iter()), - ); + let summaries = if source_schema_version < COMPACTION_SESSION_SCHEMA_VERSION { + crate::legacy_session_log::compaction_summary_ids( + plans.iter().flat_map(|plan| plan.entries.iter()), + ) + } else { + // v4 already made the explicit origin migration. Preserve its origins. + Default::default() + }; for plan in &mut plans { crate::legacy_session_log::migrate_compaction_origins(&mut plan.entries, &summaries); - for entry in std::mem::take(&mut plan.entries) { + for mut entry in std::mem::take(&mut plan.entries) { + crate::legacy_session_log::migrate_system_inputs(&mut entry)?; + entry.validate_system_inputs()?; serde_json::to_writer(&mut plan.output, &entry)?; plan.output.push(b'\n'); } @@ -1223,7 +1242,7 @@ fn migrate_segment_logs_to_v4( } for plan in plans { - atomic_write_bytes(&plan.path, &plan.output)?; + write_segment(&plan.path, &plan.output)?; } Ok(()) } @@ -1240,9 +1259,19 @@ pub(crate) fn migrate_fs_session(root: &Path, session_id: SessionId) -> Result<( Err(error) => return Err(error.into()), }; match version { - SESSION_SCHEMA_VERSION => Ok(()), - ANNOTATED_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v4(segment_log_paths_in_directory(root)?, session_id, version)?; + SESSION_SCHEMA_VERSION => { + validate_canonical_segment_paths(segment_log_paths_in_directory(root)?) + } + LEGACY_SESSION_SCHEMA_VERSION + | PREVIOUS_SESSION_SCHEMA_VERSION + | ANNOTATED_SESSION_SCHEMA_VERSION + | COMPACTION_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_v5( + segment_log_paths_in_directory(root)?, + session_id, + version, + atomic_write_bytes, + )?; atomic_write_json(&marker, &SESSION_SCHEMA_VERSION) } _ => Err(StoreError::Corrupt { @@ -1253,8 +1282,15 @@ pub(crate) fn migrate_fs_session(root: &Path, session_id: SessionId) -> Result<( } fn validate_canonical_segment_logs(root: &Path) -> Result<(), StoreError> { - for (_, path) in segment_log_paths(root)? { - let _: Vec = parse_jsonl(&fs::read(&path)?)?; + validate_canonical_segment_paths(segment_log_paths(root)?) +} + +fn validate_canonical_segment_paths(paths: Vec<(SegmentId, PathBuf)>) -> Result<(), StoreError> { + for (_, path) in paths { + let entries: Vec = parse_jsonl(&fs::read(&path)?)?; + for entry in entries { + entry.validate_system_inputs()?; + } } Ok(()) } @@ -2022,6 +2058,146 @@ mod tests { ); } + #[test] + fn v5_migration_retries_after_second_file_publication_fails() { + fn identity_and_body(entries: &[LogEntry]) -> Vec { + entries + .iter() + .flat_map(|record| match record { + LogEntry::AnnotatedSystemItem { entry, .. } => vec![entry.to_history_entry()], + LogEntry::AnnotatedUserInput { history, .. } => history.clone(), + _ => panic!("unexpected record in fixed migration fixture"), + }) + .map(|entry| { + serde_json::json!({ + "entry_id": entry.metadata.entry_id, + "origin": entry.metadata.origin, + "derivation": entry.metadata.derivation, + "item": entry.item, + }) + }) + .collect() + } + + const SOURCE: &str = + include_str!("legacy_session_log/tests/fixtures/v4-system-inputs.jsonl"); + let root = tempfile::tempdir().unwrap(); + let session_id = new_session_id(); + let first_id = SegmentId::from_u128(1); + let second_id = SegmentId::from_u128(2); + fs::create_dir_all(root.path().join(SEGMENTS_DIR)).unwrap(); + let marker_path = root.path().join(SESSION_FILE); + atomic_write_json( + &marker_path, + &SessionManifest { + schema_version: COMPACTION_SESSION_SCHEMA_VERSION, + session_id, + }, + ) + .unwrap(); + let old_marker = fs::read(&marker_path).unwrap(); + let first_path = root + .path() + .join(SEGMENTS_DIR) + .join(format!("{first_id}.jsonl")); + let second_path = root + .path() + .join(SEGMENTS_DIR) + .join(format!("{second_id}.jsonl")); + let second_source = SOURCE.replace("-id", "-second-id"); + fs::write(&first_path, SOURCE).unwrap(); + fs::write(&second_path, &second_source).unwrap(); + let old_first = fs::read(&first_path).unwrap(); + let old_second = fs::read(&second_path).unwrap(); + let first_identity = + identity_and_body(&parse_legacy_jsonl(4, session_id, first_id, &old_first).unwrap()); + let second_identity = + identity_and_body(&parse_legacy_jsonl(4, session_id, second_id, &old_second).unwrap()); + + let mut writes = 0; + let mut first_published = None; + let failure = WorkerSessionStore::new_with_migration_writer( + root.path().to_path_buf(), + |path, bytes| { + writes += 1; + match writes { + 1 => { + assert_eq!(path, first_path.as_path()); + // Both files are still old when the first publication starts. + assert_eq!(fs::read(&first_path).unwrap(), old_first); + assert_eq!(fs::read(&second_path).unwrap(), old_second); + atomic_write_bytes(path, bytes)?; + first_published = Some(bytes.to_vec()); + Ok(()) + } + 2 => { + assert_eq!(path, second_path.as_path()); + assert_eq!( + fs::read(&first_path).unwrap(), + *first_published.as_ref().unwrap() + ); + assert_eq!(fs::read(&second_path).unwrap(), old_second); + // Fail only this atomic-write boundary, after complete + // preflight and a real successful first-file publication. + Err(std::io::Error::other("injected second migration write failure").into()) + } + _ => panic!("publication continued after its write failure"), + } + }, + ); + let error = match failure { + Ok(_) => panic!("the injected publication failure must reject store open"), + Err(error) => error, + }; + assert!( + error + .to_string() + .contains("injected second migration write failure") + ); + assert_eq!(writes, 2); + let first_published = first_published.unwrap(); + assert_ne!(first_published, old_first); + assert_eq!(fs::read(&first_path).unwrap(), first_published); + assert_eq!(fs::read(&second_path).unwrap(), old_second); + assert_eq!(fs::read(&marker_path).unwrap(), old_marker); + let updated_first: Vec = parse_jsonl(&first_published).unwrap(); + for entry in &updated_first { + entry.validate_system_inputs().unwrap(); + } + assert!(matches!( + &updated_first[0], + LogEntry::AnnotatedSystemItem { + entry: crate::LoggedSystemHistoryEntry { + item: crate::SystemItem::Input { .. }, + .. + }, + .. + } + )); + assert_eq!(identity_and_body(&updated_first), first_identity); + + // Retry via the public production constructor, without any injection or + // hand-built mixed fixture. The old marker drives the same migration. + let store = WorkerSessionStore::new(root.path()).unwrap(); + let first = store.read_all(session_id, first_id).unwrap(); + let second = store.read_all(session_id, second_id).unwrap(); + assert_eq!(identity_and_body(&first), first_identity); + assert_eq!(identity_and_body(&second), second_identity); + for entry in first.iter().chain(second.iter()) { + entry.validate_system_inputs().unwrap(); + } + assert_eq!(fs::read(&first_path).unwrap(), first_published); + let second_published = fs::read(&second_path).unwrap(); + assert_ne!(second_published, old_second); + let manifest: SessionManifest = + serde_json::from_slice(&fs::read(&marker_path).unwrap()).unwrap(); + assert_eq!(manifest.schema_version, SESSION_SCHEMA_VERSION); + drop(store); + WorkerSessionStore::new(root.path()).unwrap(); + assert_eq!(fs::read(&first_path).unwrap(), first_published); + assert_eq!(fs::read(&second_path).unwrap(), second_published); + } + #[test] fn current_jsonl_requires_annotations_across_append_rewrite_and_reopen() { let root = tempfile::tempdir().unwrap(); diff --git a/crates/standalone/src/subjektiv/journal/tests/handoff.rs b/crates/standalone/src/subjektiv/journal/tests/handoff.rs index 4b571a313..03e5b6ebf 100644 --- a/crates/standalone/src/subjektiv/journal/tests/handoff.rs +++ b/crates/standalone/src/subjektiv/journal/tests/handoff.rs @@ -79,7 +79,7 @@ async fn recorded_idle_body_handoff_preserves_open_after_source_deletion_and_see let sessions = session_store::FsStore::new_with_retention(workers.sessions_dir(second.worker_id()), crate::store::session_retention(&state)).unwrap(); let pointer = session_store::FsWorkerStore::new(workers.worker_metadata_dir(second.worker_id())).unwrap().read_by_name(&second.record().storage_key).unwrap().unwrap().active.unwrap(); let entries = sessions.read_all(session, pointer.segment_id.unwrap()).unwrap(); - let journal_items = entries.iter().filter(|entry| matches!(entry, session_store::LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, session_store::SystemItem::JournalStartup { .. }))).count(); + let journal_items = entries.iter().filter(|entry| matches!(entry, session_store::LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).is_some_and(|payload| payload.schema() == "journal:startup"))).count(); assert_eq!(journal_items, 1); assert_eq!(catalog.store.subject(&subject.id).unwrap().unwrap().behavior_md, "BEHAVIOR_ORIGINAL_UNCHANGED"); second.shutdown().await.unwrap(); diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index d8be6da1a..f9d7f240f 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1656,10 +1656,10 @@ where if ordinary_subjektiv_features_enabled && lifecycle_enabled && let Some(host) = subjektiv_host.as_ref() - && host.host.journal_startup_granted() + && (host.host.journal_startup_granted() || host.host.journal_compaction_granted()) { - worker.install_initial_context_source(Arc::new( - crate::feature::builtin::journal::JournalStartupContextSource::new( + worker.install_session_context_source(Arc::new( + crate::feature::builtin::journal::JournalContextSource::new( host.clone(), worker.prompts(), ), diff --git a/crates/worker/src/feature/builtin/journal.rs b/crates/worker/src/feature/builtin/journal.rs index 990cd347a..15d2bb2b4 100644 --- a/crates/worker/src/feature/builtin/journal.rs +++ b/crates/worker/src/feature/builtin/journal.rs @@ -23,7 +23,7 @@ use crate::hook::{ use crate::subjektiv::{SubjektivHostConnection, SubjektivHostError}; use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; use async_trait::async_trait; -use serde::Deserialize; +use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; use std::sync::Arc; use subjektiv::journal::transport::*; @@ -32,12 +32,73 @@ use subjektiv::journal::{ MAX_JOURNAL_RESPONSE_BYTES, }; +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct JournalStartupInput { + pub snapshot: subjektiv::journal::startup::JournalStartupSnapshot, + pub snapshot_sha256: String, + pub prompt_provenance: session_store::PromptRenderProvenance, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +enum JournalReadStatus { + Available, + Absent, + Disabled, + Unavailable, +} +impl JournalReadStatus { + fn as_str(self) -> &'static str { + match self { + Self::Available => "available", + Self::Absent => "absent", + Self::Disabled => "disabled", + Self::Unavailable => "unavailable", + } + } +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct JournalCompactionInput { + status: JournalReadStatus, + snapshot: Option, + snapshot_sha256: Option, + prompt_provenance: session_store::PromptRenderProvenance, +} + +fn journal_input( + body: String, + payload: session_store::FeatureInputPayload, + kind: &str, +) -> Result { + use session_store::*; + let public = PublicSystemItem { + kind: kind.into(), + body: body.clone(), + presentation: PublicSystemPresentation::Context, + }; + let metadata = SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::ReferenceInput, + }, + Some(payload), + Some(public), + ) + .map_err(|e| e.to_string())?; + SystemInputEnvelope::new(body, metadata).map_err(|e| e.to_string()) +} + /// Subjektiv owns startup projection; Core only commits its typed initial items. -pub(crate) struct JournalStartupContextSource { +pub(crate) struct JournalContextSource { host: SubjektivHostConnection, prompts: Arc>, } -impl JournalStartupContextSource { +impl JournalContextSource { pub(crate) fn new( host: SubjektivHostConnection, prompts: Arc>, @@ -46,9 +107,12 @@ impl JournalStartupContextSource { } } #[async_trait] -impl crate::worker::InitialContextSource for JournalStartupContextSource { +impl crate::worker::SessionContextSource for JournalContextSource { async fn load(&self) -> Result, String> { use sha2::{Digest, Sha256}; + if !self.host.host.journal_startup_granted() { + return Ok(Vec::new()); + } let snapshot = self .host .journal_startup_context() @@ -58,8 +122,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource { 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())?; - let snapshot_json = serde_json::to_string(&snapshot).map_err(|e| e.to_string())?; + let snapshot_value = serde_json::to_value(&snapshot).map_err(|e| e.to_string())?; + let snapshot_json = serde_json::to_string(&snapshot_value).map_err(|e| e.to_string())?; if snapshot_json.len() > subjektiv::journal::startup::STARTUP_MAX_BYTES { return Err("Host Journal startup snapshot exceeds independent byte budget".into()); } @@ -74,9 +138,10 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource { return Err("Journal startup prompt exceeds rendered byte budget".into()); } let projection = prompts.projection(); - Ok(vec![session_store::LoggedSystemHistoryEntry { - item: session_store::SystemItem::JournalStartup { - body, + let payload = session_store::FeatureInputPayload::encode( + "journal:startup", + 1, + &JournalStartupInput { snapshot, snapshot_sha256: Sha256::digest(snapshot_json.as_bytes()) .iter() @@ -89,11 +154,38 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource { logical_name: "internal.subjektiv_journal_startup".into(), }, }, - metadata: crate::session_history::metadata( - session_store::LoggedSessionHistoryOrigin::JournalContext, - None, - ), - }]) + ) + .map_err(|e| e.to_string())?; + Ok(vec![session_store::LoggedSystemHistoryEntry::from_input( + session_store::LoggedSessionHistoryOrigin::Provided { + producer: "subjektiv.journal".into(), + }, + None, + journal_input(body, payload, "journal_startup")?, + )]) + } + + async fn load_compaction_context( + &self, + session: session_store::SessionId, + ) -> Result, String> { + // A changed Session must never reuse the old source's Host ownership. + let connection = if self.host.context.session_id == session.to_string() { + Ok(Some(self.host.clone())) + } else { + Err(std::io::Error::new( + std::io::ErrorKind::PermissionDenied, + "context source belongs to a different Session", + )) + }; + let input = freeze_compaction_journal(connection, &self.prompts.load_full())?; + Ok(vec![session_store::LoggedSystemHistoryEntry::from_input( + session_store::LoggedSessionHistoryOrigin::Provided { + producer: "subjektiv.journal".into(), + }, + None, + input, + )]) } } @@ -103,7 +195,8 @@ impl crate::worker::InitialContextSource for JournalStartupContextSource { pub(crate) fn freeze_compaction_journal( connection: std::io::Result>, prompts: &crate::prompt::catalog::PromptCatalog, -) -> Result { +) -> Result { + use JournalReadStatus::*; let workspace_id = connection.as_ref().ok().and_then(|connection| { connection .as_ref() @@ -116,24 +209,25 @@ pub(crate) fn freeze_compaction_journal( Ok(_) => None, Err(_) => Some(Err(())), }; + let snapshot = read.as_ref().and_then(|read| read.as_ref().ok()).cloned(); let (status, json) = match read { - None => ("disabled", String::new()), + None => (Disabled, String::new()), Some(Ok(snapshot)) => { let status = if snapshot.journals.is_empty() { - "absent" + Absent } else { - "available" + Available }; match serde_json::to_string(&snapshot) { Ok(json) if json.len() <= subjektiv::journal::startup::STARTUP_MAX_BYTES => { (status, json) } - _ => ("unavailable", String::new()), + _ => (Unavailable, String::new()), } } - Some(Err(())) => ("unavailable", String::new()), + Some(Err(())) => (Unavailable, String::new()), }; - if status == "unavailable" { + if status == Unavailable { tracing::warn!("Journal compaction snapshot unavailable; continuing without Journal"); } use sha2::{Digest, Sha256}; @@ -143,12 +237,35 @@ pub(crate) fn freeze_compaction_journal( .map(|byte| format!("{byte:02x}")) .collect() }); - prompts.subjektiv_journal_compaction( - status, - &json, - snapshot_sha256.as_deref(), - workspace_id.as_deref(), + let body = prompts + .subjektiv_journal_compaction( + status.as_str(), + &json, + snapshot_sha256.as_deref(), + workspace_id.as_deref(), + ) + .map_err(|e| e.to_string())?; + let payload = session_store::FeatureInputPayload::encode( + "journal:compaction", + 1, + &JournalCompactionInput { + status, + prompt_provenance: session_store::PromptRenderProvenance { + workspace_id, + source_digest: prompts.projection().source_digest.clone(), + projection_digest: prompts.projection().catalog_digest.clone(), + logical_name: "internal.subjektiv_journal_compaction".into(), + }, + snapshot: if status == Unavailable { + None + } else { + snapshot + }, + snapshot_sha256, + }, ) + .map_err(|e| e.to_string())?; + journal_input(body, payload, "journal_compaction") } #[derive(Clone)] diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 32155ea04..f6f5869e6 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -43,6 +43,80 @@ use crate::worker::{ WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, }; +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SubjectBehaviorInput { + pub behavior_md: String, + pub prompt_provenance: Option, +} + +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +struct ResidentSummaryInput { + prompt_provenance: Option, +} + +fn resident_input( + body: String, + schema: &str, + payload: &impl serde::Serialize, + kind: &str, +) -> Result { + use session_store::*; + let input = SystemInputEnvelope::new( + body.clone(), + SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + Some(FeatureInputPayload::encode(schema, 1, payload)?), + Some(PublicSystemItem { + kind: kind.into(), + body, + presentation: PublicSystemPresentation::Context, + }), + )?, + )?; + Ok(LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: "subjektiv.resident-context".into(), + }, + None, + input, + )) +} + +pub(crate) fn subject_behavior_input( + body: String, + behavior_md: String, + prompt_provenance: Option, +) -> Result { + resident_input( + body, + "subject:behavior", + &SubjectBehaviorInput { + behavior_md, + prompt_provenance, + }, + "subject_behavior_refresh", + ) +} + +pub(crate) fn resident_summary_input( + body: String, + prompt_provenance: Option, +) -> Result { + resident_input( + body, + "memory:resident-summary", + &ResidentSummaryInput { prompt_provenance }, + "resident_summary_refresh", + ) +} + #[derive(Clone, Debug)] pub struct WorkspaceHttpMemoryBackend { client: Arc, @@ -714,7 +788,7 @@ impl WorkspaceSubjektivResidentSummarySource { fn render( &self, output: &server_api::SubjektivResidentContextOutput, - ) -> Result { + ) -> Result<(String, session_store::PromptRenderProvenance), String> { use memory::backend::MemoryResidentSummaryAvailability as Availability; let availability = match output.memory_surface.availability { Availability::Ready => "ready", @@ -723,14 +797,24 @@ impl WorkspaceSubjektivResidentSummarySource { Availability::Failed => "failed", Availability::Unavailable => "unavailable", }; - self.prompts - .load() + let prompts = self.prompts.load_full(); + let body = prompts .subjektiv_resident_context( &output.behavior_md, availability, output.memory_surface.content.as_deref(), ) - .map_err(|error| error.to_string()) + .map_err(|error| error.to_string())?; + let projection = prompts.projection(); + Ok(( + body, + 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_resident_context".into(), + }, + )) } fn remember_behavior(&self, output: &server_api::SubjektivResidentContextOutput) { @@ -752,7 +836,7 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource async fn load(&self) -> SystemPromptContribution { match self.fetch().await { Ok(output) => match self.render(&output) { - Ok(rendered) => { + Ok((rendered, _)) => { self.remember_behavior(&output); SystemPromptContribution::Ready(rendered) } @@ -788,20 +872,30 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource if !changed { return Ok(None); } - let rendered = self.render(&output)?; + let (rendered, prompt_provenance) = self.render(&output)?; Ok(Some(ResidentContextRefresh { body: rendered, behavior_md: output.behavior_md, + prompt_provenance: Some(prompt_provenance), })) } fn system_item_committed( &self, - item: &session_store::SystemItem, - _metadata: &crate::session_history::SessionHistoryMetadata, + _item: &session_store::SystemItem, + metadata: &crate::session_history::SessionHistoryMetadata, ) { - if let session_store::SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item { - self.confirm_resident_context_behavior(behavior_md); + if let Some(input) = metadata + .system_input + .as_ref() + .and_then(|input| input.feature_payload()) + .and_then(|payload| { + payload + .decode::("subject:behavior", 1) + .ok() + }) + { + self.confirm_resident_context_behavior(&input.behavior_md); } } @@ -954,7 +1048,13 @@ impl Hook for SubjektivBehaviorRefreshHook { "Subject behavior refresh requires durable session append authority", ) })?; - system_items.append_subject_behavior_refresh(refresh.body, refresh.behavior_md); + let input = subject_behavior_input( + refresh.body, + refresh.behavior_md, + refresh.prompt_provenance, + ) + .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?; + system_items.append_input(input); } Ok(HookPreRequestAction::Continue) } @@ -979,16 +1079,18 @@ impl Hook for SubjektivResidentRestoreRefreshHook { .resident_memory_restore_section(surface_ready, summary) .map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?; let projection = prompts.projection(); - context.system_items().append_resident_summary_refresh( + let input = resident_summary_input( body, - session_store::PromptRenderProvenance { + Some(session_store::PromptRenderProvenance { workspace_id: self.workspace_id.clone(), source_digest: projection.source_digest.clone(), projection_digest: projection.catalog_digest.clone(), logical_name: WorkerPrompt::ResidentMemoryRestoreSection.key().to_string(), - }, - ); + }), + ) + .map_err(|error| HookError::new(HookErrorCategory::Dependency, error.to_string()))?; + context.system_items().append_input(input); Ok(()) } } @@ -1751,14 +1853,13 @@ permission = "write" .expect("changed behavior must produce one refresh"); assert_eq!(refresh.behavior_md, "Ask when uncertain."); assert!(refresh.body.contains("Ask when uncertain.")); - source.system_item_committed( - &session_store::SystemItem::SubjectBehaviorRefresh { - body: refresh.body.clone(), - behavior_md: refresh.behavior_md.clone(), - prompt_provenance: None, - }, - &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), - ); + let input = subject_behavior_input( + refresh.body.clone(), + refresh.behavior_md.clone(), + refresh.prompt_provenance.clone(), + ) + .unwrap(); + source.system_item_committed(&input.item, &input.metadata); assert!( source .load_changed_resident_context() @@ -1775,14 +1876,13 @@ permission = "write" .expect("cleared behavior must produce one refresh"); assert_eq!(cleared.behavior_md, ""); assert!(cleared.body.contains("No user-managed behavior is set")); - source.system_item_committed( - &session_store::SystemItem::SubjectBehaviorRefresh { - body: cleared.body.clone(), - behavior_md: cleared.behavior_md.clone(), - prompt_provenance: None, - }, - &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), - ); + let input = subject_behavior_input( + cleared.body.clone(), + cleared.behavior_md.clone(), + cleared.prompt_provenance.clone(), + ) + .unwrap(); + source.system_item_committed(&input.item, &input.metadata); assert!( source .load_changed_resident_context() @@ -1840,14 +1940,13 @@ permission = "write" .await .unwrap() .unwrap(); - source.system_item_committed( - &session_store::SystemItem::SubjectBehaviorRefresh { - body: first.body.clone(), - behavior_md: first.behavior_md.clone(), - prompt_provenance: None, - }, - &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), - ); + let input = subject_behavior_input( + first.body.clone(), + first.behavior_md.clone(), + first.prompt_provenance.clone(), + ) + .unwrap(); + source.system_item_committed(&input.item, &input.metadata); assert_eq!( source .load_changed_resident_context() @@ -1856,14 +1955,13 @@ permission = "write" .unwrap(), second ); - source.system_item_committed( - &session_store::SystemItem::SubjectBehaviorRefresh { - body: second.body.clone(), - behavior_md: second.behavior_md.clone(), - prompt_provenance: None, - }, - &session_store::LoggedSessionHistoryMetadata::legacy_unknown(), - ); + let input = subject_behavior_input( + second.body.clone(), + second.behavior_md.clone(), + second.prompt_provenance.clone(), + ) + .unwrap(); + source.system_item_committed(&input.item, &input.metadata); assert!( source .load_changed_resident_context() diff --git a/crates/worker/src/feature/builtin/task/mod.rs b/crates/worker/src/feature/builtin/task/mod.rs index e45b1b703..0d82bb437 100644 --- a/crates/worker/src/feature/builtin/task/mod.rs +++ b/crates/worker/src/feature/builtin/task/mod.rs @@ -276,7 +276,9 @@ mod tests { use super::*; use crate::hook::{PreRequestInfo, SystemItemAppendHandle}; - fn pre_request_context(pending: Arc>>) -> PreRequestContext { + fn pre_request_context( + pending: Arc>>, + ) -> PreRequestContext { PreRequestContext::new( PreRequestInfo { item_count: 1, @@ -315,7 +317,7 @@ mod tests { let queued = pending.lock().expect("pending queue poisoned"); assert_eq!(queued.len(), 1); - let SystemItem::TaskReminder { body, .. } = &queued[0] else { + let SystemItem::TaskReminder { body, .. } = &queued[0].item else { panic!("unexpected system item: {:?}", queued[0]); }; assert!(body.starts_with("Current session steps are listed below.")); @@ -339,7 +341,7 @@ mod tests { } let queued = pending.lock().expect("pending queue poisoned"); - let SystemItem::TaskReminder { source, body, .. } = &queued[0] else { + let SystemItem::TaskReminder { source, body, .. } = &queued[0].item else { panic!("unexpected system item: {:?}", queued[0]); }; assert_eq!(*source, SystemReminderSource::TaskInactivity); @@ -452,7 +454,7 @@ mod tests { } let queued = pending.lock().expect("pending queue poisoned"); assert_eq!(queued.len(), 1); - let SystemItem::TaskReminder { body, .. } = &queued[0] else { + let SystemItem::TaskReminder { body, .. } = &queued[0].item else { panic!("unexpected system item: {:?}", queued[0]); }; assert!(body.contains("taskid 1 (completed) done")); diff --git a/crates/worker/src/feature/invocation.rs b/crates/worker/src/feature/invocation.rs index de8bc6fbe..91009461e 100644 --- a/crates/worker/src/feature/invocation.rs +++ b/crates/worker/src/feature/invocation.rs @@ -39,6 +39,45 @@ const MAX_INVOCATION_DESCRIPTOR_BYTES: usize = protocol::invocation::MAX_FEATURE_INVOCATION_DESCRIPTOR_BYTES; const MAX_COMPLETION_RESPONSE_BYTES: usize = 64 * 1024; +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct InvocationResultInput { + pub result: FeatureInvocationResult, +} + +pub(crate) fn result_input( + result: FeatureInvocationResult, + body: String, +) -> Result { + use session_store::*; + let payload = FeatureInputPayload::encode( + "feature:invocation-result", + 1, + &InvocationResultInput { result }, + )?; + let metadata = SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + Some(payload), + Some(PublicSystemItem { + kind: "feature_invocation_result".into(), + body: body.clone(), + presentation: PublicSystemPresentation::Context, + }), + )?; + Ok(LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: "feature.invocation".into(), + }, + None, + SystemInputEnvelope::new(body, metadata)?, + )) +} + pub(crate) fn invocation_result_is_bounded(result: &FeatureInvocationResult) -> bool { result.message.len() <= MAX_INVOCATION_MESSAGE_BYTES && result diff --git a/crates/worker/src/hook.rs b/crates/worker/src/hook.rs index a433bb7e3..c480471dc 100644 --- a/crates/worker/src/hook.rs +++ b/crates/worker/src/hook.rs @@ -25,7 +25,9 @@ use agen::tool::{ToolOutput, ToolResult}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; use serde_json::Value; -use session_store::{SystemItem, SystemReminder}; +#[cfg(test)] +use session_store::SystemItem; +use session_store::SystemReminder; use thiserror::Error; use crate::SessionEntryRef; @@ -254,7 +256,7 @@ impl From for TurnEndAction { // Hook context handles // ============================================================================= -/// Host-created handle for appending approved durable [`SystemItem`] requests. +/// Host-created handle for appending approved durable [`session_store::SystemItem`] requests. /// /// Hook code can use this handle only when the Worker host includes it in an /// event-specific context. The handle queues typed requests; the host drains the @@ -263,11 +265,11 @@ impl From for TurnEndAction { /// raw `agen::Item`, history writer, event sender, `Worker`, `Engine`, or /// notification buffer. pub struct SystemItemAppendHandle { - pending: Arc>>, + pending: Arc>>, } impl SystemItemAppendHandle { - pub(crate) fn new(pending: Arc>>) -> Self { + pub(crate) fn new(pending: Arc>>) -> Self { Self { pending } } @@ -279,53 +281,44 @@ impl SystemItemAppendHandle { self.pending .lock() .expect("system-item append queue poisoned") - .push(item); + .push(session_store::LoggedSystemHistoryEntry { + item, + metadata: crate::session_history::metadata( + session_store::LoggedSessionHistoryOrigin::BackendInstruction { + operation_id: None, + }, + None, + ), + }); } - /// Queue a Subject behavior refresh with its exact represented behavior. The caller receives this - /// narrow operation only at a pre-request boundary, and the host commits it - /// before exposing it to the model. - pub fn append_subject_behavior_refresh( - &self, - body: impl Into, - behavior_md: impl Into, - ) { + /// Queue producer-validated System context. Metadata and body are committed together. + /// The Host exposes this capability only at a pre-request boundary. + pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) { self.pending .lock() .expect("system-item append queue poisoned") - .push(SystemItem::SubjectBehaviorRefresh { - body: body.into(), - behavior_md: behavior_md.into(), - prompt_provenance: None, - }); + .push(input); } } /// Host-created narrow append authority exposed only at a restore lifecycle -/// boundary. It cannot emit arbitrary system items. +/// boundary. No model-facing tool receives this capability. pub struct RestoreSystemItemAppendHandle { - pending: Arc>>, + pending: Arc>>, } impl RestoreSystemItemAppendHandle { - pub(crate) fn new(pending: Arc>>) -> Self { + pub(crate) fn new(pending: Arc>>) -> Self { Self { pending } } - /// Queue a Feature-rendered resident-summary refresh for durable append at - /// the restore lifecycle boundary. - pub fn append_resident_summary_refresh( - &self, - body: impl Into, - prompt_provenance: session_store::PromptRenderProvenance, - ) { + /// Queue producer-validated System context at the Host restore boundary. + pub fn append_input(&self, input: session_store::LoggedSystemHistoryEntry) { self.pending .lock() .expect("restore system-item append queue poisoned") - .push(SystemItem::ResidentSummaryRefresh { - body: body.into(), - prompt_provenance: Some(prompt_provenance), - }); + .push(input); } } @@ -983,7 +976,7 @@ mod tests { let queued = pending.lock().expect("pending queue poisoned"); assert_eq!(queued.len(), 1); - match &queued[0] { + match &queued[0].item { SystemItem::TaskReminder { body, .. } => { assert_eq!(body, "remember tasks"); } @@ -992,7 +985,7 @@ mod tests { } #[test] - fn restore_append_handle_queues_only_resident_refresh_items() { + fn restore_append_handle_preserves_producer_metadata() { let pending = Arc::new(Mutex::new(Vec::new())); let handle = RestoreSystemItemAppendHandle::new(Arc::clone(&pending)); let provenance = session_store::PromptRenderProvenance { @@ -1003,15 +996,16 @@ mod tests { logical_name: "internal.resident_memory_restore_section".into(), }; - handle.append_resident_summary_refresh("current surface", provenance.clone()); - - assert!(matches!( - pending.lock().unwrap().as_slice(), - [SystemItem::ResidentSummaryRefresh { - body, - prompt_provenance: Some(actual), - }] if body == "current surface" && actual == &provenance - )); + let input = crate::feature::builtin::memory::resident_summary_input( + "current surface".into(), + Some(provenance), + ) + .unwrap(); + let expected = serde_json::to_value(&input).unwrap(); + handle.append_input(input); + let queued = pending.lock().unwrap(); + assert_eq!(queued.len(), 1); + assert_eq!(serde_json::to_value(&queued[0]).unwrap(), expected); } #[test] diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index cbb500796..1e91f1797 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -262,6 +262,31 @@ impl WorkerInterceptor { ) } + fn commit_system_entries( + &self, + entries: &[session_store::LoggedSystemHistoryEntry], + ) -> Result<(), session_store::StoreError> { + let Some(writer) = &self.log_writer else { + return Ok(()); + }; + for entry in entries { + writer.commit_annotated_system_item_then( + entry.clone(), + Vec::new(), + &mut |history| { + self.pending_committed_history + .lock() + .expect("pending committed history poisoned") + .push_back(history.clone()); + if let Some(source) = &self.resident_context_source { + source.system_item_committed(&entry.item, &history.annotation); + } + }, + )?; + } + Ok(()) + } + fn current_turn_index(&self) -> usize { self.next_turn_index .load(Ordering::Relaxed) @@ -387,11 +412,6 @@ impl WorkerInterceptor { } if prompt_provenance.is_none() => { *prompt_provenance = Some(provenance("internal.task_reminder")); } - SystemItem::SubjectBehaviorRefresh { - prompt_provenance, .. - } if prompt_provenance.is_none() => { - *prompt_provenance = Some(provenance("internal.subjektiv_resident_context")); - } SystemItem::Interrupt { prompt_provenance, .. } if prompt_provenance.is_none() => { @@ -617,15 +637,17 @@ impl Interceptor for WorkerInterceptor { return Ok(PreRequestAction::Yield); } - let mut system_items: Vec = std::mem::take( + let mut system_items: Vec = std::mem::take( &mut *pending_hook_system_items .lock() .expect("pending hook system-item queue poisoned"), ); - self.attach_prompt_provenance(&mut system_items); + for entry in &mut system_items { + self.attach_prompt_provenance(std::slice::from_mut(&mut entry.item)); + } let mut appended_items: Vec = system_items .iter() - .map(SystemItem::to_history_item) + .map(|entry| entry.item.to_history_item()) .collect(); let effective_context = if appended_items.is_empty() { Cow::Borrowed(context.as_slice()) @@ -639,7 +661,7 @@ impl Interceptor for WorkerInterceptor { let compact_decision = self.request_compact_decision(current_tokens, effective_context.as_ref()); if !matches!(compact_decision, AutomaticCompactDecision::Continue) { - if let Err(error) = self.commit_system_items(&system_items) { + if let Err(error) = self.commit_system_entries(&system_items) { return Ok(PreRequestAction::Cancel(format!( "session persistence failed: {error}" ))); @@ -683,7 +705,7 @@ impl Interceptor for WorkerInterceptor { if appended_items.is_empty() { return Ok(PreRequestAction::Continue); } - Ok(match self.commit_system_items(&system_items) { + Ok(match self.commit_system_entries(&system_items) { Ok(()) => PreRequestAction::ContinueWith(appended_items), Err(error) => PreRequestAction::Cancel(format!("session persistence failed: {error}")), }) @@ -983,7 +1005,14 @@ mod tests { input .system_items() .expect("test interceptor has durable append authority") - .append_subject_behavior_refresh("current Subject context", "Ask when uncertain."); + .append_input( + crate::feature::builtin::memory::subject_behavior_input( + "current Subject context".into(), + "Ask when uncertain.".into(), + None, + ) + .unwrap(), + ); Ok(HookPreRequestAction::Continue) } } @@ -1000,11 +1029,23 @@ mod tests { fn system_item_committed( &self, - item: &SystemItem, - _metadata: &crate::session_history::SessionHistoryMetadata, + _item: &SystemItem, + metadata: &crate::session_history::SessionHistoryMetadata, ) { - if let SystemItem::SubjectBehaviorRefresh { behavior_md, .. } = item { - self.confirmed.lock().unwrap().push(behavior_md.to_owned()); + if let Some(input) = metadata + .system_input + .as_ref() + .and_then(|input| input.feature_payload()) + .and_then(|payload| { + payload + .decode::( + "subject:behavior", + 1, + ) + .ok() + }) + { + self.confirmed.lock().unwrap().push(input.behavior_md); } } } @@ -1549,11 +1590,7 @@ mod tests { assert_eq!(*confirmed.lock().unwrap(), ["Ask when uncertain."]); assert!(matches!( committed.lock().unwrap().as_slice(), - [SystemItem::SubjectBehaviorRefresh { - behavior_md, - prompt_provenance: Some(_), - .. - }] if behavior_md == "Ask when uncertain." + [SystemItem::Input { body }] if body == "current Subject context" )); } @@ -1654,12 +1691,15 @@ mod tests { })), ) .with_resident_context_source(Some(source)); - let items = ["raw behavior", ""].map(|behavior_md| SystemItem::SubjectBehaviorRefresh { - body: "same rendered context".into(), - behavior_md: behavior_md.into(), - prompt_provenance: None, + let items = ["raw behavior", ""].map(|behavior_md| { + crate::feature::builtin::memory::subject_behavior_input( + "same rendered context".into(), + behavior_md.into(), + None, + ) + .unwrap() }); - interceptor.commit_system_items(&items).unwrap(); + interceptor.commit_system_entries(&items).unwrap(); assert_eq!(*confirmed.lock().unwrap(), ["raw behavior", ""]); assert_eq!(committed.lock().unwrap().len(), 2); } diff --git a/crates/worker/src/ipc/protocol_session.rs b/crates/worker/src/ipc/protocol_session.rs index 038770284..8fc4a5d97 100644 --- a/crates/worker/src/ipc/protocol_session.rs +++ b/crates/worker/src/ipc/protocol_session.rs @@ -340,22 +340,13 @@ mod tests { #[test] fn system_item_log_entry_keeps_identity_in_live_event() { let entry_id = session_store::LoggedSessionHistoryEntryId::new(); + let mut input = + crate::feature::builtin::memory::resident_summary_input("refresh".into(), None) + .unwrap(); + input.metadata.entry_id = entry_id.clone(); let event = live_log_entry_event(LogEntry::AnnotatedSystemItem { ts: session_store::segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { - item: session_store::SystemItem::ResidentSummaryRefresh { - body: "refresh".to_string(), - prompt_provenance: None, - }, - metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, - entry_id: entry_id.clone(), - origin: session_store::LoggedSessionHistoryOrigin::BackendInstruction { - operation_id: None, - }, - derivation: None, - }, - }, + entry: input, extensions: Vec::new(), }) .expect("SystemItem must be live-relevant"); @@ -381,25 +372,22 @@ mod tests { logical_name: "private-template".into(), }; let items = [ - session_store::SystemItem::SubjectBehaviorRefresh { - body: "displayed context".into(), - behavior_md: "private-raw-behavior".into(), - prompt_provenance: Some(provenance.clone()), - }, - session_store::SystemItem::JournalStartup { - body: "displayed reference".into(), - snapshot: serde_json::json!({ "internal": "private-snapshot" }), - snapshot_sha256: "private-snapshot-digest".into(), - prompt_provenance: provenance, - }, + crate::feature::builtin::memory::subject_behavior_input( + "displayed context".into(), + "private-raw-behavior".into(), + Some(provenance.clone()), + ) + .unwrap(), + crate::feature::builtin::memory::resident_summary_input( + "displayed reference".into(), + Some(provenance), + ) + .unwrap(), ]; - for item in items { + for input in items { let entry = LogEntry::AnnotatedSystemItem { ts: 7, - entry: session_store::LoggedSystemHistoryEntry { - item, - metadata: session_store::LoggedSessionHistoryMetadata::legacy_unknown(), - }, + entry: input, extensions: vec![], }; let persisted: LogEntry = diff --git a/crates/worker/src/session_history.rs b/crates/worker/src/session_history.rs index 232a8c92d..472d40eb6 100644 --- a/crates/worker/src/session_history.rs +++ b/crates/worker/src/session_history.rs @@ -67,6 +67,16 @@ pub(crate) fn metadata( } } +pub(crate) fn compaction_use( + metadata: &SessionHistoryMetadata, +) -> session_store::InputCompactionUse { + metadata + .system_input + .as_ref() + .map(|input| input.usage().compaction) + .unwrap_or(session_store::InputCompactionUse::Retain) +} + #[cfg(test)] pub(crate) fn history_entry( item: Item, @@ -134,22 +144,10 @@ pub(crate) fn restore_history_entries( } *target = from_logged(entry); } - LogEntry::AnnotatedSystemItem { entry, .. } => { - history.push(HistoryEntry::new(entry.item.to_history_item(), { - let mut metadata = entry.metadata.clone(); - // 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 - })) - } + LogEntry::AnnotatedSystemItem { entry, .. } => history.push(HistoryEntry::new( + entry.item.to_history_item(), + entry.metadata.clone(), + )), _ => {} } } diff --git a/crates/worker/src/skill.rs b/crates/worker/src/skill.rs index f9b15bf88..4629e03a8 100644 --- a/crates/worker/src/skill.rs +++ b/crates/worker/src/skill.rs @@ -8,6 +8,50 @@ pub use server_api::{ SkillProjectionStatus, SkillProvenance, SkillResourceRef, SkillSourceKind, }; +#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub(crate) struct SkillActivationInput { + pub name: String, +} + +pub(crate) fn activation_input( + activation: &SkillActivationResponse, +) -> Result { + use session_store::*; + let body = format!( + "Agent Skill `{}` activated from {}.\n\n{}", + activation.name, activation.provenance.id, activation.body + ); + let payload = FeatureInputPayload::encode( + "skill:activation", + 1, + &SkillActivationInput { + name: activation.name.clone(), + }, + )?; + let metadata = SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + Some(payload), + Some(PublicSystemItem { + kind: "skill_activation".into(), + body: body.clone(), + presentation: PublicSystemPresentation::Context, + }), + )?; + Ok(LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: "skill.activation".into(), + }, + None, + SystemInputEnvelope::new(body, metadata)?, + )) +} + #[derive(Debug, thiserror::Error)] pub enum SkillClientError { #[error("workspace client is unavailable: {0}")] diff --git a/crates/worker/src/subjektiv.rs b/crates/worker/src/subjektiv.rs index 40d4836c6..eb871f992 100644 --- a/crates/worker/src/subjektiv.rs +++ b/crates/worker/src/subjektiv.rs @@ -26,6 +26,30 @@ use server_api::*; use crate::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceServerOperation}; +impl + crate::Worker +{ + /// Bind live Host capability and its context provider, not authority inherited + /// from a previous binding. Ordinary Feature/lifecycle gates also apply to embedders. + pub fn bind_subjektiv_host(&mut self, host: Option>) { + self.set_subjektiv_host(host); + if self.manifest_lifecycle_features_enabled() + && self.backend_job().is_none() + && crate::feature::builtin::memory::ordinary_subjektiv_features_enabled(self.manifest()) + && let Ok(Some(connection)) = self.subjektiv_host_connection() + && (connection.host.journal_startup_granted() + || connection.host.journal_compaction_granted()) + { + self.install_session_context_source(Arc::new( + crate::feature::builtin::journal::JournalContextSource::new( + connection, + self.prompts(), + ), + )); + } + } +} + /// Host-issued execution context, never deserialized from model arguments. #[derive(Clone, Debug, PartialEq, Eq)] pub struct SubjektivHostContext { @@ -1130,7 +1154,12 @@ enabled = true 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 { .. }) => + if entry + .metadata + .system_input + .as_ref() + .and_then(|input| input.feature_payload()) + .is_some_and(|payload| payload.schema() == "journal:startup") => { Some(entry) } @@ -1140,22 +1169,33 @@ enabled = true let body = journal.map(|journal| { assert_eq!( journal.metadata.origin, - LoggedSessionHistoryOrigin::JournalContext + LoggedSessionHistoryOrigin::Provided { + producer: "subjektiv.journal".into() + } ); - let SystemItem::JournalStartup { - body, - snapshot: saved, - snapshot_sha256, - prompt_provenance, - } = &journal.item - else { + let SystemItem::Input { body } = &journal.item else { unreachable!() }; - assert_eq!(*saved, serde_json::to_value(&snapshot).unwrap()); + let payload = journal + .metadata + .system_input + .as_ref() + .unwrap() + .feature_payload() + .unwrap() + .decode::( + "journal:startup", + 1, + ) + .unwrap(); + let saved = serde_json::to_value(&payload.snapshot).unwrap(); + let snapshot_sha256 = &payload.snapshot_sha256; + let prompt_provenance = &payload.prompt_provenance; + 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()) + Sha256::digest(serde_json::to_vec(&saved).unwrap()) .iter() .map(|byte| format!("{byte:02x}")) .collect::() @@ -1168,7 +1208,7 @@ enabled = true body.trim_end(), format!( "## Prior Journal context\n\n```json\n{}\n```", - serde_json::to_string(saved).unwrap() + serde_json::to_string(&saved).unwrap() ) ); body.as_str() @@ -1237,7 +1277,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(), usize::from(contributes_context)); + assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).is_some_and(|payload| payload.schema() == "journal:startup"))).count(), usize::from(contributes_context)); controller .handle .send(Method::Shutdown { diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index fdde548d2..805ffb044 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2757,11 +2757,25 @@ pub trait SystemItemCommitter: Send + Sync { .unwrap_or(WorkerHistoryProvenance::BackendInstruction { operation_id: None }), None, ); - let history = HistoryEntry::new(item.to_history_item(), metadata.clone()); + self.commit_annotated_system_item_then( + session_store::LoggedSystemHistoryEntry { item, metadata }, + extensions, + after_commit, + ) + } + + fn commit_annotated_system_item_then( + &self, + entry: session_store::LoggedSystemHistoryEntry, + extensions: Vec, + after_commit: &mut dyn FnMut(&HistoryEntry), + ) -> Result, StoreError> { + entry.validate_system_input()?; + let history = HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone()); self.commit_usage_boundary( LogEntry::AnnotatedSystemItem { ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { item, metadata }, + entry, extensions, }, &mut || after_commit(&history), @@ -2934,14 +2948,21 @@ pub(crate) enum SystemPromptContribution { pub(crate) struct ResidentContextRefresh { pub body: String, pub behavior_md: String, + pub prompt_provenance: Option, } -/// 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. +/// Feature-owned System context at explicit creation and compaction boundaries. +/// Core persists and replays each pair unchanged; normal restore never reloads it. #[async_trait::async_trait] -pub(crate) trait InitialContextSource: Send + Sync { +pub(crate) trait SessionContextSource: Send + Sync { async fn load(&self) -> Result, String>; + + async fn load_compaction_context( + &self, + _session: SessionId, + ) -> Result, String> { + Ok(Vec::new()) + } } #[async_trait::async_trait] @@ -3132,7 +3153,7 @@ pub struct Worker { inject_resident_summary: bool, /// Deferred resident prompt source installed by an enabled Feature. feature_resident_summary_source: Option>, - initial_context_source: Option>, + session_context_source: Option>, /// Exact complete seed retained across ambiguous Store responses. The committed /// segment remains authoritative; metadata must be repaired before accepting input. prepared_initial_seed: Option>, @@ -3404,7 +3425,7 @@ impl Worker { prompts, inject_resident_summary: true, feature_resident_summary_source: None, - initial_context_source: None, + session_context_source: None, prepared_initial_seed: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, @@ -3452,8 +3473,8 @@ impl Worker { self.inject_resident_summary = enabled; } - pub(crate) fn install_initial_context_source(&mut self, source: Arc) { - self.initial_context_source = Some(source); + pub(crate) fn install_session_context_source(&mut self, source: Arc) { + self.session_context_source = Some(source); } pub(crate) fn install_system_prompt_contribution( @@ -3629,10 +3650,12 @@ impl Worker { self.workspace_context.client_handle() } - /// Bind a live capability before Feature installation. This is deliberately - /// separate from Workspace authority and is not inherited by SubWorkers. - pub fn bind_subjektiv_host(&mut self, host: Option>) { + pub(crate) fn set_subjektiv_host( + &mut self, + host: Option>, + ) { self.subjektiv_host = host; + self.session_context_source = None; } pub fn subjektiv_host_connection( @@ -3700,26 +3723,13 @@ impl Worker { { let activation = self.workspace_client().activate_skill(name)?; self.ensure_segment_head().await?; - let body = format!( - "Agent Skill `{}` activated from {}.\n\n{}", - activation.name, activation.provenance.id, activation.body - ); - let skill_metadata = new_history_metadata( - WorkerHistoryProvenance::BackendInstruction { operation_id: None }, - None, - ); + let input = crate::skill::activation_input(&activation).map_err(StoreError::from)?; + let history_entry = HistoryEntry::new(input.item.to_history_item(), input.metadata.clone()); self.commit_entry(LogEntry::AnnotatedSystemItem { ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { - item: SystemItem::SkillActivation { - name: activation.name.clone(), - body: body.clone(), - }, - metadata: skill_metadata.clone(), - }, + entry: input, extensions: Vec::new(), })?; - let history_entry = HistoryEntry::new(agen::Item::system_message(body), skill_metadata); let mut annotate = history_annotator( self.log_writer_handle(), vec![history_entry.clone()], @@ -4594,18 +4604,12 @@ impl Worker { .lock() .expect("restore system-item queue poisoned"), ); - for item in system_items { - let history_item = item.to_history_item(); - let metadata = new_history_metadata( - WorkerHistoryProvenance::BackendInstruction { operation_id: None }, - None, - ); + for entry in system_items { + let history_item = entry.item.to_history_item(); + let metadata = entry.metadata.clone(); self.commit_entry(LogEntry::AnnotatedSystemItem { ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { - item, - metadata: metadata.clone(), - }, + entry, extensions: Vec::new(), })?; let history_entry = HistoryEntry::new(history_item, metadata); @@ -5368,17 +5372,14 @@ impl Worker { .invocation_receipts .push(InvocationReceipt::new(invocation, Some(result.clone()))); } - let item = SystemItem::FeatureInvocationResult { result, body }; - let metadata = new_history_metadata( - WorkerHistoryProvenance::BackendInstruction { operation_id: None }, - None, - ); - let history = HistoryEntry::new(item.to_history_item(), metadata.clone()); + let input = + crate::feature::invocation::result_input(result, body).map_err(StoreError::from)?; + let history = HistoryEntry::new(input.item.to_history_item(), input.metadata.clone()); handle .writer .append_entry_locked(LogEntry::AnnotatedSystemItem { ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { item, metadata }, + entry: input, extensions: vec![pending_activation_extension(&completed)], })?; *state = completed; @@ -5882,7 +5883,7 @@ impl Worker { } else { if self.prepared_initial_seed.is_none() { self.ensure_system_prompt_materialized().await?; - let items = match &self.initial_context_source { + let items = match &self.session_context_source { Some(source) => source.load().await.map_err(WorkerError::InvalidState)?, None => Vec::new(), }; @@ -6939,11 +6940,18 @@ impl Worker { // 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 frozen_context = match &self.session_context_source { + Some(source) => source + .load_compaction_context(self.session_id()) + .await + .map_err(WorkerError::InvalidState)?, + None => Vec::new(), + }; + for entry in &frozen_context { + entry + .validate_system_input() + .map_err(|e| WorkerError::InvalidState(e.to_string()))?; + } let history_entries = self.session.history().entries(); let retain_from = cut.index.min(history_entries.len()); let initial_summary = initial_compaction_summary(history_entries)?; @@ -6953,11 +6961,8 @@ impl Worker { let retained_history_entries = history_entries[retain_from..] .iter() .filter(|entry| { - !matches!( - entry.annotation.origin, - WorkerHistoryProvenance::CompactionSummary - | WorkerHistoryProvenance::JournalContext - ) + crate::session_history::compaction_use(&entry.annotation) + == session_store::InputCompactionUse::Retain }) .cloned() .collect::>(); @@ -6976,11 +6981,8 @@ impl Worker { let mut entries_to_summarise = history_entries[..retain_from] .iter() .filter(|entry| { - !matches!( - entry.annotation.origin, - WorkerHistoryProvenance::CompactionSummary - | WorkerHistoryProvenance::JournalContext - ) + crate::session_history::compaction_use(&entry.annotation) + == session_store::InputCompactionUse::Retain }) .cloned() .collect::>(); @@ -7067,24 +7069,29 @@ impl Worker { // 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) - { + for entry in history_entries.iter().filter(|entry| { + crate::session_history::compaction_use(&entry.annotation) + == session_store::InputCompactionUse::ReferenceInput + }) { if let Some(body) = entry.item.as_text() { summary_input.text.push_str( - "\n\n## Previous Journal reference (historical, not automatically retained)\n", + "\n\n## Previous reference input (historical, not automatically retained)\n", ); summary_input.text.push_str(body); summary_input .text - .push_str("\n## End previous Journal reference\n"); + .push_str("\n## End previous reference input\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); + for (index, entry) in frozen_context.iter().enumerate() { + if index != 0 { + summary_input.text.push_str("\n\n"); + } + summary_input.text.push_str(&entry.item.history_text()); + } if summary_input.warning_exceeded { self.alert( AlertLevel::Warn, @@ -7400,8 +7407,11 @@ impl Worker { 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()); + let context_messages = frozen_context + .iter() + .map(|entry| entry.item.to_history_item()) + .collect::>(); + compact_introduced_system_messages.extend(context_messages.iter().cloned()); 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()); @@ -7414,7 +7424,7 @@ impl Worker { compact_introduced_system_messages.push(task_snapshot_message.clone()); new_history.push(summary_message); - new_history.push(journal_message); + new_history.extend(context_messages); new_history.extend(auto_read_messages); if let Some(msg) = reference_message { new_history.push(msg); @@ -7454,6 +7464,32 @@ impl Worker { .iter() .map(|entry| entry.annotation.entry_id.clone()) .collect::>(); + let generic_summary_input = session_store::SystemInputEnvelope::new( + new_history[0] + .as_text() + .expect("compaction summary is text"), + session_store::SystemInputMetadata::new( + session_store::SystemInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: session_store::InputCompactionUse::OpeningSummary, + }, + None, + None, + ) + .map_err(|e| WorkerError::InvalidState(e.to_string()))?, + ) + .map_err(|e| WorkerError::InvalidState(e.to_string()))?; + let summary_entry = session_store::LoggedSystemHistoryEntry::from_input( + WorkerHistoryProvenance::Provided { + producer: "worker.compaction".into(), + }, + Some(SessionHistoryDerivation { + sources: derived_sources.clone(), + }), + generic_summary_input, + ); let mut original_cursor = 0usize; let compacted_history_entries = new_history .iter() @@ -7461,25 +7497,9 @@ impl Worker { .enumerate() .map(|(index, item)| { if index == 0 { - HistoryEntry::new( - item, - new_history_metadata( - WorkerHistoryProvenance::CompactionSummary, - Some(SessionHistoryDerivation { - sources: derived_sources.clone(), - }), - ), - ) - } else if index == 1 { - HistoryEntry::new( - item, - new_history_metadata( - WorkerHistoryProvenance::JournalContext, - Some(SessionHistoryDerivation { - sources: derived_sources.clone(), - }), - ), - ) + HistoryEntry::new(item, summary_entry.metadata.clone()) + } else if let Some(context) = frozen_context.get(index - 1) { + HistoryEntry::new(item, context.metadata.clone()) } else if let Some((offset, original)) = original_entries[original_cursor..] .iter() .enumerate() @@ -7977,7 +7997,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - initial_context_source: None, + session_context_source: None, prepared_initial_seed: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, @@ -8358,7 +8378,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - initial_context_source: None, + session_context_source: None, prepared_initial_seed: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, @@ -8490,7 +8510,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - initial_context_source: None, + session_context_source: None, prepared_initial_seed: None, subjektiv_host: None, restore_feature_lifecycle_pending: false, @@ -8901,7 +8921,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - initial_context_source: None, + session_context_source: None, prepared_initial_seed: None, subjektiv_host: None, restore_feature_lifecycle_pending: true, @@ -9296,9 +9316,10 @@ fn summary_shrink_prompt( fn initial_compaction_summary( entries: &[HistoryEntry], ) -> Result, WorkerError> { - let mut summaries = entries - .iter() - .filter(|entry| entry.annotation.origin == WorkerHistoryProvenance::CompactionSummary); + let mut summaries = entries.iter().filter(|entry| { + crate::session_history::compaction_use(&entry.annotation) + == session_store::InputCompactionUse::OpeningSummary + }); let Some(entry) = summaries.next() else { return Ok(None); }; @@ -10739,13 +10760,32 @@ mod build_summary_prompt_tests { #[test] fn malformed_or_duplicate_host_summaries_fail_instead_of_omitting_context() { use crate::session_history::history_entry; - let valid = history_entry( - Item::system_message(format!( - "{}body", - crate::session_history::COMPACTION_SUMMARY_HEADER - )), - WorkerHistoryProvenance::CompactionSummary, - ); + let opening = |item| { + let mut entry = history_entry( + item, + WorkerHistoryProvenance::Provided { + producer: "arbitrary-summary-source".into(), + }, + ); + entry.annotation.system_input = Some( + session_store::SystemInputMetadata::new( + session_store::SystemInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: session_store::InputCompactionUse::OpeningSummary, + }, + None, + None, + ) + .unwrap(), + ); + entry + }; + let valid = opening(Item::system_message(format!( + "{}body", + crate::session_history::COMPACTION_SUMMARY_HEADER + ))); assert_eq!( initial_compaction_summary(std::slice::from_ref(&valid)) .unwrap() @@ -10757,13 +10797,7 @@ mod build_summary_prompt_tests { Item::user_message("pasted"), Item::system_message("missing marker"), ] { - assert!( - initial_compaction_summary(&[history_entry( - item, - WorkerHistoryProvenance::CompactionSummary - )]) - .is_err() - ); + assert!(initial_compaction_summary(&[opening(item)]).is_err()); } } @@ -11190,7 +11224,7 @@ mod build_summary_prompt_tests { 1 } LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::FeatureInvocationResult { .. }) => + if input_schema(entry) == Some("feature:invocation-result") => { 2 } @@ -14162,13 +14196,7 @@ permission = "write" let entries = store.read_all(session_id, segment_id).unwrap(); assert!(matches!( entries.last(), - Some(LogEntry::AnnotatedSystemItem { - entry: session_store::LoggedSystemHistoryEntry { - item: SystemItem::ResidentSummaryRefresh { .. }, - .. - }, - .. - }) + Some(LogEntry::AnnotatedSystemItem { entry, .. }) if input_schema(entry) == Some("memory:resident-summary") )); } } @@ -14465,15 +14493,10 @@ permission = "write" assert!(entries.iter().any(|entry| { matches!( entry, - LogEntry::AnnotatedSystemItem { - entry: session_store::LoggedSystemHistoryEntry { - item: SystemItem::SkillActivation { name, body }, - .. - }, - .. - } if name == "triage-errors" - && body.contains("# Triage Errors") - && body == history_text + LogEntry::AnnotatedSystemItem { entry, .. } + if entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).and_then(|payload| payload.decode::("skill:activation", 1).ok()).is_some_and(|payload| payload.name == "triage-errors") + && entry.item.history_text().contains("# Triage Errors") + && entry.item.history_text() == history_text ) })); } @@ -15867,7 +15890,7 @@ permission = "write" .position(|entry| { matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::FeatureInvocationResult { .. }) + if input_schema(entry) == Some("feature:invocation-result") ) }) .expect("invocation result must be durable before the model request"); @@ -15876,7 +15899,7 @@ permission = "write" .position(|entry| { matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::SubjectBehaviorRefresh { .. }) + if input_schema(entry) == Some("subject:behavior") ) }) .expect("Subject behavior refresh must be durable before the model request"); @@ -16115,14 +16138,14 @@ permission = "write" .iter() .filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::FeatureInvocationResult { .. }) + if input_schema(entry) == Some("feature:invocation-result") )) .count(), 2 // Explicit replay displays the cached result; it does not repeat the effect. ); assert!(entries.iter().any(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::ResidentSummaryRefresh { .. }) + if input_schema(entry) == Some("memory:resident-summary") ))); } @@ -16165,7 +16188,7 @@ permission = "write" LogEntry::AnnotatedSystemItem { entry, extensions, .. } => { - matches!(entry.item, SystemItem::FeatureInvocationResult { .. }) + input_schema(entry) == Some("feature:invocation-result") && extensions.iter().any(|extension| { extension.domain == SESSION_PENDING_ACTIVATIONS_EXTENSION_DOMAIN && serde_json::from_value::( @@ -16756,34 +16779,70 @@ permission = "write" })); } - fn startup_test_item() -> SystemItem { - SystemItem::JournalStartup { - body: "PRIOR_JOURNAL_REFERENCE".into(), - snapshot: serde_json::json!({"subject_id":"subject", "journals":[], "omitted":false}), - snapshot_sha256: "snapshot-digest".into(), - prompt_provenance: PromptRenderProvenance { - workspace_id: None, - source_digest: "source".into(), - projection_digest: "projection".into(), - logical_name: "internal.subjektiv_journal_startup".into(), + fn input_schema(entry: &session_store::LoggedSystemHistoryEntry) -> Option<&str> { + entry + .metadata + .system_input + .as_ref()? + .feature_payload() + .map(|payload| payload.schema()) + } + + fn startup_test_input() -> session_store::LoggedSystemHistoryEntry { + use session_store::*; + let payload = FeatureInputPayload::encode( + "journal:startup", + 1, + &crate::feature::builtin::journal::JournalStartupInput { + snapshot: subjektiv::journal::startup::JournalStartupSnapshot { + subject_id: "subject".into(), + journals: vec![], + omitted: false, + }, + snapshot_sha256: "snapshot-digest".into(), + prompt_provenance: PromptRenderProvenance { + workspace_id: None, + source_digest: "source".into(), + projection_digest: "projection".into(), + logical_name: "internal.subjektiv_journal_startup".into(), + }, }, - } + ) + .unwrap(); + LoggedSystemHistoryEntry::from_input( + LoggedSessionHistoryOrigin::Provided { + producer: "arbitrary-context-provider".into(), + }, + None, + SystemInputEnvelope::new( + "PRIOR_JOURNAL_REFERENCE", + SystemInputMetadata::new( + SystemInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::ReferenceInput, + }, + Some(payload), + None, + ) + .unwrap(), + ) + .unwrap(), + ) } struct StartupTestSource { calls: Arc, fail: Arc, } #[async_trait] - impl InitialContextSource for StartupTestSource { + impl SessionContextSource for StartupTestSource { async fn load(&self) -> Result, String> { self.calls.fetch_add(1, Ordering::SeqCst); if self.fail.load(Ordering::SeqCst) { return Err("startup read failed".into()); } - Ok(vec![session_store::LoggedSystemHistoryEntry { - item: startup_test_item(), - metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None), - }]) + Ok(vec![startup_test_input()]) } } @@ -16791,7 +16850,7 @@ permission = "write" async fn initial_system_messages_preserve_feature_metadata_without_payload_classification() { struct Source(Vec); #[async_trait] - impl InitialContextSource for Source { + impl SessionContextSource for Source { async fn load(&self) -> Result, String> { Ok(self.0.clone()) } @@ -16807,16 +16866,19 @@ permission = "write" }, metadata: new_history_metadata(WorkerHistoryProvenance::JournalContext, None), }, - session_store::LoggedSystemHistoryEntry { - item: startup_test_item(), - metadata: new_history_metadata( + { + let mut input = startup_test_input(); + let custom = new_history_metadata( WorkerHistoryProvenance::BackendInstruction { operation_id: Some("feature-selected-origin".into()), }, Some(SessionHistoryDerivation { sources: vec![SessionHistoryEntryId::new()], }), - ), + ); + input.metadata.origin = custom.origin; + input.metadata.derivation = custom.derivation; + input }, ]; let mut worker = Worker::new( @@ -16829,7 +16891,7 @@ permission = "write" ) .await .unwrap(); - worker.install_initial_context_source(Arc::new(Source(expected.clone()))); + worker.install_session_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()) @@ -16886,7 +16948,7 @@ permission = "write" .await .unwrap(); let calls = Arc::new(AtomicUsize::new(0)); - worker.install_initial_context_source(Arc::new(StartupTestSource { + worker.install_session_context_source(Arc::new(StartupTestSource { calls: calls.clone(), fail: Arc::new(AtomicBool::new(false)), })); @@ -16907,7 +16969,9 @@ permission = "write" assert_eq!(replay[0].item.as_text(), Some("PRIOR_JOURNAL_REFERENCE")); assert_eq!( replay[0].annotation.origin, - WorkerHistoryProvenance::JournalContext + WorkerHistoryProvenance::Provided { + producer: "arbitrary-context-provider".into() + } ); let view = crate::session_capture::SessionCapture::from_history_entries("startup", replay); @@ -16918,7 +16982,7 @@ permission = "write" let entries = store .read_all(worker.session_id(), worker.segment_id()) .unwrap(); - let journal = entries.iter().position(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if matches!(entry.item, SystemItem::JournalStartup { .. }))).unwrap(); + let journal = entries.iter().position(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if input_schema(entry) == Some("journal:startup"))).unwrap(); let user = entries .iter() .position(|entry| matches!(entry, LogEntry::AnnotatedUserInput { .. })) @@ -16944,7 +17008,7 @@ permission = "write" .unwrap(); let calls = Arc::new(AtomicUsize::new(0)); let fail = Arc::new(AtomicBool::new(true)); - worker.install_initial_context_source(Arc::new(StartupTestSource { + worker.install_session_context_source(Arc::new(StartupTestSource { calls: calls.clone(), fail: fail.clone(), })); @@ -16975,7 +17039,7 @@ permission = "write" .await .unwrap(); let calls = Arc::new(AtomicUsize::new(0)); - worker.install_initial_context_source(Arc::new(StartupTestSource { + worker.install_session_context_source(Arc::new(StartupTestSource { calls: calls.clone(), fail: Arc::new(AtomicBool::new(false)), })); @@ -17030,7 +17094,7 @@ permission = "write" .unwrap(); worker.enable_worker_metadata_write_through().unwrap(); let calls = Arc::new(AtomicUsize::new(0)); - worker.install_initial_context_source(Arc::new(StartupTestSource { + worker.install_session_context_source(Arc::new(StartupTestSource { calls: calls.clone(), fail: Arc::new(AtomicBool::new(false)), })); @@ -17052,7 +17116,7 @@ permission = "write" &manifest.worker.name, manifest.clone(), store.clone(), PromptCatalogSource::builtins_only(), WorkerWorkspaceContext::no_workspace(), WorkerFilesystemAuthority::None, Some(Box::new(NoopClient))).await.unwrap(); - restored.install_initial_context_source(Arc::new(StartupTestSource { + restored.install_session_context_source(Arc::new(StartupTestSource { calls: calls.clone(), fail: Arc::new(AtomicBool::new(true)), })); @@ -17377,7 +17441,7 @@ permission = "write" .iter() .filter_map(|entry| match entry { LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::JournalStartup { .. }) => + if input_schema(entry) == Some("journal:startup") => { Some(entry.item.clone()) } @@ -17428,7 +17492,7 @@ permission = "write" .iter() .filter_map(|entry| match entry { LogEntry::AnnotatedSystemItem { entry, .. } - if matches!(entry.item, SystemItem::JournalStartup { .. }) => + if input_schema(entry) == Some("journal:startup") => { Some(&entry.item) } diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index d72150e71..e0f773abf 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -2220,10 +2220,48 @@ fn summary_entry( origin: session_store::LoggedSessionHistoryOrigin, ) -> session_store::LoggedHistoryEntry { let mut entry = annotated(item); + entry.metadata.system_input = matches!( + &entry.item, + session_store::LoggedItem::Message { + role: session_store::LoggedRole::System, + .. + } + ) + .then(|| { + session_store::SystemInputMetadata::new( + session_store::SystemInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: match origin { + session_store::LoggedSessionHistoryOrigin::CompactionSummary => { + session_store::InputCompactionUse::OpeningSummary + } + session_store::LoggedSessionHistoryOrigin::JournalContext => { + session_store::InputCompactionUse::ReferenceInput + } + _ => session_store::InputCompactionUse::Retain, + }, + }, + None, + None, + ) + .unwrap() + }); entry.metadata.origin = origin; entry } +fn compaction_use( + metadata: &session_store::LoggedSessionHistoryMetadata, +) -> session_store::InputCompactionUse { + metadata + .system_input + .as_ref() + .map(|input| input.usage().compaction) + .unwrap_or(session_store::InputCompactionUse::Retain) +} + fn long_summary(label: &str) -> String { format!( "## Completed Tasks\n{label}: investigated the transport boundary.\n\n\ @@ -2359,7 +2397,8 @@ async fn full_initial_summary_survives_small_overview_budgets_and_both_retention assert_eq!( replacement .iter() - .filter(|entry| entry.metadata.origin == CompactionSummary) + .filter(|entry| compaction_use(&entry.metadata) + == session_store::InputCompactionUse::OpeningSummary) .count(), 1 ); @@ -2604,7 +2643,9 @@ async fn three_chained_compactions_transport_each_complete_predecessor_without_r let seed = replacement_seed(&fixture.worker); let inherited = seed .iter() - .filter(|entry| entry.metadata.origin == CompactionSummary) + .filter(|entry| { + compaction_use(&entry.metadata) == session_store::InputCompactionUse::OpeningSummary + }) .collect::>(); assert_eq!(inherited.len(), 1); assert_eq!( @@ -2892,7 +2933,7 @@ impl worker::subjektiv::SubjektivHost for CompactionJournalHost { } fn journal_startup_granted(&self) -> bool { - true + false } fn journal_startup_context( @@ -2996,8 +3037,8 @@ fn compaction_snapshot(generation: usize) -> subjektiv::journal::startup::Journa fn journal_body(worker: &Worker) -> String { let seed = replacement_seed(worker); assert_eq!( - seed[1].metadata.origin, - session_store::LoggedSessionHistoryOrigin::JournalContext + compaction_use(&seed[1].metadata), + session_store::InputCompactionUse::ReferenceInput ); let item = Item::from(seed[1].item.clone()); assert!(matches!( @@ -3009,8 +3050,8 @@ fn journal_body(worker: &Worker) -> String { )); assert_eq!( seed.iter() - .filter(|entry| entry.metadata.origin - == session_store::LoggedSessionHistoryOrigin::JournalContext) + .filter(|entry| compaction_use(&entry.metadata) + == session_store::InputCompactionUse::ReferenceInput) .count(), 1 ); @@ -3119,20 +3160,34 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai 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, - }; + // Already migrated generic input with unrelated provenance/payload; Core + // must select it by explicit policy, not identify a Journal Feature. + let legacy_startup = session_store::LoggedSystemHistoryEntry::from_input( + DerivedSummary, + None, + session_store::SystemInputEnvelope::new( + "OLD_TYPED_STARTUP_JOURNAL", + session_store::SystemInputMetadata::new( + session_store::SystemInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: session_store::InputCompactionUse::ReferenceInput, + }, + Some( + session_store::FeatureInputPayload::encode( + "renamed-owner:reference", + 8, + &serde_json::json!({"private": "retained opaque context"}), + ) + .unwrap(), + ), + None, + ) + .unwrap(), + ) + .unwrap(), + ); let legacy_id = legacy_startup.metadata.entry_id.clone(); fixture .restore_seed_with_system( @@ -3166,7 +3221,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai .unwrap() .annotation .origin, - JournalContext + DerivedSummary ); let session = fixture.worker.session_id(); let original_segment = fixture.worker.segment_id(); @@ -3191,8 +3246,8 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai .unwrap() .annotation .origin, - JournalContext, - "legacy startup classification survives a normal turn" + DerivedSummary, + "generic source provenance survives a normal turn unchanged" ); } assert_eq!( @@ -3243,7 +3298,7 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai 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)") + .split("## Previous reference input (historical, not automatically retained)") .next() .unwrap(); for old in &old_bodies { @@ -3336,6 +3391,17 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai .run_text("normal post-compaction turn") .await .unwrap(); + let frozen_seed = replacement_seed(&fixture.worker) + .into_iter() + .filter(|entry| { + matches!( + compaction_use(&entry.metadata), + session_store::InputCompactionUse::ReferenceInput + | session_store::InputCompactionUse::OpeningSummary + ) + }) + .collect::>(); + assert_eq!(frozen_seed.len(), 2); let records = fixture .worker .store() @@ -3375,6 +3441,20 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai .run_text("normal turn after typed history restore") .await .unwrap(); + for frozen in frozen_seed { + let replayed = fixture + .worker + .session_history() + .entries() + .iter() + .find(|entry| entry.annotation.entry_id == frozen.metadata.entry_id) + .expect("frozen entry identity survives re-restore"); + assert_eq!(replayed.item, Item::from(frozen.item)); + assert_eq!( + replayed.annotation, frozen.metadata, + "origin, derivation, public allowlist, policy and private payload survive atomic seed/restore" + ); + } assert_eq!( host.reads.lock().unwrap().len(), 3, @@ -3445,27 +3525,41 @@ async fn journal_absent_disabled_unavailable_and_over_byte_budget_keep_conversat ) .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), - ); + let body = if matches!(case, "disabled" | "no-host") { + assert_eq!( + replacement_seed(&fixture.worker) + .iter() + .filter(|entry| compaction_use(&entry.metadata) + == session_store::InputCompactionUse::ReferenceInput) + .count(), + 0 + ); + assert!(!input.contains("Journal reference frozen at this compaction boundary")); + String::new() + } else { + let body = journal_body(&fixture.worker); + let expected = if case == "absent" { + "absent" + } else { + "unavailable" + }; + assert!( + body.contains(&format!("Journal read status: {expected}.")), + "{case}" + ); + assert_frozen_journal_transport(&input, &body); + assert_journal_provenance( + &body, + expected_json.as_deref(), + &fixture.worker.prompts().load_full(), + Some(COMPACTION_HOST_WORKSPACE_ID), + ); + body + }; 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!(input.contains("## Previous reference input (historical, not automatically retained)\nPREVIOUS_JOURNAL_DELEGATED_CONTEXT")); assert!( fixture.worker.history().iter().all(|item| !item .as_text() diff --git a/crates/worker/tests/feature_invocation_test.rs b/crates/worker/tests/feature_invocation_test.rs index a08cffda0..805f694ac 100644 --- a/crates/worker/tests/feature_invocation_test.rs +++ b/crates/worker/tests/feature_invocation_test.rs @@ -14,7 +14,7 @@ use protocol::{ InvocationArgumentDescriptor, InvocationArgumentType, InvocationArgumentValue, InvocationCompletion, InvocationValue, Segment, }; -use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry, SystemItem}; +use session_store::{CombinedStore, FsStore, FsWorkerStore, LogEntry}; use worker::feature::{ FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureInvocationContext, FeatureInvocationHandler, FeatureInvocationHandlerError, FeatureModule, FeatureRegistryBuilder, @@ -224,10 +224,20 @@ fn results(worker: &Worker) -> Vec { .0 .into_iter() .filter_map(|entry| match entry { - LogEntry::AnnotatedSystemItem { entry, .. } => match entry.item { - SystemItem::FeatureInvocationResult { result, .. } => Some(result), - _ => None, - }, + LogEntry::AnnotatedSystemItem { entry, .. } => { + #[derive(serde::Deserialize)] + struct SavedResult { + result: FeatureInvocationResult, + } + entry + .metadata + .system_input + .as_ref()? + .feature_payload()? + .decode::("feature:invocation-result", 1) + .ok() + .map(|input| input.result) + } _ => None, }) .collect() 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 c3aca207b..893720ccd 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 @@ -547,30 +547,40 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() let segments = session.list_segments(session_id).unwrap(); assert_eq!(segments.len(), 1); let entries = session.read_all_read_only(session_id, segments[0]).unwrap(); + #[derive(serde::Deserialize)] + struct SavedJournal { + snapshot: serde_json::Value, + snapshot_sha256: String, + prompt_provenance: session_store::PromptRenderProvenance, + } let startup: Vec<_> = entries .iter() .enumerate() - .filter_map(|(index, log)| match log { - session_store::LogEntry::AnnotatedSystemItem { entry, .. } => match &entry.item { - session_store::SystemItem::JournalStartup { - snapshot, - snapshot_sha256, - body, - prompt_provenance, - } => Some((index, snapshot, snapshot_sha256, body, prompt_provenance)), - _ => None, - }, - _ => None, + .filter_map(|(index, log)| { + let session_store::LogEntry::AnnotatedSystemItem { entry, .. } = log else { + return None; + }; + let payload = entry + .metadata + .system_input + .as_ref()? + .feature_payload()? + .decode::("journal:startup", 1) + .ok()?; + Some((index, payload, entry.item.history_text())) }) .collect(); assert_eq!(startup.len(), 1, "startup must be durably committed once"); - let (startup_index, snapshot, digest, body, provenance) = startup[0]; + let (startup_index, payload, body) = &startup[0]; + let snapshot = &payload.snapshot; + let digest = &payload.snapshot_sha256; + let provenance = &payload.prompt_provenance; let first_input = entries .iter() .position(|entry| matches!(entry, session_store::LogEntry::AnnotatedUserInput { .. })) .expect("new body persisted its initial human input"); assert!( - startup_index < first_input, + *startup_index < first_input, "startup reference must be persisted before the first human input/model request" ); assert_eq!(snapshot["subject_id"], subject.id); diff --git a/tools/state-contract-name-exceptions.json b/tools/state-contract-name-exceptions.json index 07fa26137..933d7a696 100644 --- a/tools/state-contract-name-exceptions.json +++ b/tools/state-contract-name-exceptions.json @@ -70,12 +70,12 @@ ] }, { - "path": "crates/session-store/src/system_item.rs", + "path": "crates/session-store/src/legacy_session_log/tests/v5.rs", "kind": "saved_data", - "reason": "frozen_subject_behavior_log_reads_old_counter_without_republishing_itの保存済みSystemItem fixture。旧JSONを読んでもexact archived bodyを維持することを証明する。削除すると過去Session log形式の読取回帰を検出できない。", + "reason": "旧Subject behavior JSONに保存されていたcounterの実fixture。v1–4専用migration境界で旧綴りを読み、本文を維持して現行payloadへcounterを再公開しないことを検証する。通常v5 decoderは旧Feature kind自体を拒否する。", "lines": [ { - "sha256": "5b5998029cfed30634e8b063b759aefd70e30dab3d02ce9defe9086977fc9811", + "sha256": "54c30cb3198eb16e57c7e580fce4dc058f89e793b77a2955be317aa38b490138", "count": 1 } ]