From 3b3381900f9141d401be6312183df05f975e6c5d Mon Sep 17 00:00:00 2001 From: Hare Date: Tue, 20 Oct 2026 04:45:50 +0900 Subject: [PATCH] fix: preserve role-neutral input policy across compaction and replay --- crates/session-store/src/fs_store.rs | 4 +- crates/session-store/src/history.rs | 92 +++++---- .../{system_input.rs => input_metadata.rs} | 115 ++++++------ .../session-store/src/legacy_session_log.rs | 116 +++++++++--- .../src/legacy_session_log/tests.rs | 2 +- .../tests/fixtures/v5-role-metadata.jsonl | 5 + .../src/legacy_session_log/tests/v5.rs | 137 ++++++++++++-- crates/session-store/src/lib.rs | 10 +- crates/session-store/src/public_index.rs | 4 +- crates/session-store/src/public_material.rs | 11 +- .../src/public_material/tests.rs | 32 ++-- crates/session-store/src/public_snapshot.rs | 35 ++-- crates/session-store/src/segment_log.rs | 10 +- crates/session-store/src/store.rs | 4 +- .../session-store/src/worker_session_store.rs | 29 +-- crates/session-store/tests/fs_store_test.rs | 17 +- crates/session-store/tests/session_test.rs | 2 +- .../standalone/src/subjektiv/journal/tests.rs | 2 +- .../src/subjektiv/journal/tests/compaction.rs | 27 +++ .../journal/tests/connected_compaction.rs | 6 +- .../src/subjektiv/journal/tests/handoff.rs | 2 +- crates/standalone/src/subjektiv_sessions.rs | 2 +- crates/tui/src/app.rs | 2 +- .../tests/session_material_tests.rs | 2 +- crates/worker-runtime/src/retention.rs | 4 +- .../src/retention/runtime_v5_tests.rs | 77 +++++++- crates/worker-runtime/src/session_public.rs | 2 +- crates/worker-runtime/src/worker_backend.rs | 2 +- crates/worker/src/feature/builtin/journal.rs | 4 +- crates/worker/src/feature/builtin/memory.rs | 12 +- crates/worker/src/feature/invocation.rs | 6 +- crates/worker/src/internal_worker.rs | 2 +- crates/worker/src/ipc/interceptor.rs | 2 +- crates/worker/src/session_capture.rs | 176 ++++++++++++++++-- crates/worker/src/session_history.rs | 10 +- crates/worker/src/skill.rs | 6 +- crates/worker/src/spawn/registry.rs | 2 +- crates/worker/src/subjektiv.rs | 6 +- crates/worker/src/wip.rs | 2 +- crates/worker/src/worker.rs | 51 +++-- crates/worker/tests/compact_events_test.rs | 14 +- .../worker/tests/feature_invocation_test.rs | 2 +- crates/workspace-server/src/server.rs | 38 ++-- .../subjektiv_journal_connected_tests.rs | 2 +- .../server/tests/subjektiv_journal_tests.rs | 4 +- .../tests/subjektiv_memory_journal_tests.rs | 2 +- 46 files changed, 779 insertions(+), 315 deletions(-) rename crates/session-store/src/{system_input.rs => input_metadata.rs} (82%) create mode 100644 crates/session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl diff --git a/crates/session-store/src/fs_store.rs b/crates/session-store/src/fs_store.rs index aa9988e8f..26226a5fc 100644 --- a/crates/session-store/src/fs_store.rs +++ b/crates/session-store/src/fs_store.rs @@ -361,7 +361,7 @@ impl Store for FsStore { segment_id: SegmentId, entry: &LogEntry, ) -> Result<(), StoreError> { - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; let _retention = self.retention().lock_live(session_id)?; let line = serde_json::to_string(entry)?; self.append_line(&self.log_path(session_id, segment_id), &line) @@ -450,7 +450,7 @@ impl Store for FsStore { entries: &[LogEntry], ) -> Result<(), StoreError> { for entry in entries { - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; } let _retention = self.retention().lock_live(session_id)?; let path = self.log_path(session_id, segment_id); diff --git a/crates/session-store/src/history.rs b/crates/session-store/src/history.rs index 67d753c13..4170cdf60 100644 --- a/crates/session-store/src/history.rs +++ b/crates/session-store/src/history.rs @@ -85,21 +85,22 @@ pub struct LoggedHistoryDerivation { } #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct LoggedSessionHistoryMetadata { pub entry_id: LoggedSessionHistoryEntryId, pub origin: LoggedSessionHistoryOrigin, #[serde(default, skip_serializing_if = "Option::is_none")] pub derivation: Option, #[serde(default, skip_serializing_if = "Option::is_none")] - pub system_input: Option, + pub input_metadata: Option, } impl LoggedSessionHistoryMetadata { - pub(crate) fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { + pub(crate) fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> { if matches!(self.origin, LoggedSessionHistoryOrigin::Provided { .. }) - && self.system_input.is_none() + && self.input_metadata.is_none() { - return Err(crate::SystemInputError::MissingMetadata); + return Err(crate::InputMetadataError::MissingMetadata); } Ok(()) } @@ -109,7 +110,7 @@ impl LoggedSessionHistoryMetadata { entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, - system_input: None, + input_metadata: None, } } } @@ -131,34 +132,40 @@ struct HistoryEntryWire { } impl TryFrom for LoggedHistoryEntry { - type Error = crate::SystemInputError; + type Error = crate::InputMetadataError; fn try_from(wire: HistoryEntryWire) -> Result { let entry = Self { item: wire.item, metadata: wire.metadata, }; - entry.validate_system_input()?; + entry.validate_input_metadata()?; Ok(entry) } } impl LoggedHistoryEntry { - pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { - self.metadata.validate_system_input()?; - if self.metadata.system_input.is_some() { - let LoggedItem::Message { - role: crate::LoggedRole::System, - content, - } = &self.item - else { - return Err(crate::SystemInputError::InconsistentRecord); - }; - let [crate::LoggedContentPart::Text { text }] = content.as_slice() else { - return Err(crate::SystemInputError::InconsistentRecord); - }; - if text.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES { - return Err(crate::SystemInputError::ResourceLimit); + pub fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> { + self.metadata.validate_input_metadata()?; + if let Some(input) = &self.metadata.input_metadata { + match &self.item { + LoggedItem::Message { + role: crate::LoggedRole::System, + content, + } => { + let [crate::LoggedContentPart::Text { text }] = content.as_slice() else { + return Err(crate::InputMetadataError::InconsistentRecord); + }; + if text.len() > crate::input_metadata::MAX_SYSTEM_INPUT_BYTES { + return Err(crate::InputMetadataError::ResourceLimit); + } + } + // Other logical roles may carry use policy, but cannot masquerade + // as the System-only public/payload extension contract. + _ if input.feature_payload().is_some() || input.public_item().is_some() => { + return Err(crate::InputMetadataError::InconsistentRecord); + } + _ => {} } } Ok(()) @@ -183,14 +190,14 @@ struct SystemEntryWire { } impl TryFrom for LoggedSystemHistoryEntry { - type Error = crate::SystemInputError; + type Error = crate::InputMetadataError; fn try_from(wire: SystemEntryWire) -> Result { let entry = Self { item: wire.item, metadata: wire.metadata, }; - entry.validate_system_input()?; + entry.validate_input_metadata()?; Ok(entry) } } @@ -201,30 +208,30 @@ impl LoggedSystemHistoryEntry { derivation: Option, input: crate::SystemInputEnvelope, ) -> Self { - let (body, system_input) = input.into_parts(); + let (body, input_metadata) = input.into_parts(); Self { item: crate::SystemItem::Input { body }, metadata: LoggedSessionHistoryMetadata { entry_id: LoggedSessionHistoryEntryId::new(), origin, derivation, - system_input: Some(system_input), + input_metadata: Some(input_metadata), }, } } - pub fn validate_system_input(&self) -> Result<(), crate::SystemInputError> { - self.metadata.validate_system_input()?; - match (&self.item, &self.metadata.system_input) { + pub fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> { + self.metadata.validate_input_metadata()?; + match (&self.item, &self.metadata.input_metadata) { (crate::SystemItem::Input { body }, Some(_)) => { - if body.len() > crate::system_input::MAX_SYSTEM_INPUT_BYTES { - return Err(crate::SystemInputError::ResourceLimit); + if body.len() > crate::input_metadata::MAX_SYSTEM_INPUT_BYTES { + return Err(crate::InputMetadataError::ResourceLimit); } } (crate::SystemItem::Input { .. }, None) => { - return Err(crate::SystemInputError::MissingMetadata); + return Err(crate::InputMetadataError::MissingMetadata); } - (_, Some(_)) => return Err(crate::SystemInputError::InconsistentRecord), + (_, Some(_)) => return Err(crate::InputMetadataError::InconsistentRecord), (_, None) => {} } Ok(()) @@ -241,9 +248,9 @@ impl LoggedSystemHistoryEntry { if matches!(self.item, crate::SystemItem::Input { .. }) { return self .metadata - .system_input + .input_metadata .as_ref() - .and_then(crate::SystemInputMetadata::public_item) + .and_then(crate::HistoryInputMetadata::public_item) .cloned() .unwrap_or_else(|| self.item.public_item()); } @@ -268,7 +275,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::HumanInput { account_id: "account-1".into(), @@ -293,7 +300,7 @@ mod tests { "item": { "kind": "input", "body": "context" }, "metadata": { "entry_id": "stable-entry", "origin": { "kind": "provided", "producer": "owner" }, - "system_input": { "usage": { + "input_metadata": { "usage": { "recording_material": "exclude", "memory_evidence": "exclude", "observation": "hidden", "compaction": "reference_input" } } @@ -315,10 +322,10 @@ mod tests { missing["metadata"] .as_object_mut() .unwrap() - .remove("system_input"); + .remove("input_metadata"); assert!(decode(missing)); let mut absent_policy = wire.clone(); - absent_policy["metadata"]["system_input"] + absent_policy["metadata"]["input_metadata"] .as_object_mut() .unwrap() .remove("usage"); @@ -327,15 +334,18 @@ mod tests { invalid_role["item"] = serde_json::to_value(LoggedItem::from(agen::Item::user_message("not system"))) .unwrap(); + invalid_role["metadata"]["input_metadata"]["public"] = serde_json::json!({ + "kind": "context", "body": "public system projection", "presentation": {"kind": "context"} + }); assert!(decode(invalid_role)); let mut oversized = wire.clone(); if generic { oversized["item"]["body"] = "x" - .repeat(crate::system_input::MAX_SYSTEM_INPUT_BYTES + 1) + .repeat(crate::input_metadata::MAX_SYSTEM_INPUT_BYTES + 1) .into(); } else { oversized["item"]["content"][0]["text"] = "x" - .repeat(crate::system_input::MAX_SYSTEM_INPUT_BYTES + 1) + .repeat(crate::input_metadata::MAX_SYSTEM_INPUT_BYTES + 1) .into(); } assert!(decode(oversized)); diff --git a/crates/session-store/src/system_input.rs b/crates/session-store/src/input_metadata.rs similarity index 82% rename from crates/session-store/src/system_input.rs rename to crates/session-store/src/input_metadata.rs index 3595c7bdb..2d29d07da 100644 --- a/crates/session-store/src/system_input.rs +++ b/crates/session-store/src/input_metadata.rs @@ -1,4 +1,4 @@ -//! Validated, Feature-neutral system-input data. +//! Validated, Feature-neutral history-input policy and System extensions. //! //! The enclosing history annotation owns identity, origin and derivation. None of //! those descriptive facts selects a use policy here. Model text, independent use @@ -23,22 +23,22 @@ const MAX_JSON_DEPTH: usize = 16; const MAX_JSON_NODES: usize = 8192; #[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)] -pub enum SystemInputError { - #[error("system input payload schema is invalid")] +pub enum InputMetadataError { + #[error("history input payload schema is invalid")] InvalidSchema, - #[error("system input payload version must be nonzero")] + #[error("history input payload version must be nonzero")] InvalidVersion, #[error("system input resource limit exceeded")] ResourceLimit, - #[error("system input payload could not be serialized")] + #[error("history input payload could not be serialized")] Serialization, - #[error("system input payload schema or version is unsupported by this consumer")] + #[error("history input payload schema or version is unsupported by this consumer")] UnsupportedPayload, - #[error("system input payload does not match the owner's typed contract")] + #[error("history input payload does not match the owner's typed contract")] InvalidTypedPayload, - #[error("generic system input is missing explicit metadata")] + #[error("provided input is missing explicit metadata")] MissingMetadata, - #[error("system input metadata does not match its logical record")] + #[error("input metadata does not match its logical record")] InconsistentRecord, } @@ -70,7 +70,7 @@ struct PayloadWire { } impl TryFrom for FeatureInputPayload { - type Error = SystemInputError; + type Error = InputMetadataError; fn try_from(wire: PayloadWire) -> Result { if wire.schema.is_empty() @@ -80,10 +80,10 @@ impl TryFrom for FeatureInputPayload { byte.is_ascii_lowercase() || byte.is_ascii_digit() || b"-_.:/".contains(&byte) }) { - return Err(SystemInputError::InvalidSchema); + return Err(InputMetadataError::InvalidSchema); } if wire.version == 0 { - return Err(SystemInputError::InvalidVersion); + return Err(InputMetadataError::InvalidVersion); } validate_json(&wire.value, MAX_FEATURE_PAYLOAD_BYTES)?; Ok(Self { @@ -99,11 +99,11 @@ impl FeatureInputPayload { schema: impl Into, version: u32, value: &T, - ) -> Result { + ) -> Result { PayloadWire { schema: schema.into(), version, - value: serde_json::to_value(value).map_err(|_| SystemInputError::Serialization)?, + value: serde_json::to_value(value).map_err(|_| InputMetadataError::Serialization)?, } .try_into() } @@ -122,12 +122,12 @@ impl FeatureInputPayload { &self, schema: &str, version: u32, - ) -> Result { + ) -> Result { if self.schema != schema || self.version != version { - return Err(SystemInputError::UnsupportedPayload); + return Err(InputMetadataError::UnsupportedPayload); } serde_json::from_value(self.value.clone()) - .map_err(|_| SystemInputError::InvalidTypedPayload) + .map_err(|_| InputMetadataError::InvalidTypedPayload) } } @@ -161,17 +161,20 @@ pub enum InputCompactionUse { /// Public presentation is deliberately not one of these use decisions. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] -pub struct SystemInputUsePolicy { +pub struct HistoryInputUsePolicy { pub recording_material: InputEvidenceUse, pub memory_evidence: InputEvidenceUse, pub observation: InputObservationUse, pub compaction: InputCompactionUse, } +/// Use policy is independent of logical role. Non-System derived Tool context +/// carries policy only; private Feature payload and public presentation belong +/// to the System envelope and are rejected on other roles at the history boundary. #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] #[serde(try_from = "MetadataWire")] -pub struct SystemInputMetadata { - usage: SystemInputUsePolicy, +pub struct HistoryInputMetadata { + usage: HistoryInputUsePolicy, feature_payload: Option, public: Option, } @@ -179,13 +182,13 @@ pub struct SystemInputMetadata { #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct MetadataWire { - usage: SystemInputUsePolicy, + usage: HistoryInputUsePolicy, feature_payload: Option, public: Option, } -impl TryFrom for SystemInputMetadata { - type Error = SystemInputError; +impl TryFrom for HistoryInputMetadata { + type Error = InputMetadataError; fn try_from(wire: MetadataWire) -> Result { if let Some(public) = &wire.public { @@ -194,10 +197,10 @@ impl TryFrom for SystemInputMetadata { || public.kind.len() > MAX_SCHEMA_BYTES || public.kind.chars().any(char::is_control) { - return Err(SystemInputError::InvalidSchema); + return Err(InputMetadataError::InvalidSchema); } let value = - serde_json::to_value(public).map_err(|_| SystemInputError::Serialization)?; + serde_json::to_value(public).map_err(|_| InputMetadataError::Serialization)?; validate_json(&value, MAX_SYSTEM_PUBLIC_BYTES)?; } Ok(Self { @@ -208,12 +211,12 @@ impl TryFrom for SystemInputMetadata { } } -impl SystemInputMetadata { +impl HistoryInputMetadata { pub fn new( - usage: SystemInputUsePolicy, + usage: HistoryInputUsePolicy, feature_payload: Option, public: Option, - ) -> Result { + ) -> Result { MetadataWire { usage, feature_payload, @@ -222,7 +225,7 @@ impl SystemInputMetadata { .try_into() } - pub fn usage(&self) -> SystemInputUsePolicy { + pub fn usage(&self) -> HistoryInputUsePolicy { self.usage } @@ -258,22 +261,22 @@ impl SystemInputMetadata { #[serde(try_from = "InputWire")] pub struct SystemInputEnvelope { body: String, - metadata: SystemInputMetadata, + metadata: HistoryInputMetadata, } #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct InputWire { body: String, - metadata: SystemInputMetadata, + metadata: HistoryInputMetadata, } impl TryFrom for SystemInputEnvelope { - type Error = SystemInputError; + type Error = InputMetadataError; fn try_from(wire: InputWire) -> Result { if wire.body.len() > MAX_SYSTEM_INPUT_BYTES { - return Err(SystemInputError::ResourceLimit); + return Err(InputMetadataError::ResourceLimit); } Ok(Self { body: wire.body, @@ -285,8 +288,8 @@ impl TryFrom for SystemInputEnvelope { impl SystemInputEnvelope { pub fn new( body: impl Into, - metadata: SystemInputMetadata, - ) -> Result { + metadata: HistoryInputMetadata, + ) -> Result { InputWire { body: body.into(), metadata, @@ -298,7 +301,7 @@ impl SystemInputEnvelope { &self.body } - pub fn metadata(&self) -> &SystemInputMetadata { + pub fn metadata(&self) -> &HistoryInputMetadata { &self.metadata } @@ -306,15 +309,15 @@ impl SystemInputEnvelope { Item::system_message(self.body.clone()) } - pub fn into_parts(self) -> (String, SystemInputMetadata) { + pub fn into_parts(self) -> (String, HistoryInputMetadata) { (self.body, self.metadata) } } -fn validate_json(value: &Value, max_bytes: usize) -> Result<(), SystemInputError> { - fn walk(value: &Value, depth: usize, remaining: &mut usize) -> Result<(), SystemInputError> { +fn validate_json(value: &Value, max_bytes: usize) -> Result<(), InputMetadataError> { + fn walk(value: &Value, depth: usize, remaining: &mut usize) -> Result<(), InputMetadataError> { if depth > MAX_JSON_DEPTH || *remaining == 0 { - return Err(SystemInputError::ResourceLimit); + return Err(InputMetadataError::ResourceLimit); } *remaining -= 1; match value { @@ -334,9 +337,9 @@ fn validate_json(value: &Value, max_bytes: usize) -> Result<(), SystemInputError } let mut remaining = MAX_JSON_NODES; walk(value, 0, &mut remaining)?; - let bytes = serde_json::to_vec(value).map_err(|_| SystemInputError::Serialization)?; + let bytes = serde_json::to_vec(value).map_err(|_| InputMetadataError::Serialization)?; if bytes.len() > max_bytes { - return Err(SystemInputError::ResourceLimit); + return Err(InputMetadataError::ResourceLimit); } Ok(()) } @@ -353,8 +356,8 @@ mod tests { omitted: bool, } - fn reference_policy() -> SystemInputUsePolicy { - SystemInputUsePolicy { + fn reference_policy() -> HistoryInputUsePolicy { + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, @@ -375,7 +378,7 @@ mod tests { .unwrap(); let envelope = SystemInputEnvelope::new( "private model context", - SystemInputMetadata::new(reference_policy(), Some(payload), None).unwrap(), + HistoryInputMetadata::new(reference_policy(), Some(payload), None).unwrap(), ) .unwrap(); let restored: SystemInputEnvelope = @@ -387,7 +390,7 @@ mod tests { assert_eq!(saved_payload.version(), 99); assert_eq!( saved_payload.decode::("known-owner:context", 1), - Err(SystemInputError::UnsupportedPayload) + Err(InputMetadataError::UnsupportedPayload) ); assert_eq!( serde_json::to_value(&restored).unwrap(), @@ -415,7 +418,7 @@ mod tests { .unwrap(); assert_eq!( invalid.decode::("owner:context", 1), - Err(SystemInputError::InvalidTypedPayload) + Err(InputMetadataError::InvalidTypedPayload) ); } @@ -462,14 +465,14 @@ mod tests { ] { let mut absent = valid.clone(); absent.as_object_mut().unwrap().remove(field); - assert!(serde_json::from_value::(absent).is_err()); + assert!(serde_json::from_value::(absent).is_err()); let mut unknown = valid.clone(); unknown[field] = Value::String("future-allow".into()); - assert!(serde_json::from_value::(unknown).is_err()); + assert!(serde_json::from_value::(unknown).is_err()); } let mut extra = valid; extra["feature_name"] = Value::String("infer-policy-from-me".into()); - assert!(serde_json::from_value::(extra).is_err()); + assert!(serde_json::from_value::(extra).is_err()); } #[test] @@ -483,7 +486,7 @@ mod tests { body: "explicit public text".into(), presentation: PublicSystemPresentation::Context, }; - let metadata = SystemInputMetadata::new( + let metadata = HistoryInputMetadata::new( reference_policy(), Some( FeatureInputPayload::encode( @@ -519,14 +522,14 @@ mod tests { #[test] fn use_policies_remain_independent_of_each_other_and_of_public_display() { - let usage = SystemInputUsePolicy { + let usage = HistoryInputUsePolicy { recording_material: InputEvidenceUse::Include, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Readable, compaction: InputCompactionUse::Retain, }; - let metadata = SystemInputMetadata::new(usage, None, None).unwrap(); - let restored: SystemInputMetadata = + let metadata = HistoryInputMetadata::new(usage, None, None).unwrap(); + let restored: HistoryInputMetadata = serde_json::from_value(serde_json::to_value(&metadata).unwrap()).unwrap(); assert_eq!(restored.usage(), usage); assert!(restored.public_item().is_none()); @@ -535,7 +538,7 @@ mod tests { #[test] fn model_body_and_public_projection_have_independent_resource_limits() { - let metadata = SystemInputMetadata::new(reference_policy(), None, None).unwrap(); + let metadata = HistoryInputMetadata::new(reference_policy(), None, None).unwrap(); assert!( SystemInputEnvelope::new("x".repeat(MAX_SYSTEM_INPUT_BYTES + 1), metadata).is_err() ); @@ -544,7 +547,7 @@ mod tests { body: "x".repeat(MAX_SYSTEM_PUBLIC_BYTES), presentation: PublicSystemPresentation::Context, }; - assert!(SystemInputMetadata::new(reference_policy(), None, Some(public)).is_err()); + assert!(HistoryInputMetadata::new(reference_policy(), None, Some(public)).is_err()); let invalid = serde_json::json!({ "body": "small", "metadata": { "public": null, "feature_payload": null } }); assert!(serde_json::from_value::(invalid).is_err()); } diff --git a/crates/session-store/src/legacy_session_log.rs b/crates/session-store/src/legacy_session_log.rs index 10e247539..dc91caf07 100644 --- a/crates/session-store/src/legacy_session_log.rs +++ b/crates/session-store/src/legacy_session_log.rs @@ -1,4 +1,4 @@ -//! Versioned decoder for Session schemas v1-4, including frozen Feature wire. +//! Versioned decoder for Session schemas v1-5, including frozen Feature wire. //! //! These types are intentionally private to `session-store`. Current writers, //! replay, and public projections use [`crate::LogEntry`] exclusively; only the @@ -93,14 +93,14 @@ impl LegacySystemItem { fn into_entry( self, mut metadata: LoggedSessionHistoryMetadata, - ) -> Result { + ) -> 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); + metadata.validate_input_metadata()?; + if metadata.input_metadata.is_some() { + return Err(crate::InputMetadataError::InconsistentRecord); } let (body, input_metadata) = match feature { LegacyFeatureSystemItem::JournalStartup { body, payload } => { @@ -150,12 +150,12 @@ impl LegacySystemItem { } }; // Identity/origin/derivation are retained, never re-inferred. - metadata.system_input = Some(input_metadata); + metadata.input_metadata = Some(input_metadata); SystemItem::Input { body } } }; let entry = LoggedSystemHistoryEntry { item, metadata }; - entry.validate_system_input()?; + entry.validate_input_metadata()?; Ok(entry) } } @@ -166,8 +166,8 @@ fn migrated_feature_input( schema: &str, payload: T, compaction: crate::InputCompactionUse, -) -> Result<(String, crate::SystemInputMetadata), crate::SystemInputError> { - let metadata = crate::SystemInputMetadata::new( +) -> Result<(String, crate::HistoryInputMetadata), crate::InputMetadataError> { + let metadata = crate::HistoryInputMetadata::new( migration_policy(compaction), Some(crate::FeatureInputPayload::encode(schema, 1, &payload)?), Some(crate::PublicSystemItem { @@ -207,7 +207,7 @@ enum LegacyAnnotatedOrCurrent { } impl LegacyAnnotatedOrCurrent { - fn into_current(self) -> Result { + fn into_current(self) -> Result { match self { Self::Annotated(LegacyAnnotatedLogEntry::AnnotatedSystemItem { ts, @@ -285,16 +285,19 @@ pub(crate) fn decode_entry( segment_id: SegmentId, line_index: usize, ) -> Result { + let mut wire: serde_json::Value = serde_json::from_str(line)?; + migrate_legacy_metadata_field(&mut wire)?; let entry = match schema_version { - 1 => match serde_json::from_str::(line)? { + 1 => match serde_json::from_value::(wire)? { LegacySessionLogEntryV1::History(entry) => Entry::History(entry), LegacySessionLogEntryV1::Current(entry) => Entry::Current(entry), }, - 2 => match serde_json::from_str::(line)? { + 2 => match serde_json::from_value::(wire)? { LegacySessionLogEntryV2::History(entry) => Entry::History(entry), LegacySessionLogEntryV2::Current(entry) => Entry::Current(entry), }, - 3 | 4 => Entry::Current(serde_json::from_str(line)?), + 3 | 4 => Entry::Current(serde_json::from_value(wire)?), + 5 => return serde_json::from_value(wire), _ => unreachable!("legacy decoder called for unsupported schema {schema_version}"), }; match entry { @@ -306,6 +309,53 @@ pub(crate) fn decode_entry( .map_err(serde::de::Error::custom) } +// The v5 System-only metadata field was widened to role-neutral input metadata +// in v6. Touch only known history slots, never arbitrary Feature payload objects. +// Both keys are an ambiguous source, not permission to discard either value. +fn migrate_legacy_metadata_field(record: &mut serde_json::Value) -> Result<(), serde_json::Error> { + fn entry(value: &mut serde_json::Value) -> Result<(), serde_json::Error> { + if let Some(metadata) = value + .get_mut("metadata") + .and_then(serde_json::Value::as_object_mut) + { + if let Some(old) = metadata.remove("system_input") { + if metadata.contains_key("input_metadata") { + return Err(serde::de::Error::custom( + "ambiguous legacy input metadata fields", + )); + } + metadata.insert("input_metadata".into(), old); + } + } + Ok(()) + } + let kind = record + .get("kind") + .and_then(serde_json::Value::as_str) + .unwrap_or_default() + .to_owned(); + if matches!( + kind.as_str(), + "annotated_system_item" + | "annotated_assistant_item" + | "annotated_tool_result" + | "tool_result_corrected" + ) { + if let Some(value) = record.get_mut("entry") { + entry(value)?; + } + } + if let Some(history) = record + .get_mut("history") + .and_then(serde_json::Value::as_array_mut) + { + for value in history { + entry(value)?; + } + } + Ok(()) +} + /// Schema 3 used DerivedSummary for Host compaction seeds. Only the trusted /// creation boundary may identify one; arbitrary text and later seed items do not. /// Include already-upgraded entries so an interrupted multi-file migration resumes. @@ -361,15 +411,15 @@ pub(crate) fn migrate_compaction_origins( } } -/// Explicit v1-4 -> v5 boundary. No Feature inference belongs in ordinary +/// Explicit legacy-to-current 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()?; +pub(crate) fn migrate_system_inputs(entry: &mut LogEntry) -> Result<(), crate::InputMetadataError> { + entry.validate_input_metadata()?; 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::AnnotatedSystemItem { entry, .. } => entry.validate_input_metadata()?, LogEntry::AnnotatedSegmentStart { history, .. } | LogEntry::AnnotatedUserInput { history, .. } => { for entry in history { @@ -383,11 +433,11 @@ pub(crate) fn migrate_system_inputs(entry: &mut LogEntry) -> Result<(), crate::S } _ => {} } - entry.validate_system_inputs() + entry.validate_input_metadata() } -fn migration_policy(compaction: crate::InputCompactionUse) -> crate::SystemInputUsePolicy { - crate::SystemInputUsePolicy { +fn migration_policy(compaction: crate::InputCompactionUse) -> crate::HistoryInputUsePolicy { + crate::HistoryInputUsePolicy { recording_material: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude, observation: crate::InputObservationUse::Hidden, @@ -397,18 +447,22 @@ fn migration_policy(compaction: crate::InputCompactionUse) -> crate::SystemInput fn migrate_flattened_system_input( entry: &mut LoggedHistoryEntry, -) -> Result<(), crate::SystemInputError> { +) -> Result<(), crate::InputMetadataError> { // 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.validate_input_metadata()?; + if entry.metadata.input_metadata.is_none() + && (matches!( entry.item, LoggedItem::Message { role: crate::LoggedRole::System, .. } - ) + ) || matches!( + entry.metadata.origin, + LoggedSessionHistoryOrigin::DerivedSummary + | LoggedSessionHistoryOrigin::CompactionSummary + )) { let compaction = match entry.metadata.origin { LoggedSessionHistoryOrigin::JournalContext => crate::InputCompactionUse::ReferenceInput, @@ -417,14 +471,16 @@ fn migrate_flattened_system_input( } _ => crate::InputCompactionUse::Retain, }; - // No Feature payload or public projection can be recovered from text. - entry.metadata.system_input = Some(crate::SystemInputMetadata::new( + // Legacy derived context was excluded from independent evidence. Preserve + // that established decision explicitly, including synthetic Tool roles; + // never guess from Tool names or body text. Payload/public stay absent. + entry.metadata.input_metadata = Some(crate::HistoryInputMetadata::new( migration_policy(compaction), None, None, )?); } - entry.validate_system_input() + entry.validate_input_metadata() } enum Entry { @@ -442,7 +498,7 @@ fn legacy_metadata( identity.extend_from_slice(&(line_index as u64).to_be_bytes()); identity.extend_from_slice(&(item_index as u64).to_be_bytes()); LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(format!( "l-{}", base64::Engine::encode(&base64::engine::general_purpose::URL_SAFE_NO_PAD, identity) @@ -457,7 +513,7 @@ fn canonicalize_history_entry( segment_id: SegmentId, line_index: usize, entry: LegacyHistoryLogEntry, -) -> Result { +) -> Result { Ok(match entry { LegacyHistoryLogEntry::SegmentStart { ts, diff --git a/crates/session-store/src/legacy_session_log/tests.rs b/crates/session-store/src/legacy_session_log/tests.rs index d7914c194..e879d0ab1 100644 --- a/crates/session-store/src/legacy_session_log/tests.rs +++ b/crates/session-store/src/legacy_session_log/tests.rs @@ -24,7 +24,7 @@ fn seed( content: vec![crate::LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("summary".into()), origin, derivation: Some(crate::LoggedHistoryDerivation { diff --git a/crates/session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl b/crates/session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl new file mode 100644 index 000000000..533bf518d --- /dev/null +++ b/crates/session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl @@ -0,0 +1,5 @@ +{"kind":"annotated_system_item","ts":201,"entry":{"item":{"kind":"input","body":"frozen v5 reference"},"metadata":{"entry_id":"reference-v5","origin":{"kind":"provided","producer":"owner-v5"},"system_input":{"usage":{"recording_material":"exclude","memory_evidence":"exclude","observation":"hidden","compaction":"reference_input"},"feature_payload":{"schema":"future:owner","version":8,"value":{"system_input":"opaque owner field must not be renamed"}},"public":null}}}} +{"kind":"annotated_assistant_item","ts":202,"entry":{"item":{"kind":"tool_call","call_id":"original-call","name":"ArbitraryTool","arguments":"{}"},"metadata":{"entry_id":"original-call-entry","origin":{"kind":"model_output","worker":{"worker_id":"body"}}}}} +{"kind":"annotated_tool_result","ts":203,"entry":{"item":{"kind":"tool_result","call_id":"original-call","summary":"original result"},"metadata":{"entry_id":"original-result-entry","origin":{"kind":"tool_output","worker":{"worker_id":"body"}}}}} +{"kind":"annotated_assistant_item","ts":204,"entry":{"item":{"kind":"tool_call","call_id":"derived-call","name":"ArbitraryTool","arguments":"{}"},"metadata":{"entry_id":"derived-call-entry","origin":{"kind":"derived_summary"},"derivation":{"sources":["original-call-entry","original-result-entry"]}}}} +{"kind":"annotated_tool_result","ts":205,"entry":{"item":{"kind":"tool_result","call_id":"derived-call","summary":"derived retrieval result"},"metadata":{"entry_id":"derived-result-entry","origin":{"kind":"derived_summary"},"derivation":{"sources":["original-call-entry","original-result-entry"]}}}} diff --git a/crates/session-store/src/legacy_session_log/tests/v5.rs b/crates/session-store/src/legacy_session_log/tests/v5.rs index f197eeec9..847b81bf0 100644 --- a/crates/session-store/src/legacy_session_log/tests/v5.rs +++ b/crates/session-store/src/legacy_session_log/tests/v5.rs @@ -27,7 +27,7 @@ fn encode(entries: &[LogEntry]) -> Vec { fn migrate(entries: &mut [LogEntry]) { for entry in entries { migrate_system_inputs(entry).unwrap(); - entry.validate_system_inputs().unwrap(); + entry.validate_input_metadata().unwrap(); } } @@ -100,7 +100,7 @@ fn v4_fixed_wire_preserves_identity_body_payload_public_and_independent_policy() 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 metadata = entry.metadata.input_metadata.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()); @@ -182,7 +182,7 @@ fn v4_fixed_wire_preserves_identity_body_payload_public_and_independent_policy() history[index].metadata.derivation, old[index].metadata.derivation ); - let metadata = history[index].metadata.system_input.as_ref().unwrap(); + let metadata = history[index].metadata.input_metadata.as_ref().unwrap(); assert_eq!(metadata.usage(), migration_policy(compaction)); assert!(metadata.public_item().is_none()); assert!(metadata.feature_payload().is_none()); @@ -225,7 +225,7 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() { fs::write(&old_path, source).unwrap(); let store = WorkerSessionStore::new(root.path()).unwrap(); - assert_marker(root.path(), 5); + assert_marker(root.path(), 6); let restored = store.read_all(session, segment).unwrap(); assert_eq!(encode(&restored), new_bytes); assert_eq!(fs::read(&new_path).unwrap(), new_bytes); @@ -243,7 +243,7 @@ fn v1_through_v4_worker_migration_reopen_and_mixed_file_retry() { 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)}); + json!({"large": "x".repeat(crate::input_metadata::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(); @@ -295,7 +295,7 @@ fn all_segments_preflight_keeps_old_bytes_on_decode_or_conversion_failure() { #[test] fn unsupported_schema_preserves_all_old_bytes() { - for version in [0, 6, 99] { + for version in [0, 7, 99] { let root = tempfile::tempdir().unwrap(); let (_, segment) = write_worker(root.path(), version, V4.as_bytes()); let path = root @@ -331,13 +331,13 @@ fn standalone_store_uses_same_v1_through_v4_migration_and_reopens() { let store = FsStore::new(root.path()).unwrap(); let entries = store.read_all(session, segment).unwrap(); for entry in entries { - entry.validate_system_inputs().unwrap(); + entry.validate_input_metadata().unwrap(); } assert_eq!( fs::read_to_string(dir.join("session-schema.json")) .unwrap() .trim(), - "5" + "6" ); let once = fs::read(&path).unwrap(); drop(store); @@ -381,7 +381,7 @@ fn compaction_origin_upgrade_is_retained_only_for_pre_v4_sources() { } ); assert_eq!( - metadata.system_input.as_ref().unwrap().usage().compaction, + metadata.input_metadata.as_ref().unwrap().usage().compaction, if version == 3 { InputCompactionUse::OpeningSummary } else { @@ -423,7 +423,7 @@ fn current_v5_rejects_all_legacy_feature_kinds_without_rewriting_sources() { 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 (session, segment) = write_worker(root.path(), 6, bytes.as_bytes()); let path = root .path() .join("segments") @@ -495,7 +495,7 @@ fn v1_v2_feature_wire_uses_the_same_typed_payloads_and_stable_legacy_identity() else { panic!() }; - assert_eq!(entry.metadata.system_input, v4.metadata.system_input); + assert_eq!(entry.metadata.input_metadata, v4.metadata.input_metadata); } } } @@ -555,7 +555,7 @@ fn frozen_behavior_counter_is_accepted_only_at_the_old_schema_boundary() { assert_eq!(entry.item.history_text(), "exact archived body"); let payload = entry .metadata - .system_input + .input_metadata .as_ref() .unwrap() .feature_payload() @@ -573,7 +573,7 @@ fn frozen_behavior_counter_is_accepted_only_at_the_old_schema_boundary() { } #[test] -fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() { +fn current_schema_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() @@ -581,7 +581,7 @@ fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() { .remove("system_input"); let bytes = format!("{}\n", missing); let root = tempfile::tempdir().unwrap(); - let (_, segment) = write_worker(root.path(), 5, bytes.as_bytes()); + let (_, segment) = write_worker(root.path(), 6, bytes.as_bytes()); assert!(WorkerSessionStore::new(root.path()).is_err()); assert_eq!( fs::read( @@ -592,5 +592,112 @@ fn current_v5_rejects_generic_input_without_metadata_instead_of_migrating_it() { .unwrap(), bytes.as_bytes() ); - assert_marker(root.path(), 5); + assert_marker(root.path(), 6); +} + +const V5: &str = include_str!("fixtures/v5-role-metadata.jsonl"); + +#[test] +fn v5_metadata_and_derived_tool_context_upgrade_without_changing_roles_or_original_evidence() { + for standalone in [false, true] { + let root = tempfile::tempdir().unwrap(); + let (session, segment) = if standalone { + let session = SessionId::now_v7(); + let segment = SegmentId::now_v7(); + let dir = root.path().join(session.to_string()); + fs::create_dir_all(&dir).unwrap(); + fs::write(dir.join("session-schema.json"), b"5").unwrap(); + fs::write(dir.join(format!("{segment}.jsonl")), V5).unwrap(); + (session, segment) + } else { + write_worker(root.path(), 5, V5.as_bytes()) + }; + let entries = if standalone { + FsStore::new(root.path()) + .unwrap() + .read_all(session, segment) + .unwrap() + } else { + WorkerSessionStore::new(root.path()) + .unwrap() + .read_all(session, segment) + .unwrap() + }; + let originals: Vec = V5 + .lines() + .map(|line| serde_json::from_str(line).unwrap()) + .collect(); + assert_eq!(entries.len(), originals.len()); + for (index, (entry, old)) in entries.iter().zip(originals).enumerate() { + let wire = serde_json::to_value(entry).unwrap(); + assert_eq!(wire["ts"], old["ts"]); + assert_eq!(wire["entry"]["item"], old["entry"]["item"]); + for field in ["entry_id", "origin", "derivation"] { + assert_eq!( + wire["entry"]["metadata"][field], + old["entry"]["metadata"][field] + ); + } + let metadata = &wire["entry"]["metadata"]; + if index == 0 { + assert_eq!( + metadata["input_metadata"], + old["entry"]["metadata"]["system_input"] + ); + assert_eq!( + metadata["input_metadata"]["feature_payload"]["value"]["system_input"], + "opaque owner field must not be renamed" + ); + } else if index < 3 { + assert!( + metadata.get("input_metadata").is_none(), + "ordinary model Tool originals must not acquire exclusion policy" + ); + } else { + assert_eq!( + metadata["input_metadata"]["usage"], + json!({ + "recording_material":"exclude", "memory_evidence":"exclude", "observation":"hidden", "compaction":"retain" + }) + ); + } + } + let before = encode(&entries); + let reopened = if standalone { + FsStore::new(root.path()) + .unwrap() + .read_all(session, segment) + .unwrap() + } else { + WorkerSessionStore::new(root.path()) + .unwrap() + .read_all(session, segment) + .unwrap() + }; + assert_eq!(encode(&reopened), before); + } +} + +#[test] +fn legacy_metadata_alias_conflict_fails_preflight_without_dropping_either_value() { + let mut bad: Value = serde_json::from_str(V5.lines().next().unwrap()).unwrap(); + bad["entry"]["metadata"]["input_metadata"] = bad["entry"]["metadata"]["system_input"].clone(); + for version in [4, 5] { + let root = tempfile::tempdir().unwrap(); + let (_, segment) = write_worker(root.path(), version, V5.as_bytes()); + let path = root + .path() + .join("segments") + .join(format!("{segment}.jsonl")); + let later = root + .path() + .join("segments") + .join(format!("{}.jsonl", SegmentId::from_u128(2))); + let bad_bytes = format!("{bad}\n"); + fs::write(&later, &bad_bytes).unwrap(); + assert!(WorkerSessionStore::new(root.path()).is_err()); + assert_eq!(fs::read(path).unwrap(), V5.as_bytes()); + assert_eq!(fs::read(later).unwrap(), bad_bytes.as_bytes()); + assert_marker(root.path(), version); + } } diff --git a/crates/session-store/src/lib.rs b/crates/session-store/src/lib.rs index 7750f87e0..d767defca 100644 --- a/crates/session-store/src/lib.rs +++ b/crates/session-store/src/lib.rs @@ -41,6 +41,7 @@ pub(crate) fn read_without_atime_bounded(path: &Path, max_bytes: u64) -> io::Res pub mod event_trace; pub mod fs_store; pub mod history; +pub mod input_metadata; mod legacy_session_log; pub mod logged_item; mod paste_artifact; @@ -51,7 +52,6 @@ pub mod retention; pub mod segment; pub mod segment_log; pub mod store; -pub mod system_input; pub mod system_item; pub mod uploaded_file; pub mod worker_metadata; @@ -66,6 +66,10 @@ pub use history::{ LoggedSessionHistoryMetadata, LoggedSessionHistoryOrigin, LoggedSystemHistoryEntry, LoggedWorkerSubject, }; +pub use input_metadata::{ + FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse, + InputEvidenceUse, InputMetadataError, InputObservationUse, SystemInputEnvelope, +}; pub use logged_item::{LoggedContentPart, LoggedItem, LoggedRole, from_logged, to_logged}; pub use paste_artifact::PasteArtifactLimits; pub use public_index::{ @@ -98,10 +102,6 @@ pub use segment::{ }; pub use segment_log::{LogEntry, RestoredState, SegmentOrigin, SessionExtension, collect_state}; pub use store::{Store, StoreError}; -pub use system_input::{ - FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse, - SystemInputEnvelope, SystemInputError, SystemInputMetadata, SystemInputUsePolicy, -}; pub use system_item::{ PromptRenderProvenance, PublicSystemItem, PublicSystemPresentation, SystemItem, SystemReminder, SystemReminderSource, render_worker_event, diff --git a/crates/session-store/src/public_index.rs b/crates/session-store/src/public_index.rs index cba29bc87..eff4c04ad 100644 --- a/crates/session-store/src/public_index.rs +++ b/crates/session-store/src/public_index.rs @@ -1296,7 +1296,7 @@ mod tests { LoggedHistoryEntry { item, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.to_owned()), origin, derivation: None, @@ -1538,7 +1538,7 @@ mod tests { prompt_provenance: None, }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("system-item".into()), origin: LoggedSessionHistoryOrigin::BackendInstruction { operation_id: None, diff --git a/crates/session-store/src/public_material.rs b/crates/session-store/src/public_material.rs index 2c7d66293..a7cdccaf2 100644 --- a/crates/session-store/src/public_material.rs +++ b/crates/session-store/src/public_material.rs @@ -371,11 +371,10 @@ impl SessionPublicMaterialReader { let source_ids: Option> = if compact { let summaries: Vec<_> = history .iter() - .filter(|e| { - matches!( - e.metadata.origin, - LoggedSessionHistoryOrigin::CompactionSummary - ) + .filter(|entry| { + entry.metadata.input_metadata.as_ref().is_some_and(|input| { + input.usage().compaction == crate::InputCompactionUse::OpeningSummary + }) }) .collect(); if summaries.len() != 1 { @@ -895,7 +894,7 @@ fn canonical_body( entry: &LoggedHistoryEntry, purpose: PublicMaterialUse, ) -> Option> { - if let Some(input) = &entry.metadata.system_input { + if let Some(input) = &entry.metadata.input_metadata { let usage = input.usage(); let allowed = match purpose { PublicMaterialUse::Recording => usage.recording_material, diff --git a/crates/session-store/src/public_material/tests.rs b/crates/session-store/src/public_material/tests.rs index 5bf7ce7ea..230b81055 100644 --- a/crates/session-store/src/public_material/tests.rs +++ b/crates/session-store/src/public_material/tests.rs @@ -9,7 +9,7 @@ fn entry(id: &str, item: LoggedItem) -> LoggedHistoryEntry { LoggedHistoryEntry { item, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.into()), origin: LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None, @@ -31,9 +31,9 @@ fn system_context( compaction: crate::InputCompactionUse, ) -> LoggedHistoryEntry { let mut entry = message(id, LoggedRole::System, text); - entry.metadata.system_input = Some( - crate::SystemInputMetadata::new( - crate::SystemInputUsePolicy { + entry.metadata.input_metadata = Some( + crate::HistoryInputMetadata::new( + crate::HistoryInputUsePolicy { recording_material: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude, observation: crate::InputObservationUse::Hidden, @@ -613,9 +613,9 @@ fn reasoning_proves_rollback_safety_but_private_and_derived_items_are_not_materi body: "previous Journal must not become new recording material".into(), }, metadata: LoggedSessionHistoryMetadata { - system_input: Some( - crate::SystemInputMetadata::new( - crate::SystemInputUsePolicy { + input_metadata: Some( + crate::HistoryInputMetadata::new( + crate::HistoryInputUsePolicy { recording_material: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude, observation: crate::InputObservationUse::Hidden, @@ -812,9 +812,9 @@ 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 { + summary.metadata.input_metadata = Some( + crate::HistoryInputMetadata::new( + crate::HistoryInputUsePolicy { recording_material: crate::InputEvidenceUse::Exclude, memory_evidence: crate::InputEvidenceUse::Exclude, observation: crate::InputObservationUse::Hidden, @@ -878,7 +878,9 @@ fn midturn_compaction_uses_exact_source_ids_and_preserves_discarded_originals() "never use as original", crate::InputCompactionUse::OpeningSummary, ); - summary.metadata.origin = LoggedSessionHistoryOrigin::CompactionSummary; + summary.metadata.origin = LoggedSessionHistoryOrigin::Provided { + producer: "renamed-summary-owner".into(), + }; summary.metadata.derivation = Some(crate::LoggedHistoryDerivation { sources: vec![ LoggedSessionHistoryEntryId("u".into()), @@ -1311,8 +1313,8 @@ fn compaction_without_exact_derivation_or_with_invalid_turn_is_rejected() { #[test] fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() { use crate::{ - FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse, - LoggedSystemHistoryEntry, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy, + FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse, + InputEvidenceUse, InputObservationUse, LoggedSystemHistoryEntry, SystemInputEnvelope, }; for worker_layout in [false, true] { let temp = tempfile::tempdir().unwrap(); @@ -1328,8 +1330,8 @@ fn generic_system_material_uses_explicit_evidence_purpose_not_producer_or_body() None, SystemInputEnvelope::new( format!("private model text: {body}"), - SystemInputMetadata::new( - SystemInputUsePolicy { + HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material, memory_evidence, observation: InputObservationUse::Hidden, diff --git a/crates/session-store/src/public_snapshot.rs b/crates/session-store/src/public_snapshot.rs index 1baa121c8..ce23ee221 100644 --- a/crates/session-store/src/public_snapshot.rs +++ b/crates/session-store/src/public_snapshot.rs @@ -1625,9 +1625,9 @@ pub(crate) fn project_session_snapshot_for_segment( derivation_ids(entry), entry .metadata - .system_input + .input_metadata .as_ref() - .map(crate::SystemInputMetadata::public_use_policy), + .map(crate::HistoryInputMetadata::public_use_policy), )), LogEntry::RunYielded { ts, @@ -1797,9 +1797,9 @@ fn history_entry( SessionSnapshotEntry { use_policy: entry .metadata - .system_input + .input_metadata .as_ref() - .map(crate::SystemInputMetadata::public_use_policy), + .map(crate::HistoryInputMetadata::public_use_policy), entry_id: entry.metadata.entry_id.0.clone(), timestamp, provenance: provenance(&entry.metadata.origin), @@ -1926,7 +1926,16 @@ pub fn session_tool_attachment_id( } fn project_history_item(entry: &LoggedHistoryEntry) -> Option { - if let Some(input) = &entry.metadata.system_input { + if let Some(input) = &entry.metadata.input_metadata { + if !matches!( + entry.item, + LoggedItem::Message { + role: LoggedRole::System, + .. + } + ) { + return None; + } let public = input.public_projection(); return Some(SessionSnapshotEntryData::SystemItem { item_kind: public.kind.clone(), @@ -2310,7 +2319,7 @@ mod tests { content: vec![LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(entry_id.into()), origin, derivation: None, @@ -3609,7 +3618,7 @@ mod tests { .map(|item| LoggedHistoryEntry { item, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -3664,7 +3673,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: user_entry_id, origin: LoggedSessionHistoryOrigin::HumanInput { account_id: "account-1".into(), @@ -3734,7 +3743,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::FlowInstruction { selector: "builtin:coder-review".into(), @@ -3754,7 +3763,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: user_entry_id.clone(), origin: origin.clone(), derivation: Some(LoggedHistoryDerivation { @@ -3785,7 +3794,7 @@ mod tests { fn annotated_projection_preserves_identity_and_provenance() { let session_id = crate::new_session_id(); let metadata = LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::ModelOutput { worker: LoggedWorkerSubject { @@ -4348,7 +4357,7 @@ mod tests { is_error: false, }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("ancestor-image".into()), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -4443,7 +4452,7 @@ mod tests { is_error: false, }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("tool-images".into()), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/session-store/src/segment_log.rs b/crates/session-store/src/segment_log.rs index f76ea783c..d54a2659d 100644 --- a/crates/session-store/src/segment_log.rs +++ b/crates/session-store/src/segment_log.rs @@ -255,18 +255,18 @@ pub struct SegmentOrigin { impl LogEntry { /// Validate generic System records before a Store publishes any bytes. - pub fn validate_system_inputs(&self) -> Result<(), crate::SystemInputError> { + pub fn validate_input_metadata(&self) -> Result<(), crate::InputMetadataError> { match self { Self::AnnotatedSegmentStart { history, .. } | Self::AnnotatedUserInput { history, .. } => { for entry in history { - entry.validate_system_input()?; + entry.validate_input_metadata()?; } } Self::AnnotatedAssistantItem { entry, .. } | Self::AnnotatedToolResult { entry, .. } - | Self::ToolResultCorrected { entry, .. } => entry.validate_system_input()?, - Self::AnnotatedSystemItem { entry, .. } => entry.validate_system_input()?, + | Self::ToolResultCorrected { entry, .. } => entry.validate_input_metadata()?, + Self::AnnotatedSystemItem { entry, .. } => entry.validate_input_metadata()?, _ => {} } Ok(()) @@ -554,7 +554,7 @@ mod tests { LoggedHistoryEntry { item: LoggedItem::from(item), metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/session-store/src/store.rs b/crates/session-store/src/store.rs index bf29ccb63..ef26359c7 100644 --- a/crates/session-store/src/store.rs +++ b/crates/session-store/src/store.rs @@ -27,8 +27,8 @@ pub enum StoreError { #[error("serialization error: {0}")] Serde(#[from] serde_json::Error), - #[error("invalid system input: {0}")] - InvalidSystemInput(#[from] crate::SystemInputError), + #[error("invalid history input metadata: {0}")] + InvalidInputMetadata(#[from] crate::InputMetadataError), #[error("segment not found: {0}")] NotFound(SegmentId), diff --git a/crates/session-store/src/worker_session_store.rs b/crates/session-store/src/worker_session_store.rs index 17a2691d1..9c6d8db00 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 = 5; +const SESSION_SCHEMA_VERSION: u32 = 6; +const SYSTEM_INPUT_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; @@ -225,8 +226,9 @@ impl WorkerSessionStore { LEGACY_SESSION_SCHEMA_VERSION | PREVIOUS_SESSION_SCHEMA_VERSION | ANNOTATED_SESSION_SCHEMA_VERSION - | COMPACTION_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v5( + | COMPACTION_SESSION_SCHEMA_VERSION + | SYSTEM_INPUT_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_current( segment_log_paths(&root)?, manifest.session_id, manifest.schema_version, @@ -753,7 +755,7 @@ impl Store for WorkerSessionStore { segment_id: SegmentId, entry: &LogEntry, ) -> Result<(), StoreError> { - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; let _retention = self.lock_publication(session_id)?; self.ensure_session(session_id, true)?; self.append_log_entry(&self.log_path(segment_id), entry) @@ -813,7 +815,7 @@ impl Store for WorkerSessionStore { entries: &[LogEntry], ) -> Result<(), StoreError> { for entry in entries { - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; } let _retention = self.lock_publication(session_id)?; self.ensure_session(session_id, true)?; @@ -1173,7 +1175,7 @@ fn segment_log_paths_in_directory( Ok(paths) } -fn migrate_segment_logs_to_v5( +fn migrate_segment_logs_to_current( paths: Vec<(SegmentId, PathBuf)>, session_id: SessionId, source_schema_version: u32, @@ -1220,7 +1222,7 @@ fn migrate_segment_logs_to_v5( crate::legacy_session_log::migrate_compaction_origins(&mut plan.entries, &summaries); for mut entry in std::mem::take(&mut plan.entries) { crate::legacy_session_log::migrate_system_inputs(&mut entry)?; - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; serde_json::to_writer(&mut plan.output, &entry)?; plan.output.push(b'\n'); } @@ -1265,8 +1267,9 @@ pub(crate) fn migrate_fs_session(root: &Path, session_id: SessionId) -> Result<( LEGACY_SESSION_SCHEMA_VERSION | PREVIOUS_SESSION_SCHEMA_VERSION | ANNOTATED_SESSION_SCHEMA_VERSION - | COMPACTION_SESSION_SCHEMA_VERSION => { - migrate_segment_logs_to_v5( + | COMPACTION_SESSION_SCHEMA_VERSION + | SYSTEM_INPUT_SESSION_SCHEMA_VERSION => { + migrate_segment_logs_to_current( segment_log_paths_in_directory(root)?, session_id, version, @@ -1289,7 +1292,7 @@ fn validate_canonical_segment_paths(paths: Vec<(SegmentId, PathBuf)>) -> Result< for (_, path) in paths { let entries: Vec = parse_jsonl(&fs::read(&path)?)?; for entry in entries { - entry.validate_system_inputs()?; + entry.validate_input_metadata()?; } } Ok(()) @@ -1470,7 +1473,7 @@ mod tests { LoggedHistoryEntry { item: LoggedItem::from(item), metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -2162,7 +2165,7 @@ mod tests { 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(); + entry.validate_input_metadata().unwrap(); } assert!(matches!( &updated_first[0], @@ -2184,7 +2187,7 @@ mod tests { 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(); + entry.validate_input_metadata().unwrap(); } assert_eq!(fs::read(&first_path).unwrap(), first_published); let second_published = fs::read(&second_path).unwrap(); diff --git a/crates/session-store/tests/fs_store_test.rs b/crates/session-store/tests/fs_store_test.rs index af4e940ec..9673980c8 100644 --- a/crates/session-store/tests/fs_store_test.rs +++ b/crates/session-store/tests/fs_store_test.rs @@ -11,7 +11,7 @@ fn annotated(item: Item) -> LoggedHistoryEntry { LoggedHistoryEntry { item: LoggedItem::from(item), metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -428,10 +428,9 @@ fn lookup_session_of_finds_owning_session() { #[test] fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak() { use session_store::{ - FeatureInputPayload, InputCompactionUse, InputEvidenceUse, InputObservationUse, - LoggedHistoryDerivation, LoggedSystemHistoryEntry, PublicSystemItem, - PublicSystemPresentation, SystemInputEnvelope, SystemInputMetadata, SystemInputUsePolicy, - WorkerSessionStore, + FeatureInputPayload, HistoryInputMetadata, HistoryInputUsePolicy, InputCompactionUse, + InputEvidenceUse, InputObservationUse, LoggedHistoryDerivation, LoggedSystemHistoryEntry, + PublicSystemItem, PublicSystemPresentation, SystemInputEnvelope, WorkerSessionStore, }; for worker_layout in [false, true] { let dir = tempfile::tempdir().unwrap(); @@ -455,8 +454,8 @@ fn generic_system_append_and_fork_preserve_metadata_without_public_payload_leak( }), SystemInputEnvelope::new( "private model body", - SystemInputMetadata::new( - SystemInputUsePolicy { + HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, @@ -584,11 +583,11 @@ fn invalid_generic_system_input_rejects_append_create_and_external_json_without_ }; assert!(matches!( store.append(sid, segment, &invalid), - Err(StoreError::InvalidSystemInput(_)) + Err(StoreError::InvalidInputMetadata(_)) )); assert!(matches!( store.create_segment(sid, segment, &[seed, invalid.clone()]), - Err(StoreError::InvalidSystemInput(_)) + Err(StoreError::InvalidInputMetadata(_)) )); assert!( serde_json::from_value::(serde_json::to_value(invalid).unwrap()).is_err() diff --git a/crates/session-store/tests/session_test.rs b/crates/session-store/tests/session_test.rs index e12d3ff29..dd6346be9 100644 --- a/crates/session-store/tests/session_test.rs +++ b/crates/session-store/tests/session_test.rs @@ -24,7 +24,7 @@ fn annotated(items: &[Item]) -> Vec { .map(|item| session_store::LoggedHistoryEntry { item: session_store::LoggedItem::from(item), metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId::new(), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/standalone/src/subjektiv/journal/tests.rs b/crates/standalone/src/subjektiv/journal/tests.rs index bd24ef67d..c488b4879 100644 --- a/crates/standalone/src/subjektiv/journal/tests.rs +++ b/crates/standalone/src/subjektiv/journal/tests.rs @@ -212,7 +212,7 @@ fn message(id: &str, text: &str) -> session_store::LoggedHistoryEntry { content: vec![session_store::LoggedContentPart::Text { text: text.into() }], }, metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId(id.into()), origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None, diff --git a/crates/standalone/src/subjektiv/journal/tests/compaction.rs b/crates/standalone/src/subjektiv/journal/tests/compaction.rs index b1aba547f..c9ebe8352 100644 --- a/crates/standalone/src/subjektiv/journal/tests/compaction.rs +++ b/crates/standalone/src/subjektiv/journal/tests/compaction.rs @@ -215,6 +215,7 @@ async fn public_mid_turn_compactions_preserve_originals_and_drain_after_cancel_a } protocol::Event::TextDelta { text } if text.contains(PARTIAL) => break, protocol::Event::RunEnd { .. } => panic!("dialog ended before cancellation"), + protocol::Event::SessionEntryCommitted { entry } if matches!(entry.data, protocol::SessionSnapshotEntryData::RunError { .. }) => panic!("dialog failed before cancellation: {entry:?}"), protocol::Event::Error { code, message } => panic!("{code:?}: {message}"), _ => {} } @@ -294,6 +295,32 @@ async fn public_mid_turn_compactions_preserve_originals_and_drain_after_cancel_a .iter() .flat_map(|segment| source_store.read_all(session, *segment).unwrap()) .collect(); + for history in raw.iter().filter_map(|record| match record { + session_store::LogEntry::AnnotatedSegmentStart { history, compacted_from: Some(_), .. } => Some(history), + _ => None, + }) { + let mut calls = std::collections::HashSet::new(); + let mut results = 0; + for entry in history { + let is_tool = matches!(entry.item, session_store::LoggedItem::ToolCall { .. } | session_store::LoggedItem::ToolResult { .. }); + let Some(metadata) = entry.metadata.input_metadata.as_ref().filter(|_| is_tool) else { continue; }; + assert_eq!(metadata.usage().recording_material, session_store::InputEvidenceUse::Exclude); + assert_eq!(metadata.usage().memory_evidence, session_store::InputEvidenceUse::Exclude); + assert_eq!(metadata.usage().observation, session_store::InputObservationUse::Hidden); + assert!(metadata.public_item().is_none()); + assert!(entry.metadata.derivation.is_some()); + match &entry.item { + session_store::LoggedItem::ToolCall { call_id, .. } => { calls.insert(call_id.clone()); } + session_store::LoggedItem::ToolResult { call_id, .. } => { + assert!(calls.contains(call_id), "derived Tool result must follow its real logical call"); + results += 1; + } + _ => unreachable!(), + } + } + assert!(!calls.is_empty()); + assert!(results >= calls.len()); + } assert_eq!( raw.iter() .filter(|entry| matches!( diff --git a/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs b/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs index 04032a8d7..c78ff4a66 100644 --- a/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs +++ b/crates/standalone/src/subjektiv/journal/tests/connected_compaction.rs @@ -11,7 +11,7 @@ use agen::llm_client::{ types::Item, }; use futures::Stream; -use session_store::{CombinedStore, FsWorkerStore, LogEntry, LoggedSessionHistoryOrigin}; +use session_store::{CombinedStore, FsWorkerStore, LogEntry}; use std::pin::Pin; const BEFORE: &str = "FROZEN_POSITION: B partially implemented; tests not run; result unconfirmed"; @@ -139,14 +139,14 @@ async fn local_host_compaction_freezes_pending_journal_through_request_and_durab let worker = prepared.worker(); assert_ne!(worker.segment_id(), source_segment); let seed = worker.store().read_all(worker.session_id(), worker.segment_id()).unwrap().into_iter().find_map(|entry| match entry { LogEntry::AnnotatedSegmentStart { history, .. } => Some(history), _ => None }).expect("durable compacted seed"); - let journal_items: Vec<_> = seed.iter().filter(|entry| entry.metadata.origin == LoggedSessionHistoryOrigin::JournalContext).collect(); + let journal_items: Vec<_> = seed.iter().filter(|entry| entry.metadata.input_metadata.as_ref().is_some_and(|input| input.usage().compaction == session_store::InputCompactionUse::ReferenceInput)).collect(); assert_eq!(journal_items.len(), 1); let journal_item = Item::from(journal_items[0].item.clone()); let body = journal_item.as_text().unwrap(); assert!(body.contains(&expected_json), "durable seed contains the full selected JSON and processing checkpoint"); assert!(body.contains(BEFORE)); assert!(!body.contains(AFTER)); - assert!(seed.iter().any(|entry| entry.metadata.origin == LoggedSessionHistoryOrigin::CompactionSummary && Item::from(entry.item.clone()).as_text().is_some_and(|text| text.contains(COMPLEMENT)))); + assert!(seed.iter().any(|entry| entry.metadata.input_metadata.as_ref().is_some_and(|input| input.usage().compaction == session_store::InputCompactionUse::OpeningSummary) && Item::from(entry.item.clone()).as_text().is_some_and(|text| text.contains(COMPLEMENT)))); let requests = model.requests.lock().unwrap(); let input = requests[0].items.iter().find_map(|item| match item { Item::Message { role: agen::llm_client::types::Role::User, .. } => item.as_text().filter(|text| text.contains("## Conversation overview/index")), diff --git a/crates/standalone/src/subjektiv/journal/tests/handoff.rs b/crates/standalone/src/subjektiv/journal/tests/handoff.rs index 03e5b6ebf..62d511c8f 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 entry.metadata.system_input.as_ref().and_then(|input| input.feature_payload()).is_some_and(|payload| payload.schema() == "journal:startup"))).count(); + let journal_items = entries.iter().filter(|entry| matches!(entry, session_store::LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.input_metadata.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/standalone/src/subjektiv_sessions.rs b/crates/standalone/src/subjektiv_sessions.rs index 71dc48631..9eee16adc 100644 --- a/crates/standalone/src/subjektiv_sessions.rs +++ b/crates/standalone/src/subjektiv_sessions.rs @@ -1621,7 +1621,7 @@ allow = [] let history = |id: &str, item: LoggedItem| LoggedHistoryEntry { item, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.into()), origin: LoggedSessionHistoryOrigin::BackendInstruction { operation_id: None }, derivation: None, diff --git a/crates/tui/src/app.rs b/crates/tui/src/app.rs index e537c5385..9a56b4c90 100644 --- a/crates/tui/src/app.rs +++ b/crates/tui/src/app.rs @@ -3567,7 +3567,7 @@ mod completion_flow_tests { session_store::LoggedHistoryEntry { item: session_store::LoggedItem::from(item), metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId::new(), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs index 9c0e5f9e2..733f658ac 100644 --- a/crates/worker-runtime/src/http_server/tests/session_material_tests.rs +++ b/crates/worker-runtime/src/http_server/tests/session_material_tests.rs @@ -35,7 +35,7 @@ fn message(id: &str, role: LoggedRole, text: &str) -> LoggedHistoryEntry { content: vec![LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.into()), origin: LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None, diff --git a/crates/worker-runtime/src/retention.rs b/crates/worker-runtime/src/retention.rs index d488c9608..6e2496246 100644 --- a/crates/worker-runtime/src/retention.rs +++ b/crates/worker-runtime/src/retention.rs @@ -1867,7 +1867,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("original".into()), origin: LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None, @@ -1881,7 +1881,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId("summary".into()), origin: LoggedSessionHistoryOrigin::DerivedSummary, derivation: Some(session_store::LoggedHistoryDerivation { diff --git a/crates/worker-runtime/src/retention/runtime_v5_tests.rs b/crates/worker-runtime/src/retention/runtime_v5_tests.rs index 8c0123606..5527dba73 100644 --- a/crates/worker-runtime/src/retention/runtime_v5_tests.rs +++ b/crates/worker-runtime/src/retention/runtime_v5_tests.rs @@ -155,7 +155,7 @@ fn assert_generic(root: &Path, session: SessionId, segment: SegmentId) { entry.item.history_text(), old["entry"]["item"]["body"].as_str().unwrap() ); - let input = entry.metadata.system_input.as_ref().unwrap(); + let input = entry.metadata.input_metadata.as_ref().unwrap(); let usage = input.usage(); assert_eq!(usage.recording_material, InputEvidenceUse::Exclude); assert_eq!(usage.memory_evidence, InputEvidenceUse::Exclude); @@ -557,3 +557,78 @@ fn current_runtime_session_migration_still_rejects_live_worker_sockets() { assert_eq!(tree_bytes(&root), before); assert!(!temp.path().join(".runtime.schema-v12-staging").exists()); } + +#[test] +fn current_runtime_upgrades_v5_role_metadata_through_staging_without_reserializing_owner_evidence() +{ + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("runtime"); + let (worker, session, segment) = runtime_fixture(&root, 12); + let aggregate = root.join("workers").join(worker.to_string()); + let path = aggregate + .join("session/segments") + .join(format!("{segment}.jsonl")); + let old = fs::read_to_string(&path).unwrap(); + let header = old.lines().next().unwrap(); + let source = include_str!( + "../../../session-store/src/legacy_session_log/tests/fixtures/v5-role-metadata.jsonl" + ); + fs::write(&path, format!("{header}\n{source}")).unwrap(); + write_json( + &aggregate.join("session/session.json"), + &json!({"schema_version":5, "session_id": session}), + ); + let owner_before = fs::read(root.join("runtime.json")).unwrap(); + let worker_before = fs::read(aggregate.join("worker.json")).unwrap(); + let metadata_before = fs::read(aggregate.join("metadata.json")).unwrap(); + assert!(WorkerSessionStore::open_read_only(aggregate.join("session")).is_err()); + let options = FsRuntimeStoreOptions::new(&root).with_runtime_id("runtime-a"); + assert!( + FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + FsRuntimeStore::migrate(&options).unwrap(); + let store = WorkerSessionStore::open_read_only(aggregate.join("session")).unwrap(); + let entries = store.read_all_read_only(session, segment).unwrap(); + assert_eq!(entries.len(), 6); + for (current, old) in entries[1..].iter().zip(source.lines()) { + let current = serde_json::to_value(current).unwrap(); + let old: Value = serde_json::from_str(old).unwrap(); + assert_eq!(current["ts"], old["ts"]); + assert_eq!(current["entry"]["item"], old["entry"]["item"]); + for field in ["entry_id", "origin", "derivation"] { + assert_eq!( + current["entry"]["metadata"][field], + old["entry"]["metadata"][field] + ); + } + if current["entry"]["metadata"]["origin"]["kind"] == "derived_summary" { + assert_eq!( + current["entry"]["metadata"]["input_metadata"]["usage"]["recording_material"], + "exclude" + ); + assert_eq!( + current["entry"]["metadata"]["input_metadata"]["usage"]["memory_evidence"], + "exclude" + ); + } + } + assert_eq!(fs::read(root.join("runtime.json")).unwrap(), owner_before); + assert_eq!( + fs::read(aggregate.join("worker.json")).unwrap(), + worker_before + ); + assert_eq!( + fs::read(aggregate.join("metadata.json")).unwrap(), + metadata_before + ); + let once = tree_bytes(&aggregate.join("session")); + assert!( + !FsRuntimeStore::migration_plan(&options) + .unwrap() + .migration_required + ); + FsRuntimeStore::migrate(&options).unwrap(); + assert_eq!(tree_bytes(&aggregate.join("session")), once); +} diff --git a/crates/worker-runtime/src/session_public.rs b/crates/worker-runtime/src/session_public.rs index 93f337f6d..ebdc2fa87 100644 --- a/crates/worker-runtime/src/session_public.rs +++ b/crates/worker-runtime/src/session_public.rs @@ -398,7 +398,7 @@ mod tests { content: vec![LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.to_string()), origin: LoggedSessionHistoryOrigin::HumanInput { account_id: "account-1".to_string(), diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index ca598cd80..b4c585cc1 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -5033,7 +5033,7 @@ mod tests { }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id, origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/worker/src/feature/builtin/journal.rs b/crates/worker/src/feature/builtin/journal.rs index 15d2bb2b4..e5de44038 100644 --- a/crates/worker/src/feature/builtin/journal.rs +++ b/crates/worker/src/feature/builtin/journal.rs @@ -79,8 +79,8 @@ fn journal_input( body: body.clone(), presentation: PublicSystemPresentation::Context, }; - let metadata = SystemInputMetadata::new( - SystemInputUsePolicy { + let metadata = HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index ed90b8622..270de8430 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -61,12 +61,12 @@ fn resident_input( schema: &str, payload: &impl serde::Serialize, kind: &str, -) -> Result { +) -> Result { use session_store::*; let input = SystemInputEnvelope::new( body.clone(), - SystemInputMetadata::new( - SystemInputUsePolicy { + HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, @@ -93,7 +93,7 @@ pub(crate) fn subject_behavior_input( body: String, behavior_md: String, prompt_provenance: Option, -) -> Result { +) -> Result { resident_input( body, "subject:behavior", @@ -108,7 +108,7 @@ pub(crate) fn subject_behavior_input( pub(crate) fn resident_summary_input( body: String, prompt_provenance: Option, -) -> Result { +) -> Result { resident_input( body, "memory:resident-summary", @@ -891,7 +891,7 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource metadata: &crate::session_history::SessionHistoryMetadata, ) { if let Some(input) = metadata - .system_input + .input_metadata .as_ref() .and_then(|input| input.feature_payload()) .and_then(|payload| { diff --git a/crates/worker/src/feature/invocation.rs b/crates/worker/src/feature/invocation.rs index 91009461e..523be6068 100644 --- a/crates/worker/src/feature/invocation.rs +++ b/crates/worker/src/feature/invocation.rs @@ -48,15 +48,15 @@ pub(crate) struct InvocationResultInput { pub(crate) fn result_input( result: FeatureInvocationResult, body: String, -) -> Result { +) -> Result { use session_store::*; let payload = FeatureInputPayload::encode( "feature:invocation-result", 1, &InvocationResultInput { result }, )?; - let metadata = SystemInputMetadata::new( - SystemInputUsePolicy { + let metadata = HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, diff --git a/crates/worker/src/internal_worker.rs b/crates/worker/src/internal_worker.rs index 46400d52d..2c6a52a13 100644 --- a/crates/worker/src/internal_worker.rs +++ b/crates/worker/src/internal_worker.rs @@ -1499,7 +1499,7 @@ permission = "write" is_error: false, }, metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/worker/src/ipc/interceptor.rs b/crates/worker/src/ipc/interceptor.rs index 1e91f1797..95860b27b 100644 --- a/crates/worker/src/ipc/interceptor.rs +++ b/crates/worker/src/ipc/interceptor.rs @@ -1033,7 +1033,7 @@ mod tests { metadata: &crate::session_history::SessionHistoryMetadata, ) { if let Some(input) = metadata - .system_input + .input_metadata .as_ref() .and_then(|input| input.feature_payload()) .and_then(|payload| { diff --git a/crates/worker/src/session_capture.rs b/crates/worker/src/session_capture.rs index 7e78a12fc..e53da9105 100644 --- a/crates/worker/src/session_capture.rs +++ b/crates/worker/src/session_capture.rs @@ -352,11 +352,19 @@ impl SessionCapture { let entries = entries .into_iter() .map(|entry| { - let (item, use_policy) = if let Some(input) = &entry.annotation.system_input { - ( - Item::system_message(input.public_projection().body), - Some(input.public_use_policy()), - ) + let (item, use_policy) = if let Some(input) = &entry.annotation.input_metadata { + let item = if matches!( + entry.item, + Item::Message { + role: Role::System, + .. + } + ) { + Item::system_message(input.public_projection().body) + } else { + entry.item + }; + (item, Some(input.public_use_policy())) } else { (entry.item, None) }; @@ -982,10 +990,10 @@ mod tests { #[test] fn generic_public_input_observation_and_memory_policies_survive_store_restore_independently() { use session_store::{ - FeatureInputPayload, FsStore, InputCompactionUse, InputEvidenceUse, - InputObservationUse, LogEntry, LoggedSystemHistoryEntry, PublicSystemItem, - PublicSystemPresentation, Store, SystemInputEnvelope, SystemInputMetadata, - SystemInputUsePolicy, + FeatureInputPayload, FsStore, HistoryInputMetadata, HistoryInputUsePolicy, + InputCompactionUse, InputEvidenceUse, InputObservationUse, LogEntry, + LoggedSystemHistoryEntry, PublicSystemItem, PublicSystemPresentation, Store, + SystemInputEnvelope, }; let root = tempfile::tempdir().unwrap(); let store = FsStore::new(root.path()).unwrap(); @@ -1046,8 +1054,8 @@ mod tests { None, SystemInputEnvelope::new( "private model text", - SystemInputMetadata::new( - SystemInputUsePolicy { + HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence, observation, @@ -1121,6 +1129,152 @@ mod tests { } } + // Real FsStore seed/fork/reopen -> ordinary history replay, public snapshot, + // capture and both evidence consumers; no Feature or Tool execution occurs. + #[test] + fn derived_tool_context_keeps_roles_identity_and_policy_through_fork_without_becoming_evidence() + { + use session_store::public_material::{ + PublicMaterialLimits, PublicMaterialUse, SessionPublicMaterialReader, + }; + use session_store::*; + let root = tempfile::tempdir().unwrap(); + let store = FsStore::new(root.path()).unwrap(); + let session = new_session_id(); + let first = new_segment_id(); + let fork = new_segment_id(); + let items = vec![ + Item::tool_call("original", "ArbitraryTool", "{}"), + Item::tool_result("original", "original result"), + Item::tool_call("derived", "ArbitraryTool", "PRIVATE_SYNTHETIC_CONTEXT"), + Item::tool_result("derived", "PRIVATE_SYNTHETIC_RESULT"), + ]; + let mut history: Vec<_> = items + .iter() + .cloned() + .map(|item| LoggedHistoryEntry { + item: LoggedItem::from(item), + metadata: LoggedSessionHistoryMetadata::legacy_unknown(), + }) + .collect(); + let sources = history[..2] + .iter() + .map(|entry| entry.metadata.entry_id.clone()) + .collect::>(); + for entry in &mut history[2..] { + entry.metadata.derivation = Some(LoggedHistoryDerivation { + sources: sources.clone(), + }); + entry.metadata.input_metadata = Some( + HistoryInputMetadata::new( + HistoryInputUsePolicy { + recording_material: InputEvidenceUse::Exclude, + memory_evidence: InputEvidenceUse::Exclude, + observation: InputObservationUse::Hidden, + compaction: InputCompactionUse::Retain, + }, + None, + None, + ) + .unwrap(), + ); + } + for segment in [first, fork] { + store + .create_segment( + session, + segment, + &[LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: None, + config: Default::default(), + history: history.clone(), + forked_from: (segment == fork).then_some(SegmentOrigin { + segment_id: first, + at_turn_index: 0, + }), + compacted_from: None, + }], + ) + .unwrap(); + } + let reopened = FsStore::new(root.path()).unwrap(); + let saved = reopened.read_all(session, fork).unwrap(); + let restored = + crate::session_history::restore_history_entries(session, fork, &saved).unwrap(); + for (index, entry) in restored.iter().enumerate() { + assert_eq!(entry.item, items[index]); + assert_eq!(entry.annotation, history[index].metadata); + } + let capture = SessionCapture::from_history_entries(fork.to_string(), restored); + assert!( + matches!(&capture.entries[2].item, Item::ToolCall { call_id, .. } if call_id == "derived") + ); + assert!( + matches!(&capture.entries[3].item, Item::ToolResult { call_id, .. } if call_id == "derived") + ); + let snapshot = public_snapshot::project_session_snapshot(session, &saved); + assert!( + !serde_json::to_string(&snapshot) + .unwrap() + .contains("PRIVATE_SYNTHETIC") + ); + for capture in [ + capture, + SessionCapture::from_session_snapshot(fork.to_string(), snapshot), + ] { + assert_eq!( + capture + .search(&SearchOptions { + kind: Some(ReferenceKind::Tool), + ..Default::default() + }) + .len(), + 2 + ); + for (index, entry) in history.iter().enumerate() { + let reference = SessionEntryRef::from_history_entry_id(&entry.metadata.entry_id); + assert_eq!( + !capture + .read(ReadSelector::Id(reference.as_str()), ReadOptions::default()) + .entries + .is_empty(), + index < 2 + ); + assert_eq!( + capture.evidence_for(reference.as_str()).is_some(), + index < 2 + ); + } + } + for purpose in [ + PublicMaterialUse::Recording, + PublicMaterialUse::MemoryEvidence, + ] { + let reader = SessionPublicMaterialReader::open_fs_for_use( + root.path(), + session, + fork, + PublicMaterialLimits::default(), + reopened.retention(), + purpose, + ) + .unwrap(); + let ids = reader + .descriptors() + .map(|entry| entry.identity.entry_ref.clone()) + .collect::>(); + assert_eq!( + ids, + sources + .iter() + .map(|id| format!("E{}", id.0)) + .collect::>() + ); + } + } + #[test] fn overview_contains_user_and_assistant_only() { let view = SessionCapture::new( diff --git a/crates/worker/src/session_history.rs b/crates/worker/src/session_history.rs index 472d40eb6..dda081a38 100644 --- a/crates/worker/src/session_history.rs +++ b/crates/worker/src/session_history.rs @@ -24,7 +24,7 @@ pub(crate) fn test_logged_history_entry(item: impl Into) -> LoggedHistoryE LoggedHistoryEntry { item: session_store::LoggedItem::from(item.into()), metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -39,7 +39,7 @@ pub(crate) fn test_logged_system_entry( session_store::LoggedSystemHistoryEntry { item, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId::new(), origin: LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -60,7 +60,7 @@ pub(crate) fn metadata( derivation: Option, ) -> SessionHistoryMetadata { SessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: SessionHistoryEntryId::new(), origin, derivation, @@ -71,7 +71,7 @@ pub(crate) fn compaction_use( metadata: &SessionHistoryMetadata, ) -> session_store::InputCompactionUse { metadata - .system_input + .input_metadata .as_ref() .map(|input| input.usage().compaction) .unwrap_or(session_store::InputCompactionUse::Retain) @@ -93,7 +93,7 @@ pub(crate) fn history_entry_with_id( HistoryEntry::new( item, SessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id, origin, derivation: None, diff --git a/crates/worker/src/skill.rs b/crates/worker/src/skill.rs index 4629e03a8..2ad347a6d 100644 --- a/crates/worker/src/skill.rs +++ b/crates/worker/src/skill.rs @@ -16,7 +16,7 @@ pub(crate) struct SkillActivationInput { pub(crate) fn activation_input( activation: &SkillActivationResponse, -) -> Result { +) -> Result { use session_store::*; let body = format!( "Agent Skill `{}` activated from {}.\n\n{}", @@ -29,8 +29,8 @@ pub(crate) fn activation_input( name: activation.name.clone(), }, )?; - let metadata = SystemInputMetadata::new( - SystemInputUsePolicy { + let metadata = HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 1d5b7dfb3..9315c3532 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -1338,7 +1338,7 @@ mod tests { is_error: false, }, metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId(entry_id.into()), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/worker/src/subjektiv.rs b/crates/worker/src/subjektiv.rs index eb871f992..0593e00ad 100644 --- a/crates/worker/src/subjektiv.rs +++ b/crates/worker/src/subjektiv.rs @@ -1156,7 +1156,7 @@ enabled = true LogEntry::AnnotatedSystemItem { entry, .. } if entry .metadata - .system_input + .input_metadata .as_ref() .and_then(|input| input.feature_payload()) .is_some_and(|payload| payload.schema() == "journal:startup") => @@ -1178,7 +1178,7 @@ enabled = true }; let payload = journal .metadata - .system_input + .input_metadata .as_ref() .unwrap() .feature_payload() @@ -1277,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 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)); + assert_eq!(logs.iter().filter(|entry| matches!(entry, LogEntry::AnnotatedSystemItem { entry, .. } if entry.metadata.input_metadata.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/wip.rs b/crates/worker/src/wip.rs index 1f12478d3..0965571bf 100644 --- a/crates/worker/src/wip.rs +++ b/crates/worker/src/wip.rs @@ -5502,7 +5502,7 @@ mod tests { entry: session_store::LoggedHistoryEntry { item: session_store::LoggedItem::from(saved), metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId::new(), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index aabb6314f..0e29f0cbd 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2739,7 +2739,7 @@ pub trait SystemItemCommitter: Send + Sync { extensions: Vec, after_commit: &mut dyn FnMut(&HistoryEntry), ) -> Result, StoreError> { - entry.validate_system_input()?; + entry.validate_input_metadata()?; let history = HistoryEntry::new(entry.item.to_history_item(), entry.metadata.clone()); self.commit_usage_boundary( LogEntry::AnnotatedSystemItem { @@ -6950,7 +6950,7 @@ impl Worker { }; for entry in &frozen_context { entry - .validate_system_input() + .validate_input_metadata() .map_err(|e| WorkerError::InvalidState(e.to_string()))?; } let history_entries = self.session.history().entries(); @@ -7469,8 +7469,8 @@ impl Worker { new_history[0] .as_text() .expect("compaction summary is text"), - session_store::SystemInputMetadata::new( - session_store::SystemInputUsePolicy { + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { recording_material: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude, observation: session_store::InputObservationUse::Hidden, @@ -7509,15 +7509,26 @@ impl Worker { original_cursor += offset + 1; HistoryEntry::new(item, original.annotation.clone()) } else { - HistoryEntry::new( - item, - new_history_metadata( - WorkerHistoryProvenance::DerivedSummary, - Some(SessionHistoryDerivation { - sources: derived_sources.clone(), - }), - ), - ) + let mut metadata = new_history_metadata( + WorkerHistoryProvenance::DerivedSummary, + Some(SessionHistoryDerivation { + sources: derived_sources.clone(), + }), + ); + metadata.input_metadata = Some( + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { + recording_material: session_store::InputEvidenceUse::Exclude, + memory_evidence: session_store::InputEvidenceUse::Exclude, + observation: session_store::InputObservationUse::Hidden, + compaction: session_store::InputCompactionUse::Retain, + }, + None, + None, + ) + .expect("bounded private derived context policy"), + ); + HistoryEntry::new(item, metadata) } }) .collect::>(); @@ -10790,9 +10801,9 @@ mod build_summary_prompt_tests { producer: "arbitrary-summary-source".into(), }, ); - entry.annotation.system_input = Some( - session_store::SystemInputMetadata::new( - session_store::SystemInputUsePolicy { + entry.annotation.input_metadata = Some( + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { recording_material: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude, observation: session_store::InputObservationUse::Hidden, @@ -14631,7 +14642,7 @@ permission = "write" matches!( entry, 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") + if entry.metadata.input_metadata.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 ) @@ -16919,7 +16930,7 @@ permission = "write" fn input_schema(entry: &session_store::LoggedSystemHistoryEntry) -> Option<&str> { entry .metadata - .system_input + .input_metadata .as_ref()? .feature_payload() .map(|payload| payload.schema()) @@ -16953,8 +16964,8 @@ permission = "write" None, SystemInputEnvelope::new( "PRIOR_JOURNAL_REFERENCE", - SystemInputMetadata::new( - SystemInputUsePolicy { + HistoryInputMetadata::new( + HistoryInputUsePolicy { recording_material: InputEvidenceUse::Exclude, memory_evidence: InputEvidenceUse::Exclude, observation: InputObservationUse::Hidden, diff --git a/crates/worker/tests/compact_events_test.rs b/crates/worker/tests/compact_events_test.rs index e0f773abf..faeea461b 100644 --- a/crates/worker/tests/compact_events_test.rs +++ b/crates/worker/tests/compact_events_test.rs @@ -119,7 +119,7 @@ fn annotated(item: Item) -> session_store::LoggedHistoryEntry { session_store::LoggedHistoryEntry { item: session_store::LoggedItem::from(item), metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId::new(), origin: session_store::LoggedSessionHistoryOrigin::LegacyUnknown, derivation: None, @@ -2220,7 +2220,7 @@ fn summary_entry( origin: session_store::LoggedSessionHistoryOrigin, ) -> session_store::LoggedHistoryEntry { let mut entry = annotated(item); - entry.metadata.system_input = matches!( + entry.metadata.input_metadata = matches!( &entry.item, session_store::LoggedItem::Message { role: session_store::LoggedRole::System, @@ -2228,8 +2228,8 @@ fn summary_entry( } ) .then(|| { - session_store::SystemInputMetadata::new( - session_store::SystemInputUsePolicy { + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { recording_material: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude, observation: session_store::InputObservationUse::Hidden, @@ -2256,7 +2256,7 @@ fn compaction_use( metadata: &session_store::LoggedSessionHistoryMetadata, ) -> session_store::InputCompactionUse { metadata - .system_input + .input_metadata .as_ref() .map(|input| input.usage().compaction) .unwrap_or(session_store::InputCompactionUse::Retain) @@ -3167,8 +3167,8 @@ async fn journal_three_compactions_freeze_one_snapshot_preserve_full_summary_tai None, session_store::SystemInputEnvelope::new( "OLD_TYPED_STARTUP_JOURNAL", - session_store::SystemInputMetadata::new( - session_store::SystemInputUsePolicy { + session_store::HistoryInputMetadata::new( + session_store::HistoryInputUsePolicy { recording_material: session_store::InputEvidenceUse::Exclude, memory_evidence: session_store::InputEvidenceUse::Exclude, observation: session_store::InputObservationUse::Hidden, diff --git a/crates/worker/tests/feature_invocation_test.rs b/crates/worker/tests/feature_invocation_test.rs index 805f694ac..91f3969b7 100644 --- a/crates/worker/tests/feature_invocation_test.rs +++ b/crates/worker/tests/feature_invocation_test.rs @@ -231,7 +231,7 @@ fn results(worker: &Worker) -> Vec { } entry .metadata - .system_input + .input_metadata .as_ref()? .feature_payload()? .decode::("feature:invocation-result", 1) diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 4bdf8a7f6..4ffb70865 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -44108,7 +44108,7 @@ mod tests { content: vec![LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.to_string()), origin: LoggedSessionHistoryOrigin::HumanInput { account_id: "account-1".to_string(), @@ -44298,7 +44298,7 @@ mod tests { content: vec![LoggedContentPart::Text { text: text.into() }], }, metadata: LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: LoggedSessionHistoryEntryId(id.to_string()), origin: LoggedSessionHistoryOrigin::HumanInput { account_id: "account-1".to_string(), @@ -48550,21 +48550,25 @@ mod tests { let worker_root = root .join(".test-embedded-runtime-store/workers") .join(&worker.worker_id); - fs::create_dir_all(worker_root.join("session/segments")).unwrap(); - fs::write( - worker_root.join("session/session.json"), - serde_json::to_vec_pretty(&json!({ - "schema_version": 1, - "session_id": session_store::new_session_id() - })) - .unwrap(), - ) - .unwrap(); - fs::write( - worker_root.join("session/segments/segment-a.jsonl"), - b"backend job retained session evidence\n", - ) - .unwrap(); + use session_store::Store; + let store = session_store::WorkerSessionStore::new(worker_root.join("session")).unwrap(); + let session = session_store::new_session_id(); + let segment = session_store::new_segment_id(); + store + .create_segment( + session, + segment, + &[session_store::LogEntry::AnnotatedSegmentStart { + ts: 1, + session_id: session, + system_prompt: Some("backend job retained session evidence".into()), + config: Default::default(), + history: vec![], + forked_from: None, + compacted_from: None, + }], + ) + .unwrap(); } fn job_notification_evidence( 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 893720ccd..a1447eece 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 @@ -562,7 +562,7 @@ async fn backend_real_runtime_journal_release_and_next_body_persisted_startup() }; let payload = entry .metadata - .system_input + .input_metadata .as_ref()? .feature_payload()? .decode::("journal:startup", 1) diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index fd0c89fea..4b0500eb8 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs @@ -479,7 +479,7 @@ fn seed_session( vec![session_store::LoggedHistoryEntry { item, metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId("read-source-original".into()), origin, derivation: None, @@ -1707,7 +1707,7 @@ impl SubjectRemovalFixture { .map(|text| session_store::LoggedHistoryEntry { item: message(text), metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId(text.into()), origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None, diff --git a/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs index efe44156f..190924f52 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs +++ b/crates/workspace-server/src/server/tests/subjektiv_memory_journal_tests.rs @@ -67,7 +67,7 @@ impl Fixture { content: vec![session_store::LoggedContentPart::Text { text: text.into() }], }, metadata: session_store::LoggedSessionHistoryMetadata { - system_input: None, + input_metadata: None, entry_id: session_store::LoggedSessionHistoryEntryId(format!("original-{index}")), origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput, derivation: None,